diff --git a/crates/punktfunk-core/src/packet/header.rs b/crates/punktfunk-core/src/packet/header.rs index 7f039ed4..1346c028 100644 --- a/crates/punktfunk-core/src/packet/header.rs +++ b/crates/punktfunk-core/src/packet/header.rs @@ -42,6 +42,19 @@ pub const USER_FLAG_RECOVERY_ANCHOR: u32 = 0x20; /// COMPLETE frame must be consumed window-by-window (the padding is not part of the stream). pub const USER_FLAG_CHUNK_ALIGNED: u32 = 0x40; +/// `user_flags` bit: this AU was packetized as a **slice-streamed** frame (the P2 slice +/// pipeline): its sentinel blocks (`block_count == 0`) are SLICE-granularity and carry their +/// shard-aligned BASE byte offset in `frame_bytes` (the legacy fixed-geometry sentinel is the +/// degenerate base-0 case), and its FINAL block's base derives from the totals as +/// `(total_data_shards − final_data_shards) × shard_bytes` — variable-size blocks tile the +/// frame in shard units, so the uniform-geometry offset formula does not apply to ANY of its +/// blocks. On every packet of the AU (not just sentinels) because reorder can deliver the final +/// block first and its placement rule differs. Only emitted toward peers advertising +/// [`VIDEO_CAP_STREAMED_AU`](crate::quic::VIDEO_CAP_STREAMED_AU) ∧ +/// [`VIDEO_CAP_MULTI_SLICE`](crate::quic::VIDEO_CAP_MULTI_SLICE) — the pair whose receivers +/// know this contract. +pub const USER_FLAG_SLICE_STREAM: u32 = 0x80; + /// 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, diff --git a/crates/punktfunk-core/src/packet/packetize.rs b/crates/punktfunk-core/src/packet/packetize.rs index b004b460..807c3630 100644 --- a/crates/punktfunk-core/src/packet/packetize.rs +++ b/crates/punktfunk-core/src/packet/packetize.rs @@ -52,31 +52,49 @@ pub struct Packetizer { /// streamed AU's size isn't known up front, so this is the only pre-emission guard against /// producing a frame the receiver must reject). max_blocks: usize, + /// The streamed path's block-count ceiling in SLICE mode ([`USER_FLAG_SLICE_STREAM`]) — + /// variable-K blocks, floored at `min(MIN_STREAM_BLOCK_SHARDS, max_data_per_block)` shards. + slice_block_cap: usize, } -/// One in-progress **streamed** access unit (design/nvenc-subframe-slice-output.md Phase 2): -/// the caller feeds encoder chunks in as they exist ([`Packetizer::push_streamed`]) and every -/// completed `max_data_per_block × shard_payload` block leaves under SENTINEL headers -/// (`frame_bytes = 0`, `block_count = 0` — "not final yet") before the AU's total size is +/// One in-progress **streamed** access unit (design/nvenc-subframe-slice-output.md Phase 2 + +/// the P2 slice pipeline): the caller feeds encoder chunks in as they exist +/// ([`Packetizer::push_streamed`]) and slice-granularity blocks leave under SENTINEL headers +/// (`block_count = 0`, `frame_bytes` = the block's shard-aligned base byte offset — the first +/// block's base 0 is byte-identical to the legacy sentinel) before the AU's total size is /// known; [`Packetizer::finish_streamed`] seals the tail block with the real totals and -/// `FLAG_EOF`. Only ever sent to a peer that advertised -/// [`crate::quic::VIDEO_CAP_STREAMED_AU`]. +/// `FLAG_EOF`, which retro-validate the whole layout. Only ever sent to a peer that advertised +/// [`crate::quic::VIDEO_CAP_STREAMED_AU`] — and slice-granularity (non-zero-base) sentinels +/// additionally require [`crate::quic::VIDEO_CAP_MULTI_SLICE`], whose receivers know the +/// base-offset contract (stable pre-multi-slice clients reject a non-zero sentinel +/// `frame_bytes`, and their hosts never fired this path at real bitrates anyway). pub struct StreamedAu { frame_index: u32, pts_ns: u64, user_flags: u32, - /// Bytes not yet sealed into a block. Kept ≤ one block by `push_streamed` (it flushes only - /// while STRICTLY more than a block is buffered, so the final block always has ≥ 1 byte and - /// a sentinel block is never retroactively the frame's last). + /// Bytes not yet sealed into a block: the sub-shard remainder plus anything below the + /// slice-flush threshold. The final block always has ≥ 1 byte (flushes emit only whole + /// shards and never drain to empty on a slice that ends the AU — `finish_streamed` seals + /// whatever remains). pending: Vec, /// Sentinel blocks already emitted. blocks_out: u16, /// Total AU bytes accumulated so far (`pending` included). total_bytes: u64, + /// WHOLE SHARDS already emitted in sentinel blocks — the next block's base offset in shard + /// units (bases are always shard-aligned so the frame layout tiles in shard units; the + /// receiver derives the final block's base from the totals the same way). + emitted_shards: u64, /// The frame's first packet (block 0, shard 0 — carries `FLAG_SOF`) has been emitted. opened: bool, } +/// The slice-flush floor: a sentinel block below this many data shards costs disproportionate +/// per-block FEC parity (`ceil(k × pct/100)` ≥ 1 whatever `k`), so slice boundaries only flush +/// once this much has accumulated (~22 KB at the standard shard payload). Small slices simply +/// ride with the next one; the wire is never worse than one flush per slice. +pub const MIN_STREAM_BLOCK_SHARDS: usize = 16; + impl StreamedAu { /// The wire frame index this AU is sealed with (the caller's RFI bookkeeping domain). pub fn frame_index(&self) -> u32 { @@ -104,6 +122,10 @@ impl Packetizer { 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), + // Every non-final SLICE block carries at least `min(MIN_STREAM_BLOCK_SHARDS, K)` + // data shards (the flush floor, clamped by the block size), so a max-size frame + // bounds the block count. Mirrors the receiver's slice firewall — keep in step. + slice_block_cap: total_data_max / MIN_STREAM_BLOCK_SHARDS.min(max_data.max(1)) + 2, } } @@ -339,55 +361,94 @@ impl Packetizer { pending: Vec::new(), blocks_out: 0, total_bytes: 0, + emitted_shards: 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. + /// Feed one encoder chunk into a streamed AU, emitting every SLICE-GRANULARITY block that + /// completes under sentinel headers (`block_count = 0`; `frame_bytes` = the block's BASE + /// SHARD OFFSET × shard size — see [`PacketHeader`]'s sentinel contract). `slice_end` marks + /// an encoder-chunk boundary (an Annex-B slice cut): only there may a partial-frame block + /// flush, and only WHOLE shards flush (the sub-shard remainder stays pending so every + /// sentinel base is shard-aligned and the layout tiles in shard units — the receiver's + /// placement + retro-validation contract). Small tails ride along until + /// [`MIN_STREAM_BLOCK_SHARDS`] accumulate: per-block parity makes tiny blocks + /// FEC-expensive. The frame's last block — whose header must carry the real totals — is + /// never emitted here. Packets reach `emit` in wire order, data then parity per block. pub fn push_streamed( &mut self, au: &mut StreamedAu, chunk: &[u8], + slice_end: bool, 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) { + let payload = self.shard_payload; + let block_bytes = self.fec.max_data_per_block as usize * payload; + // The AU's own [`USER_FLAG_SLICE_STREAM`] bit is the single gate for the slice wire + // shape: without it, `slice_end` is inert and every sentinel goes out byte-identical + // to the legacy contract (full-K, `frame_bytes = 0`) — which shipped receivers REQUIRE + // (their firewall drops any other sentinel). Callers therefore pass `slice_end` + // unconditionally and gate by setting the flag on the AU. + let slice_wire = au.user_flags & USER_FLAG_SLICE_STREAM != 0; + // A chunk can complete SEVERAL blocks (a slice bigger than one FEC block, or a huge + // legacy chunk) — keep cutting until neither flush condition holds, so the final block + // can never end up oversized. + loop { + let whole = au.pending.len() / payload; + // Legacy full-block flush stays as the hard ceiling (a frame bigger than one FEC + // block must flush regardless of slice geometry); slice boundaries flush earlier. + let must_flush = au.pending.len() > block_bytes; + let slice_flush = slice_wire && slice_end && whole >= MIN_STREAM_BLOCK_SHARDS; + if !(must_flush || slice_flush) { + return Ok(()); + } + // Room for this sentinel block AND the final block after it, within the u16 wire + // field and the mode's block-count ceiling (the receiver enforces its mirror). + let cap = if slice_wire { + self.slice_block_cap + } else { + self.max_blocks + }; + if au.blocks_out as usize + 2 > cap.min(u16::MAX as usize) { return Err(PunktfunkError::Unsupported( "streamed AU exceeds the negotiated max_frame_bytes", )); } + let k = whole.min(self.fec.max_data_per_block as usize); let sof = !au.opened; let (bi, pts, uf) = (au.blocks_out, au.pts_ns, au.user_flags); let fi = au.frame_index; + let base_bytes = if slice_wire { + au.emitted_shards + .checked_mul(payload as u64) + .and_then(|b| u32::try_from(b).ok()) + .ok_or(PunktfunkError::Unsupported("streamed AU exceeds u32 bytes"))? + } else { + 0 // the legacy sentinel contract: uniform full-K blocks, no base on the wire + }; + let flush_len = k * payload; self.emit_streamed_block( fi, pts, uf, bi, - &au.pending[..block_bytes], - 0, + &au.pending[..flush_len], + base_bytes, 0, sof, false, coder, &mut emit, )?; - au.pending.drain(..block_bytes); + au.pending.drain(..flush_len); + au.emitted_shards += k as u64; au.blocks_out += 1; au.opened = true; } - Ok(()) } /// Close a streamed AU: seal the final block — its headers carry the REAL @@ -419,8 +480,10 @@ impl Packetizer { } /// 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. + /// `block_count = 0` and REUSE `frame_bytes` as the block's base byte offset (shard-aligned + /// by the caller; the first block's base is 0 — byte-identical to the legacy sentinel); + /// 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, diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index c571e972..4ac5dbfb 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -53,6 +53,12 @@ struct BlockState { /// against every packet of the block. data_shards: usize, recovery_shards: usize, + /// The block's base offset in SHARD units — where its data shards land in the frame buffer. + /// Uniform (legacy) frames: `block_index × max_data_shards`. Slice-streamed frames + /// ([`USER_FLAG_SLICE_STREAM`]): sentinels carry it on the wire (`frame_bytes ÷ + /// shard_bytes`), the final block derives it from the totals + /// (`total_data − final_data_shards`). + base_shard: usize, /// Per-data-shard presence: which ranges of the frame buffer hold received bytes (also the /// FEC input map — the codec reads only present slots). have_data: Vec, @@ -296,13 +302,36 @@ impl Reassembler { // 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; + // Slice-streamed frames (the P2 pipeline): variable-size, base-addressed blocks — the + // uniform-geometry rules below don't apply to ANY of their blocks (see + // [`USER_FLAG_SLICE_STREAM`]). Flagged on every packet because reorder can deliver the + // final block first. + let slice_stream = hdr.user_flags & crate::packet::USER_FLAG_SLICE_STREAM != 0; let block_idx = hdr.block_index as usize; // 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); + // The slice pipeline's per-frame block ceiling: every non-final slice block carries at + // least `min(MIN_STREAM_BLOCK_SHARDS, max_data_per_block)` data shards (the sender's + // flush floor, clamped by the block size), so a max-size frame bounds the block count + // (+ the final block and one rounding block). Mirrors `Packetizer::slice_block_cap`. + let slice_block_cap = total_data_max + / super::packetize::MIN_STREAM_BLOCK_SHARDS.min(lim.max_data_shards.max(1)) + + 2; let total_data = frame_bytes.div_ceil(shard_bytes).max(1); - if sentinel { + if sentinel && slice_stream { + // Slice-granularity sentinel: `frame_bytes` is the block's BASE byte offset — + // shard-aligned by contract, and the block's whole range must fit the negotiated + // frame budget (placement re-checks against the live buffer too). + if frame_bytes % shard_bytes != 0 + || frame_bytes + data_shards * shard_bytes > lim.max_frame_bytes + || block_idx + 1 >= slice_block_cap + { + drop(stats); + return Ok(None); + } + } else if sentinel { if frame_bytes != 0 || data_shards != lim.max_data_shards || block_idx + 1 >= lim.max_blocks @@ -311,28 +340,48 @@ impl Reassembler { return Ok(None); } } else { - if block_count > lim.max_blocks || block_idx >= block_count { + let block_cap = if slice_stream { + slice_block_cap + } else { + lim.max_blocks + }; + if block_count > block_cap || 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 + if slice_stream { + // Slice-streamed frames: the earlier blocks are variable-size, so the uniform + // derivation below is meaningless — only the FINAL block is ever non-sentinel, + // and its geometry must be self-consistent: it IS the last block, its K fits + // the negotiated block size, and its shards fit inside the frame (its base + // derives as `total_data − data_shards`). + if block_idx + 1 != block_count + || data_shards > lim.max_data_shards + || data_shards > total_data + { + 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); + // 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]; @@ -399,7 +448,9 @@ impl Reassembler { } *in_flight_bytes += buf_len; e.insert(FrameBuf { - frame_bytes, + // A slice-stream sentinel's `frame_bytes` is its block's BASE offset, not a + // frame size — the unpinned marker stays 0 until the final block's totals. + frame_bytes: if sentinel { 0 } else { frame_bytes }, block_count, pts_ns: hdr.pts_ns, user_flags: hdr.user_flags, @@ -409,6 +460,13 @@ impl Reassembler { }) } }; + // The slice marker must be frame-consistent: a mixed frame would firewall under one + // placement rule and place under the other. The per-packet checks above and the + // placement bounds guard below stay memory-safe without this — it's the tighter drop. + if (frame.user_flags ^ hdr.user_flags) & crate::packet::USER_FLAG_SLICE_STREAM != 0 { + drop(stats); + return Ok(None); + } 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 — @@ -417,9 +475,26 @@ impl Reassembler { // 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); + if frame.block_count != 0 { + if block_idx + 1 >= frame.block_count { + drop(stats); + return Ok(None); + } + if slice_stream { + // The final block pinned the totals, so it exists in `blocks` (the pinning + // packet created it) — a sentinel arriving after the pin must fit strictly + // below its base or it could scribble over the final block's landed shards. + let final_k = frame + .blocks + .get(&((frame.block_count - 1) as u16)) + .map(|b| b.data_shards) + .unwrap_or(0); + let pinned_total = frame.frame_bytes.div_ceil(shard_bytes).max(1); + if frame_bytes / shard_bytes + data_shards > pinned_total - final_k { + 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 @@ -429,14 +504,28 @@ impl Reassembler { // 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) - }); + let lied = if slice_stream { + // Slice frames have no uniform shape to demand — the invariant is positional: + // every sentinel block must sit strictly below the final block's base + // (`total_data − data_shards`; the firewall already proved the subtraction + // safe) and be a non-final index. Sentinel-vs-sentinel overlap is not policed — + // the sender is AEAD-authenticated, so a lying base can only corrupt this + // frame's own pixels, never memory (placement stays in-bounds by these checks). + let final_base = total_data - data_shards; + frame.blocks.iter().any(|(&bi, b)| { + let bi = bi as usize; + bi + 1 >= block_count || b.base_shard + b.data_shards > final_base + }) + } else { + 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; + 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 @@ -483,9 +572,19 @@ impl Reassembler { // First packet of a block sizes its state; `data_shards` is already pinned by the // derived geometry above, but `recovery_shards` is per-block wire input (adaptive FEC // varies it per frame) — later packets must match the block's first. + let base_shard = if slice_stream { + if sentinel { + frame_bytes / shard_bytes // the sentinel's wire base (firewall: shard-aligned) + } else { + total_data - data_shards // the final block sits at the end of the frame + } + } else { + block_idx * lim.max_data_shards + }; let block = blocks.entry(hdr.block_index).or_insert_with(|| BlockState { data_shards, recovery_shards, + base_shard, have_data: vec![false; data_shards], data_received: 0, recovery: vec![None; recovery_shards], @@ -497,6 +596,13 @@ impl Reassembler { drop(stats); return Ok(None); } + // A slice-stream block's base is per-block wire input like `recovery_shards` — later + // packets must agree with the block's first, or two packets of "one" block would place + // their shards at different frame offsets. + if block.base_shard != base_shard { + drop(stats); + return Ok(None); + } // Defense-in-depth (2026-07 security review): the geometry invariants above guarantee // every packet of a block agrees on K, so this can't fire today — but `have_data` // indexing and the recovery-slot math below assume it, and an explicit check keeps a @@ -525,7 +631,13 @@ impl Reassembler { // A data shard lands at its final AU offset — the only copy its bytes ever make // past decrypt. if !block.have_data[shard_index] { - let off = (block_idx * lim.max_data_shards + shard_index) * shard_bytes; + let off = (block.base_shard + shard_index) * shard_bytes; + // The firewall + pin checks keep every accepted base in range; this guard is + // the same future-refactor insurance as the `data_shards` check above. + if off + shard_bytes > buf.len() { + drop(stats); + return Ok(None); + } buf[off..off + shard_bytes].copy_from_slice(body); block.have_data[shard_index] = true; block.data_received += 1; @@ -547,7 +659,7 @@ impl Reassembler { let outcome = if missing == 0 { Ok(()) // every original arrived — its bytes are already in place } else { - let base = block_idx * lim.max_data_shards * shard_bytes; + let base = block.base_shard * shard_bytes; let region = &mut buf[base..base + block.data_shards * shard_bytes]; let mut slots: Vec<&mut [u8]> = region.chunks_mut(shard_bytes).collect(); let parity: Vec<(usize, &[u8])> = block diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 47e502f5..3097eafb 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -661,13 +661,21 @@ fn streamed_packets( 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(()) - }) + // slice_end = true with USER_FLAG_SLICE_STREAM unset: must be inert (the flag is the + // gate) — every legacy-shape assertion downstream proves it. + pk.push_streamed( + &mut au, + c, + true, + 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]| { @@ -906,6 +914,388 @@ fn streamed_lying_final_totals_kill_the_frame_wholesale() { ); } +// --------------------------------------------------------------------------- +// Slice-granularity streamed AUs (USER_FLAG_SLICE_STREAM — nvenc-subframe P2b) +// --------------------------------------------------------------------------- + +/// A block geometry big enough that slice cuts land INSIDE a block — variable-K sentinel +/// blocks with non-uniform bases, the shape the uniform derivation can't describe. +fn slice_config() -> Config { + use crate::config::{FecConfig, ProtocolPhase, Role}; + Config { + role: Role::Host, + phase: ProtocolPhase::P2Punktfunk, + fec: FecConfig { + scheme: FecScheme::Gf16, + fec_percent: 50, + max_data_per_block: 64, + }, + shard_payload: 16, + max_frame_bytes: 4096, + encrypt: false, + key: SessionKey::Aes128Gcm([0u8; 16]), + salt: [0u8; 4], + loopback_drop_period: 0, + } +} + +/// Slice chunks chosen to exercise every packetizer path: an exact-shard slice, a slice with +/// a sub-shard remainder, a slice below [`MIN_STREAM_BLOCK_SHARDS`] that must accumulate, +/// and a finish tail. 1023 B total → blocks (K, base-shard): (20, 0), (25, 20), (18, 45), +/// final (1, 63) with block_count 4. +fn slice_chunks() -> Vec> { + [320usize, 403, 100, 200] + .iter() + .enumerate() + .map(|(c, &n)| (0..n).map(|i| (c * 57 + i * 131 + 7) as u8).collect()) + .collect() +} + +/// Packetize one SLICE-streamed AU (every chunk is a slice boundary), returning the wire +/// packets and concatenated source. +fn slice_streamed_packets() -> (Vec>, Vec) { + let cfg = slice_config(); + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let mut au = pk.begin_streamed(12345, USER_FLAG_SLICE_STREAM, Some(0)); + let mut pkts: Vec> = Vec::new(); + let mut src = Vec::new(); + for c in slice_chunks() { + src.extend_from_slice(&c); + pk.push_streamed(&mut au, &c, true, coder.as_ref(), |h, b| { + 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, b| { + 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) +} + +fn push_all( + r: &mut Reassembler, + coder: &dyn crate::fec::ErasureCoder, + stats: &StatsCounters, + delivery: &[Vec], +) -> Option { + let mut got = None; + for p in delivery { + if let Some(f) = r.push(p, coder, stats).unwrap() { + assert!(got.is_none(), "frame must complete exactly once"); + got = Some(f); + } + } + got +} + +/// The slice wire shape: the flag on EVERY packet, sentinel `frame_bytes` = shard-aligned +/// block base, variable K per block, real totals only on the final block — and the AU +/// reassembles byte-identically from in-order delivery. +#[test] +fn slice_streamed_wire_shape_and_roundtrip() { + let (pkts, src) = slice_streamed_packets(); + assert_eq!(src.len(), 1023); + // (block_index, K, base bytes) — chunk 2 (100 B) accumulated instead of flushing (6 + // whole shards < MIN_STREAM_BLOCK_SHARDS) and rode into block 2 with chunk 3's bytes. + let expect = [(0u16, 20u16, 0u32), (1, 25, 320), (2, 18, 720)]; + for p in &pkts { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + assert_ne!( + h.user_flags & USER_FLAG_SLICE_STREAM, + 0, + "the marker must ride EVERY packet — reorder can deliver any of them first" + ); + if h.block_count == 0 { + let (_, k, base) = expect[h.block_index as usize]; + assert_eq!(h.data_shards, k, "block {} K", h.block_index); + assert_eq!(h.frame_bytes, base, "block {} base", h.block_index); + assert_eq!(base % 16, 0, "sentinel bases are shard-aligned"); + } else { + assert_eq!(h.block_index, 3); + assert_eq!(h.block_count, 4); + assert_eq!(h.frame_bytes as usize, src.len()); + assert_eq!(h.data_shards, 1); + } + } + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts) + .expect("slice-streamed frame must complete"); + assert_eq!(f.data, src, "reassembled slice AU must be byte-identical"); + assert_ne!(f.flags & USER_FLAG_SLICE_STREAM, 0); +} + +/// Loss inside two different variable-K blocks, delivered fully REVERSED (the final block's +/// totals arrive first and pin the frame; every sentinel is then accepted against them) with +/// a duplicate — still byte-identical. +#[test] +fn slice_streamed_survives_loss_and_reorder() { + let (pkts, src) = slice_streamed_packets(); + let mut delivery: Vec> = pkts + .iter() + .filter(|p| { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + // Kill data shards 2/5/17 of block 0 and 3/7 of block 1 — within the 50% parity. + !(h.block_count == 0 + && ((h.block_index == 0 && [2, 5, 17].contains(&h.shard_index)) + || (h.block_index == 1 && [3, 7].contains(&h.shard_index)))) + }) + .cloned() + .collect(); + delivery.reverse(); + let dup = delivery.first().cloned().unwrap(); + delivery.push(dup); + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &delivery) + .expect("slice-streamed frame must complete within the FEC budget"); + assert_eq!(f.data, src); +} + +/// Post-pin range firewall: once the final block pinned the totals, a sentinel whose base +/// would reach into (or past) the final block's range is dropped — and the honest copy of +/// that block still assembles the frame. +#[test] +fn slice_streamed_post_pin_out_of_range_sentinel_dropped() { + let (pkts, src) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + + // Final block first — pins totals (64 data shards, final K = 1). + let finals: Vec> = pkts + .iter() + .filter(|p| hdr_of(p).block_count != 0) + .cloned() + .collect(); + assert!(push_all(&mut r, coder.as_ref(), &stats, &finals).is_none()); + + // A block-0 packet whose base claims shard 60: 60 + 20 > 63 (the final block's base) — + // it would overlap the final block's landed shards. Must drop WITHOUT killing the frame. + let mut evil = pkts + .iter() + .find(|p| { + let h = hdr_of(p); + h.block_count == 0 && h.block_index == 0 && h.shard_index == 0 + }) + .cloned() + .unwrap(); + let mut h = PacketHeader::read_from_bytes(&evil[..HEADER_LEN]).unwrap(); + h.frame_bytes = 60 * 16; + evil[..HEADER_LEN].copy_from_slice(h.as_bytes()); + let before = stats.snapshot().packets_dropped; + assert!(r.push(&evil, coder.as_ref(), &stats).unwrap().is_none()); + assert_eq!(stats.snapshot().packets_dropped, before + 1); + + // The honest packets (including the real block-0) still complete the frame. + let rest: Vec> = pkts + .iter() + .filter(|p| hdr_of(p).block_count == 0) + .cloned() + .collect(); + let f = push_all(&mut r, coder.as_ref(), &stats, &rest) + .expect("the honest blocks must still complete the frame"); + assert_eq!(f.data, src); +} + +/// Slice retro-validation: final totals under which an already-landed sentinel block would +/// overlap the final block's range kill the WHOLE frame, and stragglers can't resurrect it. +#[test] +fn slice_streamed_lying_final_kills_frame() { + let (pkts, _) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + + // All sentinel blocks land first. + let sentinels: Vec> = pkts + .iter() + .filter(|p| hdr_of(p).block_count == 0) + .cloned() + .collect(); + assert!(push_all(&mut r, coder.as_ref(), &stats, &sentinels).is_none()); + + // A final header claiming K = 30 puts the final base at shard 34 — under block 2's + // landed range (base 45, K 18 → needs base ≥ 63). The frame dies wholesale. + let mut lying = pkts + .iter() + .find(|p| { + let h = hdr_of(p); + h.block_count != 0 && h.shard_index == 0 + }) + .cloned() + .unwrap(); + let mut h = PacketHeader::read_from_bytes(&lying[..HEADER_LEN]).unwrap(); + h.data_shards = 30; + lying[..HEADER_LEN].copy_from_slice(h.as_bytes()); + assert!(r.push(&lying, coder.as_ref(), &stats).unwrap().is_none()); + assert_eq!( + stats.snapshot().frames_dropped, + 1, + "the lying frame must be counted lost" + ); + + // The honest final packets are stragglers for a killed index now — no resurrection. + let finals: Vec> = pkts + .iter() + .filter(|p| hdr_of(p).block_count != 0) + .cloned() + .collect(); + assert!(push_all(&mut r, coder.as_ref(), &stats, &finals).is_none()); + assert_eq!(stats.snapshot().frames_dropped, 1); +} + +/// One slice bigger than a whole FEC block must cut MULTIPLE blocks from a single push (the +/// flush loop) — the final block can never be left oversized. +#[test] +fn slice_streamed_giant_slice_cuts_multiple_blocks() { + let cfg = slice_config(); // max_data_per_block 64 + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let mut au = pk.begin_streamed(1, USER_FLAG_SLICE_STREAM, Some(0)); + let src: Vec = (0..70 * 16).map(|i| (i * 131 + 7) as u8).collect(); + let mut pkts: Vec> = Vec::new(); + pk.push_streamed(&mut au, &src, true, coder.as_ref(), |h, b| { + 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, b| { + 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(); + for p in &pkts { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + if h.block_count == 0 { + assert_eq!((h.block_index, h.data_shards, h.frame_bytes), (0, 64, 0)); + } else { + assert_eq!((h.block_index, h.block_count, h.data_shards), (1, 2, 6)); + assert_eq!(h.frame_bytes as usize, src.len()); + } + } + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts).expect("must complete"); + assert_eq!(f.data, src); +} + +/// `max_data_per_block` SMALLER than [`MIN_STREAM_BLOCK_SHARDS`]: one slice cut shatters into +/// several full-K blocks (the flush floor clamps to the block size) and the mirrored +/// block-count ceilings must still admit the frame end to end. +#[test] +fn slice_streamed_small_kmax_roundtrip() { + let cfg = e2e_config(FecScheme::Gf16, 50); // max_data_per_block 4 < 16 + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let mut au = pk.begin_streamed(1, USER_FLAG_SLICE_STREAM, Some(0)); + let mut pkts: Vec> = Vec::new(); + let mut src = Vec::new(); + for c in 0..2usize { + let chunk: Vec = (0..320 + c * 83) + .map(|i| (c * 57 + i * 131 + 7) as u8) + .collect(); + src.extend_from_slice(&chunk); + pk.push_streamed(&mut au, &chunk, true, coder.as_ref(), |h, b| { + 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, b| { + 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(); + for p in &pkts { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + if h.block_count == 0 { + assert_eq!(h.data_shards, 4, "floor clamps to full-K blocks"); + assert_eq!(h.frame_bytes % (4 * 16), 0, "bases advance block-wise"); + } + } + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts).expect("must complete"); + assert_eq!(f.data, src); +} + +/// A frame's packets must agree on the slice marker: a legacy-shaped header for a +/// slice-opened frame (or vice versa) is dropped before it can pin or place anything under +/// the wrong rule. +#[test] +fn slice_streamed_mixed_flag_packet_dropped() { + let (pkts, _) = slice_streamed_packets(); + let hdr_of = |p: &Vec| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + + let cfg = slice_config(); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let coder = coder_for(FecScheme::Gf16); + let stats = StatsCounters::default(); + + // Open the frame with one honest slice sentinel packet. + let first = pkts + .iter() + .find(|p| hdr_of(p).block_count == 0) + .cloned() + .unwrap(); + assert!(r.push(&first, coder.as_ref(), &stats).unwrap().is_none()); + + // A LEGACY final for the same frame index that PASSES the legacy firewall (one 16-byte + // shard, one block): only the flag-consistency check stands between it and pinning the + // slice-opened frame under uniform rules — which would then "catch" the sentinel lying + // and kill the frame. It must be dropped as a packet, neither pinning nor killing. + let mut h = hdr_of(&first); + h.user_flags &= !USER_FLAG_SLICE_STREAM; + h.block_index = 0; + h.block_count = 1; + h.frame_bytes = 16; + h.data_shards = 1; + h.recovery_shards = 0; + h.shard_index = 0; + let mut legacy = Vec::with_capacity(HEADER_LEN + 16); + legacy.extend_from_slice(h.as_bytes()); + legacy.extend_from_slice(&[0xEE; 16]); + let before = stats.snapshot().packets_dropped; + assert!(r.push(&legacy, coder.as_ref(), &stats).unwrap().is_none()); + assert_eq!(stats.snapshot().packets_dropped, before + 1); + assert_eq!(stats.snapshot().frames_dropped, 0, "dropped, not killed"); +} + /// 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] diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 30bf27fc..544c0962 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -305,17 +305,19 @@ impl Session { .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. + /// Feed one encoder chunk into a streamed AU, sealing slice-granularity blocks under + /// sentinel headers (see [`begin_streamed_frame_at`](Self::begin_streamed_frame_at)). + /// `slice_end` marks an encoder slice boundary — the flush points of the P2 slice pipeline + /// (with it false the legacy full-FEC-block granularity applies). The returned batch is + /// often EMPTY (the chunk is buffered until enough accumulates) — that's normal. pub fn seal_streamed_chunk( &mut self, au: &mut StreamedAu, chunk: &[u8], + slice_end: bool, ) -> Result>> { self.seal_run(false, |p, coder, emit| { - p.push_streamed(au, chunk, coder, emit) + p.push_streamed(au, chunk, slice_end, coder, emit) }) } diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 3770bc2d..4974b382 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -623,6 +623,7 @@ fn handle_chunk( session: &mut Session, open: &mut Option, c: ChunkMsg, + slice_wire: bool, burst_cap: Option, pace_rate_bps: u64, ) -> Result> { @@ -635,9 +636,19 @@ fn handle_chunk( "streamed AU abandoned mid-flight (encoder rebuild) — client ages it out" ); } + // The AU's own flag bit is what switches the wire to slice-granularity blocks — + // set only toward a client that negotiated BOTH streamed AUs and multi-slice (the + // flag's contract in `punktfunk_core::packet`); without it the sealer stays on the + // legacy full-FEC-block shape shipped receivers require. + let flags = c.flags + | if slice_wire { + punktfunk_core::packet::USER_FLAG_SLICE_STREAM + } else { + 0 + }; *open = Some(StreamedOpen { au: session - .begin_streamed_frame_at(c.capture_ns, c.flags, c.frame_index) + .begin_streamed_frame_at(c.capture_ns, flags, c.frame_index) .map_err(|e| anyhow!("begin_streamed_frame: {e:?}"))?, spread_us: 0, paced: false, @@ -648,8 +659,11 @@ fn handle_chunk( "streamed chunk without an open AU (encode-loop bug)" )); }; + // Every ChunkMsg IS an encoder slice boundary (the chunked poll returns per-slice + // readbacks), so `slice_end` is unconditionally true — the AU's flag gates whether the + // sealer may actually cut a block there. let wires = session - .seal_streamed_chunk(&mut s.au, &c.data) + .seal_streamed_chunk(&mut s.au, &c.data, true) .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)?; @@ -743,6 +757,9 @@ fn send_loop( probe_result_tx: tokio::sync::mpsc::UnboundedSender, stop: Arc, perf: bool, + // Streamed AUs go out as slice-granularity blocks ([`USER_FLAG_SLICE_STREAM`]'s contract) + // instead of the legacy full-FEC-block shape. + slice_wire: bool, burst_cap: Option, fec_target: Arc, stats: SendStats, @@ -822,9 +839,14 @@ fn send_loop( pace_rate, ) .map(|stat| Some((msg, stat))), - SendMsg::Chunk(c) => { - handle_chunk(&mut session, &mut streamed, c, burst_cap, pace_rate) - } + SendMsg::Chunk(c) => handle_chunk( + &mut session, + &mut streamed, + c, + slice_wire, + burst_cap, + pace_rate, + ), }; match outcome { // Mid-AU chunk: sealed + on the wire; the per-AU accounting runs at `last`. @@ -1384,8 +1406,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option