diff --git a/crates/punktfunk-core/src/client/pump/handshake.rs b/crates/punktfunk-core/src/client/pump/handshake.rs index d3392a5e..9f0d45fc 100644 --- a/crates/punktfunk-core/src/client/pump/handshake.rs +++ b/crates/punktfunk-core/src/client/pump/handshake.rs @@ -83,10 +83,13 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result, + /// Sentinel blocks already emitted. + blocks_out: u16, + /// Total AU bytes accumulated so far (`pending` included). + total_bytes: u64, + /// The frame's first packet (block 0, shard 0 — carries `FLAG_SOF`) has been emitted. + opened: bool, +} + +impl StreamedAu { + /// The wire frame index this AU is sealed with (the caller's RFI bookkeeping domain). + pub fn frame_index(&self) -> u32 { + self.frame_index + } } impl Packetizer { pub fn new(config: &Config) -> Self { let max_data = config.fec.max_data_per_block as usize; + let total_data_max = config + .max_frame_bytes + .div_ceil(config.shard_payload.max(1)) + .max(1); Packetizer { next_frame_index: 0, next_probe_index: 0, @@ -64,6 +103,7 @@ impl Packetizer { // Mirrors `ReassemblerLimits::from_config` — keep the two in step. max_total_shards: (max_data + config.fec.recovery_for(max_data)) .min(config.fec.scheme.max_total_shards()), + max_blocks: total_data_max.div_ceil(max_data).max(1), } } @@ -277,4 +317,202 @@ impl Packetizer { self.next_seq = next_seq; Ok(()) } + + /// Open a **streamed** access unit (see [`StreamedAu`]). `frame_index` follows the same + /// contract as [`packetize_each`](Self::packetize_each): `Some(i)` = the caller owns the + /// video numbering; `None` draws from the internal counter. + pub fn begin_streamed( + &mut self, + pts_ns: u64, + user_flags: u32, + frame_index: Option, + ) -> StreamedAu { + let frame_index = frame_index.unwrap_or_else(|| { + let i = self.next_frame_index; + self.next_frame_index = i.wrapping_add(1); + i + }); + StreamedAu { + frame_index, + pts_ns, + user_flags, + pending: Vec::new(), + blocks_out: 0, + total_bytes: 0, + opened: false, + } + } + + /// Feed one encoder chunk into a streamed AU, emitting every block that COMPLETES under + /// sentinel headers (`frame_bytes = 0`, `block_count = 0`, exactly `max_data_per_block` + /// data shards — the receiver's offset formula needs no total). Flushes only while + /// STRICTLY more than one block is buffered, so the frame's last block — whose header must + /// carry the real totals — is never emitted here. Packets reach `emit` in wire order (the + /// caller's nonce order), data then parity per block. + pub fn push_streamed( + &mut self, + au: &mut StreamedAu, + chunk: &[u8], + coder: &dyn ErasureCoder, + mut emit: impl FnMut(&PacketHeader, &[u8]) -> Result<()>, + ) -> Result<()> { + au.total_bytes += chunk.len() as u64; + au.pending.extend_from_slice(chunk); + let block_bytes = self.fec.max_data_per_block as usize * self.shard_payload; + while au.pending.len() > block_bytes { + // Room for this sentinel block AND the final block after it, within the peer's + // per-frame block ceiling and the u16 wire field. + if au.blocks_out as usize + 2 > self.max_blocks.min(u16::MAX as usize) { + return Err(PunktfunkError::Unsupported( + "streamed AU exceeds the negotiated max_frame_bytes", + )); + } + let sof = !au.opened; + let (bi, pts, uf) = (au.blocks_out, au.pts_ns, au.user_flags); + let fi = au.frame_index; + self.emit_streamed_block( + fi, + pts, + uf, + bi, + &au.pending[..block_bytes], + 0, + 0, + sof, + false, + coder, + &mut emit, + )?; + au.pending.drain(..block_bytes); + au.blocks_out += 1; + au.opened = true; + } + Ok(()) + } + + /// Close a streamed AU: seal the final block — its headers carry the REAL + /// `frame_bytes`/`block_count`, which retro-validate the whole frame at the receiver — with + /// `FLAG_EOF` on the last emitted packet. An empty AU degenerates to today's single + /// zero-padded-shard frame (`block_count = 1`, never a sentinel). + pub fn finish_streamed( + &mut self, + au: StreamedAu, + coder: &dyn ErasureCoder, + mut emit: impl FnMut(&PacketHeader, &[u8]) -> Result<()>, + ) -> Result<()> { + let frame_bytes = u32::try_from(au.total_bytes) + .map_err(|_| PunktfunkError::Unsupported("streamed AU exceeds u32 bytes"))?; + let block_count = au.blocks_out + 1; + self.emit_streamed_block( + au.frame_index, + au.pts_ns, + au.user_flags, + au.blocks_out, + &au.pending, + frame_bytes, + block_count, + !au.opened, + true, + coder, + &mut emit, + ) + } + + /// Seal ONE streamed block (data shards + its parity, in wire order). Sentinel blocks pass + /// `frame_bytes = 0` / `block_count = 0`; the final block passes the real totals. `sof` + /// marks the frame's very first packet, `eof` its very last. + #[allow(clippy::too_many_arguments)] + fn emit_streamed_block( + &mut self, + frame_index: u32, + pts_ns: u64, + user_flags: u32, + block_index: u16, + bytes: &[u8], + frame_bytes: u32, + block_count: u16, + sof: bool, + eof: bool, + coder: &dyn ErasureCoder, + emit: &mut impl FnMut(&PacketHeader, &[u8]) -> Result<()>, + ) -> Result<()> { + let payload = self.shard_payload; + if payload > u16::MAX as usize { + return Err(PunktfunkError::InvalidArg("shard_payload exceeds u16")); + } + // At least one (zero-padded) data shard even for an empty final block (empty AU). + let k = bytes.len().div_ceil(payload).max(1); + let m = self + .fec + .recovery_for(k) + .min(self.max_total_shards.saturating_sub(k)); + if k + m > u16::MAX as usize { + return Err(PunktfunkError::Unsupported("block shard count exceeds u16")); + } + // Stage the one possibly-partial (or empty-frame) shard in the zero-padded scratch. + let full_shards = bytes.len() / payload; + self.tail.clear(); + self.tail.resize(payload, 0); + let rem = bytes.len() % payload; + if rem > 0 { + self.tail[..rem].copy_from_slice(&bytes[full_shards * payload..]); + } + let tail = &self.tail; + let shard_at = |s: usize| -> &[u8] { + if s < full_shards { + &bytes[s * payload..(s + 1) * payload] + } else { + tail.as_slice() + } + }; + let data_shards: Vec<&[u8]> = (0..k).map(shard_at).collect(); + if self.recovery.is_empty() { + self.recovery.push(Vec::new()); + } + coder.encode_into(&data_shards, m, &mut self.recovery[0])?; + + let mut next_seq = self.next_seq; + let mut emit_one = |next_seq: &mut u32, shard_index: usize, body: &[u8], flags: u8| { + let seq = *next_seq; + *next_seq = next_seq.wrapping_add(1); + let hdr = PacketHeader { + pts_ns, + frame_index, + stream_seq: seq, + frame_bytes, + user_flags, + block_index, + block_count, + data_shards: k as u16, + recovery_shards: m as u16, + shard_index: shard_index as u16, + shard_bytes: payload as u16, + magic: PUNKTFUNK_MAGIC, + version: self.version, + fec_scheme: coder.scheme() as u8, + flags, + }; + emit(&hdr, body) + }; + for (shard_index, body) in data_shards.iter().enumerate() { + let mut flags = FLAG_PIC; + if sof && shard_index == 0 { + flags |= FLAG_SOF; + } + if eof && m == 0 && shard_index + 1 == k { + flags |= FLAG_EOF; + } + emit_one(&mut next_seq, shard_index, body, flags)?; + } + for r in 0..m { + let mut flags = FLAG_PIC; + if eof && r + 1 == m { + flags |= FLAG_EOF; + } + let body: &[u8] = &self.recovery[0][r]; + emit_one(&mut next_seq, k + r, body, flags)?; + } + self.next_seq = next_seq; + Ok(()) + } } diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index fda0de27..5514eb08 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -70,7 +70,12 @@ struct BlockState { } struct FrameBuf { + /// Exact AU size. 0 = unknown: the frame was opened by a streamed-AU SENTINEL packet + /// ([`crate::quic::VIDEO_CAP_STREAMED_AU`]) and the final block's real totals haven't + /// arrived yet — the frame can't complete before they do (and retro-validate). frame_bytes: usize, + /// Block count; 0 = unknown (sentinel-opened, totals not yet pinned). A legacy-opened + /// frame always has ≥ 1 here, so 0 doubles as the "unpinned streamed" marker. block_count: usize, pts_ns: u64, user_flags: u32, @@ -277,33 +282,58 @@ impl Reassembler { || total == 0 || total > lim.max_total_shards || shard_index >= total - || block_count == 0 - || block_count > lim.max_blocks - || hdr.block_index as usize >= block_count || frame_bytes > lim.max_frame_bytes { drop(stats); return Ok(None); } - // Derived-geometry firewall: every sender (our Packetizer, any version) slices a frame - // into consecutive blocks of exactly `max_data_per_block` data shards with only the LAST - // block smaller, and stamps the exact `frame_bytes` in every header. That makes every - // data shard's final AU offset computable on arrival — - // offset = (block_index × max_data_per_block + shard_index) × shard_bytes - // — which is what lets shards land straight in the frame buffer below. Enforce the - // invariant so a header lying about its geometry is dropped instead of scribbling into - // another shard's range. - let total_data = frame_bytes.div_ceil(shard_bytes).max(1); - let expect_blocks = total_data.div_ceil(lim.max_data_shards).max(1); + // Streamed-AU sentinel ([`crate::quic::VIDEO_CAP_STREAMED_AU`]): `block_count == 0` — a + // value no legacy sender ever emits — marks a NON-FINAL block of an AU whose total size + // doesn't exist yet (the host is still encoding its tail). The exact derived-geometry + // check below can't run without a total, so a sentinel is bounded by the negotiated + // limits instead — and by construction: full-K exactly (the offset formula needs it), + // never the last block the limits allow (the real final block must still fit after it), + // and no total to lie about. The frame-wide exact check runs retroactively the moment + // the final block's real totals arrive (the pinning arm below). + let sentinel = block_count == 0; let block_idx = hdr.block_index as usize; - let expect_data_shards = if block_idx + 1 == expect_blocks { - total_data - (expect_blocks - 1) * lim.max_data_shards + // For a sentinel-opened frame the buffer must hold ANY final geometry the totals may + // later pin — the maximum the negotiated limits allow (the design's "allocate at + // max_frame_bytes"; the existing in-flight budget bounds the amplification). + let total_data_max = lim.max_frame_bytes.div_ceil(shard_bytes).max(1); + let total_data = frame_bytes.div_ceil(shard_bytes).max(1); + if sentinel { + if frame_bytes != 0 + || data_shards != lim.max_data_shards + || block_idx + 1 >= lim.max_blocks + { + drop(stats); + return Ok(None); + } } else { - lim.max_data_shards - }; - if block_count != expect_blocks || data_shards != expect_data_shards { - drop(stats); - return Ok(None); + if block_count > lim.max_blocks || block_idx >= block_count { + drop(stats); + return Ok(None); + } + // Derived-geometry firewall: every sender (our Packetizer, any version) slices a + // frame into consecutive blocks of exactly `max_data_per_block` data shards with + // only the LAST block smaller, and stamps the exact `frame_bytes` in every + // non-sentinel header. That makes every data shard's final AU offset computable on + // arrival — + // offset = (block_index × max_data_per_block + shard_index) × shard_bytes + // — which is what lets shards land straight in the frame buffer below. Enforce the + // invariant so a header lying about its geometry is dropped instead of scribbling + // into another shard's range. + let expect_blocks = total_data.div_ceil(lim.max_data_shards).max(1); + let expect_data_shards = if block_idx + 1 == expect_blocks { + total_data - (expect_blocks - 1) * lim.max_data_shards + } else { + lim.max_data_shards + }; + if block_count != expect_blocks || data_shards != expect_data_shards { + drop(stats); + return Ok(None); + } } let body = &pkt[HEADER_LEN..HEADER_LEN + shard_bytes]; @@ -350,8 +380,13 @@ impl Reassembler { } // First packet of a frame allocates its whole (zeroed) buffer, budget-gated; later - // packets must agree with its geometry. - let buf_len = total_data * shard_bytes; + // packets must agree with its geometry. A sentinel-opened (streamed) frame allocates at + // the limits' maximum — its real size doesn't exist yet. + let buf_len = if sentinel { + total_data_max * shard_bytes + } else { + total_data * shard_bytes + }; let frame = match win.frames.entry(hdr.frame_index) { std::collections::hash_map::Entry::Occupied(e) => e.into_mut(), std::collections::hash_map::Entry::Vacant(e) => { @@ -374,7 +409,66 @@ impl Reassembler { }) } }; - if frame.block_count != block_count || frame.frame_bytes != frame_bytes { + if sentinel { + // Sentinel packets carry no totals to cross-check: while the frame is unpinned + // (`block_count == 0`) they match by construction, and once the totals are pinned — + // whichever packet arrived first, sentinel or final (reorder is normal) — a + // sentinel block must still be NON-final under them. Its full-K shape was already + // enforced by the firewall, which is exactly what the pinned geometry demands of + // every non-final block, and its shard offsets are within the pinned range by + // `block_idx + 1 < block_count`. + if frame.block_count != 0 && block_idx + 1 >= frame.block_count { + drop(stats); + return Ok(None); + } + } else if frame.block_count == 0 { + // A streamed frame meets its FINAL block's real totals: retro-validate every block + // the sentinels created against the exact geometry these totals derive (the same + // invariant the firewall enforces per-packet for legacy frames), then PIN them. A + // header that lies — totals under which an already-received sentinel block is + // out of range or not full-K — kills the WHOLE frame: its landed shards can't be + // trusted to be at the offsets this geometry means, so delivering any of it would + // hand the decoder spliced bytes. + let expect_blocks = total_data.div_ceil(lim.max_data_shards).max(1); + let final_k = total_data - (expect_blocks - 1) * lim.max_data_shards; + let lied = frame.blocks.iter().any(|(&bi, b)| { + let bi = bi as usize; + bi >= expect_blocks + || (bi + 1 < expect_blocks && b.data_shards != lim.max_data_shards) + || (bi + 1 == expect_blocks && b.data_shards != final_k) + }); + if lied { + let mut f = win + .frames + .remove(&hdr.frame_index) + .expect("frame entry exists"); + *in_flight_bytes -= f.buf.len(); + // Remember the index (with its late-shard memory, exactly like an aged-out + // frame) so stragglers can't resurrect it, reclaim the parity buffers, and + // count the loss — the client's recovery request is the right outcome for a + // frame that was destroyed by a lying header. + win.completed.insert( + hdr.frame_index, + reconstructed_shards(&f.blocks, lim.max_data_shards), + ); + for block in f.blocks.values_mut() { + for slot in block.recovery.iter_mut() { + if let Some(rb) = slot.take() { + if recovery_pool.len() < RECOVERY_POOL_MAX { + recovery_pool.push(rb); + } + } + } + } + if !is_probe { + StatsCounters::add(&stats.frames_dropped, 1); + } + drop(stats); + return Ok(None); + } + frame.frame_bytes = frame_bytes; + frame.block_count = block_count; + } else if frame.block_count != block_count || frame.frame_bytes != frame_bytes { drop(stats); return Ok(None); } @@ -382,6 +476,7 @@ impl Reassembler { buf, blocks, blocks_ok, + block_count: frame_block_count, .. } = frame; @@ -485,8 +580,12 @@ impl Reassembler { } } - // Whole frame ready? - if *blocks_ok == block_count { + // Whole frame ready? Judged against the FRAME's pinned block count, not this packet's + // header — a streamed frame can complete on a reordered sentinel packet (header says 0) + // after the final block already pinned the real totals, and can never complete before + // they're pinned (`0` never equals a non-zero `blocks_ok`). + let block_count = *frame_block_count; + if block_count != 0 && *blocks_ok == block_count { let mut done = win.frames.remove(&hdr.frame_index).unwrap(); win.completed.insert( hdr.frame_index, diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 003fb0bc..e456eaa5 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -640,3 +640,294 @@ fn adaptive_fec_ramp_keeps_maximal_blocks_within_the_peers_ceiling() { src ); } + +// --------------------------------------------------------------------------- +// Streamed access units (VIDEO_CAP_STREAMED_AU — nvenc-subframe-slice-output.md Phase 2) +// --------------------------------------------------------------------------- + +/// Packetize one streamed AU from `chunks` via begin/push/finish, returning the emitted wire +/// packets (header ++ shard) and the concatenated source bytes. +fn streamed_packets( + scheme: FecScheme, + fec_percent: u8, + chunks: &[&[u8]], +) -> (Vec>, Vec) { + let cfg = e2e_config(scheme, fec_percent); + let coder = coder_for(scheme); + let mut pk = Packetizer::new(&cfg); + let mut au = pk.begin_streamed(12345, 0, Some(0)); + let mut pkts: Vec> = Vec::new(); + let mut src = Vec::new(); + for c in chunks { + src.extend_from_slice(c); + pk.push_streamed(&mut au, c, coder.as_ref(), |h: &PacketHeader, b: &[u8]| { + let mut p = Vec::with_capacity(HEADER_LEN + b.len()); + p.extend_from_slice(h.as_bytes()); + p.extend_from_slice(b); + pkts.push(p); + Ok(()) + }) + .unwrap(); + } + pk.finish_streamed(au, coder.as_ref(), |h: &PacketHeader, b: &[u8]| { + let mut p = Vec::with_capacity(HEADER_LEN + b.len()); + p.extend_from_slice(h.as_bytes()); + p.extend_from_slice(b); + pkts.push(p); + Ok(()) + }) + .unwrap(); + (pkts, src) +} + +/// Deliver a streamed AU's packets (optionally with kills within the FEC budget, reversed +/// order, and a duplicate) and assert byte-identical completion. Reversed order is the +/// critical case: the FINAL block's real-total headers arrive FIRST, the frame opens +/// legacy-shaped, and the sentinels must still be accepted against the pinned totals. +fn streamed_roundtrip(scheme: FecScheme, kill: &[usize], reverse: bool) { + let chunks: Vec> = (0..3) + .map(|c| (0..50).map(|i| (c * 57 + i * 131 + 7) as u8).collect()) + .collect(); + let chunk_refs: Vec<&[u8]> = chunks.iter().map(|c| c.as_slice()).collect(); + let (pkts, src) = streamed_packets(scheme, 50, &chunk_refs); + // 150 B / 16 B shards / 4-shard blocks → sentinel blocks 0,1 (4 data + 2 rec each) + + // final block (2 data + 1 rec) = 15 packets. + assert_eq!( + pkts.len(), + 15, + "expected geometry changed — update the kills" + ); + + let mut delivery: Vec> = pkts + .iter() + .enumerate() + .filter(|(i, _)| !kill.contains(i)) + .map(|(_, p)| p.clone()) + .collect(); + if reverse { + delivery.reverse(); + } + if let Some(dup) = delivery.first().cloned() { + delivery.push(dup); + } + + let cfg = e2e_config(scheme, 50); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let coder = coder_for(scheme); + let stats = StatsCounters::default(); + let mut got = None; + for p in &delivery { + if let Some(f) = r.push(p, coder.as_ref(), &stats).unwrap() { + assert!(got.is_none(), "frame must complete exactly once"); + got = Some(f); + } + } + let f = got.expect("streamed frame must complete within the FEC budget"); + assert_eq!( + f.data, src, + "reassembled streamed AU must be byte-identical" + ); + assert_eq!(f.pts_ns, 12345); + assert!(f.complete); +} + +#[test] +fn streamed_roundtrip_clean_and_reversed() { + streamed_roundtrip(FecScheme::Gf16, &[], false); + streamed_roundtrip(FecScheme::Gf16, &[], true); + streamed_roundtrip(FecScheme::Gf8, &[], true); +} + +/// Loss within each block's FEC budget: one data shard from a sentinel block, one from the +/// final block — in both delivery orders. +#[test] +fn streamed_roundtrip_survives_loss_and_reorder() { + // Wire order: blk0 = 0..4 data + 4..6 rec, blk1 = 6..10 data + 10..12 rec, + // final = 12..14 data + 14 rec. + streamed_roundtrip(FecScheme::Gf16, &[1, 12], false); + streamed_roundtrip(FecScheme::Gf16, &[1, 12], true); +} + +/// The wire shape of a streamed AU: sentinel headers (block_count = 0, frame_bytes = 0, +/// full-K) on every non-final block, real totals + EOF on the final block, SOF on the very +/// first packet only. +#[test] +fn streamed_headers_sentinel_then_final() { + let chunks: Vec> = (0..3).map(|_| vec![0xA5u8; 50]).collect(); + let chunk_refs: Vec<&[u8]> = chunks.iter().map(|c| c.as_slice()).collect(); + let (pkts, src) = streamed_packets(FecScheme::Gf16, 50, &chunk_refs); + let mut saw_final = false; + for (i, p) in pkts.iter().enumerate() { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + assert_eq!( + h.flags & FLAG_SOF != 0, + i == 0, + "SOF exactly on the first packet" + ); + assert_eq!( + h.flags & FLAG_EOF != 0, + i + 1 == pkts.len(), + "EOF exactly on the last packet" + ); + if h.block_index < 2 { + assert_eq!( + h.block_count, 0, + "non-final block must ride sentinel headers" + ); + assert_eq!(h.frame_bytes, 0); + assert_eq!(h.data_shards, 4, "sentinel blocks are exactly full-K"); + } else { + saw_final = true; + assert_eq!(h.block_count, 3, "final block carries the real block count"); + assert_eq!(h.frame_bytes as usize, src.len(), "and the real AU size"); + } + } + assert!(saw_final); +} + +/// A streamed AU smaller than one block emits NO sentinels — its single (final) block is +/// byte-identical in shape to a legacy frame, so small frames pay zero streaming overhead +/// and any receiver accepts them. +#[test] +fn streamed_small_frame_degenerates_to_legacy() { + let (pkts, src) = streamed_packets(FecScheme::Gf16, 50, &[&[0x5Au8; 40]]); + for p in &pkts { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + assert_eq!( + h.block_count, 1, + "single-block streamed AU must be legacy-shaped" + ); + assert_eq!(h.frame_bytes as usize, src.len()); + } +} + +/// Sentinel firewall: a sentinel header that is not exactly full-K, or claims a non-zero +/// total, or sits where the final block could no longer follow, is dropped before any +/// allocation happens. +#[test] +fn streamed_sentinel_firewall_bounds() { + let mut r = Reassembler::new(limits()); + let coder = coder_for(FecScheme::Gf8); + let stats = StatsCounters::default(); + let sentinel = |f: fn(&mut PacketHeader)| { + let mut h = base_header(); + h.block_count = 0; + h.frame_bytes = 0; + h.data_shards = 8; // limits().max_data_shards — the only legal sentinel K + h.recovery_shards = 0; + f(&mut h); + h + }; + // Not full-K. + let h = sentinel(|h| h.data_shards = 7); + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + // Claims a total. + let h = sentinel(|h| h.frame_bytes = 64); + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + // Sits on the last block the limits allow (no room for the final block after it). + let h = sentinel(|h| h.block_index = 3); // limits().max_blocks == 4 + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + assert_eq!(stats.snapshot().packets_dropped, 3); + // A conformant sentinel IS accepted (proves the rejections above weren't vacuous). + let h = sentinel(|_| {}); + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + assert_eq!( + stats.snapshot().packets_dropped, + 3, + "conformant sentinel accepted" + ); +} + +/// Retro-validation: final-block totals under which an already-received sentinel block is +/// out of range (or mis-sized) kill the WHOLE frame — no spliced delivery — and the killed +/// index cannot be resurrected by stragglers. +#[test] +fn streamed_lying_final_totals_kill_the_frame_wholesale() { + let mut r = Reassembler::new(limits()); + let coder = coder_for(FecScheme::Gf8); + let stats = StatsCounters::default(); + // Two sentinel blocks (indexes 0 and 1) open the frame and land shards. + for bi in 0..2u16 { + let mut h = base_header(); + h.block_count = 0; + h.frame_bytes = 0; + h.data_shards = 8; + h.recovery_shards = 0; + h.block_index = bi; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + } + // A "final" header claiming the whole AU is ONE 16-byte shard: geometry-valid on its own + // (expect_blocks = 1, K = 1), but it disowns both sentinel blocks → the frame dies. + let mut lying = base_header(); + lying.block_count = 1; + lying.frame_bytes = 16; + lying.data_shards = 1; + lying.recovery_shards = 0; + assert!(r + .push(&packet(lying), coder.as_ref(), &stats) + .unwrap() + .is_none()); + let snap = stats.snapshot(); + assert_eq!( + snap.frames_dropped, 1, + "the lying frame must be counted lost" + ); + // A straggler sentinel for the killed index must not resurrect it. + let mut h = base_header(); + h.block_count = 0; + h.frame_bytes = 0; + h.data_shards = 8; + h.recovery_shards = 0; + let before = stats.snapshot().packets_dropped; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + assert_eq!( + stats.snapshot().packets_dropped, + before + 1, + "straggler for a killed frame must be dropped, not re-open it" + ); +} + +/// A sentinel first-packet commits a MAX-sized frame buffer, so the in-flight budget must +/// bite after IN_FLIGHT_BUF_FACTOR frames — the amplification bound for one-datagram opens. +#[test] +fn streamed_open_amplification_is_budget_bounded() { + let mut r = Reassembler::new(limits()); + let coder = coder_for(FecScheme::Gf8); + let stats = StatsCounters::default(); + // limits(): max_frame_bytes 4096 → each sentinel open commits 4096 B; budget = 4×4096. + for fi in 0..5u32 { + let mut h = base_header(); + h.block_count = 0; + h.frame_bytes = 0; + h.data_shards = 8; + h.recovery_shards = 0; + h.frame_index = fi; + assert!(r + .push(&packet(h), coder.as_ref(), &stats) + .unwrap() + .is_none()); + } + assert_eq!( + stats.snapshot().packets_dropped, + 1, + "the fifth max-sized open must be refused by the in-flight budget" + ); +} diff --git a/crates/punktfunk-core/src/quic/caps.rs b/crates/punktfunk-core/src/quic/caps.rs index 46bce69d..0c89cbcf 100644 --- a/crates/punktfunk-core/src/quic/caps.rs +++ b/crates/punktfunk-core/src/quic/caps.rs @@ -32,6 +32,18 @@ pub const VIDEO_CAP_HOST_TIMING: u8 = 0x08; /// 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; +/// [`Hello::video_caps`] bit: the client's reassembler accepts **streamed access units** +/// (design/nvenc-subframe-slice-output.md Phase 2): the host may ship an AU's early FEC blocks +/// before the AU's total size exists — while the tail of the frame is still encoding — so the +/// AU's last packet leaves the host sooner (latency plan §7 LN1). Non-final blocks ride +/// SENTINEL headers (`block_count == 0` — a value no legacy sender emits — with +/// `frame_bytes == 0` and exactly `max_data_per_block` data shards, so the shard-offset +/// formula needs no total); the FINAL block's headers carry the real +/// `frame_bytes`/`block_count` (+ `FLAG_EOF`), which retro-validate the whole frame's geometry +/// — a mismatch drops the frame wholesale. The host streams ONLY to clients advertising this +/// bit; every other client gets today's whole-AU path (chunks concatenated before sealing), so +/// the fallback is zero-risk. +pub const VIDEO_CAP_STREAMED_AU: u8 = 0x20; /// [`Welcome::host_caps`] bit: the host applies [`InputKind::GamepadState`] /// (crate::input::InputKind::GamepadState) snapshot events — full per-pad state with a reorder diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 6195fa5b..eb72a3d6 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -13,7 +13,9 @@ use crate::crypto::SessionCrypto; use crate::error::{PunktfunkError, Result}; use crate::fec::{coder_for, ErasureCoder}; use crate::input::InputEvent; -use crate::packet::{Packetizer, Reassembler, ReassemblerLimits, MAX_DATAGRAM_BYTES}; +use crate::packet::{ + PacketHeader, Packetizer, Reassembler, ReassemblerLimits, StreamedAu, MAX_DATAGRAM_BYTES, +}; use crate::stats::{Stats, StatsCounters}; use crate::transport::Transport; use zerocopy::IntoBytes; @@ -274,6 +276,68 @@ impl Session { pts_ns: u64, user_flags: u32, frame_index: Option, + ) -> Result>> { + self.seal_run(true, |p, coder, emit| { + p.packetize_each(data, pts_ns, user_flags, frame_index, coder, emit) + }) + } + + /// Host: open a **streamed** access unit ([`crate::quic::VIDEO_CAP_STREAMED_AU`] — only + /// toward a client that advertised it; anyone else must get the whole-AU + /// [`seal_frame_at`](Self::seal_frame_at) path). The AU's bytes are fed in with + /// [`seal_streamed_chunk`](Self::seal_streamed_chunk) as the encoder produces them and + /// closed with [`seal_streamed_finish`](Self::seal_streamed_finish); the three calls' + /// returned wire batches are ONE frame, and the nonce order is emission order across the + /// calls — send each batch as it is returned, before sealing the next. + pub fn begin_streamed_frame_at( + &mut self, + pts_ns: u64, + user_flags: u32, + frame_index: u32, + ) -> Result { + if self.config.role != Role::Host { + return Err(PunktfunkError::InvalidArg( + "seal_frame called on a client session", + )); + } + Ok(self + .packetizer + .begin_streamed(pts_ns, user_flags, Some(frame_index))) + } + + /// Feed one encoder chunk into a streamed AU, sealing every FEC block it completes under + /// sentinel headers (see [`begin_streamed_frame_at`](Self::begin_streamed_frame_at)). The + /// returned batch is often EMPTY (the chunk is buffered until a block fills) — that's + /// normal, not an error. + pub fn seal_streamed_chunk( + &mut self, + au: &mut StreamedAu, + chunk: &[u8], + ) -> Result>> { + self.seal_run(false, |p, coder, emit| { + p.push_streamed(au, chunk, coder, emit) + }) + } + + /// Close a streamed AU: seal the final block with the real totals (+ `FLAG_EOF`), which + /// retro-validate the frame at the receiver. Counts the frame as submitted. + pub fn seal_streamed_finish(&mut self, au: StreamedAu) -> Result>> { + self.seal_run(true, |p, coder, emit| p.finish_streamed(au, coder, emit)) + } + + /// The shared packetize → pooled-wire → seal machinery behind [`seal_frame`](Self::seal_frame) + /// and the streamed sealers: `run` drives the packetizer against an emit sink that writes + /// each packet's plaintext at its final wire offset; the seal pass (two-lane for large runs) + /// then encrypts in place. `count_frame` gates the per-frame stats — a streamed AU counts + /// once, at its finish call. + fn seal_run( + &mut self, + count_frame: bool, + run: impl FnOnce( + &mut Packetizer, + &dyn ErasureCoder, + &mut dyn FnMut(&PacketHeader, &[u8]) -> Result<()>, + ) -> Result<()>, ) -> Result>> { if self.config.role != Role::Host { return Err(PunktfunkError::InvalidArg( @@ -320,10 +384,10 @@ impl Session { // sealing itself is a separate pass so it can split across lanes. let seq_base = *next_seq; let encrypting = crypto.is_some(); - let result = packetizer.packetize_each(data, pts_ns, user_flags, frame_index, coder_ref, { + let result = { let wires = &mut wires; let used = &mut used; - move |hdr, body| { + let mut emit = move |hdr: &PacketHeader, body: &[u8]| -> Result<()> { if *used == wires.len() { wires.push(Vec::new()); } @@ -342,8 +406,9 @@ impl Session { wire.extend_from_slice(body); } Ok(()) - } - }); + }; + run(packetizer, coder_ref, &mut emit) + }; result?; // A smaller frame uses fewer buffers than the pool holds: drop the unused tail, same // as the previous `resize_with(packets.len(), ..)` did. (Before the seal phase, so a @@ -421,10 +486,12 @@ impl Session { if let Some(p) = self.seal_perf.as_mut() { p.fec_ns += fec_ns.load(std::sync::atomic::Ordering::Relaxed); p.seal_ns += seal_ns; - p.frames += 1; + p.frames += count_frame as u64; p.packets += used as u64; } - StatsCounters::add(&self.stats.frames_submitted, 1); + if count_frame { + StatsCounters::add(&self.stats.frames_submitted, 1); + } let bytes: u64 = wires.iter().map(|w| w.len() as u64).sum(); StatsCounters::add(&self.stats.packets_sent, wires.len() as u64); StatsCounters::add(&self.stats.bytes_sent, bytes); diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 4bfe60f3..c3478755 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -1238,6 +1238,9 @@ async fn serve_session( // 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; + // Streamed-AU capability: the client's reassembler accepts sentinel-headed streamed blocks, + // so a chunked encoder session may ship an AU's early FEC blocks while its tail encodes. + let streamed_au = hello.video_caps & punktfunk_core::quic::VIDEO_CAP_STREAMED_AU != 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). @@ -1330,6 +1333,7 @@ async fn serve_session( conn: conn_stream, timing_conn, probe_seq, + streamed_au, stats: stats_dp, client_label, launch: launch_for_dp, diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index e3b3383f..4f4986ce 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -252,6 +252,19 @@ fn paced_submit( let wires = session .seal_frame_at(data, pts_ns, flags, frame_index) .map_err(|e| anyhow!("seal_frame: {e:?}"))?; + pace_sealed(session, wires, deadline, burst_cap, pace_rate_bps) +} + +/// The pace-and-send half of [`paced_submit`], for wires that are ALREADY sealed — shared with +/// the streamed-AU path, whose seal happens per encoder chunk ([`handle_chunk`]) under the same +/// microburst policy and frame deadline. +fn pace_sealed( + session: &mut Session, + wires: Vec>, + deadline: std::time::Instant, + burst_cap: Option, + pace_rate_bps: u64, +) -> Result { 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. crate::send_pacing::inject_video_drop(&mut refs); @@ -335,6 +348,116 @@ struct FrameMsg { was_measured: bool, } +/// What the encode thread hands the send thread: a whole AU (the legacy path — every session +/// shape except a chunked encoder toward a streamed-capable client), or one slice-boundary +/// chunk of a streamed AU (§7 LN1 Phase 2 — the send thread seals/paces each chunk's completed +/// FEC blocks while the encoder still produces the AU's tail). +enum SendMsg { + Frame(FrameMsg), + Chunk(ChunkMsg), +} + +/// One encoder chunk of a streamed AU. AU-level fields (`capture_ns`/`flags`/`frame_index`/ +/// `deadline`) are identical on every chunk of one AU (the send thread opens the streamed seal +/// at `first`); the perf split fields are meaningful on `last` (whole-AU figures, exactly like +/// [`FrameMsg`]'s). +struct ChunkMsg { + data: Vec, + first: bool, + last: bool, + capture_ns: u64, + flags: u32, + frame_index: u32, + deadline: std::time::Instant, + encode_us: u32, + queue_us: u32, + cap_us: u32, + submit_us: u32, + wait_us: u32, + repeat: bool, + was_measured: bool, +} + +/// A streamed AU the send thread has open: the core's incremental sealer plus the pace +/// aggregation across its per-chunk flushes (the accounting the whole-AU path reads off one +/// [`paced_submit`] call). +struct StreamedOpen { + au: punktfunk_core::packet::StreamedAu, + spread_us: u32, + paced: bool, +} + +/// Feed one [`ChunkMsg`] through the streamed sealer: open at `first`, seal + pace every FEC +/// block the chunk completes, close (+ final block, real totals) at `last`. Returns +/// `Some((accounting, aggregated PaceStat))` when the AU finished — the caller runs the same +/// per-AU accounting as the whole-frame path — and `None` mid-AU. +fn handle_chunk( + session: &mut Session, + open: &mut Option, + c: ChunkMsg, + burst_cap: Option, + pace_rate_bps: u64, +) -> Result> { + if c.first { + if open.take().is_some() { + // The encode loop abandoned a mid-flight AU (an encoder stall/rebuild forfeits the + // in-flight frame). Its sentinel packets are already on the wire — the client ages + // that frame out and the rebuild's IDR re-anchors; just don't leak the open state. + tracing::warn!( + "streamed AU abandoned mid-flight (encoder rebuild) — client ages it out" + ); + } + *open = Some(StreamedOpen { + au: session + .begin_streamed_frame_at(c.capture_ns, c.flags, c.frame_index) + .map_err(|e| anyhow!("begin_streamed_frame: {e:?}"))?, + spread_us: 0, + paced: false, + }); + } + let Some(s) = open.as_mut() else { + return Err(anyhow!( + "streamed chunk without an open AU (encode-loop bug)" + )); + }; + let wires = session + .seal_streamed_chunk(&mut s.au, &c.data) + .map_err(|e| anyhow!("seal_streamed_chunk: {e:?}"))?; + if !wires.is_empty() { + let stat = pace_sealed(session, wires, c.deadline, burst_cap, pace_rate_bps)?; + s.spread_us = s.spread_us.saturating_add(stat.spread_us); + s.paced |= stat.paced; + } + if !c.last { + return Ok(None); + } + let s = open.take().expect("checked above"); + let tail = session + .seal_streamed_finish(s.au) + .map_err(|e| anyhow!("seal_streamed_finish: {e:?}"))?; + let stat = pace_sealed(session, tail, c.deadline, burst_cap, pace_rate_bps)?; + Ok(Some(( + FrameMsg { + data: Vec::new(), // already on the wire — accounting only + capture_ns: c.capture_ns, + flags: c.flags, + frame_index: c.frame_index, + deadline: c.deadline, + encode_us: c.encode_us, + queue_us: c.queue_us, + cap_us: c.cap_us, + submit_us: c.submit_us, + wait_us: c.wait_us, + repeat: c.repeat, + was_measured: c.was_measured, + }, + PaceStat { + spread_us: s.spread_us.saturating_add(stat.spread_us), + paced: s.paced || stat.paced, + }, + ))) +} + /// The dedicated send thread: it owns the whole [`Session`] (so no socket clone or shared stats are /// needed) and does FEC+seal + microburst-paced send OFF the capture/encode thread, plus the /// speed-test probe bursts (which also need the Session). Decoupling the paced send from encoding @@ -380,7 +503,7 @@ pub(super) fn reconfig_allowed( #[allow(clippy::too_many_arguments)] fn send_loop( mut session: Session, - frame_rx: std::sync::mpsc::Receiver, + frame_rx: std::sync::mpsc::Receiver, probe_rx: std::sync::mpsc::Receiver, probe_result_tx: tokio::sync::mpsc::UnboundedSender, stop: Arc, @@ -425,96 +548,119 @@ fn send_loop( let mut last_frames_dropped = 0u64; let mut last_packets_dropped = 0u64; let mut last_fec_recovered = 0u64; + // The streamed AU currently open (VIDEO_CAP_STREAMED_AU chunked sends) — `Some` strictly + // between a `ChunkMsg::first` and its `last`. + let mut streamed: Option = None; loop { if stop.load(Ordering::SeqCst) { break; } // 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, probe_seq); + // on the full frame channel meanwhile, which is exactly the intended pause. Never mid-AU: + // a streamed frame's chunks are already leaving the socket, so a burst spliced between + // them would push the AU's tail past its deadline (the exact latency the mode removes). + if streamed.is_none() { + 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. match frame_rx.recv_timeout(std::time::Duration::from_millis(50)) { - Ok(msg) => match paced_submit( - &mut session, - &msg.data, - msg.capture_ns, - msg.flags, - msg.frame_index, - msg.deadline, - burst_cap, + Ok(send_msg) => { // Live ABR-tracked encoder bitrate → pace rate; 0 (not yet known) = uncapped. - (stats.bitrate_kbps.load(Ordering::Relaxed) as f64 * 1000.0 * pace_factor) as u64, - ) { - Ok(stat) => { - // First VIDEO packets are on the wire — complete the bring-up trace (P0.1; - // once-only, no-op on every later frame). Speed-test filler isn't video. - if msg.flags & FLAG_PROBE as u32 == 0 { - stats.bringup.finish("first_packet"); + let pace_rate = (stats.bitrate_kbps.load(Ordering::Relaxed) as f64 + * 1000.0 + * pace_factor) as u64; + // `Ok(Some(..))` = an AU fully left the socket (a whole frame, or a streamed + // AU's last chunk) — run the per-AU accounting; `Ok(None)` = mid-AU chunk. + let outcome = match send_msg { + SendMsg::Frame(msg) => paced_submit( + &mut session, + &msg.data, + msg.capture_ns, + msg.flags, + msg.frame_index, + msg.deadline, + burst_cap, + pace_rate, + ) + .map(|stat| Some((msg, stat))), + SendMsg::Chunk(c) => { + handle_chunk(&mut session, &mut streamed, c, burst_cap, pace_rate) } - // Host timing (0xCF): stamped now — the AU's packets have fully left the - // socket — against the same capture anchor the wire pts carries, so the - // client's per-frame math tiles exactly (network = its host+network − this). - // Best-effort like every side-plane datagram; skipped for speed-test filler - // (FLAG_PROBE isn't video and its pts is the burst clock). - if let Some(tc) = &timing_conn { + }; + match outcome { + // Mid-AU chunk: sealed + on the wire; the per-AU accounting runs at `last`. + Ok(None) => {} + Ok(Some((msg, stat))) => { + // First VIDEO packets are on the wire — complete the bring-up trace (P0.1; + // once-only, no-op on every later frame). Speed-test filler isn't video. if msg.flags & FLAG_PROBE as u32 == 0 { - let host_us = (now_ns().saturating_sub(msg.capture_ns) / 1000) - .min(u32::MAX as u64) - as u32; - let t = punktfunk_core::quic::HostTiming { - pts_ns: msg.capture_ns, - host_us, - // T0.1 stage split: queue + encode ride the FrameMsg (always - // measured), pace is this send's spread. The client derives - // seal/FEC + channel-wait as the residual against host_us. - stages: Some(punktfunk_core::quic::HostStages { - queue_us: msg.queue_us, - encode_us: msg.encode_us, - pace_us: stat.spread_us, - }), - }; - let _ = tc.send_datagram( - punktfunk_core::quic::encode_host_timing_datagram(&t).into(), - ); + stats.bringup.finish("first_packet"); } - } - if perf || stats.rec.is_armed() { - // `encode_us`/`pace_us`/fps are valid for every frame (always measured), - // including the Windows relay + tail-drain frames. The cap/submit/wait splits - // are only real when the frame was measured at capture time — a frame captured - // before this capture armed carries zeroed splits, so skip those (an empty - // window → `percentile()` returns 0) rather than pull the percentiles down. - encode_us.push(msg.encode_us); - pace_us.push(stat.spread_us); - if msg.was_measured { - cap_v.push(msg.cap_us); - submit_v.push(msg.submit_us); - wait_v.push(msg.wait_us); - // Queue age is only meaningful for fresh frames (repeats/tail carry 0 - // by construction — including those would drag the percentiles down). - if !msg.repeat { - queue_v.push(msg.queue_us); + // Host timing (0xCF): stamped now — the AU's packets have fully left the + // socket — against the same capture anchor the wire pts carries, so the + // client's per-frame math tiles exactly (network = its host+network − this). + // Best-effort like every side-plane datagram; skipped for speed-test filler + // (FLAG_PROBE isn't video and its pts is the burst clock). + if let Some(tc) = &timing_conn { + if msg.flags & FLAG_PROBE as u32 == 0 { + let host_us = (now_ns().saturating_sub(msg.capture_ns) / 1000) + .min(u32::MAX as u64) + as u32; + let t = punktfunk_core::quic::HostTiming { + pts_ns: msg.capture_ns, + host_us, + // T0.1 stage split: queue + encode ride the FrameMsg (always + // measured), pace is this send's spread. The client derives + // seal/FEC + channel-wait as the residual against host_us. + stages: Some(punktfunk_core::quic::HostStages { + queue_us: msg.queue_us, + encode_us: msg.encode_us, + pace_us: stat.spread_us, + }), + }; + let _ = tc.send_datagram( + punktfunk_core::quic::encode_host_timing_datagram(&t).into(), + ); } } - if msg.repeat { - repeat_frames += 1; - } else { - new_frames += 1; - } - if stat.paced { - paced_frames += 1; - } else { - immediate_frames += 1; + if perf || stats.rec.is_armed() { + // `encode_us`/`pace_us`/fps are valid for every frame (always measured), + // including the Windows relay + tail-drain frames. The cap/submit/wait splits + // are only real when the frame was measured at capture time — a frame captured + // before this capture armed carries zeroed splits, so skip those (an empty + // window → `percentile()` returns 0) rather than pull the percentiles down. + encode_us.push(msg.encode_us); + pace_us.push(stat.spread_us); + if msg.was_measured { + cap_v.push(msg.cap_us); + submit_v.push(msg.submit_us); + wait_v.push(msg.wait_us); + // Queue age is only meaningful for fresh frames (repeats/tail carry 0 + // by construction — including those would drag the percentiles down). + if !msg.repeat { + queue_v.push(msg.queue_us); + } + } + if msg.repeat { + repeat_frames += 1; + } else { + new_frames += 1; + } + if stat.paced { + paced_frames += 1; + } else { + immediate_frames += 1; + } } } + Err(e) => { + tracing::error!(error = %format!("{e:#}"), "send failed — stopping stream"); + break; + } } - Err(e) => { - tracing::error!(error = %format!("{e:#}"), "send failed — stopping stream"); - break; - } - }, + } Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, // encode thread done } @@ -798,6 +944,11 @@ pub(super) struct SessionContext { /// 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). pub(super) probe_seq: bool, + /// The client advertised [`punktfunk_core::quic::VIDEO_CAP_STREAMED_AU`]: when the session's + /// encoder runs chunked poll (multi-slice sub-frame readback, §7 LN1), the host streams each + /// AU's FEC blocks under sentinel headers as the slices complete instead of waiting for the + /// whole AU. `false` = older client — chunks (if any) are drained whole-AU, zero wire change. + pub(super) streamed_au: 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. @@ -866,6 +1017,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option(3); + let (frame_tx, frame_rx) = std::sync::mpsc::sync_channel::(3); // Live encoder bitrate, shared with the send thread's stats sample: a mid-stream adaptive // bitrate change (bitrate_rx below) updates it so the console shows the actual target. let live_bitrate = Arc::new(AtomicU32::new(bitrate_kbps)); @@ -1932,6 +2094,116 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option = None; while inflight.len() >= depth { + // Streamed chunked drain (§7 LN1 Phase 2): toward a STREAMED_AU client with the + // encoder's chunked poll live, forward each slice chunk to the send thread the + // moment it's readable, so packetize/FEC/pacing overlap the encode tail. Re-queried + // per AU (never cached): a pipelined-retrieve escalation or a session rebuild turns + // the mode off and the next AU falls back to the whole-AU path below. + if streamed_wire && enc.supports_chunked_poll() { + let t_wait = std::time::Instant::now(); + let mut first_chunk_us = 0u32; + let mut au_flags = 0u32; + let mut au_done = false; + loop { + let c = match enc.poll_chunk() { + Ok(Some(c)) => c, + Ok(None) => break, // defensive: nothing in flight + Err(e) => { + poll_err = Some(e); + break; + } + }; + // Every chunk proves the encoder is alive. + last_au_at = std::time::Instant::now(); + encoder_resets = 0; + if c.first { + first_chunk_us = t_wait.elapsed().as_micros() as u32; + au_flags = if c.keyframe { + (FLAG_PIC | FLAG_SOF) as u32 + } else { + FLAG_PIC as u32 + }; + let caps = enc.caps(); + if caps.intra_refresh_recovery + && caps.intra_refresh_period > 0 + && mark_recovery_boundary( + &mut ir_wave_pos, + c.keyframe, + caps.intra_refresh_period, + ) + { + au_flags |= punktfunk_core::packet::USER_FLAG_RECOVERY_POINT; + } + if c.recovery_anchor { + au_flags |= punktfunk_core::packet::USER_FLAG_RECOVERY_ANCHOR; + } + if c.chunk_aligned { + au_flags |= punktfunk_core::packet::USER_FLAG_CHUNK_ALIGNED; + } + if let Some(m) = last_hdr_meta { + if c.keyframe || resend_meta { + let _ = conn.send_datagram( + punktfunk_core::quic::encode_hdr_meta_datagram(&m).into(), + ); + resend_meta = false; + } + } + bringup.mark("first_au"); + } + let last = c.last; + let (cap_ns, sub_ns, deadline) = *inflight.front().expect("inflight non-empty"); + let wait_total_us = t_wait.elapsed().as_micros() as u32; + let encode_us = (now_ns().saturating_sub(sub_ns) / 1000) as u32; + let msg = ChunkMsg { + data: c.data, + first: c.first, + last, + capture_ns: cap_ns, + flags: au_flags, + frame_index: au_seq, + deadline, + encode_us, + queue_us, + cap_us, + submit_us, + wait_us: if measure { wait_total_us } else { 0 }, + repeat, + was_measured: measure, + }; + if frame_tx.send(SendMsg::Chunk(msg)).is_err() { + send_gone = true; + break; + } + if last { + inflight.pop_front(); + au_seq = au_seq.wrapping_add(1); + sent += 1; + au_done = true; + if perf { + st_wait.push(wait_total_us); + // The overlap measurement the Phase-3 gate needs (sampled): how + // early the first slice reached the send thread vs. the whole + // encode — the win is roughly their difference per AU. + if sent % 120 == 0 { + tracing::info!( + first_slice_us = first_chunk_us, + encode_us, + "streamed AU (sampled): first slice handed to send at \ + first_slice_us; encode finished at encode_us" + ); + } + } + break; + } + } + if send_gone || poll_err.is_some() { + break; + } + if au_done { + continue; // drain the next in-flight frame, if depth allows + } + break; // defensive Ok(None): leave the frame in flight, re-poll next tick + } let t_wait = std::time::Instant::now(); let polled = enc.poll(); let wait_us = if measure { @@ -2012,7 +2284,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option