diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index c5b9d9e3..6863afd9 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -462,12 +462,21 @@ fn pump( // paths (one recovery ask per 100 ms) so a burst of gaps — a full-screen // pan shedding shards — can't storm the control stream. This fires ~120 ms // before frames_dropped would, so recovery also starts sooner. + // + // A gap wider than RFI_MAX_RANGE is beyond any encoder's reference + // history (a seconds-long outage — or a phantom index jump, e.g. the + // first real AU after an old host's speed-test burst consumed video + // indexes): RFI is hopeless there, so ask for the IDR resync directly. if last_kf_req .is_none_or(|t| now.duration_since(t) >= Duration::from_millis(100)) { last_kf_req = Some(now); - let _ = - connector.request_rfi(exp, frame.frame_index.wrapping_sub(1)); + if gap > punktfunk_core::packet::RFI_MAX_RANGE { + let _ = connector.request_keyframe(); + } else { + let _ = connector + .request_rfi(exp, frame.frame_index.wrapping_sub(1)); + } } tracing::trace!( gap, diff --git a/crates/punktfunk-core/src/client.rs b/crates/punktfunk-core/src/client.rs index 78020d22..82a170c2 100644 --- a/crates/punktfunk-core/src/client.rs +++ b/crates/punktfunk-core/src/client.rs @@ -375,14 +375,29 @@ struct RfiRecovery { last_req: Option, } +/// What a forward gap should ask the host for: a precise RFI for a recoverable range, a plain +/// keyframe for a range wider than any encoder's reference history +/// ([`crate::packet::RFI_MAX_RANGE`] — a seconds-long outage, or a phantom index jump such as an +/// old host's speed-test burst consuming video indexes), or nothing (contiguous / straggler / +/// throttled). +#[derive(Debug, PartialEq, Eq)] +enum RecoveryAsk { + None, + Rfi(u32, u32), + Keyframe, +} + 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, rfi_range)`. `gap` is whether this frame revealed a forward gap; `rfi_range` is - /// `Some((first_missing, last_missing))` when a (throttled) RFI should fire for the lost span, or - /// `None` when contiguous, a straggler, or throttled. Split out from the connection so the wrapping - /// arithmetic + [`RFI_THROTTLE`] are unit-testable without a live session (see the tests below). - fn observe(&mut self, frame_index: u32, now: Instant) -> (bool, Option<(u32, u32)>) { + /// 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). + fn observe(&mut self, frame_index: u32, now: Instant) -> (bool, RecoveryAsk) { match self.next_expected { Some(exp) => { // Wrapping split at the half-space: a small positive delta is a forward gap @@ -390,10 +405,10 @@ impl RfiRecovery { let ahead = frame_index.wrapping_sub(exp); if ahead == 0 { self.next_expected = Some(frame_index.wrapping_add(1)); // contiguous - (false, None) + (false, 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 RFI for the lost range. + // 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 @@ -401,15 +416,22 @@ impl RfiRecovery { if send { self.last_req = Some(now); } - let range = send.then(|| (exp, frame_index.wrapping_sub(1))); - (true, range) + let ask = if !send { + RecoveryAsk::None + } else if ahead > crate::packet::RFI_MAX_RANGE { + RecoveryAsk::Keyframe + } else { + RecoveryAsk::Rfi(exp, frame_index.wrapping_sub(1)) + }; + (true, ask) } else { - (false, None) // straggler behind the delivery point — leave the expectation + // Straggler behind the delivery point — leave the expectation. + (false, RecoveryAsk::None) } } None => { self.next_expected = Some(frame_index.wrapping_add(1)); - (false, None) + (false, RecoveryAsk::None) } } } @@ -970,13 +992,21 @@ impl NativeClient { /// signal and shares one throttle across RFI + keyframe requests.) pub fn note_frame_index(&self, frame_index: u32) -> bool { // Decide (and update state) under the lock; fire the request after releasing it. - let (gap, rfi_range) = self + let (gap, ask) = self .rfi .lock() .unwrap() .observe(frame_index, Instant::now()); - if let Some((first, last)) = rfi_range { - let _ = self.request_rfi(first, last); + match ask { + RecoveryAsk::Rfi(first, last) => { + let _ = self.request_rfi(first, last); + } + // A gap wider than any encoder's reference history (RFI_MAX_RANGE) — a seconds-long + // outage or a phantom index jump: RFI can't repair it, resync on a keyframe instead. + RecoveryAsk::Keyframe => { + let _ = self.request_keyframe(); + } + RecoveryAsk::None => {} } gap } @@ -1410,7 +1440,12 @@ async fn worker_main(args: WorkerArgs) { // when the matching bit is set, so `0` stays an 8-bit BT.709 stream. HOST_TIMING is // OR'd in unconditionally: every NativeClient build demuxes the 0xCF plane, and the // bit only asks the host for observability datagrams (never changes the encode). - video_caps: video_caps | crate::quic::VIDEO_CAP_HOST_TIMING, + // PROBE_SEQ likewise: the shared reassembler keeps probe filler in its own window + // (every embedder inherits it), so the host may burst speed tests without consuming + // video frame indexes. + video_caps: video_caps + | crate::quic::VIDEO_CAP_HOST_TIMING + | crate::quic::VIDEO_CAP_PROBE_SEQ, // Requested surround channel count; the host echoes the resolved value in Welcome. audio_channels, // The codecs this client can decode + its soft preference (0 = auto). The host @@ -2071,7 +2106,7 @@ mod rfi_recovery_tests { //! The client-side loss-range detector shared by every embedder (Android, the C-ABI Apple //! client, the Windows shell pump). `observe` is pure over `(frame_index, now)`, so the wrapping //! frame arithmetic and the RFI throttle are exercised here without a live session. - use super::{RfiRecovery, RFI_THROTTLE}; + use super::{RecoveryAsk, RfiRecovery, RFI_THROTTLE}; use std::time::{Duration, Instant}; // A fixed base instant; offsets model the throttle window deterministically (no sleeping). @@ -2083,7 +2118,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, None)); + assert_eq!(r.observe(100, base()), (false, RecoveryAsk::None)); assert_eq!(r.next_expected, Some(101)); } @@ -2092,9 +2127,9 @@ mod rfi_recovery_tests { let mut r = RfiRecovery::default(); let t = base(); r.observe(100, t); - assert_eq!(r.observe(101, t), (false, None)); - assert_eq!(r.observe(102, t), (false, None)); - assert_eq!(r.observe(103, t), (false, None)); + 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.next_expected, Some(104)); } @@ -2104,7 +2139,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, Some((101, 104)))); + assert_eq!(r.observe(105, t), (true, 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)); } @@ -2115,7 +2150,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, Some((101, 101)))); + assert_eq!(r.observe(102, t), (true, RecoveryAsk::Rfi(101, 101))); } #[test] @@ -2124,13 +2159,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, Some((101, 104)))); + assert_eq!(r.observe(105, t0), (true, 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, None)); + assert_eq!( + r.observe(110, t0 + Duration::from_millis(50)), + (true, 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, Some((111, 119))) + (true, RecoveryAsk::Rfi(111, 119)) ); } @@ -2142,7 +2180,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, None)); + assert_eq!(r.observe(103, t), (false, RecoveryAsk::None)); assert_eq!(r.next_expected, Some(106)); } @@ -2151,9 +2189,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, None)); // contiguous, expectation wraps to 0 + assert_eq!(r.observe(u32::MAX, t), (false, RecoveryAsk::None)); // contiguous, wraps to 0 assert_eq!(r.next_expected, Some(0)); - assert_eq!(r.observe(0, t), (false, None)); // still contiguous across the wrap + assert_eq!(r.observe(0, t), (false, RecoveryAsk::None)); // still contiguous across the wrap assert_eq!(r.next_expected, Some(1)); } @@ -2163,9 +2201,29 @@ 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, Some((u32::MAX, 0)))); + assert_eq!(r.observe(1, t), (true, RecoveryAsk::Rfi(u32::MAX, 0))); assert_eq!(r.next_expected, Some(2)); } + + #[test] + fn huge_gap_resyncs_via_keyframe_not_rfi() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(100, t); // expecting 101 next + // A jump wider than any encoder's reference history (RFI_MAX_RANGE): no valid + // 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)); + // 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)); + // 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) + ); + } } #[cfg(test)] diff --git a/crates/punktfunk-core/src/packet.rs b/crates/punktfunk-core/src/packet.rs index 981efe6f..6eab5976 100644 --- a/crates/punktfunk-core/src/packet.rs +++ b/crates/punktfunk-core/src/packet.rs @@ -54,6 +54,15 @@ pub const USER_FLAG_RECOVERY_POINT: u32 = 0x10; /// `AV_FRAME_FLAG_KEY` — this host flag is the only signal. pub const USER_FLAG_RECOVERY_ANCHOR: u32 = 0x20; +/// Widest lost-frame range (frames, wrapping `last - first`) a reference-frame-invalidation +/// recovery may be asked to repair; anything wider goes straight to the keyframe path on BOTH +/// ends. RFI can only re-reference history the encoder still holds — NVENC keeps a 5-frame DPB, +/// AMD LTR ~1 s of marks — and a genuine loss this wide (>1 s even at 240 fps) has no valid +/// reference anywhere, so an RFI request for it is either hopeless or (worse) a phantom range +/// from a desynced counter. Shared by the host's RFI dispatch (range → keyframe fallback) and the +/// client-side gap detectors (huge gap → resync + keyframe request, no RFI). +pub const RFI_MAX_RANGE: u32 = 256; + /// Crypto framing overhead [`Session`](crate::session::Session) adds when encrypting: /// an 8-byte sequence prefix plus the GCM tag. pub const CRYPTO_OVERHEAD: usize = 8 + crate::crypto::TAG_LEN; @@ -117,8 +126,18 @@ const _: () = assert!(HEADER_LEN == 40, "PacketHeader must be 40 bytes / unpadde // --------------------------------------------------------------------------- /// Splits encoded access units into FEC-protected shard packets. Host-side only. +/// +/// Frame numbering: a caller can pass an **explicit** `frame_index` to +/// [`packetize_each`](Self::packetize_each) (the punktfunk/1 encode loop owns the video numbering +/// so the encoder's reference-frame-invalidation bookkeeping stays 1:1 with the wire across +/// encoder rebuilds/resets), or pass `None` to draw from the internal counter (the legacy path — +/// synthetic/spike/ABI sessions where no encoder cares). Speed-test probe filler draws from a +/// **separate** index space ([`alloc_probe_index`](Self::alloc_probe_index)) so a burst never +/// consumes video indexes — see [`crate::quic::VIDEO_CAP_PROBE_SEQ`]. pub struct Packetizer { next_frame_index: u32, + /// Probe-space frame counter (see [`alloc_probe_index`](Self::alloc_probe_index)). + next_probe_index: u32, next_seq: u32, shard_payload: usize, fec: crate::config::FecConfig, @@ -134,6 +153,7 @@ impl Packetizer { pub fn new(config: &Config) -> Self { Packetizer { next_frame_index: 0, + next_probe_index: 0, next_seq: 0, shard_payload: config.shard_payload, fec: config.fec, @@ -142,6 +162,17 @@ impl Packetizer { } } + /// Allocate the next **probe-space** frame index (speed-test filler). A separate counter from + /// the video `frame_index`es so a multi-thousand-AU probe burst never advances the video + /// numbering — the client routes [`FLAG_PROBE`]-flagged shards into its own reassembly window + /// (see [`Reassembler`]), so the two spaces never collide. Only used against clients that + /// advertise [`crate::quic::VIDEO_CAP_PROBE_SEQ`]. + pub fn alloc_probe_index(&mut self) -> u32 { + let i = self.next_probe_index; + self.next_probe_index = i.wrapping_add(1); + i + } + /// Live-adjust the FEC recovery percentage (adaptive FEC). Takes effect on the next /// [`packetize`](Self::packetize); the wire is self-describing (each packet carries its block's /// data/recovery counts), so the receiver needs no notification. Clamped to ≤ 90. @@ -165,7 +196,7 @@ impl Packetizer { coder: &dyn ErasureCoder, ) -> Result>> { let mut packets = Vec::new(); - self.packetize_each(frame, pts_ns, user_flags, coder, |hdr, body| { + self.packetize_each(frame, pts_ns, user_flags, None, coder, |hdr, body| { let mut pkt = Vec::with_capacity(HEADER_LEN + body.len()); pkt.extend_from_slice(hdr.as_bytes()); pkt.extend_from_slice(body); @@ -181,17 +212,27 @@ impl Packetizer { /// shard straight into a pooled wire buffer and seal in place /// ([`Session::seal_frame`](crate::session::Session::seal_frame)). An `emit` error aborts /// the frame mid-way (packet numbering has already advanced — callers treat it as fatal). + /// + /// `frame_index`: `Some(i)` stamps the AU with the caller's index — the punktfunk/1 encode + /// loop numbers video AUs itself so the encoder's RFI bookkeeping (LTR marks, DPB timestamps) + /// is 1:1 with what the client sees, surviving encoder rebuilds/resets that restart internal + /// counters. `None` draws from the internal counter (the legacy/self-numbering path). A + /// session must not mix the two styles for the same index space. pub fn packetize_each( &mut self, frame: &[u8], pts_ns: u64, user_flags: u32, + frame_index: Option, coder: &dyn ErasureCoder, mut emit: impl FnMut(&PacketHeader, &[u8]) -> Result<()>, ) -> Result<()> { let payload = self.shard_payload; - let frame_index = self.next_frame_index; - self.next_frame_index = self.next_frame_index.wrapping_add(1); + let frame_index = frame_index.unwrap_or_else(|| { + let i = self.next_frame_index; + self.next_frame_index = i.wrapping_add(1); + i + }); // At least one (zero-padded) data shard even for an empty frame. let total_data = frame.len().div_ceil(payload).max(1); @@ -343,10 +384,13 @@ impl ReassemblerLimits { } } -/// Buffers incoming shards, recovers lost ones via FEC, and emits whole access units. -/// Client-side only. -pub struct Reassembler { - limits: ReassemblerLimits, +/// One frame-index space's reassembly state: the in-flight frames, the recently-emitted memory, +/// and the loss-window anchor. The [`Reassembler`] keeps two — video and speed-test probe filler — +/// because the two ride **separate index counters** on a [`VIDEO_CAP_PROBE_SEQ`]-aware host +/// (a probe burst must neither advance the video loss window nor be dropped as "stale" against +/// it). [`VIDEO_CAP_PROBE_SEQ`]: crate::quic::VIDEO_CAP_PROBE_SEQ +#[derive(Default)] +struct ReassemblyWindow { frames: HashMap, /// Recently-emitted frames, so stray/late shards can't resurrect them. Pruned to /// the reorder window alongside `frames`. @@ -357,13 +401,27 @@ pub struct Reassembler { newest_frame: Option<(u32, u64)>, } +/// Buffers incoming shards, recovers lost ones via FEC, and emits whole access units. +/// Client-side only. +pub struct Reassembler { + limits: ReassemblerLimits, + /// The video stream's window — its aged-out incomplete frames count into `frames_dropped` + /// (the client's loss-recovery trigger). + video: ReassemblyWindow, + /// Speed-test probe filler ([`FLAG_PROBE`] in `user_flags`). Routed by the flag, so it also + /// captures an OLD host's probe frames (which still carry video-space indexes — they complete + /// fine here, and keeping them out of the video window means a burst can no longer advance the + /// video loss anchor). Aged-out probe frames are NOT `frames_dropped` — probe loss is measured + /// bytes-wise by the probe accumulator and must not fire video recovery. + probe: ReassemblyWindow, +} + impl Reassembler { pub fn new(limits: ReassemblerLimits) -> Self { Reassembler { limits, - frames: HashMap::new(), - completed: HashSet::new(), - newest_frame: None, + video: ReassemblyWindow::default(), + probe: ReassemblyWindow::default(), } } @@ -424,18 +482,28 @@ impl Reassembler { } let payload = pkt[HEADER_LEN..HEADER_LEN + shard_bytes].to_vec(); - self.advance_window(hdr.frame_index, hdr.pts_ns, stats); + // Route by index space: speed-test probe filler (FLAG_PROBE in user_flags) reassembles in + // its own window so its indexes never interact with the video loss window — a probe burst + // can neither advance the video anchor nor be dropped as stale against it (and its aged-out + // frames never count as `frames_dropped`, which would fire video loss recovery). + let is_probe = hdr.user_flags & (FLAG_PROBE as u32) != 0; + let win = if is_probe { + &mut self.probe + } else { + &mut self.video + }; + win.advance_window(hdr.frame_index, hdr.pts_ns, stats, !is_probe); // Drop shards for frames we've already emitted (e.g. the recovery shards of a // frame that completed early via the all-originals-present fast path) or that // have fallen out of the loss window. - if self.completed.contains(&hdr.frame_index) || self.is_stale(hdr.frame_index, hdr.pts_ns) { + if win.completed.contains(&hdr.frame_index) || win.is_stale(hdr.frame_index, hdr.pts_ns) { drop(stats); return Ok(None); } // First packet of a frame establishes its geometry; later packets must agree. - let frame = self + let frame = win .frames .entry(hdr.frame_index) .or_insert_with(|| FrameBuf { @@ -521,8 +589,8 @@ impl Reassembler { // Whole frame ready? if frame.block_data.len() == frame.block_count { - let frame = self.frames.remove(&hdr.frame_index).unwrap(); - self.completed.insert(hdr.frame_index); + let frame = win.frames.remove(&hdr.frame_index).unwrap(); + win.completed.insert(hdr.frame_index); // Reserve based on the bytes we actually hold, not the (already-bounded but // still caller-supplied) frame_bytes, so a small frame can't over-reserve. let actual: usize = frame.block_data.values().map(|b| b.len()).sum(); @@ -541,11 +609,30 @@ impl Reassembler { Ok(None) } + /// Drop all in-flight state — every partially-assembled frame and the completed/abandoned + /// index memory, in both index spaces — as if the session just started. Used by the client's + /// backlog flush ([`Session::flush_backlog`](crate::session::Session::flush_backlog)): after + /// the socket backlog is discarded wholesale, the partial frames here can never complete + /// (their remaining shards were just thrown away) and the window anchors (`newest_frame`) + /// point into the discarded past. + pub fn reset(&mut self) { + self.video = ReassemblyWindow::default(); + self.probe = ReassemblyWindow::default(); + } +} + +impl ReassemblyWindow { /// Track the newest frame, declare incomplete frames that fell out of the loss window - /// ([`LOSS_WINDOW_NS`] behind the newest pts, or [`HARD_LOSS_WINDOW`] indices) lost — counting - /// them dropped, which is what drives the client's recovery-keyframe request — and prune the - /// completed-index memory to [`REORDER_WINDOW`]. - fn advance_window(&mut self, frame_index: u32, pts_ns: u64, stats: &StatsCounters) { + /// ([`LOSS_WINDOW_NS`] behind the newest pts, or [`HARD_LOSS_WINDOW`] indices) lost — for the + /// video window (`count_drops`) counting them dropped, which is what drives the client's + /// recovery-keyframe request — and prune the completed-index memory to [`REORDER_WINDOW`]. + fn advance_window( + &mut self, + frame_index: u32, + pts_ns: u64, + stats: &StatsCounters, + count_drops: bool, + ) { let (newest, newest_pts) = match self.newest_frame { // `frame_index` is newer iff it's within the forward half of the index space. Some((n, p)) if frame_index.wrapping_sub(n) > u32::MAX / 2 => (n, p), @@ -567,29 +654,17 @@ impl Reassembler { keep }); let pruned = before - self.frames.len(); - if pruned > 0 { + if pruned > 0 && count_drops { StatsCounters::add(&stats.frames_dropped, pruned as u64); } self.completed .retain(|&idx| newest.wrapping_sub(idx) <= REORDER_WINDOW); } - /// Drop all in-flight state — every partially-assembled frame and the completed/abandoned - /// index memory — as if the session just started. Used by the client's backlog flush - /// ([`Session::flush_backlog`](crate::session::Session::flush_backlog)): after the socket - /// backlog is discarded wholesale, the partial frames here can never complete (their remaining - /// shards were just thrown away) and the window anchor (`newest_frame`) points into the - /// discarded past. - pub fn reset(&mut self) { - self.frames.clear(); - self.completed.clear(); - self.newest_frame = None; - } - /// True if this packet's frame lies outside the loss window (behind the newest frame by more /// than [`LOSS_WINDOW_NS`] of capture time or [`HARD_LOSS_WINDOW`] indices) — its shards /// arrive too late to be useful, and accepting one would only create a frame buffer the next - /// [`advance_window`] immediately declares lost. + /// [`advance_window`](Self::advance_window) immediately declares lost. fn is_stale(&self, frame_index: u32, pts_ns: u64) -> bool { match self.newest_frame { Some((n, newest_pts)) => { @@ -769,6 +844,119 @@ mod tests { assert_eq!(stats.snapshot().frames_dropped, 1, "no double-count"); } + /// The explicit-index path stamps the caller's `frame_index` and leaves the internal video + /// counter untouched — the punktfunk/1 encode loop owns the numbering, and mixing must not + /// perturb the legacy self-numbering path (tests/ABI/synthetic). + #[test] + fn explicit_frame_index_is_stamped_and_internal_counter_untouched() { + use crate::config::{FecConfig, FecScheme, ProtocolPhase, Role}; + let cfg = Config { + role: Role::Host, + phase: ProtocolPhase::P2Punktfunk, + fec: FecConfig { + scheme: FecScheme::Gf16, + fec_percent: 0, + max_data_per_block: 8, + }, + shard_payload: 16, + max_frame_bytes: 4096, + encrypt: false, + key: [0u8; 16], + salt: [0u8; 4], + loopback_drop_period: 0, + }; + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let mut seen = Vec::new(); + pk.packetize_each(&[1u8; 16], 0, 0, Some(4242), coder.as_ref(), |hdr, _| { + seen.push(hdr.frame_index); + Ok(()) + }) + .unwrap(); + assert_eq!(seen, vec![4242]); + // The legacy wrapper still numbers from the untouched internal counter. + let pkts = pk.packetize(&[1u8; 16], 0, 0, coder.as_ref()).unwrap(); + let hdr = PacketHeader::read_from_bytes(&pkts[0][..HEADER_LEN]).unwrap(); + assert_eq!(hdr.frame_index, 0); + // The probe space is a third, independent counter. + assert_eq!(pk.alloc_probe_index(), 0); + assert_eq!(pk.alloc_probe_index(), 1); + } + + /// Probe filler (FLAG_PROBE in user_flags) reassembles in its OWN window: a probe frame whose + /// index is far behind the video stream's completes anyway (an old client's single window + /// would drop it as stale), and video frames complete undisturbed around it. + #[test] + fn probe_frames_reassemble_in_their_own_window() { + let mut r = Reassembler::new(limits()); + let coder = coder_for(FecScheme::Gf8); + let stats = StatsCounters::default(); + + // Establish a video stream far into its index space. + let mut v = base_header(); + v.frame_index = 100_000; + v.pts_ns = 1_000_000_000; + assert!(r + .push(&packet(v), coder.as_ref(), &stats) + .unwrap() + .is_some()); + + // A probe frame at index 0 — 100k "behind" the video window — must still complete. + let mut p = base_header(); + p.frame_index = 0; + p.pts_ns = 1_000_000_100; + p.user_flags = FLAG_PROBE as u32; + let got = r.push(&packet(p), coder.as_ref(), &stats).unwrap(); + assert!(got.is_some(), "probe frame must complete in its own window"); + assert_eq!(got.unwrap().flags & FLAG_PROBE as u32, FLAG_PROBE as u32); + + // The probe burst must not have advanced the VIDEO window: the next video frame is + // contiguous and completes, with nothing counted dropped. + let mut v2 = base_header(); + v2.frame_index = 100_001; + v2.pts_ns = 1_000_000_200; + assert!(r + .push(&packet(v2), coder.as_ref(), &stats) + .unwrap() + .is_some()); + assert_eq!(stats.snapshot().frames_dropped, 0); + } + + /// An incomplete probe frame aging out of the probe window is NOT a video `frames_dropped` + /// (which would fire the client's loss recovery) — probe loss is measured bytes-wise by the + /// probe accumulator. + #[test] + fn aged_out_probe_frames_do_not_count_as_dropped() { + let mut r = Reassembler::new(limits()); + let coder = coder_for(FecScheme::Gf8); + let stats = StatsCounters::default(); + + // Probe frame 0: one of two shards — incomplete. + let mut p = base_header(); + p.user_flags = FLAG_PROBE as u32; + p.data_shards = 2; + p.frame_bytes = 32; + assert!(r + .push(&packet(p), coder.as_ref(), &stats) + .unwrap() + .is_none()); + + // A much newer probe frame ages it out of the probe window. + let mut p2 = base_header(); + p2.user_flags = FLAG_PROBE as u32; + p2.frame_index = 1; + p2.pts_ns = LOSS_WINDOW_NS + 1; + assert!(r + .push(&packet(p2), coder.as_ref(), &stats) + .unwrap() + .is_some()); + assert_eq!( + stats.snapshot().frames_dropped, + 0, + "probe-window drops must not fire video loss recovery" + ); + } + #[test] fn rejects_wrong_shard_bytes_and_oversized_frame() { let coder = coder_for(FecScheme::Gf8); diff --git a/crates/punktfunk-core/src/quic/msgs.rs b/crates/punktfunk-core/src/quic/msgs.rs index 36c69d16..32e0ee1c 100644 --- a/crates/punktfunk-core/src/quic/msgs.rs +++ b/crates/punktfunk-core/src/quic/msgs.rs @@ -107,6 +107,18 @@ pub const VIDEO_CAP_444: u8 = 0x04; /// host ignores it and simply never sends any); a client that doesn't set it keeps the combined /// stage. Purely observability — never changes what the host encodes. pub const VIDEO_CAP_HOST_TIMING: u8 = 0x08; +/// [`Hello::video_caps`] bit: the client's reassembler keeps **speed-test probe filler in its own +/// frame-index space** (a second reassembly window keyed on the [`crate::packet::FLAG_PROBE`] +/// user-flag), so probe bursts no longer consume video `frame_index`es. Without this, a mid-session +/// speed test burns thousands of video indexes that are invisible to every client-side gap detector +/// (probe frames are filtered before the pump sees them) — the first real AU afterwards reads as a +/// phantom multi-thousand-frame loss (spurious freeze + a nonsense RFI). It also lets the host's +/// encode loop own the video numbering outright (the wire-index contract +/// [`crate::packet::Packetizer::packetize_each`] documents), which reference-frame invalidation +/// depends on. The host runs mid-session probe bursts ONLY against clients that set this bit — an +/// older client gets a declined (zeroed) [`ProbeResult`] instead of a measurement its single-window +/// reassembler would silently drop as stale. +pub const VIDEO_CAP_PROBE_SEQ: u8 = 0x10; /// QUIC application error code a punktfunk/1 client closes the control connection with on a /// **deliberate quit** (a user "stop", not a network drop). The host reads it off the connection's diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 89065dec..73a3e5f2 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -161,6 +161,31 @@ impl Session { data: &[u8], pts_ns: u64, user_flags: u32, + ) -> Result>> { + self.seal_frame_inner(data, pts_ns, user_flags, None) + } + + /// [`seal_frame`](Self::seal_frame) with the caller's **explicit** `frame_index` instead of the + /// packetizer's internal counter. The punktfunk/1 encode loop owns the video numbering (one + /// session-lifetime counter, stamped per AU) so the encoder's reference-frame-invalidation + /// bookkeeping stays 1:1 with the wire across encoder rebuilds/resets — see + /// [`Packetizer::packetize_each`]. A session must use ONE numbering style per index space. + pub fn seal_frame_at( + &mut self, + data: &[u8], + pts_ns: u64, + user_flags: u32, + frame_index: u32, + ) -> Result>> { + self.seal_frame_inner(data, pts_ns, user_flags, Some(frame_index)) + } + + fn seal_frame_inner( + &mut self, + data: &[u8], + pts_ns: u64, + user_flags: u32, + frame_index: Option, ) -> Result>> { if self.config.role != Role::Host { return Err(PunktfunkError::InvalidArg( @@ -184,35 +209,36 @@ impl Session { } = self; let mut wires = std::mem::take(wire_pool); let mut used = 0usize; - let result = packetizer.packetize_each(data, pts_ns, user_flags, coder.as_ref(), { - let wires = &mut wires; - let used = &mut used; - move |hdr, body| { - if *used == wires.len() { - wires.push(Vec::new()); - } - let wire = &mut wires[*used]; - *used += 1; - let seq = *next_seq; - *next_seq = next_seq.wrapping_add(1); - wire.clear(); - match crypto { - Some(c) => { - // seq(8) ‖ header(40) ‖ shard ‖ tag scratch(16), sealed over [8..]. - wire.extend_from_slice(&seq.to_be_bytes()); - wire.extend_from_slice(hdr.as_bytes()); - wire.extend_from_slice(body); - wire.resize(wire.len() + crate::crypto::TAG_LEN, 0); - c.seal_in_place(seq, &mut wire[8..])?; + let result = + packetizer.packetize_each(data, pts_ns, user_flags, frame_index, coder.as_ref(), { + let wires = &mut wires; + let used = &mut used; + move |hdr, body| { + if *used == wires.len() { + wires.push(Vec::new()); } - None => { - wire.extend_from_slice(hdr.as_bytes()); - wire.extend_from_slice(body); + let wire = &mut wires[*used]; + *used += 1; + let seq = *next_seq; + *next_seq = next_seq.wrapping_add(1); + wire.clear(); + match crypto { + Some(c) => { + // seq(8) ‖ header(40) ‖ shard ‖ tag scratch(16), sealed over [8..]. + wire.extend_from_slice(&seq.to_be_bytes()); + wire.extend_from_slice(hdr.as_bytes()); + wire.extend_from_slice(body); + wire.resize(wire.len() + crate::crypto::TAG_LEN, 0); + c.seal_in_place(seq, &mut wire[8..])?; + } + None => { + wire.extend_from_slice(hdr.as_bytes()); + wire.extend_from_slice(body); + } } + Ok(()) } - Ok(()) - } - }); + }); result?; // A smaller frame uses fewer buffers than the pool holds: drop the unused tail, same // as the previous `resize_with(packets.len(), ..)` did. @@ -258,6 +284,23 @@ impl Session { r.map(|_| ()) } + /// Host: seal + send one **speed-test probe filler** access unit in the probe index space + /// (its own frame counter + the [`crate::packet::FLAG_PROBE`] user-flag) so a burst never + /// consumes video `frame_index`es — the client reassembles probe frames in a separate window + /// and its gap detectors never see them. Only call this against a client that advertised + /// [`crate::quic::VIDEO_CAP_PROBE_SEQ`]; an older client's single-window reassembler would + /// drop probe-space indexes as stale against the video stream. + pub fn submit_probe_frame(&mut self, data: &[u8], pts_ns: u64) -> Result<()> { + let idx = self.packetizer.alloc_probe_index(); + let wires = + self.seal_frame_inner(data, pts_ns, crate::packet::FLAG_PROBE as u32, Some(idx))?; + let refs: Vec<&[u8]> = wires.iter().map(|w| w.as_slice()).collect(); + let r = self.send_sealed(&refs); + drop(refs); + self.reclaim_wires(wires); + r.map(|_| ()) + } + /// Host: live-adjust the FEC recovery percentage (adaptive FEC). Affects the next /// [`submit_frame`](Self::submit_frame)/[`seal_frame`](Self::seal_frame); the receiver needs no /// notification (each packet's header carries its block's data/recovery shard counts). diff --git a/crates/punktfunk-host/src/encode.rs b/crates/punktfunk-host/src/encode.rs index fc288ad2..f1aef906 100644 --- a/crates/punktfunk-host/src/encode.rs +++ b/crates/punktfunk-host/src/encode.rs @@ -21,10 +21,13 @@ pub struct EncodedFrame { pub keyframe: bool, /// True when this AU is a **reference-frame-invalidation recovery frame** — a clean P-frame the /// encoder coded against a known-good reference in response to - /// [`invalidate_ref_frames`](Encoder::invalidate_ref_frames) (AMD LTR force-reference). The pump - /// tags it [`punktfunk_core::packet::USER_FLAG_RECOVERY_ANCHOR`] so the client lifts its post-loss - /// freeze on it without an IDR. Only the native-AMF LTR path sets it; every other backend leaves - /// it `false` (their RFI, when present, re-references transparently with no distinct clean-point AU). + /// [`invalidate_ref_frames`](Encoder::invalidate_ref_frames). The pump tags it + /// [`punktfunk_core::packet::USER_FLAG_RECOVERY_ANCHOR`] so the client lifts its post-loss + /// freeze on it without an IDR. Set by BOTH RFI backends: native AMF (the LTR force-reference + /// frame) and Windows direct-NVENC (the first frame encoded after `nvEncInvalidateRefFrames` — + /// the invalidation applies at the next `encode_picture`, so that AU is by construction the + /// clean re-anchor). Without it the client's freeze can only lift on an IDR — which the host + /// suppresses after a successful RFI (the cooldown), a ~1 s frozen stall per loss event. pub recovery_anchor: bool, } @@ -201,8 +204,9 @@ pub struct EncoderCaps { /// The encoder can perform real reference-frame invalidation — i.e. /// [`invalidate_ref_frames`](Encoder::invalidate_ref_frames) can return `true`. When `false` /// the caller skips that always-`false` call and forces a keyframe directly on loss recovery. - /// Only the Windows direct-NVENC path implements RFI; libavcodec (Linux NVENC), VAAPI and - /// AMF/QSV always keyframe. + /// Two backends implement RFI: Windows direct-NVENC (`nvEncInvalidateRefFrames`) and native + /// AMF (user-LTR force-reference, when the driver accepted the LTR slots at open). The + /// libavcodec paths (Linux NVENC, VAAPI, QSV) can't express it and always keyframe. pub supports_rfi: bool, /// The encoder emits in-band HDR mastering/CLL SEI from [`set_hdr_meta`](Encoder::set_hdr_meta). /// When `false`, `set_hdr_meta` is a no-op and no in-band grade reaches the client. Only the @@ -242,6 +246,22 @@ pub struct EncoderCaps { /// A hardware encoder. One per session; runs on the encode thread. pub trait Encoder: Send { fn submit(&mut self, frame: &CapturedFrame) -> Result<()>; + /// [`submit`](Self::submit) with the **wire frame index** this frame's AU will carry — the + /// number the packetizer stamps on it and the client's loss reports/RFI requests name. The + /// session glue predicts it exactly as `AUs sent so far + frames in flight` (AUs are emitted + /// FIFO, one per submission; anything that would break the prediction — an in-place reset, a + /// device-change teardown, an encoder rebuild — forfeits the in-flight frames on BOTH sides + /// and clears the encoder's reference state, so stale predictions die with it). The RFI + /// backends pin their frame numbering (LTR marks, DPB timestamps) to this so + /// [`invalidate_ref_frames`](Self::invalidate_ref_frames) compares client frame numbers + /// against the same domain — an encoder-internal counter desyncs from the wire on the first + /// mid-stream rebuild (adaptive bitrate steps do this under congestion, exactly when losses + /// happen). Default: ignore the index and delegate to `submit` (backends without per-frame + /// reference bookkeeping don't care). + fn submit_indexed(&mut self, frame: &CapturedFrame, wire_index: u32) -> Result<()> { + let _ = wire_index; + self.submit(frame) + } /// This encoder's static [capabilities](EncoderCaps) (RFI, HDR SEI), so the session glue can /// route by query rather than rely on the no-op/`false` defaults of /// [`invalidate_ref_frames`](Self::invalidate_ref_frames) / [`set_hdr_meta`](Self::set_hdr_meta). @@ -259,13 +279,14 @@ pub trait Encoder: Send { /// Default: no-op (SDR encoders / libavcodec paths that don't attach it yet). Cheap to call /// every frame; only the direct-NVENC path consumes it. fn set_hdr_meta(&mut self, _meta: Option) {} - /// Invalidate a contiguous range of previously-encoded reference frames (client frame numbers, - /// as reported in a loss-recovery request) so the encoder re-references an older still-valid - /// frame instead of emitting a full IDR. Returns `true` if a real reference invalidation was - /// performed; `false` means the encoder couldn't (range older than the DPB, or the backend has - /// no RFI) and the caller should fall back to [`request_keyframe`](Self::request_keyframe). - /// Default: `false` — only the Windows direct-NVENC path implements true RFI; libavcodec - /// (Linux NVENC) and VAAPI can't express `nvEncInvalidateRefFrames`, so they keyframe. + /// Invalidate a contiguous range of previously-encoded reference frames (client frame numbers + /// — WIRE frame indexes, the domain [`submit_indexed`](Self::submit_indexed) pins the encoder's + /// bookkeeping to) so the encoder re-references an older still-valid frame instead of emitting + /// a full IDR. Returns `true` if a real reference invalidation was performed; `false` means the + /// encoder couldn't (range older than the DPB/LTR history, or the backend has no RFI) and the + /// caller should fall back to [`request_keyframe`](Self::request_keyframe). Default: `false` — + /// the Windows direct-NVENC path (`nvEncInvalidateRefFrames`) and native AMF (LTR + /// force-reference) implement true RFI; the libavcodec paths can't express it, so they keyframe. fn invalidate_ref_frames(&mut self, _first_frame: i64, _last_frame: i64) -> bool { false } @@ -441,6 +462,9 @@ impl Encoder for TrackedEncoder { fn submit(&mut self, frame: &CapturedFrame) -> Result<()> { self.inner.submit(frame) } + fn submit_indexed(&mut self, frame: &CapturedFrame, wire_index: u32) -> Result<()> { + self.inner.submit_indexed(frame, wire_index) + } fn caps(&self) -> EncoderCaps { self.inner.caps() } diff --git a/crates/punktfunk-host/src/encode/linux/mod.rs b/crates/punktfunk-host/src/encode/linux/mod.rs index 4cf950e0..15991caf 100644 --- a/crates/punktfunk-host/src/encode/linux/mod.rs +++ b/crates/punktfunk-host/src/encode/linux/mod.rs @@ -149,6 +149,22 @@ fn nvenc_input(format: PixelFormat) -> (Pixel, bool) { } } +/// The [`NvencEncoder::open`] arguments, kept on the encoder so [`Encoder::reset`] can rebuild it +/// in place with the session's negotiated parameters — the encode-stall watchdog's recovery lever +/// (drop the wedged libavcodec encoder, reopen fresh, forfeit the owed AUs, restart at an IDR). +#[derive(Clone, Copy)] +struct OpenArgs { + codec: Codec, + format: PixelFormat, + width: u32, + height: u32, + fps: u32, + bitrate_bps: u64, + cuda: bool, + bit_depth: u8, + chroma: ChromaFormat, +} + pub struct NvencEncoder { enc: encoder::video::Encoder, /// Reusable 4-bpp CPU input frame (CPU path only; `None` for the zero-copy/CUDA path). @@ -181,6 +197,8 @@ pub struct NvencEncoder { /// open so the pump's per-AU `caps()` doesn't re-read `PUNKTFUNK_IR_PERIOD_FRAMES`; the pump marks /// every Nth AU with `USER_FLAG_RECOVERY_POINT` for the client's clean re-anchor. intra_refresh_period: u32, + /// The open arguments, for the in-place [`reset`](Encoder::reset) rebuild. + args: OpenArgs, } // `CudaHw` holds raw `AVBufferRef`s and `sws_444` a raw `SwsContext`; the encoder lives on a single @@ -534,6 +552,17 @@ impl NvencEncoder { } else { 0 }, + args: OpenArgs { + codec, + format, + width, + height, + fps, + bitrate_bps, + cuda, + bit_depth, + chroma, + }, }) } } @@ -582,6 +611,35 @@ impl Encoder for NvencEncoder { self.force_kf = true; } + /// Encode-stall recovery: drop the wedged libavcodec encoder and reopen it fresh with the + /// session's negotiated parameters (the stored [`OpenArgs`]) — the drop-and-reopen lever the + /// QSV/VAAPI paths use, so the encode-stall watchdog can heal a wedged NVENC/driver instead of + /// ending the session. Owed AUs are forfeited; the fresh encoder opens on an IDR. + fn reset(&mut self) -> bool { + let a = self.args; + match Self::open( + a.codec, + a.format, + a.width, + a.height, + a.fps, + a.bitrate_bps, + a.cuda, + a.bit_depth, + a.chroma, + ) { + Ok(mut fresh) => { + fresh.force_kf = true; + *self = fresh; // drops the wedged encoder (frees its contexts) in the same step + true + } + Err(e) => { + tracing::error!(error = %format!("{e:#}"), "NVENC in-place reopen failed"); + false + } + } + } + fn poll(&mut self) -> Result> { let mut pkt = Packet::empty(); match self.enc.receive_packet(&mut pkt) { diff --git a/crates/punktfunk-host/src/encode/linux/vaapi.rs b/crates/punktfunk-host/src/encode/linux/vaapi.rs index 62972455..a696ca0a 100644 --- a/crates/punktfunk-host/src/encode/linux/vaapi.rs +++ b/crates/punktfunk-host/src/encode/linux/vaapi.rs @@ -1126,6 +1126,19 @@ impl Encoder for VaapiEncoder { self.force_kf = true; } + /// Encode-stall recovery: drop the wedged libavcodec encoder (its `Drop` releases the VA + /// surfaces/filter graph/devices) and let the next `submit` rebuild it lazily from the first + /// frame's payload, exactly like first-frame bring-up — the same drop-and-reopen lever the + /// Windows QSV path has. The owed AUs are forfeited (`in_flight` zeroed) and the rebuilt + /// encoder's first frame is forced IDR so the client resyncs immediately. Without this the + /// encode-stall watchdog had no lever on Linux AMD/Intel and a wedged driver ended the session. + fn reset(&mut self) -> bool { + self.inner = None; + self.in_flight = 0; + self.force_kf = true; + true + } + fn poll(&mut self) -> Result> { // With `async_depth > 1`, `submit` no longer waits for the ASIC — the AU for the frame we // just sent lands ~one hardware-encode-time later. Wait for it (bounded) so it still ships diff --git a/crates/punktfunk-host/src/encode/sw.rs b/crates/punktfunk-host/src/encode/sw.rs index d9017e32..691738b3 100644 --- a/crates/punktfunk-host/src/encode/sw.rs +++ b/crates/punktfunk-host/src/encode/sw.rs @@ -20,6 +20,7 @@ use openh264::encoder::{ }; use openh264::formats::YUVSlices; use openh264::OpenH264API; +use std::collections::VecDeque; pub struct OpenH264Encoder { enc: Oh264, @@ -34,8 +35,11 @@ pub struct OpenH264Encoder { v_plane: Vec, frame_idx: i64, force_kf: bool, - /// At most one AU per submit (no lookahead), handed back by the next `poll`. - pending: Option, + /// One AU per submit (no lookahead), handed back FIFO by `poll`. A queue, not an `Option`: + /// the session loop pipelines up to `capturer.pipeline_depth()` submits before polling, and a + /// single-slot pending would silently overwrite (lose) the older AUs — including the opening + /// IDR — and permanently skew the loop's FIFO pts pairing. + pending: VecDeque, } // openh264's Encoder holds a raw C handle (not auto-Send); it lives on the single encode thread. @@ -88,7 +92,7 @@ impl OpenH264Encoder { v_plane: vec![0; (w / 2) * (h / 2)], frame_idx: 0, force_kf: false, - pending: None, + pending: VecDeque::new(), }) } @@ -207,7 +211,7 @@ impl Encoder for OpenH264Encoder { if !data.is_empty() { let keyframe = matches!(bs.frame_type(), FrameType::IDR | FrameType::I); let pts_ns = self.frame_idx as u64 * 1_000_000_000 / self.fps.max(1) as u64; - self.pending = Some(EncodedFrame { + self.pending.push_back(EncodedFrame { data, pts_ns, keyframe, @@ -223,7 +227,7 @@ impl Encoder for OpenH264Encoder { } fn poll(&mut self) -> Result> { - Ok(self.pending.take()) + Ok(self.pending.pop_front()) } fn flush(&mut self) -> Result<()> { diff --git a/crates/punktfunk-host/src/encode/windows/amf.rs b/crates/punktfunk-host/src/encode/windows/amf.rs index cbdf177c..417b6241 100644 --- a/crates/punktfunk-host/src/encode/windows/amf.rs +++ b/crates/punktfunk-host/src/encode/windows/amf.rs @@ -1840,7 +1840,12 @@ impl Encoder for AmfEncoder { ); self.ensure_inner(&frame.device)?; let cur_idx = self.frame_idx; - let forced = std::mem::take(&mut self.force_kf) || self.frame_idx == 0; + // A component's FIRST submission must be a forced IDR (stream-start contract: in-band + // headers + LTR re-anchor). Detected via the fresh ring counter, NOT `frame_idx == 0`: + // `submit_indexed` pins frame_idx to the wire index, which is non-zero when a mid-session + // rebuild (bitrate step / reset escalation) brings a new component up. + let opening = self.inner.as_ref().is_none_or(|i| i.next == 0); + let forced = std::mem::take(&mut self.force_kf) || opening; let pts_100ns = self.frame_idx * 10_000_000 / self.fps.max(1) as i64; self.frame_idx += 1; // --- LTR-RFI per-frame decisions (design: the AMD twin of NVENC intra-refresh recovery) --- @@ -2118,6 +2123,21 @@ impl Encoder for AmfEncoder { Ok(()) } + /// Pin this submission's frame number to the wire frame index its AU will carry (see the + /// trait doc): the LTR slots then store WIRE indexes, so [`invalidate_ref_frames`]'s + /// pre-loss check (`slot < first`, both in client frame numbers) stays correct across every + /// encoder rebuild/reset — an internal counter desyncs on the first adaptive-bitrate rebuild, + /// making the check vacuously true and risking a force-reference to an LTR marked INSIDE the + /// lost range (a corrupted frame shipped as a clean recovery anchor). `frame_idx` also feeds + /// the AMF SetPts; a re-pin only ever moves it backward across a reset (fresh component, so a + /// pts restart is harmless) and forward on a rebuild (monotonic within any one component). + /// + /// [`invalidate_ref_frames`]: Encoder::invalidate_ref_frames + fn submit_indexed(&mut self, frame: &CapturedFrame, wire_index: u32) -> Result<()> { + self.frame_idx = wire_index as i64; + self.submit(frame) + } + fn request_keyframe(&mut self) { self.force_kf = true; } @@ -2145,8 +2165,10 @@ impl Encoder for AmfEncoder { } // Pick the newest LTR strictly OLDER than the loss: the most recent known-good reference the // client still holds, so re-referencing it costs the least (smallest recovery-frame residual). - // Frame numbers are 1:1 with the client's (both count submissions in order — see the NVENC - // path), so `ltr_slots` (which store `frame_idx`) compare directly against `first`. + // `ltr_slots` store the WIRE frame index of the marked frame (`submit_indexed` pins + // `frame_idx` to it per submission), so they compare directly against the client's `first` + // — and stay comparable across encoder rebuilds/resets, where an internal counter would + // make this check vacuous and risk force-referencing an LTR marked INSIDE the lost range. let mut best: Option<(usize, i64)> = None; for (slot, marked) in self.ltr_slots.iter().enumerate() { if let Some(idx) = *marked { diff --git a/crates/punktfunk-host/src/encode/windows/nvenc.rs b/crates/punktfunk-host/src/encode/windows/nvenc.rs index 73ba818c..923d7509 100644 --- a/crates/punktfunk-host/src/encode/windows/nvenc.rs +++ b/crates/punktfunk-host/src/encode/windows/nvenc.rs @@ -429,10 +429,25 @@ pub struct NvencD3d11Encoder { async_rt: Option, /// `NV_ENC_CAPS_ASYNC_ENCODE_SUPPORT` from the caps probe — gates the async retrieve mode. async_supported: bool, - /// (bitstream, mapped input resource to unmap after retrieval, pts_ns) per in-flight encode. - pending: VecDeque<(nv::NV_ENC_OUTPUT_PTR, nv::NV_ENC_INPUT_PTR, u64)>, + /// (bitstream, mapped input resource to unmap after retrieval, pts_ns, recovery-anchor) per + /// in-flight encode. The fourth field tags the first frame encoded after a successful + /// [`invalidate_ref_frames`](Encoder::invalidate_ref_frames) — the clean re-anchor P-frame the + /// client lifts its post-loss freeze on (see [`EncodedFrame::recovery_anchor`]). + pending: VecDeque<(nv::NV_ENC_OUTPUT_PTR, nv::NV_ENC_INPUT_PTR, u64, bool)>, + /// The frame number of the NEXT submission (also its `inputTimeStamp`). Pinned per frame by + /// [`Encoder::submit_indexed`] to the WIRE frame index the AU will carry, so the DPB timestamps + /// `invalidate_ref_frames` compares client frame numbers against stay 1:1 with the wire across + /// encoder rebuilds/resets (an internal counter desyncs on the first adaptive-bitrate rebuild — + /// RFI then never matches again). Self-increments as a fallback for un-indexed callers (tests). frame_idx: i64, force_kf: bool, + /// A successful [`invalidate_ref_frames`](Encoder::invalidate_ref_frames) arms this; the next + /// `submit` consumes it into `pending` so that AU ships as the recovery anchor. NVENC applies + /// the invalidation at the next `encode_picture`, so that frame is by construction the first + /// one coded against only-valid references — without tagging it the client's freeze can only + /// lift on an IDR, which the session glue suppresses after an RFI success (the cooldown): + /// a ~1 s frozen stall per loss event on NVIDIA hosts. + pending_anchor: bool, inited: bool, /// GPU capabilities probed once via `nvEncGetEncodeCaps` before configuring (Apollo's /// `get_encoder_cap`): gates 10-bit/custom-VBV/RFI on what this card actually supports instead @@ -507,6 +522,7 @@ impl NvencD3d11Encoder { pending: VecDeque::new(), frame_idx: 0, force_kf: false, + pending_anchor: false, inited: false, rfi_supported: false, custom_vbv: false, @@ -536,7 +552,7 @@ impl NvencD3d11Encoder { while rt.done_rx.try_recv().is_ok() {} } // Unmap any in-flight inputs, then unregister every cached texture and destroy the bitstreams. - for (_, map, _) in &self.pending { + for (_, map, _, _) in &self.pending { if !map.is_null() { let _ = (api().unmap_input_resource)(self.encoder, *map); } @@ -569,8 +585,10 @@ impl NvencD3d11Encoder { self.inited = false; self.next = 0; // The new session starts with an empty DPB (its first frame is an IDR), so any prior - // invalidation range is meaningless against it. + // invalidation range is meaningless against it — and the IDR is itself the re-anchor, + // so a pending anchor tag from a pre-teardown RFI is stale too. self.last_rfi_range = None; + self.pending_anchor = false; } /// Query one `NV_ENC_CAPS` value for this codec on an open session; 0 on any error (treat an @@ -1133,7 +1151,7 @@ impl NvencD3d11Encoder { /// error surfaces AFTER the unmap (the resource is retired either way) so the session glue's /// rebuild path starts from clean state. fn absorb_done(&mut self, done: RetrieveDone) -> Result<()> { - let Some((bs, map, pts_ns)) = self.pending.pop_front() else { + let Some((bs, map, pts_ns, anchor)) = self.pending.pop_front() else { bail!("NVENC async: completion with no in-flight frame (pairing bug)"); }; if bs as usize != done.bs { @@ -1157,7 +1175,7 @@ impl NvencD3d11Encoder { data, pts_ns, keyframe, - recovery_anchor: false, + recovery_anchor: anchor, }); Ok(()) } @@ -1249,6 +1267,11 @@ impl Encoder for NvencD3d11Encoder { self.init_session(&device)?; self.init_device = dev_raw; } + // The session's opening frame — NVENC emits it as an IDR regardless of pic flags, so the + // in-band HDR SEI must ride it too. Detected via the still-empty output slot counter + // (`teardown` zeroes it), NOT via `pts == 0`: `submit_indexed` pins pts to the wire frame + // index, which is non-zero on a mid-session encoder rebuild's first frame. + let opening = self.next == 0; // Async backpressure: never hand NVENC an output bitstream that is still in flight, and // keep in-flight depth within the capturer's texture ring (see `async_inflight_cap`). At // the cap, block on the OLDEST completion (the retrieve thread is already waiting on its @@ -1323,6 +1346,10 @@ impl Encoder for NvencD3d11Encoder { } else { 0 }; + // Recovery anchor (armed by a successful invalidate_ref_frames): THIS frame is the + // first one encoded after the invalidation — the clean re-anchor. A simultaneous + // forced IDR is itself the re-anchor, so the tag is dropped in that case. + let anchor = std::mem::take(&mut self.pending_anchor) && flags == 0; let mut pic = nv::NV_ENC_PIC_PARAMS { version: nv::NV_ENC_PIC_PARAMS_VER, inputWidth: self.width, @@ -1349,7 +1376,7 @@ impl Encoder for NvencD3d11Encoder { // built from the source display's metadata. Any decoder — incl. stock Moonlight — then // tone-maps from the real grade. HEVC/H.264 carry SEI; AV1 uses metadata OBUs (follow-up). // The scratch buffers must outlive `encode_picture`, so they live in this scope. - let is_idr = flags != 0 || pts == 0; + let is_idr = flags != 0 || opening; let mastering_sei = self .hdr_meta .map(|m| crate::hdr::hevc_mastering_display_sei(&m)); @@ -1391,8 +1418,12 @@ impl Encoder for NvencD3d11Encoder { (api().encode_picture)(self.encoder, &mut pic) .nv_ok() .map_err(|e| anyhow!("encode_picture: {e:?}"))?; - self.pending - .push_back((self.bitstreams[slot], mp.mappedResource, captured.pts_ns)); + self.pending.push_back(( + self.bitstreams[slot], + mp.mappedResource, + captured.pts_ns, + anchor, + )); // Async: hand the in-flight encode to the retrieve thread (channel capacity = POOL ≥ // in-flight, so this send never blocks). The pending entry above pairs with its // completion FIFO in `absorb_done`. @@ -1409,6 +1440,16 @@ impl Encoder for NvencD3d11Encoder { Ok(()) } + /// Pin this submission's frame number (= its `inputTimeStamp`) to the wire frame index the AU + /// will carry, so the DPB timestamps `invalidate_ref_frames` matches client frame numbers + /// against are the wire's — 1:1 across every rebuild/reset (see the trait doc). Within a + /// session the loop's prediction is nondecreasing; a repeat after a reset lands on a fresh + /// session (teardown cleared the DPB and `last_rfi_range`), so re-pinning is always sound. + fn submit_indexed(&mut self, frame: &CapturedFrame, wire_index: u32) -> Result<()> { + self.frame_idx = wire_index as i64; + self.submit(frame) + } + fn request_keyframe(&mut self) { self.force_kf = true; } @@ -1442,9 +1483,13 @@ impl Encoder for NvencD3d11Encoder { if self.encoder.is_null() || !self.rfi_supported || first < 0 || first > last { return false; } - // Already invalidated a covering range for this loss event — nothing more to do, no IDR. + // Already invalidated a covering range for this loss event — no new driver calls needed, + // no IDR. RE-ARM the anchor though: the client re-asking means the previous recovery + // anchor AU may itself have been lost, and the next frame is just as clean a re-anchor + // (it too references only valid frames). if let Some((pf, pl)) = self.last_rfi_range { if first >= pf && last <= pl { + self.pending_anchor = true; return true; } } @@ -1460,9 +1505,11 @@ impl Encoder for NvencD3d11Encoder { if first > last { return false; } - // We tag each input with `inputTimeStamp = frame_idx` (0,1,2,…), which is also the client's - // frame number (the packetizer numbers frames in submit order), so the client's lost-frame - // range maps 1:1 onto the timestamps NVENC invalidates here. + // Each input's `inputTimeStamp` is `frame_idx`, which `submit_indexed` pins to the WIRE + // frame index the AU carries — so the client's lost-frame range maps 1:1 onto the + // timestamps NVENC invalidates here, and stays 1:1 across encoder rebuilds/resets (an + // internal counter would desync on the first adaptive-bitrate rebuild and RFI would then + // clamp every range into first > last, silently degrading to IDR-only forever). // SAFETY: `invalidate_ref_frames` is a function pointer from the runtime-loaded `EncodeApi` table. // `self.encoder` was checked non-null at the top of this fn and is the live session; this runs // on the encode thread (like submit/poll), so there is no concurrent NVENC use. Each `ts` was @@ -1480,6 +1527,11 @@ impl Encoder for NvencD3d11Encoder { } } self.last_rfi_range = Some((first, last)); + // The next submitted frame is the first one encoded after the invalidation — the clean + // re-anchor P-frame. Arm the tag so its AU ships with `recovery_anchor` and the client + // lifts its post-loss freeze on it (instead of waiting ~1 s for the cooldown-suppressed + // IDR fallback). + self.pending_anchor = true; true } @@ -1504,7 +1556,7 @@ impl Encoder for NvencD3d11Encoder { .ready .pop_front()); } - let Some((bs, map, pts_ns)) = self.pending.pop_front() else { + let Some((bs, map, pts_ns, anchor)) = self.pending.pop_front() else { return Ok(None); }; // SAFETY: a non-empty `pending` implies `submit` ran, so `self.encoder` is the live session @@ -1545,11 +1597,28 @@ impl Encoder for NvencD3d11Encoder { data, pts_ns, keyframe, - recovery_anchor: false, + recovery_anchor: anchor, })) } } + /// Encode-stall recovery: tear the whole session down (the same teardown a capture-device + /// change uses) and let the next `submit` rebuild it lazily on the current device — the owed + /// AUs are forfeited and the fresh session opens on an IDR. Gives the encode-stall watchdog a + /// healing lever on NVENC instead of ending the session. Caveat: the SYNC retrieve mode blocks + /// inside `lock_bitstream`, so a driver wedge that hangs the lock never returns to the loop + /// for the watchdog to fire — this lever fully protects the async retrieve mode (5 s event + /// timeouts surface as poll errors) and the submit-side failure paths. + fn reset(&mut self) -> bool { + // SAFETY: `teardown` (an `unsafe fn`) requires the encode thread with no NVENC call in + // flight and a session whose cached resources belong to `self.encoder` — all hold here + // (reset is called from the session loop between submit/poll, like every other method), + // and it early-returns on an already-null session. + unsafe { self.teardown() }; + self.force_kf = true; + true + } + fn flush(&mut self) -> Result<()> { Ok(()) // P1/ULL + frameIntervalP=1: each submit yields its AU; no internal queue to drain. } diff --git a/crates/punktfunk-host/src/gamestream/stream.rs b/crates/punktfunk-host/src/gamestream/stream.rs index 1e5f8905..60b9c059 100644 --- a/crates/punktfunk-host/src/gamestream/stream.rs +++ b/crates/punktfunk-host/src/gamestream/stream.rs @@ -436,7 +436,10 @@ fn sendmmsg_all(sock: &UdpSocket, pkts: &[Vec]) -> std::io::Result<()> { /// behind encode (measured ~3 ms/frame at 4K, which capped GameStream's frame rate well below what /// the encoder alone can sustain). struct RawFrame { - aus: Vec<(Vec, FrameType)>, + /// `(bitstream, type, wire frameIndex)` per AU. The stream loop assigns the index (it owns + /// the numbering — see its `au_seq`), so the encoder's RFI bookkeeping stays 1:1 with what + /// Moonlight sees across mid-stream encoder rebuilds. + aus: Vec<(Vec, FrameType, u32)>, ts: u32, } @@ -460,8 +463,8 @@ fn spawn_packetizer( crate::punktfunk1::boost_thread_priority(false); while let Ok(frame) = rx.recv() { let mut batch: PacketBatch = Vec::new(); - for (au, ft) in frame.aus { - batch.extend(pk.packetize(&au, ft, frame.ts)); + for (au, ft, idx) in frame.aus { + batch.extend(pk.packetize(&au, ft, frame.ts, Some(idx))); } if batch.is_empty() { continue; @@ -660,6 +663,16 @@ fn stream_body( // routed through the same coalesce gate as client IDR requests so a burst of drops (congestion) // can't become an IDR storm. let mut recover_after_drop = false; + // The stream's wire frameIndex numbering, owned HERE (the index of the next AU handed to the + // packetizer thread; a dropped-at-the-queue frame consumes none). A submission's future index + // is `au_seq + enc_inflight` (AUs are emitted FIFO, one per submission); passing it to + // `Encoder::submit_indexed` keeps the encoder's RFI bookkeeping 1:1 with Moonlight's frame + // numbers across the in-place encoder rebuild above (an internal counter would desync there). + // A pipeline-head drop desyncs the prediction by the dropped AU count for the frames already + // in flight — bounded and self-healing: the drop arms `recover_after_drop`, whose forced IDR + // resets the encoder's reference state (stale LTR/DPB bookkeeping dies with it). + let mut au_seq: u32 = 0; + let mut enc_inflight: u32 = 0; while running.load(Ordering::SeqCst) { let tick = Instant::now(); @@ -728,6 +741,10 @@ fn stream_body( enc.request_keyframe(); last_keyframe = Some(Instant::now()); next_frame = Instant::now(); + // The old encoder died with its in-flight submissions — their AUs will never + // arrive, so the numbering prediction restarts at `au_seq` (the fresh encoder's + // reference state is empty, so the reused predictions meet no stale bookkeeping). + enc_inflight = 0; tracing::info!("gamestream: source rebuilt — stream continues"); continue; } @@ -742,7 +759,13 @@ fn stream_body( if let Some((first, last)) = rfi_range.lock().unwrap().take() { // Prefer reference-frame invalidation when the encoder supports it (no costly IDR // spike); otherwise — or if the range is too old to invalidate — fall back to a keyframe. - if !(supports_rfi && enc.invalidate_ref_frames(first, last)) { + // Sanity-cap the range first: wider than RFI_MAX_RANGE exceeds any encoder's reference + // history (or is a phantom range from a desynced counter) — keyframe, never a + // force-reference that could ship corruption as a clean frame. + let width = (last as u32).wrapping_sub(first as u32); + if width > punktfunk_core::packet::RFI_MAX_RANGE + || !(supports_rfi && enc.invalidate_ref_frames(first, last)) + { want_keyframe = true; } } @@ -766,21 +789,27 @@ fn stream_body( tracing::debug!("video: keyframe request coalesced (IDR still in flight)"); } } - enc.submit(&frame).context("encoder submit")?; + enc.submit_indexed(&frame, au_seq.wrapping_add(enc_inflight)) + .context("encoder submit")?; + enc_inflight = enc_inflight.wrapping_add(1); let t_enc = tick.elapsed(); // 90 kHz RTP timestamp from wall-clock, so a variable capture rate stays correct. let ts = (stream_start.elapsed().as_secs_f64() * 90_000.0) as u32; // Drain the encoder's access units (owned buffers) — FEC/packetization runs on the - // packetizer thread, off this loop, so it never serializes behind encode. - let mut aus: Vec<(Vec, FrameType)> = Vec::new(); + // packetizer thread, off this loop, so it never serializes behind encode. Each AU is + // stamped with its wire frameIndex here (`au_seq + position`); the numbering only + // ADVANCES if the batch is actually enqueued below (a dropped batch consumes none). + let mut aus: Vec<(Vec, FrameType, u32)> = Vec::new(); while let Some(au) = enc.poll().context("encoder poll")? { let ft = if au.keyframe { FrameType::Idr } else { FrameType::P }; - aus.push((au.data, ft)); + let idx = au_seq.wrapping_add(aus.len() as u32); + aus.push((au.data, ft, idx)); + enc_inflight = enc_inflight.saturating_sub(1); } let t_pkt = tick.elapsed(); @@ -788,9 +817,11 @@ fn stream_body( // (packetizer, or the paced sender behind it) is behind — drop this frame (FEC/RFI covers the // client) and keep encoding, so a downstream stall can never cap the encode rate. if !aus.is_empty() { + let batch_len = aus.len() as u32; match raw_tx.try_send(RawFrame { aus, ts }) { Ok(()) => { sent_batches += 1; + au_seq = au_seq.wrapping_add(batch_len); } Err(std::sync::mpsc::TrySendError::Full(_)) => { dropped_batches += 1; diff --git a/crates/punktfunk-host/src/gamestream/video.rs b/crates/punktfunk-host/src/gamestream/video.rs index 561296e2..bb917cb3 100644 --- a/crates/punktfunk-host/src/gamestream/video.rs +++ b/crates/punktfunk-host/src/gamestream/video.rs @@ -69,14 +69,22 @@ impl VideoPacketizer { } /// Packetize one encoded AU into wire datagrams (data shards + Cauchy RS parity shards). + /// + /// `frame_index`: `Some(i)` stamps the caller's index (the stream loop owns the numbering so + /// the encoder's RFI bookkeeping stays 1:1 with the wire across mid-stream encoder rebuilds — + /// see `Encoder::submit_indexed`); `None` draws from the internal counter (tests/harnesses). pub fn packetize( &mut self, au: &[u8], frame_type: FrameType, timestamp_90k: u32, + frame_index: Option, ) -> Vec> { - let frame_index = self.frame_index; - self.frame_index = self.frame_index.wrapping_add(1); + let frame_index = frame_index.unwrap_or_else(|| { + let i = self.frame_index; + self.frame_index = i.wrapping_add(1); + i + }); let pps = self.payload_per_shard; let blocksize = SHARD_HEADER + pps; // = packet_size + 16 let pct = self.fec_percentage; @@ -235,7 +243,7 @@ mod tests { let mut pk = VideoPacketizer::new(1392, 0, 0); // data-only; pps = 1392+16-32 = 1376 assert_eq!(pk.payload_per_shard, 1376); let au = vec![0xABu8; 4000]; // 8+4000 = 4008 → ceil(4008/1376) = 3 data shards - let pkts = pk.packetize(&au, FrameType::Idr, 90_000); + let pkts = pk.packetize(&au, FrameType::Idr, 90_000, None); assert_eq!(pkts.len(), 3); for p in &pkts { assert_eq!(p.len(), SHARD_HEADER + 1376); @@ -266,7 +274,7 @@ mod tests { for ps in [0usize, 15, 16, 17, 32] { let mut pk = VideoPacketizer::new(ps, 20, 2); assert!(pk.payload_per_shard >= 1, "pps must never be 0 (ps={ps})"); - let _ = pk.packetize(&[0xCDu8; 200], FrameType::Idr, 0); // must not panic + let _ = pk.packetize(&[0xCDu8; 200], FrameType::Idr, 0, None); // must not panic } } @@ -274,7 +282,7 @@ mod tests { fn multi_block_split() { let mut pk = VideoPacketizer::new(1392, 0, 0); // data-only let au = vec![0u8; 600_000]; - let pkts = pk.packetize(&au, FrameType::P, 0); + let pkts = pk.packetize(&au, FrameType::P, 0, None); let total = (8 + au.len()).div_ceil(1376); assert_eq!(pkts.len(), total); let n_blocks = total.div_ceil(255).clamp(1, 4); @@ -286,7 +294,7 @@ mod tests { fn emits_parity_shards() { let mut pk = VideoPacketizer::new(1392, 20, 0); // pps = 1376, 20% FEC let au = vec![0xABu8; 4000]; // 8+4000 = 4008 → 3 data shards (k=3) - let pkts = pk.packetize(&au, FrameType::Idr, 0); + let pkts = pk.packetize(&au, FrameType::Idr, 0, None); // m = ceil(3*20/100) = 1 parity shard → 4 packets; wire_pct = 100*1/3 = 33. assert_eq!(pkts.len(), 4); for p in &pkts { @@ -313,7 +321,7 @@ mod tests { fn parity_recovers_full_datagram_incl_flags() { let mut pk = VideoPacketizer::new(1392, 50, 0); // high pct → plenty of parity let au = vec![0x5Au8; 4000]; // k = 3 - let pkts = pk.packetize(&au, FrameType::Idr, 0); + let pkts = pk.packetize(&au, FrameType::Idr, 0, None); let k = 3usize; let m = pkts.len() - k; assert!(m >= 1); diff --git a/crates/punktfunk-host/src/punktfunk1.rs b/crates/punktfunk-host/src/punktfunk1.rs index 955f428d..c7e42932 100644 --- a/crates/punktfunk-host/src/punktfunk1.rs +++ b/crates/punktfunk-host/src/punktfunk1.rs @@ -1540,6 +1540,10 @@ async fn serve_session( // and gets no extra datagrams. let timing_conn = (hello.video_caps & punktfunk_core::quic::VIDEO_CAP_HOST_TIMING != 0).then(|| conn.clone()); + // Probe-sequence capability: the client reassembles speed-test filler in its own index window, + // so mid-session bursts don't consume video frame indexes. An older client (bit clear) gets + // mid-session probes declined instead — see `run_probe_burst`. + let probe_seq = hello.video_caps & punktfunk_core::quic::VIDEO_CAP_PROBE_SEQ != 0; let stats_dp = stats; // data-plane handle to the shared stats recorder // Short label for web-console stats captures: the client's cert-fingerprint prefix, else its // peer IP (no fingerprint = anonymous TOFU/--open client). @@ -1595,6 +1599,7 @@ async fn serve_session( &probe_result_tx, &fec_target_dp, timing_conn.as_ref(), + probe_seq, ), Punktfunk1Source::Virtual => { let compositor = compositor @@ -1620,6 +1625,7 @@ async fn serve_session( fec_target: fec_target_dp, conn: conn_stream, timing_conn, + probe_seq, stats: stats_dp, client_label, launch: launch_for_dp, @@ -2437,6 +2443,7 @@ fn mark_recovery_boundary(ir_wave_pos: &mut u32, is_keyframe: bool, period: u32) } } +#[allow(clippy::too_many_arguments)] fn synthetic_stream( session: &mut Session, frames: u32, @@ -2445,6 +2452,7 @@ fn synthetic_stream( probe_result_tx: &tokio::sync::mpsc::UnboundedSender, fec_target: &AtomicU8, timing_conn: Option<&quinn::Connection>, + probe_seq: bool, ) -> Result<()> { let interval = std::time::Duration::from_millis(1000 / 60); for idx in 0..frames { @@ -2453,7 +2461,7 @@ fn synthetic_stream( } apply_fec_target(session, fec_target); // Service speed-test probes between synthetic frames (loopback bandwidth tests). - service_probes(session, stop, probe_rx, probe_result_tx); + service_probes(session, stop, probe_rx, probe_result_tx, probe_seq); let data = test_frame(idx, 64 * 1024); let pts_ns = now_ns(); session @@ -2795,9 +2803,34 @@ const MAX_PROBE_MS: u32 = 5_000; /// was actually offered so the client can compute delivery ratio (`received / bytes_sent`) and /// throughput. Video is paused for the duration (the caller's loop is blocked here) — a speed test /// is a deliberate, short interruption the client initiates. -fn run_probe_burst(session: &mut Session, req: ProbeRequest, stop: &AtomicBool) -> ProbeResult { +fn run_probe_burst( + session: &mut Session, + req: ProbeRequest, + stop: &AtomicBool, + probe_seq: bool, +) -> ProbeResult { let target_kbps = req.target_kbps.min(MAX_PROBE_KBPS); let duration_ms = req.duration_ms.min(MAX_PROBE_MS); + // Probe filler is sealed in the PROBE index space (its own frame counter — video indexes are + // owned by the encode loop and must stay 1:1 with the encoder's RFI bookkeeping). A client + // that didn't advertise VIDEO_CAP_PROBE_SEQ reassembles everything in one window and would + // drop probe-space frames as stale against the video stream — measuring garbage — so its + // mid-session probe is DECLINED (zeroed result) instead. Old sealing (probe filler consuming + // video indexes) is not an option anymore: those indexes are invisible to every client gap + // detector and read as a phantom multi-thousand-frame loss after the burst. + if !probe_seq { + tracing::info!( + "declining speed-test probe: client predates VIDEO_CAP_PROBE_SEQ (its reassembler \ + cannot window probe-space frames)" + ); + return ProbeResult { + bytes_sent: 0, + packets_sent: 0, + duration_ms: 0, + wire_packets_sent: 0, + send_dropped: 0, + }; + } if target_kbps == 0 || duration_ms == 0 { return ProbeResult { bytes_sent: 0, @@ -2831,8 +2864,9 @@ fn run_probe_burst(session: &mut Session, req: ProbeRequest, stop: &AtomicBool) let allowed = (start.elapsed().as_secs_f64() * bytes_per_sec as f64) as u64; if bytes_sent < allowed { // A full send buffer drops on WouldBlock/ENOBUFS (UdpTransport returns Ok) — that loss is - // part of what the probe measures (it surfaces as send_dropped), so keep going. - let _ = session.submit_frame(&filler, now_ns(), FLAG_PROBE as u32); + // part of what the probe measures (it surfaces as send_dropped), so keep going. Sealed + // in the probe index space (FLAG_PROBE + its own counter) — never a video frame_index. + let _ = session.submit_probe_frame(&filler, now_ns()); bytes_sent += chunk as u64; packets_sent += 1; } else { @@ -2863,15 +2897,17 @@ fn run_probe_burst(session: &mut Session, req: ProbeRequest, stop: &AtomicBool) } /// Drain any pending speed-test requests and run each burst, replying with its [`ProbeResult`]. -/// Called once per data-plane loop iteration so a probe runs between frames. +/// Called once per data-plane loop iteration so a probe runs between frames. `probe_seq` = the +/// client advertised [`punktfunk_core::quic::VIDEO_CAP_PROBE_SEQ`] (see [`run_probe_burst`]). fn service_probes( session: &mut Session, stop: &AtomicBool, probe_rx: &std::sync::mpsc::Receiver, probe_result_tx: &tokio::sync::mpsc::UnboundedSender, + probe_seq: bool, ) { while let Ok(req) = probe_rx.try_recv() { - let result = run_probe_burst(session, req, stop); + let result = run_probe_burst(session, req, stop, probe_seq); let _ = probe_result_tx.send(result); } } @@ -2886,16 +2922,18 @@ fn service_probes( /// buffer → EAGAIN drop → under infinite GOP, a freeze until the next keyframe). With no slack /// (encode ≈ interval) the budget collapses to 0 and even the overflow goes out immediately, so /// this is never slower than unpaced. +#[allow(clippy::too_many_arguments)] fn paced_submit( session: &mut Session, data: &[u8], pts_ns: u64, flags: u32, + frame_index: u32, deadline: std::time::Instant, burst_cap: usize, ) -> Result { let wires = session - .seal_frame(data, pts_ns, flags) + .seal_frame_at(data, pts_ns, flags, frame_index) .map_err(|e| anyhow!("seal_frame: {e:?}"))?; let mut refs: Vec<&[u8]> = wires.iter().map(|w| w.as_slice()).collect(); // FEC/recovery test knob (PUNKTFUNK_VIDEO_DROP) — same knob the GameStream plane honors. @@ -2925,6 +2963,12 @@ struct FrameMsg { data: Vec, capture_ns: u64, flags: u32, + /// The wire `frame_index` this AU is sealed with. Assigned by the encode loop's + /// session-lifetime counter (`au_seq`) — the loop owns the video numbering so the index it + /// PREDICTED at submit time (`au_seq + inflight`, handed to `Encoder::submit_indexed`) is + /// exactly what the packetizer stamps, keeping the encoder's RFI bookkeeping 1:1 with the + /// wire across encoder rebuilds/resets. Sealed via `Session::seal_frame_at`. + frame_index: u32, /// When this frame's packets should have fully left (the next frame's due time) = the pacing /// budget. In the past when the send thread is behind → immediate send (catch up). deadline: std::time::Instant, @@ -3117,6 +3161,9 @@ fn send_loop( // `Some` = the client advertised VIDEO_CAP_HOST_TIMING: emit one 0xCF datagram per AU right // after its last packet left the socket (capture→sent, the whole host pipeline incl. pacing). timing_conn: Option, + // The client advertised VIDEO_CAP_PROBE_SEQ — mid-session speed-test bursts may run in the + // probe index space (else they're declined; see `run_probe_burst`). + probe_seq: bool, ) { boost_thread_priority(false); // transmit thread: above-normal (Apollo's encoder-thread level) let mut last_perf = std::time::Instant::now(); @@ -3145,7 +3192,7 @@ fn send_loop( } // Probes run here (they need the Session); a burst pauses video — the encode thread blocks // on the full frame channel meanwhile, which is exactly the intended pause. - service_probes(&mut session, &stop, &probe_rx, &probe_result_tx); + service_probes(&mut session, &stop, &probe_rx, &probe_result_tx, probe_seq); // Adaptive FEC: pick up any new recovery target the control task set from client LossReports. apply_fec_target(&mut session, &fec_target); // Short timeout so we keep re-checking `stop` + probes when no frames are flowing. @@ -3155,6 +3202,7 @@ fn send_loop( &msg.data, msg.capture_ns, msg.flags, + msg.frame_index, msg.deadline, burst_cap, ) { @@ -3472,6 +3520,12 @@ struct SessionContext { /// thread emits one 0xCF datagram per AU (capture→sent µs) on it, so the client can split its /// `host+network` latency stage. `None` = older client, no emission. timing_conn: Option, + /// The client advertised [`punktfunk_core::quic::VIDEO_CAP_PROBE_SEQ`]: speed-test bursts may + /// run mid-session in the probe index space (its reassembler keeps a separate probe window). + /// `false` = older client whose single-window reassembler would drop probe-space frames as + /// stale — mid-session probes are DECLINED for it (a zeroed [`ProbeResult`]) rather than + /// consuming video frame indexes its gap detectors can't see (the phantom-gap freeze). + probe_seq: bool, /// Shared streaming-stats recorder. The capture loop reads `is_armed()` per frame to decide /// whether to measure the per-stage split; the send thread builds + pushes the aggregated /// `StatsSample` at its 2 s boundary. @@ -3527,6 +3581,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { fec_target, conn, timing_conn, + probe_seq, stats, client_label, launch, @@ -3664,6 +3719,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { fec_target, send_stats, timing_conn, + probe_seq, ) } }) @@ -3689,6 +3745,16 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(seconds as u64); let mut next = std::time::Instant::now(); let mut sent: u64 = 0; + // The session's video frame numbering, owned HERE (the wire `frame_index` of the next AU this + // loop hands to the send thread; the packetizer seals with exactly this via `seal_frame_at`). + // A submission's future index is predicted as `au_seq + inflight.len()` — exact because AUs + // are emitted FIFO, one per submission, and every event that forfeits in-flight frames + // (reset/rebuild/teardown) clears `inflight` AND the encoder's reference state, so the reused + // predictions can never meet stale bookkeeping. Passing it to `Encoder::submit_indexed` keeps + // the RFI backends' frame numbers 1:1 with the client's across encoder rebuilds — an + // encoder-internal counter desyncs on the first adaptive-bitrate rebuild (NVENC RFI then + // silently dies; AMF may anchor onto a post-loss LTR). + let mut au_seq: u32 = 0; // Rebuild-in-place on capture loss: track the live mode (a mode switch updates it) so a rebuild // targets the CURRENT mode, and cap consecutive rebuilds so a flapping source can't loop the // client through endless cold restarts. @@ -3967,7 +4033,19 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { } if !want_kf { if let Some((first, last)) = rfi_range { - if enc.caps().supports_rfi && enc.invalidate_ref_frames(first as i64, last as i64) { + // Sanity-cap the range before consulting the encoder: RFI can only re-reference + // history the encoder still holds (NVENC: a 5-frame DPB; AMD LTR: ~1 s of marks). + // A range wider than RFI_MAX_RANGE is either a seconds-long outage (no valid + // reference anywhere) or a phantom jump from a desynced counter — both belong on + // the keyframe path, never a force-reference that could ship corruption as a + // recovery anchor. Wrapping width: frame indexes are u32 counters. + let width = last.wrapping_sub(first); + if width > punktfunk_core::packet::RFI_MAX_RANGE { + tracing::debug!(first, last, width, "RFI range too wide — keyframe instead"); + want_kf = true; + } else if enc.caps().supports_rfi + && enc.invalidate_ref_frames(first as i64, last as i64) + { // The RFI recovered the loss with a clean re-anchor P-frame (no IDR). Anchor the // keyframe cooldown so the client's echo of the SAME loss — its frames_dropped- // driven keyframe request, arriving ~one loss-window later — is coalesced away @@ -4234,7 +4312,11 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { st_queue.push(queue_us); } let t_submit = std::time::Instant::now(); - if let Err(e) = enc.submit(&frame) { + // This submission's future wire frame index (see `au_seq`): AUs are emitted FIFO one per + // submission, so it lands `inflight.len()` AUs after the `au_seq` the loop is about to + // assign next. The RFI backends pin their frame numbering to it. + let wire_index = au_seq.wrapping_add(inflight.len() as u32); + if let Err(e) = enc.submit_indexed(&frame, wire_index) { // The input half of an encode stall: once the driver stops draining AUs, libavcodec's // one-frame buffer fills and avcodec_send_frame starts failing (EAGAIN) — the same // wedge the watchdog below catches, seen from submit. Rebuild the encoder in place @@ -4339,6 +4421,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { data: au.data, capture_ns: cap_ns, flags, + frame_index: au_seq, deadline, encode_us, queue_us, @@ -4354,6 +4437,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { send_gone = true; break; } + au_seq = au_seq.wrapping_add(1); sent += 1; } if send_gone { @@ -4414,6 +4498,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { data: au.data, capture_ns: cap_ns, flags, + frame_index: au_seq, deadline, encode_us, queue_us: 0, @@ -4426,6 +4511,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> { if frame_tx.send(msg).is_err() { break; } + au_seq = au_seq.wrapping_add(1); sent += 1; } // Signal the send thread to drain + exit (drop the channel), then join it. diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 9483b9e6..5956f52c 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -274,6 +274,15 @@ // `AV_FRAME_FLAG_KEY` — this host flag is the only signal. #define USER_FLAG_RECOVERY_ANCHOR 32 +// Widest lost-frame range (frames, wrapping `last - first`) a reference-frame-invalidation +// recovery may be asked to repair; anything wider goes straight to the keyframe path on BOTH +// ends. RFI can only re-reference history the encoder still holds — NVENC keeps a 5-frame DPB, +// AMD LTR ~1 s of marks — and a genuine loss this wide (>1 s even at 240 fps) has no valid +// reference anywhere, so an RFI request for it is either hopeless or (worse) a phantom range +// from a desynced counter. Shared by the host's RFI dispatch (range → keyframe fallback) and the +// client-side gap detectors (huge gap → resync + keyframe request, no RFI). +#define RFI_MAX_RANGE 256 + // Largest UDP datagram the core will send or accept. `Config::validate` bounds // `shard_payload` so `HEADER_LEN + shard_payload + CRYPTO_OVERHEAD ≤ MAX_DATAGRAM_BYTES`. #define MAX_DATAGRAM_BYTES 2048 @@ -382,6 +391,21 @@ #define VIDEO_CAP_HOST_TIMING 8 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// [`Hello::video_caps`] bit: the client's reassembler keeps **speed-test probe filler in its own +// frame-index space** (a second reassembly window keyed on the [`crate::packet::FLAG_PROBE`] +// user-flag), so probe bursts no longer consume video `frame_index`es. Without this, a mid-session +// speed test burns thousands of video indexes that are invisible to every client-side gap detector +// (probe frames are filtered before the pump sees them) — the first real AU afterwards reads as a +// phantom multi-thousand-frame loss (spurious freeze + a nonsense RFI). It also lets the host's +// encode loop own the video numbering outright (the wire-index contract +// [`crate::packet::Packetizer::packetize_each`] documents), which reference-frame invalidation +// depends on. The host runs mid-session probe bursts ONLY against clients that set this bit — an +// older client gets a declined (zeroed) [`ProbeResult`] instead of a measurement its single-window +// reassembler would silently drop as stale. +#define VIDEO_CAP_PROBE_SEQ 16 +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // QUIC application error code a punktfunk/1 client closes the control connection with on a // **deliberate quit** (a user "stop", not a network drop). The host reads it off the connection's