diff --git a/clients/android/native/src/decode/async_loop.rs b/clients/android/native/src/decode/async_loop.rs index 69ace83c..0bc4123a 100644 --- a/clients/android/native/src/decode/async_loop.rs +++ b/clients/android/native/src/decode/async_loop.rs @@ -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 { diff --git a/clients/android/native/src/decode/sync_loop.rs b/clients/android/native/src/decode/sync_loop.rs index 9885adc8..4f5e1197 100644 --- a/clients/android/native/src/decode/sync_loop.rs +++ b/clients/android/native/src/decode/sync_loop.rs @@ -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. diff --git a/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift b/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift index e209cbae..b525fa25 100644 --- a/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift +++ b/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift @@ -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 host→client reassembler dropped as unrecoverable (FEC couldn't diff --git a/clients/apple/Sources/PunktfunkKit/Video/ReanchorGate.swift b/clients/apple/Sources/PunktfunkKit/Video/ReanchorGate.swift index 95fdf680..a3988d6e 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/ReanchorGate.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/ReanchorGate.swift @@ -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. diff --git a/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift b/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift index bd7e9d52..5330a35b 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift @@ -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) diff --git a/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift b/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift index e647f44f..ecc213cd 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift @@ -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 { diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 74240748..f881ea9d 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -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 diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index 73f9f2f7..e74810f1 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -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 diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index b49afde8..c7bc8ddc 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -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 diff --git a/crates/punktfunk-core/src/client/recovery.rs b/crates/punktfunk-core/src/client/recovery.rs index bf2184a2..dba14d67 100644 --- a/crates/punktfunk-core/src/client/recovery.rs +++ b/crates/punktfunk-core/src/client/recovery.rs @@ -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) ); } } diff --git a/crates/punktfunk-core/src/reanchor.rs b/crates/punktfunk-core/src/reanchor.rs index d005cc6c..b5977d1f 100644 --- a/crates/punktfunk-core/src/reanchor.rs +++ b/crates/punktfunk-core/src/reanchor.rs @@ -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, } 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); diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 2f62ab3d..30a188e1 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -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