Compare commits

...
Author SHA1 Message Date
enricobuehler 2a62fe7857 fix(client): stop the double-arm race re-freezing RFI-healed streams
ci / bun-nix (pull_request) Successful in 30s
ci / docs-site (pull_request) Successful in 1m22s
ci / rust-arm64 (pull_request) Successful in 1m42s
ci / web (pull_request) Successful in 1m51s
apple / swift (pull_request) Successful in 1m50s
apple / screenshots (pull_request) Skipped
windows / build (aarch64-pc-windows-msvc) (pull_request) Successful in 1m28s
android / android (pull_request) Successful in 6m39s
windows / build (x86_64-pc-windows-msvc) (pull_request) Failing after 13m7s
ci / rust (pull_request) Successful in 14m4s
Every unrecoverable loss armed the client's freeze gate twice: instantly at
frame-index-gap detection (which fires the RFI), and ~120 ms later when the
reassembler ages the lost frame into frames_dropped and poll() re-armed
unconditionally. An LTR-RFI recovery anchor lands in ~60 ms — between the two
signals — so the stale climb re-froze a bit-exact-healed stream, the host
swallowed the re-ask as an RFI echo, and the picture stayed frozen until the
overdue backstop extracted a full IDR: the field 'H265 freezes on every loss,
AV1 fine' signature on AMD hosts (AMF is the only LTR-RFI backend; the slower
IDR path usually lands after the climb and dodged the race).

The gap-arm now pre-credits the expected climb (ReanchorGate::arm_expecting_drops;
credit expires after DROP_CREDIT_WINDOW so a straggler-filled gap can't mask a
later real loss), and poll() consumes credited climbs instead of re-arming.
Plumbed through every embedder: pf-client-core's session pump, Android's
sync/async loops (note_frame_index now returns the gap width), and the Swift
client via new ABI exports punktfunk_connection_note_frame_index_ex +
punktfunk_reanchor_gate_arm_expecting_drops (additive; the bool ABI stays).
2026-08-12 08:11:35 +02:00
12 changed files with 342 additions and 58 deletions
@@ -43,10 +43,12 @@ struct OutputReady {
/// internal looper thread) push the codec ones; the feeder thread pushes `Au`. Each carries only
/// owned/`Copy` data so the callback closures satisfy the `Send` bound and never touch the codec.
enum DecodeEvent {
/// A received access unit from the feeder, ready to queue into the decoder. The `bool` is the
/// feeder's [`NativeClient::note_frame_index`] verdict — `true` when this AU revealed a forward
/// frame-index gap, so the loop arms the freeze gate (the feeder already fired the RFI request).
Au(Frame, bool),
/// A received access unit from the feeder, ready to queue into the decoder. The `u32` is the
/// feeder's [`NativeClient::note_frame_index`] verdict — the forward frame-index gap's WIDTH
/// (0 = none), so the loop arms the freeze gate with the same signal and pre-credits the
/// reassembler's later `frames_dropped` climb for the loss (the feeder already fired the RFI
/// request).
Au(Frame, u32),
/// An input buffer slot freed (index) — we can queue an AU into it.
InputAvailable(usize),
/// A decoded frame is ready (buffer index + echoed pts + the callback-time `decoded` stamp).
@@ -603,7 +605,11 @@ fn feeder_loop(
// AU's first piece (or a whole delivery), so the RFI gap detector keeps
// counting AUs.
let au_first = frame.part.is_none_or(|p| p.first);
let gap = au_first && client.note_frame_index(frame.frame_index);
let gap = if au_first {
client.note_frame_index(frame.frame_index)
} else {
0
};
// Park the receipt stamp (keyed by the pts the codec echoes) whenever the `decode`
// stage is consumed: the HUD, or the ABR decode signal (`measure_decode`). The
// HUD-only `received` point + host/network split stay gated on the overlay.
@@ -691,9 +697,12 @@ fn dispatch_event(
match ev {
DecodeEvent::Au(f, gap) => {
// A forward frame-index gap arms the freeze; park this AU's flags for the present side to
// fold `on_decoded` (keyed by the pts the codec will echo).
if gap {
gate.arm(Instant::now());
// fold `on_decoded` (keyed by the pts the codec will echo). Credited arm: the gap width
// pre-covers the reassembler's ~120 ms-later `frames_dropped` climb for the same loss,
// so a fast RFI anchor that heals in between isn't re-frozen by it (the double-arm
// race — see `ReanchorGate::arm_expecting_drops`).
if gap > 0 {
gate.arm_expecting_drops(Instant::now(), u64::from(gap));
}
// One entry per AU (parts share the pts): the completing delivery carries it.
if f.complete {
@@ -222,8 +222,13 @@ pub(super) fn run_sync(
// recovers with a cheap clean P-frame instead of a full IDR. The same forward gap
// arms the freeze gate so the decoder's concealment is held off the screen until the
// recovery re-anchors. The frames_dropped keyframe path below stays the backstop.
if client.note_frame_index(frame.frame_index) {
gate.arm(Instant::now());
// Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
// `frames_dropped` climb for the same loss, so a fast RFI anchor that heals in
// between isn't re-frozen by it (the double-arm race — see
// `ReanchorGate::arm_expecting_drops`).
let gap = client.note_frame_index(frame.frame_index);
if gap > 0 {
gate.arm_expecting_drops(Instant::now(), u64::from(gap));
}
// Park this AU's re-anchor flags for the present side (keyed by the pts the codec
// echoes on the output buffer) — unconditional, unlike the HUD's `in_flight` map.
@@ -774,12 +774,22 @@ public final class PunktfunkConnection {
/// `noteFrameIndex` (the throttled RFI request); call it for every received AU. Returns false
/// after close.
public func noteFrameIndexGap(_ frameIndex: UInt32) -> Bool {
noteFrameIndexGapWidth(frameIndex) > 0
}
/// Like `noteFrameIndexGap`, but reports the gap's WIDTH how many frames this arrival revealed
/// as missing (0 = none). The post-loss re-anchor gate arms with the width
/// (`ReanchorGate.arm(expectingDrops:)`) so the reassembler's later `framesDropped` climb for
/// the SAME loss cannot re-freeze a stream an RFI anchor already healed (the double-arm race).
/// Same core side effect as `noteFrameIndex` (the throttled RFI request); call it for every
/// received AU. Returns 0 after close.
public func noteFrameIndexGapWidth(_ frameIndex: UInt32) -> UInt32 {
abiLock.lock()
defer { abiLock.unlock() }
guard let h = handle, !closeRequested else { return false }
var gap = false
_ = punktfunk_connection_note_frame_index(h, frameIndex, &gap)
return gap
guard let h = handle, !closeRequested else { return 0 }
var width: UInt32 = 0
_ = punktfunk_connection_note_frame_index_ex(h, frameIndex, &width)
return width
}
/// Cumulative access units the hostclient reassembler dropped as unrecoverable (FEC couldn't
@@ -55,6 +55,16 @@ final class ReanchorGate: @unchecked Sendable {
lock.unlock()
}
/// `arm()` for a loss detected as a frame-index gap of a known width
/// (`PunktfunkConnection.noteFrameIndexGapWidth`). Pre-credits the reassembler's later
/// `framesDropped` climb for the same lost frames, so `poll` doesn't re-freeze a stream an
/// RFI anchor already healed (the double-arm race the Rust gate's docs tell the story).
func arm(expectingDrops: UInt64) {
lock.lock()
punktfunk_reanchor_gate_arm_expecting_drops(ptr, expectingDrops)
lock.unlock()
}
/// Fold one decoded frame. `flags` is the AU's wire `user_flags`. Returns true to PRESENT the
/// frame, false to WITHHOLD it as a post-loss concealment (hold the last good picture). Pass
/// `decoderKeyframe: false` VideoToolbox doesn't flag IDRs, so the wire `FLAG_SOF` covers it.
@@ -921,7 +921,11 @@ public final class Stage2Pipeline {
// recovery above stays the backstop for when the recovery frame itself is lost.
// The same gap is the earliest, most precise signal to ARM the display freeze
// the following concealed frames are withheld until a clean re-anchor.
if connection.noteFrameIndexGap(au.frameIndex) { reanchorGate.arm() }
// Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
// framesDropped climb for the same loss, so a fast RFI anchor that heals in
// between isn't re-frozen by it (the double-arm race).
let gapWidth = connection.noteFrameIndexGapWidth(au.frameIndex)
if gapWidth > 0 { reanchorGate.arm(expectingDrops: UInt64(gapWidth)) }
onFrame?(au)
if let f = connection.videoCodec.formatDescription(fromKeyframe: au.data) {
format = f // refreshed on every IDR (mode changes included)
@@ -100,7 +100,11 @@ final class StreamPump {
// with a cheap clean P-frame instead of a full IDR. The framesDropped-driven
// recovery above stays the backstop for when the recovery frame itself is lost.
// The same gap is the earliest, most precise signal to ARM the display freeze.
if connection.noteFrameIndexGap(au.frameIndex) { gate.arm() }
// Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
// framesDropped climb for the same loss, so a fast RFI anchor that heals in
// between isn't re-frozen by it (the double-arm race).
let gapWidth = connection.noteFrameIndexGapWidth(au.frameIndex)
if gapWidth > 0 { gate.arm(expectingDrops: UInt64(gapWidth)) }
onFrame?(au)
let idrFormat = connection.videoCodec.formatDescription(fromKeyframe: au.data)
if let f = idrFormat {
+6 -1
View File
@@ -885,7 +885,12 @@ fn pump(
Some(exp) => {
if let Some(gap) = index_gap(exp, frame.frame_index) {
let now = Instant::now();
gate.arm(now);
// Credited arm: the reassembler books these same lost frames into
// `frames_dropped` up to ~120 ms from now; the credit keeps that
// delayed climb from re-freezing a stream the RFI anchor healed in
// between (the double-arm race — see
// `ReanchorGate::arm_expecting_drops`).
gate.arm_expecting_drops(now, u64::from(gap));
next_expected_index = Some(frame.frame_index.wrapping_add(1));
// The gap carries the PRECISE lost range — [first missing, newest
// received - 1] — so this is the one recovery signal that can drive true
+60 -1
View File
@@ -4346,7 +4346,41 @@ pub unsafe extern "C" fn punktfunk_connection_note_frame_index(
if !gap_out.is_null() {
// SAFETY: per the ABI contract - a caller-owned out-param, non-null on this path,
// written once by value.
unsafe { *gap_out = gap };
unsafe { *gap_out = gap > 0 };
}
PunktfunkStatus::Ok
})
}
/// [`punktfunk_connection_note_frame_index`] with the gap WIDTH instead of a yes/no: writes to
/// `gap_width_out` how many frames this arrival revealed as missing (0 = contiguous/straggler).
/// A client with a post-loss display freeze passes the width to
/// [`punktfunk_reanchor_gate_arm_expecting_drops`] so the reassembler's later `frames_dropped`
/// climb for the SAME loss cannot re-freeze a stream an RFI anchor already healed (the double-arm
/// race — see the gate function's doc).
///
/// # Safety
/// `c` is a valid connection handle; `gap_width_out` is writable or NULL.
#[cfg(feature = "quic")]
#[no_mangle]
pub unsafe extern "C" fn punktfunk_connection_note_frame_index_ex(
c: *const PunktfunkConnection,
frame_index: u32,
gap_width_out: *mut u32,
) -> PunktfunkStatus {
guard(|| {
// SAFETY: per the ABI contract - an opaque handle from a `*_new`/`*_pair` that the caller
// has not yet freed, or null, which `as_mut`/`as_ref` reports as `None` and the `match`
// here handles.
let c = match unsafe { c.as_ref() } {
Some(c) => c,
None => return PunktfunkStatus::NullPointer,
};
let gap = c.inner.note_frame_index(frame_index);
if !gap_width_out.is_null() {
// SAFETY: per the ABI contract - a caller-owned out-param, non-null on this path,
// written once by value.
unsafe { *gap_width_out = gap };
}
PunktfunkStatus::Ok
})
@@ -4697,6 +4731,31 @@ pub unsafe extern "C" fn punktfunk_reanchor_gate_arm(g: *mut ReanchorGate) {
});
}
/// [`punktfunk_reanchor_gate_arm`] for a loss detected as a **frame-index gap**, where the caller
/// knows how many frames the gap skipped ([`punktfunk_connection_note_frame_index_ex`]). On top of
/// arming, the gate pre-credits the reassembler's `frames_dropped` climb those same lost frames
/// will produce up to ~120 ms later, so [`punktfunk_reanchor_gate_poll`] does not treat that
/// delayed bookkeeping as a SECOND loss — without the credit, a fast LTR-RFI anchor lifts the
/// freeze between the two signals and the stale climb re-freezes a healed stream (the double-arm
/// race). Use the plain arm for non-gap loss signals (decoder wedge/demotion). NULL is a no-op.
///
/// # Safety
/// `g` is a valid gate handle.
#[no_mangle]
pub unsafe extern "C" fn punktfunk_reanchor_gate_arm_expecting_drops(
g: *mut ReanchorGate,
expected_drops: u64,
) {
guard_void(|| {
// SAFETY: per the ABI contract - an opaque handle from a `*_new`/`*_pair` that the caller has
// not yet freed, or null, which `as_mut`/`as_ref` reports as `None` and the `match` here
// handles.
if let Some(g) = unsafe { g.as_mut() } {
g.arm_expecting_drops(std::time::Instant::now(), expected_drops);
}
});
}
/// Fold one decoded frame and write to `out_present` whether to display it (`true`) or withhold it as
/// a post-loss concealment (`false`). `flags` is the AU's `user_flags` word ([`PunktfunkFrame::flags`]):
/// the gate reads `FLAG_SOF` (the host's IDR marker), `USER_FLAG_RECOVERY_ANCHOR` and
+10 -4
View File
@@ -881,14 +881,20 @@ impl NativeClient {
///
/// Call it for EVERY received frame; it is cheap and idempotent, and the
/// [`frames_dropped`](Self::frames_dropped)-driven [`request_keyframe`](Self::request_keyframe)
/// loop stays the backstop for when the recovery frame itself is lost. Returns `true` when a
/// forward gap was detected on this call (whether or not the RFI was throttled), so a client with
/// a post-loss display freeze can (re-)arm it on the same signal.
/// loop stays the backstop for when the recovery frame itself is lost. Returns the gap WIDTH —
/// how many frames this arrival revealed as missing, `0` when none (contiguous or straggler),
/// whether or not the RFI was throttled — so a client with a post-loss display freeze can
/// (re-)arm it on the same signal AND pre-credit the reassembler's later `frames_dropped` climb
/// for the same loss ([`ReanchorGate::arm_expecting_drops`] — without the credit, a fast
/// LTR-RFI anchor lifts the freeze before the climb books the loss, and the stale climb then
/// re-freezes the healed stream).
///
/// This centralizes the loss-range detection so every embedder gets identical behavior. (The
/// in-process Vulkan session pump keeps its own copy because it gates a display freeze on the same
/// signal and shares one throttle across RFI + keyframe requests.)
pub fn note_frame_index(&self, frame_index: u32) -> bool {
///
/// [`ReanchorGate::arm_expecting_drops`]: crate::reanchor::ReanchorGate::arm_expecting_drops
pub fn note_frame_index(&self, frame_index: u32) -> u32 {
// Decide (and update state) under the lock; fire the request after releasing it.
let (gap, ask) = self
.rfi
+33 -30
View File
@@ -32,14 +32,16 @@ pub(crate) enum RecoveryAsk {
impl RfiRecovery {
/// Pure decision behind [`NativeClient::note_frame_index`]: fold one received `frame_index` (in
/// receive order) observed at `now`, advancing the expectation and returning `(gap, ask)`.
/// `gap` is whether this frame revealed a forward gap (the embedder arms its post-loss display
/// freeze on it); `ask` is the (throttled) recovery request to fire — an RFI naming the exact
/// lost span, or a keyframe when the span exceeds [`crate::packet::RFI_MAX_RANGE`] (RFI is
/// hopeless there: no encoder holds references that old, and a huge jump is more likely a
/// resync — e.g. the first real AU after an old host's speed test — than a real loss). Split
/// out from the connection so the wrapping arithmetic + [`RFI_THROTTLE`] are unit-testable
/// without a live session (see the tests below).
pub(crate) fn observe(&mut self, frame_index: u32, now: Instant) -> (bool, RecoveryAsk) {
/// `gap` is how many frames this arrival revealed as missing — 0 for contiguous/straggler; the
/// embedder arms its post-loss display freeze on a non-zero gap, and the WIDTH lets it
/// pre-credit the reassembler's later `frames_dropped` climb for the same loss
/// ([`crate::reanchor::ReanchorGate::arm_expecting_drops`] — the double-arm race). `ask` is the
/// (throttled) recovery request to fire — an RFI naming the exact lost span, or a keyframe when
/// the span exceeds [`crate::packet::RFI_MAX_RANGE`] (RFI is hopeless there: no encoder holds
/// references that old, and a huge jump is more likely a resync — e.g. the first real AU after
/// an old host's speed test — than a real loss). Split out from the connection so the wrapping
/// arithmetic + [`RFI_THROTTLE`] are unit-testable without a live session (see the tests below).
pub(crate) fn observe(&mut self, frame_index: u32, now: Instant) -> (u32, RecoveryAsk) {
match self.next_expected {
Some(exp) => {
// Wrapping split at the half-space: a small positive delta is a forward gap
@@ -47,10 +49,11 @@ impl RfiRecovery {
let ahead = frame_index.wrapping_sub(exp);
if ahead == 0 {
self.next_expected = Some(frame_index.wrapping_add(1)); // contiguous
(false, RecoveryAsk::None)
(0, RecoveryAsk::None)
} else if ahead < u32::MAX / 2 {
// Forward gap: [exp, frame_index-1] lost. Advance past this frame so the same
// gap isn't re-detected, then fire a throttled recovery ask for the lost range.
// Forward gap: [exp, frame_index-1] lost (`ahead` frames). Advance past this
// frame so the same gap isn't re-detected, then fire a throttled recovery ask
// for the lost range.
self.next_expected = Some(frame_index.wrapping_add(1));
let send = self
.last_req
@@ -65,15 +68,15 @@ impl RfiRecovery {
} else {
RecoveryAsk::Rfi(exp, frame_index.wrapping_sub(1))
};
(true, ask)
(ahead, ask)
} else {
// Straggler behind the delivery point — leave the expectation.
(false, RecoveryAsk::None)
(0, RecoveryAsk::None)
}
}
None => {
self.next_expected = Some(frame_index.wrapping_add(1));
(false, RecoveryAsk::None)
(0, RecoveryAsk::None)
}
}
}
@@ -96,7 +99,7 @@ mod rfi_recovery_tests {
fn first_frame_arms_without_a_gap() {
let mut r = RfiRecovery::default();
// The opening frame only seeds the expectation — there is no prior frame to be missing.
assert_eq!(r.observe(100, base()), (false, RecoveryAsk::None));
assert_eq!(r.observe(100, base()), (0, RecoveryAsk::None));
assert_eq!(r.next_expected, Some(101));
}
@@ -105,9 +108,9 @@ mod rfi_recovery_tests {
let mut r = RfiRecovery::default();
let t = base();
r.observe(100, t);
assert_eq!(r.observe(101, t), (false, RecoveryAsk::None));
assert_eq!(r.observe(102, t), (false, RecoveryAsk::None));
assert_eq!(r.observe(103, t), (false, RecoveryAsk::None));
assert_eq!(r.observe(101, t), (0, RecoveryAsk::None));
assert_eq!(r.observe(102, t), (0, RecoveryAsk::None));
assert_eq!(r.observe(103, t), (0, RecoveryAsk::None));
assert_eq!(r.next_expected, Some(104));
}
@@ -117,7 +120,7 @@ mod rfi_recovery_tests {
let t = base();
r.observe(100, t); // expecting 101 next
// 101..=104 were lost; 105 arrived. The RFI must name exactly the missing span.
assert_eq!(r.observe(105, t), (true, RecoveryAsk::Rfi(101, 104)));
assert_eq!(r.observe(105, t), (4, RecoveryAsk::Rfi(101, 104)));
// The expectation advances past the delivered frame so the same gap can't re-fire.
assert_eq!(r.next_expected, Some(106));
}
@@ -128,7 +131,7 @@ mod rfi_recovery_tests {
let t = base();
r.observe(100, t);
// Exactly one frame (101) lost → range is the single index [101, 101].
assert_eq!(r.observe(102, t), (true, RecoveryAsk::Rfi(101, 101)));
assert_eq!(r.observe(102, t), (1, RecoveryAsk::Rfi(101, 101)));
}
#[test]
@@ -137,16 +140,16 @@ mod rfi_recovery_tests {
let t0 = base();
r.observe(100, t0);
// First gap fires the request and stamps the throttle.
assert_eq!(r.observe(105, t0), (true, RecoveryAsk::Rfi(101, 104)));
assert_eq!(r.observe(105, t0), (4, RecoveryAsk::Rfi(101, 104)));
// A second gap 50 ms later is still a gap, but the request is throttled away.
assert_eq!(
r.observe(110, t0 + Duration::from_millis(50)),
(true, RecoveryAsk::None)
(4, RecoveryAsk::None)
);
// Past the window, the request re-opens for the still-accurate lost span.
assert_eq!(
r.observe(120, t0 + RFI_THROTTLE + Duration::from_millis(1)),
(true, RecoveryAsk::Rfi(111, 119))
(9, RecoveryAsk::Rfi(111, 119))
);
}
@@ -158,7 +161,7 @@ mod rfi_recovery_tests {
r.observe(105, t); // expecting 106 next
// A reordered late arrival (103, well behind 106) is neither a gap nor a request, and it
// must not rewind the expectation — otherwise the next in-order frame would false-gap.
assert_eq!(r.observe(103, t), (false, RecoveryAsk::None));
assert_eq!(r.observe(103, t), (0, RecoveryAsk::None));
assert_eq!(r.next_expected, Some(106));
}
@@ -167,9 +170,9 @@ mod rfi_recovery_tests {
let mut r = RfiRecovery::default();
let t = base();
r.observe(u32::MAX - 1, t); // expecting u32::MAX next
assert_eq!(r.observe(u32::MAX, t), (false, RecoveryAsk::None)); // contiguous, wraps to 0
assert_eq!(r.observe(u32::MAX, t), (0, RecoveryAsk::None)); // contiguous, wraps to 0
assert_eq!(r.next_expected, Some(0));
assert_eq!(r.observe(0, t), (false, RecoveryAsk::None)); // still contiguous across the wrap
assert_eq!(r.observe(0, t), (0, RecoveryAsk::None)); // still contiguous across the wrap
assert_eq!(r.next_expected, Some(1));
}
@@ -179,7 +182,7 @@ mod rfi_recovery_tests {
let t = base();
r.observe(u32::MAX - 1, t); // expecting u32::MAX next
// u32::MAX was lost and 1 arrived → the lost span wraps: [u32::MAX, 0].
assert_eq!(r.observe(1, t), (true, RecoveryAsk::Rfi(u32::MAX, 0)));
assert_eq!(r.observe(1, t), (2, RecoveryAsk::Rfi(u32::MAX, 0)));
assert_eq!(r.next_expected, Some(2));
}
@@ -192,14 +195,14 @@ mod rfi_recovery_tests {
// reference exists for an RFI, and the jump may be a phantom (an old host's
// speed-test burst consuming video indexes) — ask for the IDR resync instead.
let jump = 100 + crate::packet::RFI_MAX_RANGE + 2;
assert_eq!(r.observe(jump, t), (true, RecoveryAsk::Keyframe));
assert_eq!(r.observe(jump, t), (jump - 101, RecoveryAsk::Keyframe));
// The expectation still advances past the delivered frame (no re-fire on the next one).
assert_eq!(r.next_expected, Some(jump + 1));
assert_eq!(r.observe(jump + 1, t), (false, RecoveryAsk::None));
assert_eq!(r.observe(jump + 1, t), (0, RecoveryAsk::None));
// A huge gap consumes the shared throttle too — an immediate follow-up gap stays quiet.
assert_eq!(
r.observe(jump + 10, t + Duration::from_millis(1)),
(true, RecoveryAsk::None)
(8, RecoveryAsk::None)
);
}
}
+148 -6
View File
@@ -64,6 +64,26 @@ pub const REANCHOR_MARKS_TO_LIFT: u32 = 2;
/// floor fires, so a real stall still recovers.
pub const RECOVERY_MARK_PATIENCE: Duration = Duration::from_millis(1500);
/// How long a frame-index-gap arm's expected `frames_dropped` climb stays pre-credited in
/// [`ReanchorGate::poll`]. One loss arms the gate through TWO signals: the frame-index gap the
/// instant the AU after the loss is delivered ([`ReanchorGate::arm_expecting_drops`]), and the
/// reassembler's `frames_dropped` climb once the lost frame ages out of its loss window (~120 ms
/// later, and only when at least one of its packets arrived). Without the credit, a *fast* recovery
/// — an LTR-RFI anchor typically lands within ~60 ms — lifts the freeze between the two signals,
/// and the stale climb then re-freezes a stream that is already bit-exact healed; the host swallows
/// the resulting keyframe request as an echo of the very RFI that healed it, so the picture stays
/// frozen until the [`REANCHOR_FREEZE_MAX`] overdue re-ask extracts a full IDR (the field
/// "H265 freezes on every loss, AV1 fine" signature — the slower IDR path usually lands after the
/// climb and dodged the race).
///
/// Sized to cover the reassembler's 120 ms loss window plus delivery jitter with a wide margin,
/// while staying short enough that a leftover credit (a straggler that filled the gap late, so no
/// climb ever came; or a whole-frame vanish the reassembler never saw a packet of) cannot mask a
/// genuinely unrelated future climb for long. A masked climb is also never silent in practice:
/// every unrecoverable loss reveals itself as a frame-index gap on the next delivered frame, which
/// re-arms (and re-credits) through [`ReanchorGate::arm_expecting_drops`] on its own.
pub const DROP_CREDIT_WINDOW: Duration = Duration::from_millis(1000);
/// Frames skipped when `got` arrives while `expected` was the next index, or `None` if `got` is
/// contiguous (`== expected`) or a straggler we have already passed. Frame indices are u32 counters
/// that wrap, so the "ahead" test is a wrapping subtraction split at the half-space: a small positive
@@ -185,6 +205,14 @@ pub struct ReanchorGate {
/// a client stamps the decoder's decode-order watermark whenever this counter moves and
/// discards the local recovery of anything older. Every other client ignores it.
arms: u64,
/// `frames_dropped` climb still expected from losses that already armed via a frame-index gap
/// ([`Self::arm_expecting_drops`]). [`Self::poll`] consumes climbs against this before treating
/// them as a NEW loss, so the reassembler's delayed bookkeeping of a gap-armed (and possibly
/// already anchor-healed) loss cannot re-freeze the stream — see [`DROP_CREDIT_WINDOW`].
drop_credit: u64,
/// When the outstanding [`Self::drop_credit`] lapses ([`DROP_CREDIT_WINDOW`] after the latest
/// credited arm). `None` when no credit is outstanding.
drop_credit_expiry: Option<Instant>,
}
impl ReanchorGate {
@@ -199,6 +227,8 @@ impl ReanchorGate {
last_dropped: frames_dropped,
local_sei_since_arm: false,
arms: 0,
drop_credit: 0,
drop_credit_expiry: None,
}
}
@@ -230,6 +260,20 @@ impl ReanchorGate {
self.deadline = Some(now + REANCHOR_FREEZE_MAX);
}
/// [`arm`](Self::arm) for a loss detected as a **frame-index gap**, where the caller knows how
/// many frames the gap skipped. On top of arming, it pre-credits the reassembler's
/// `frames_dropped` climb those same lost frames will produce up to ~120 ms later (its
/// loss-window age-out), so [`poll`](Self::poll) does not treat that delayed bookkeeping as a
/// SECOND loss. Without the credit a fast LTR-RFI anchor lifts the freeze between the two
/// signals and the stale climb re-freezes a healed stream — the double-arm race
/// ([`DROP_CREDIT_WINDOW`] tells the whole story). Use plain [`arm`](Self::arm) for every
/// non-gap loss signal (decoder wedge/demotion), which has no climb to credit.
pub fn arm_expecting_drops(&mut self, now: Instant, expected_drops: u64) {
self.arm(now);
self.drop_credit = self.drop_credit.saturating_add(expected_drops);
self.drop_credit_expiry = Some(now + DROP_CREDIT_WINDOW);
}
/// Fold the client's OWN recovery-point observation for one decoded frame, BEFORE handing that
/// frame to [`on_decoded`](Self::on_decoded). Returns `true` when it lifted the freeze.
///
@@ -333,16 +377,36 @@ impl ReanchorGate {
}
/// Periodic fold of the session's `frames_dropped` counter plus the overdue backstop. Returns
/// `true` when the client should (throttled) request a keyframe: either the drop count climbed (a
/// fresh unrecoverable loss — arm the freeze) or the freeze has held a full [`REANCHOR_FREEZE_MAX`]
/// window with no re-anchor (re-ask and keep holding — NEVER resume to the concealed picture; a
/// genuinely dead stream is the QUIC idle-timeout watchdog's job, not the gate's).
/// `true` when the client should (throttled) request a keyframe: either the drop count climbed by
/// more than the outstanding gap-arm credit (a fresh unrecoverable loss — arm the freeze) or the
/// freeze has held a full [`REANCHOR_FREEZE_MAX`] window with no re-anchor (re-ask and keep
/// holding — NEVER resume to the concealed picture; a genuinely dead stream is the QUIC
/// idle-timeout watchdog's job, not the gate's).
///
/// A climb covered by [`arm_expecting_drops`](Self::arm_expecting_drops)' credit is the
/// reassembler's delayed bookkeeping of a loss this gate already armed for — it must neither
/// re-arm (an LTR-RFI anchor may have healed the stream in the meantime; re-freezing it is the
/// double-arm race) nor ask again (the gap already fired the precise RFI, and if THAT recovery
/// was lost the overdue backstop still re-asks at the [`REANCHOR_FREEZE_MAX`] deadline the
/// gap-arm set — which is also sooner than the deadline a re-arm here would push out to).
pub fn poll(&mut self, frames_dropped: u64, now: Instant) -> bool {
let mut want_keyframe = false;
if frames_dropped > self.last_dropped {
let climb = frames_dropped - self.last_dropped;
self.last_dropped = frames_dropped;
self.arm(now);
want_keyframe = true;
if self.drop_credit_expiry.is_some_and(|e| now >= e) {
self.drop_credit = 0;
self.drop_credit_expiry = None;
}
let credited = climb.min(self.drop_credit);
self.drop_credit -= credited;
if self.drop_credit == 0 {
self.drop_credit_expiry = None;
}
if climb > credited {
self.arm(now);
want_keyframe = true;
}
}
if self.awaiting && self.deadline.is_some_and(|d| now >= d) {
self.deadline = Some(now + REANCHOR_FREEZE_MAX);
@@ -542,6 +606,84 @@ mod tests {
);
}
#[test]
fn an_rfi_anchor_is_not_refrozen_by_the_same_losss_drop_climb() {
// The double-arm race (field: "H265 freezes on every loss, AV1 fine"): a loss arms via
// the frame-index gap at T+10ms, the LTR-RFI anchor heals at T+60ms, and the reassembler
// books the SAME loss into frames_dropped at ~T+130ms. The credited arm must keep that
// stale climb from re-freezing the healed stream (and from re-asking — the host would
// swallow the ask as an RFI echo and the picture would freeze until a forced IDR).
let mut g = ReanchorGate::new(0);
let t = t0();
g.arm_expecting_drops(t + Duration::from_millis(10), 1); // gap of one lost frame + RFI
assert_eq!(
g.on_decoded(ANCHOR, false, t + Duration::from_millis(60)),
GateVerdict::Present,
"the anchor lifts"
);
assert!(
!g.poll(1, t + Duration::from_millis(130)),
"the credited climb must not ask again"
);
assert!(!g.is_holding(), "and must not re-freeze the healed stream");
assert_eq!(
g.on_decoded(0, false, t + Duration::from_millis(141)),
GateVerdict::Present,
"healthy P-frames keep presenting"
);
}
#[test]
fn a_climb_beyond_the_credit_is_a_fresh_loss_and_arms() {
// The credit covers exactly the gap's frames; a bigger climb means MORE loss than the gap
// accounted for (an interleaved partial-frame loss) — that part must still arm and ask.
let mut g = ReanchorGate::new(0);
let t = t0();
g.arm_expecting_drops(t, 2);
g.on_decoded(ANCHOR, false, t + Duration::from_millis(50)); // healed the credited loss
assert!(
g.poll(3, t + Duration::from_millis(130)),
"one uncredited drop → ask"
);
assert!(g.is_holding(), "and re-arm for the uncredited part");
}
#[test]
fn the_drop_credit_expires_so_a_late_climb_still_arms() {
// A straggler can fill the gap late (no climb ever comes) — the leftover credit must not
// linger and mask a genuinely NEW loss later. Past DROP_CREDIT_WINDOW the credit is void.
let mut g = ReanchorGate::new(0);
let t = t0();
g.arm_expecting_drops(t, 1);
g.on_decoded(ANCHOR, false, t + Duration::from_millis(50));
let late = t + DROP_CREDIT_WINDOW + Duration::from_millis(1);
assert!(
g.poll(1, late),
"an expired credit no longer absorbs climbs"
);
assert!(g.is_holding());
}
#[test]
fn a_credited_climb_keeps_the_unhealed_freezes_original_deadline() {
// When the recovery never arrives, consuming the climb must not silence the gate: the
// overdue backstop still re-asks — at the deadline the GAP arm set, which is sooner than
// the deadline a climb re-arm would have pushed out to.
let mut g = ReanchorGate::new(0);
let t = t0();
g.arm_expecting_drops(t, 1); // RFI fired here; assume its anchor is lost in transit
assert!(
!g.poll(1, t + Duration::from_millis(130)),
"credited climb: no early re-ask"
);
assert!(g.is_holding(), "still frozen — nothing healed it");
assert!(
g.poll(1, t + REANCHOR_FREEZE_MAX + Duration::from_millis(1)),
"the overdue backstop still re-asks on the gap-arm's own deadline"
);
assert!(g.is_holding(), "and keeps holding, never resuming to gray");
}
#[test]
fn the_no_output_streak_trips_at_three() {
let mut g = ReanchorGate::new(0);
+27
View File
@@ -3319,6 +3319,21 @@ PunktfunkStatus punktfunk_connection_note_frame_index(const PunktfunkConnection
bool *gap_out);
#endif
#if defined(PUNKTFUNK_FEATURE_QUIC)
// [`punktfunk_connection_note_frame_index`] with the gap WIDTH instead of a yes/no: writes to
// `gap_width_out` how many frames this arrival revealed as missing (0 = contiguous/straggler).
// A client with a post-loss display freeze passes the width to
// [`punktfunk_reanchor_gate_arm_expecting_drops`] so the reassembler's later `frames_dropped`
// climb for the SAME loss cannot re-freeze a stream an RFI anchor already healed (the double-arm
// race — see the gate function's doc).
//
// # Safety
// `c` is a valid connection handle; `gap_width_out` is writable or NULL.
PunktfunkStatus punktfunk_connection_note_frame_index_ex(const PunktfunkConnection *c,
uint32_t frame_index,
uint32_t *gap_width_out);
#endif
#if defined(PUNKTFUNK_FEATURE_QUIC)
// Cumulative access units the host→client reassembler dropped as unrecoverable (FEC couldn't
// rebuild them). A video loop polls this and calls [`punktfunk_connection_request_keyframe`]
@@ -3442,6 +3457,18 @@ void punktfunk_reanchor_gate_free(ReanchorGate *g);
// `g` is a valid gate handle.
void punktfunk_reanchor_gate_arm(ReanchorGate *g);
// [`punktfunk_reanchor_gate_arm`] for a loss detected as a **frame-index gap**, where the caller
// knows how many frames the gap skipped ([`punktfunk_connection_note_frame_index_ex`]). On top of
// arming, the gate pre-credits the reassembler's `frames_dropped` climb those same lost frames
// will produce up to ~120 ms later, so [`punktfunk_reanchor_gate_poll`] does not treat that
// delayed bookkeeping as a SECOND loss — without the credit, a fast LTR-RFI anchor lifts the
// freeze between the two signals and the stale climb re-freezes a healed stream (the double-arm
// race). Use the plain arm for non-gap loss signals (decoder wedge/demotion). NULL is a no-op.
//
// # Safety
// `g` is a valid gate handle.
void punktfunk_reanchor_gate_arm_expecting_drops(ReanchorGate *g, uint64_t expected_drops);
// Fold one decoded frame and write to `out_present` whether to display it (`true`) or withhold it as
// a post-loss concealment (`false`). `flags` is the AU's `user_flags` word ([`PunktfunkFrame::flags`]):
// the gate reads `FLAG_SOF` (the host's IDR marker), `USER_FLAG_RECOVERY_ANCHOR` and