diff --git a/clients/probe/src/main.rs b/clients/probe/src/main.rs index e0f60bbb..112fd4ce 100644 --- a/clients/probe/src/main.rs +++ b/clients/probe/src/main.rs @@ -558,6 +558,10 @@ async fn session(args: Args) -> Result<()> { } else { 0 }, + // Like STREAMED_AU above: the shared-core reassembler pins geometry per-frame, so + // the probe accepts a mid-session shard change (and jumbo growth) up to the + // receive ceiling — and it's exactly the tool to measure both. + max_shard_payload: punktfunk_core::config::max_shard_payload() as u16, } .encode(), ) diff --git a/crates/punktfunk-core/src/client/pump/control_task.rs b/crates/punktfunk-core/src/client/pump/control_task.rs index 291f47a1..2af7b82e 100644 --- a/crates/punktfunk-core/src/client/pump/control_task.rs +++ b/crates/punktfunk-core/src/client/pump/control_task.rs @@ -243,6 +243,38 @@ impl ControlTask { seq: offer.seq, kinds: offer.kinds, }); + } else if let Ok(chg) = crate::quic::ShardPayloadChanged::decode(&msg) { + // Mid-session shard renegotiation (design/shard-payload-reneg.md): the + // host re-keys the sealed video geometry. Per-frame pinning means there + // is nothing to re-key on the receive path — the reassembler follows + // each frame's own header and every buffer is statically sized for the + // ceiling — so the dispatch is validate + ack. The ack is telemetry for + // a shrink and the GATE for a grow (the host emits nothing above the + // old size until it lands). Validate against our own receive bounds — + // the same ceiling we advertised in `Hello::max_shard_payload` — and + // answer an out-of-bounds request with SILENCE, not an ack: a buggy + // host must never read a granted grow out of garbage. + let n = chg.shard_payload as usize; + if (crate::config::MIN_SHARD_PAYLOAD..=crate::config::max_shard_payload()) + .contains(&n) + && n % 2 == 0 + { + tracing::info!( + shard_payload = n, + "host re-keyed the wire shard payload — acking" + ); + let ack = crate::quic::ShardPayloadAck { + shard_payload: chg.shard_payload, + }; + if io::write_msg(&mut ctrl_send, &ack.encode()).await.is_err() { + break; + } + } else { + tracing::warn!( + shard_payload = n, + "out-of-bounds shard-payload change — ignoring (no ack)" + ); + } } else if let Ok(shape) = crate::quic::CursorShape::decode(&msg) { // Pointer bitmap changed (cursor channel, only when negotiated). try_send: // an overflowing ring drops the newest shape — the next change resends. diff --git a/crates/punktfunk-core/src/client/pump/handshake.rs b/crates/punktfunk-core/src/client/pump/handshake.rs index c8f79b04..57485d5a 100644 --- a/crates/punktfunk-core/src/client/pump/handshake.rs +++ b/crates/punktfunk-core/src/client/pump/handshake.rs @@ -156,6 +156,12 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result usize { + match peer { + core::net::IpAddr::V4(_) => 28, + core::net::IpAddr::V6(v6) if v6.to_ipv4_mapped().is_some() => 28, + core::net::IpAddr::V6(_) => 48, + } +} + /// [`shard_payload_for_udp_budget`] for an operator-supplied ON-WIRE IP MTU (the number /// `netsh interface ipv4 show subinterfaces` / `ip link` shows): subtracts the family's IP+UDP /// headers first — 28 for IPv4 (and IPv4-mapped), 48 for IPv6. pub fn shard_payload_for_wire_mtu(wire_mtu: usize, peer: core::net::IpAddr) -> usize { - let ip_udp = match peer { - core::net::IpAddr::V4(_) => 28, - core::net::IpAddr::V6(v6) if v6.to_ipv4_mapped().is_some() => 28, - core::net::IpAddr::V6(_) => 48, - }; - shard_payload_for_udp_budget(wire_mtu.saturating_sub(ip_udp), peer) + shard_payload_for_udp_budget(wire_mtu.saturating_sub(ip_udp_overhead(peer)), peer) +} + +/// The operator's jumbo-frames opt-in (design/shard-payload-reneg.md Phase 2): the target +/// on-wire IP MTU, or `None` = no opt-in (nothing above the 1500-default wire is ever probed +/// or grown to). One knob, one code path: a `PUNKTFUNK_WIRE_MTU` above the standard 1500 +/// derives the target from the operator's number; `PUNKTFUNK_JUMBO=1` is the fixed 9000 +/// profile for operators who don't want to think in MTUs. Raising the wire above 1500 is +/// only ever an ACK-GATED mid-session grow toward a client that advertised +/// [`max_shard_payload`] headroom — sessions still START at the family default. +pub fn jumbo_wire_mtu() -> Option { + if let Ok(v) = std::env::var("PUNKTFUNK_WIRE_MTU") { + if let Ok(mtu) = v.trim().parse::() { + if mtu > 1500 { + return Some(mtu); + } + } + } + match std::env::var("PUNKTFUNK_JUMBO") { + Ok(v) if v.trim() == "1" => Some(9000), + _ => None, + } +} + +/// The jumbo sibling of [`shard_payload_for_wire_mtu`]: the largest even shard payload whose +/// sealed datagram fits `wire_mtu`, clamped to the RECEIVE ceiling ([`max_shard_payload`]) +/// instead of the family 1500-default — the up-leg's grow target. Still floored at +/// [`MIN_SHARD_PAYLOAD`]. +pub fn jumbo_shard_payload_for(wire_mtu: usize, peer: core::net::IpAddr) -> usize { + let p = wire_mtu + .saturating_sub(ip_udp_overhead(peer)) + .saturating_sub(HEADER_LEN + CRYPTO_OVERHEAD); + let p = p - p % 2; // FEC requires even shards + p.clamp(MIN_SHARD_PAYLOAD, max_shard_payload()) } /// Everything needed to construct a [`Session`](crate::session::Session). @@ -626,6 +664,29 @@ mod tests { assert_eq!(shard_payload_for_wire_mtu(1280, v6), 1168); } + /// Jumbo grow-target sizing (the up-leg, design/shard-payload-reneg.md): even, sealed + /// fits the wire, clamped to the RECEIVE ceiling instead of the family 1500-default — + /// and the standard 9000 profile lands on the exact documented value. + #[test] + fn jumbo_shard_payload_math() { + use core::net::IpAddr; + let v4: IpAddr = "192.168.1.50".parse().unwrap(); + let v6: IpAddr = "fd00::50".parse().unwrap(); + // 9000 − 28 (IPv4+UDP) − 64 (header+crypto) = 8908 even; sealed 8972 ≤ the 9216 + // datagram ceiling. The v6 sibling: 9000 − 48 − 64 = 8888. + assert_eq!(jumbo_shard_payload_for(9000, v4), 8908); + assert_eq!(sealed_datagram_bytes(8908), 8972); + assert!(sealed_datagram_bytes(8908) <= MAX_DATAGRAM_BYTES); + assert_eq!(jumbo_shard_payload_for(9000, v6), 8888); + // An operator MTU larger than the receive path clamps to the ceiling, smaller ones + // track the wire, and degenerate ones floor at MIN_SHARD_PAYLOAD. + assert_eq!(jumbo_shard_payload_for(64_000, v4), max_shard_payload()); + let p = jumbo_shard_payload_for(4000, v4); + assert_eq!(p % 2, 0); + assert!(sealed_datagram_bytes(p) <= 4000 - 28); + assert_eq!(jumbo_shard_payload_for(100, v4), MIN_SHARD_PAYLOAD); + } + /// Family selection: genuine v6 remotes get the v6 size; v4 — including the IPv4-mapped v6 /// form a dual-stack `[::]` socket reports for a v4 client — keeps the v4 size. #[test] diff --git a/crates/punktfunk-core/src/packet/header.rs b/crates/punktfunk-core/src/packet/header.rs index 1346c028..c9b7e83c 100644 --- a/crates/punktfunk-core/src/packet/header.rs +++ b/crates/punktfunk-core/src/packet/header.rs @@ -70,7 +70,17 @@ pub const CRYPTO_OVERHEAD: usize = 8 + crate::crypto::TAG_LEN; /// Largest UDP datagram the core will send or accept. `Config::validate` bounds /// `shard_payload` so `HEADER_LEN + shard_payload + CRYPTO_OVERHEAD ≤ MAX_DATAGRAM_BYTES`. -pub const MAX_DATAGRAM_BYTES: usize = 2048; +/// +/// Sized for **jumbo frames** (design/shard-payload-reneg.md W0.2): a 9000-MTU LAN carries +/// ~8908-byte shards (sealed 8972-byte UDP payloads), and every receive path — the transport +/// `RECV_BUF`, the session's `recvmmsg` ring — is sized from this constant, so a deployed +/// client can accept a jumbo geometry the moment its host negotiates one. The ring cost is +/// 128 × ~9 KiB ≈ 1.1 MiB per **client** session (lazily allocated on first poll; hosts never +/// allocate it) — measured against the ~256 KiB it was at 2048, an acceptable static price +/// for never having to resize buffers on a mid-session grow. Senders still derive their +/// shard payload from the path MTU (`config::mtu1500_shard_payload*`, the wire-MTU clamps); +/// this is the acceptance ceiling, not a transmit size. +pub const MAX_DATAGRAM_BYTES: usize = 9216; /// Fixed per-packet header. `#[repr(C)]`, no padding, zero-copy (de)serializable. #[repr(C)] diff --git a/crates/punktfunk-core/src/packet/packetize.rs b/crates/punktfunk-core/src/packet/packetize.rs index 05bc0f7f..12240871 100644 --- a/crates/punktfunk-core/src/packet/packetize.rs +++ b/crates/punktfunk-core/src/packet/packetize.rs @@ -25,6 +25,10 @@ pub struct Packetizer { next_probe_index: u32, next_seq: u32, shard_payload: usize, + /// The negotiated frame-size cap — kept so a live shard-payload swap + /// ([`set_shard_payload`](Self::set_shard_payload)) can re-derive the per-frame block + /// ceilings from the same formulas construction used. + max_frame_bytes: usize, fec: crate::config::FecConfig, version: u8, /// Reusable zero-padded scratch for the frame's final data shard when the frame isn't an @@ -47,10 +51,12 @@ pub struct Packetizer { /// where every packet of the block is dropped wholesale, the frame never completes, and the /// resulting loss pushes adaptive FEC *higher*. See the `recovery_for` clamp in `packetize_each`. max_total_shards: usize, - /// The peer's per-frame block ceiling, mirroring [`ReassemblerLimits::from_config`]'s - /// `max_blocks` — the streamed path's bound on how many sentinel blocks it may emit (a - /// 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). + /// The peer's per-frame block ceiling — the streamed path's bound on how many sentinel + /// blocks it may emit (a 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). The receiver + /// derives the same ceiling per packet from the packet's own `shard_bytes` + /// (`Reassembler::push` — geometry is per-frame), so this stays in step as long as it is + /// computed from the shard size this packetizer actually stamps. 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. @@ -105,15 +111,12 @@ impl StreamedAu { 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 { + let mut p = Packetizer { next_frame_index: 0, next_probe_index: 0, next_seq: 0, shard_payload: config.shard_payload, + max_frame_bytes: config.max_frame_bytes, fec: config.fec, version: config.phase as u8, tail: Vec::new(), @@ -121,12 +124,37 @@ 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), - // 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, - } + // Derived from the shard size below (single source of truth for the formulas). + max_blocks: 0, + slice_block_cap: 0, + }; + p.set_shard_payload(config.shard_payload); + p + } + + /// Live-swap the wire shard payload (mid-session shard renegotiation, + /// design/shard-payload-reneg.md Phase 1). Takes effect on the next packetized AU — call + /// ONLY between AUs, never with a [`StreamedAu`] in flight: an open streamed AU's + /// shard-aligned tiling derives from the size it began with, and re-keying under it would + /// corrupt the frame's layout. The per-frame block ceilings follow the new size here; the + /// receiver re-derives its side per packet from the header's own `shard_bytes` (geometry + /// is per-frame there), so the two stay in step by construction. Bounds are the caller's + /// contract — go through [`Session::set_shard_payload`](crate::session::Session::set_shard_payload), + /// which enforces the `Config::validate` rules. + pub fn set_shard_payload(&mut self, shard_payload: usize) { + let max_data = self.fec.max_data_per_block as usize; + let total_data_max = self.max_frame_bytes.div_ceil(shard_payload.max(1)).max(1); + self.shard_payload = shard_payload; + self.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. + self.slice_block_cap = total_data_max / MIN_STREAM_BLOCK_SHARDS.min(max_data.max(1)) + 2; + } + + /// The wire shard payload AUs are currently packetized at. + pub fn shard_payload(&self) -> usize { + self.shard_payload } /// Allocate the next **probe-space** frame index (speed-test filler). A separate counter from diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index 5e9d595a..3f12e26f 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -76,6 +76,12 @@ struct BlockState { } struct FrameBuf { + /// The frame's PINNED shard payload — set by its first-arriving packet (bounds-checked by + /// the firewall), matched by every later packet of the frame. Geometry is per-frame so a + /// mid-session `shard_payload` change (design/shard-payload-reneg.md) is safe on an + /// unordered wire: frames in flight complete under their own pin while new frames arrive + /// under the new one, and no cross-geometry splice can land in one buffer. + shard_bytes: usize, /// 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). @@ -105,16 +111,28 @@ struct FrameBuf { /// Per-session bounds the reassembler enforces on every packet header *before* /// allocating, so a hostile or corrupt header cannot drive unbounded memory use. All /// derived from the negotiated [`Config`]. +/// +/// Shard geometry is PER-FRAME, not per-session (mid-session shard-payload renegotiation, +/// design/shard-payload-reneg.md W0.1): a frame's first-arriving packet pins the frame's +/// `shard_bytes` within `[min_shard_bytes, max_shard_bytes]`, later packets must match the +/// pin, and the per-frame block ceiling derives from the pinned size (a shrunk shard needs +/// more blocks for the same bytes). The reorder race between an ordered control-stream +/// geometry change and the unordered video datagrams is thereby killed structurally — every +/// frame is wholly one geometry, whichever order its packets and the change arrive in. #[derive(Clone, Copy, Debug)] pub struct ReassemblerLimits { - /// Expected shard payload length; every shard in the stream must match exactly. - pub shard_bytes: usize, + /// Floor for a frame's pinned shard payload — [`crate::config::MIN_SHARD_PAYLOAD`] in + /// production (or the negotiated value when a session legitimately starts below it). + pub min_shard_bytes: usize, + /// Ceiling for a frame's pinned shard payload — what this receive path accepts and what + /// the client advertises in `Hello::max_shard_payload` + /// ([`crate::config::max_shard_payload`]): the transport recv buffers are sized for a + /// sealed datagram of exactly this shard size. + pub max_shard_bytes: usize, /// Max data shards per block (the negotiated `max_data_per_block`). pub max_data_shards: usize, /// Max total shards per block (data + recovery), capped by the FEC scheme ceiling. pub max_total_shards: usize, - /// Max FEC blocks per frame. - pub max_blocks: usize, /// Max accepted access-unit size. pub max_frame_bytes: usize, } @@ -135,12 +153,13 @@ impl ReassemblerLimits { // snapshot of it. let max_total = (max_data + (max_data * 90).div_ceil(100)).min(c.fec.scheme.max_total_shards()); - let total_data = c.max_frame_bytes.div_ceil(c.shard_payload.max(1)).max(1); ReassemblerLimits { - shard_bytes: c.shard_payload, + // `.min(c.shard_payload)`: never reject the session's own negotiated value — a + // hand-configured session below the production floor still reassembles itself. + min_shard_bytes: crate::config::MIN_SHARD_PAYLOAD.min(c.shard_payload), + max_shard_bytes: crate::config::max_shard_payload(), max_data_shards: max_data, max_total_shards: max_total, - max_blocks: total_data.div_ceil(max_data).max(1), max_frame_bytes: c.max_frame_bytes, } } @@ -179,6 +198,9 @@ const IN_FLIGHT_BUF_FACTOR: usize = 4; /// Recovery-shard buffer pool ceiling (shard-sized buffers): enough for several max-recovery /// blocks in flight, small enough (~720 KB at a 1408-byte shard) to keep after a loss burst. +/// Entries size themselves to the largest shard they ever held, so a jumbo session (opt-in, +/// desktop-LAN — shards up to [`ReassemblerLimits::max_shard_bytes`]) retains proportionally +/// more; it also needs ~6× fewer buffers per block, so the pool rarely fills there. const RECOVERY_POOL_MAX: usize = 512; /// Buffers incoming shards, recovers lost ones via FEC, and emits whole access units. @@ -295,11 +317,16 @@ impl Reassembler { // Bound every attacker-controllable header field against the negotiated limits // BEFORE allocating anything keyed on it — this is the firewall against a tiny // datagram triggering a huge `vec![None; total]` / `Vec::with_capacity`. + // `shard_bytes` is bounds-checked (not equality-checked) because geometry is + // per-frame — the frame-pin check below is what rejects a size CHANGE mid-frame; + // the even requirement mirrors `Config::validate` (FEC requires even shards). let drop = |stats: &StatsCounters| { StatsCounters::add(&stats.packets_dropped, 1); }; if hdr.magic != PUNKTFUNK_MAGIC - || shard_bytes != lim.shard_bytes + || shard_bytes < lim.min_shard_bytes + || shard_bytes > lim.max_shard_bytes + || shard_bytes % 2 != 0 || pkt.len() < HEADER_LEN + shard_bytes || data_shards == 0 || data_shards > lim.max_data_shards @@ -330,6 +357,11 @@ impl Reassembler { // 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 per-frame FEC-block ceiling under THIS packet's shard size (geometry is + // per-frame: a shrunk shard needs more blocks for the same bytes, so a session-level + // cap from the negotiated size would reject legitimate post-shrink frames). Mirrors + // the sender's `Packetizer::new` for whatever size it currently packetizes at. + let max_blocks = total_data_max.div_ceil(lim.max_data_shards).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 @@ -350,9 +382,7 @@ impl Reassembler { return Ok(None); } } else if sentinel { - if frame_bytes != 0 - || data_shards != lim.max_data_shards - || block_idx + 1 >= lim.max_blocks + if frame_bytes != 0 || data_shards != lim.max_data_shards || block_idx + 1 >= max_blocks { drop(stats); return Ok(None); @@ -361,7 +391,7 @@ impl Reassembler { let block_cap = if slice_stream { slice_block_cap } else { - lim.max_blocks + max_blocks }; if block_count > block_cap || block_idx >= block_count { drop(stats); @@ -513,6 +543,7 @@ impl Reassembler { } *in_flight_bytes += buf_len; e.insert(FrameBuf { + shard_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 }, @@ -527,6 +558,15 @@ impl Reassembler { }) } }; + // Per-frame geometry pin: the frame's first packet pinned its shard size; a later + // packet claiming a different (even in-bounds) size is dropped — otherwise two + // geometries would compute different offsets into one buffer (a splice). This is + // also what makes a mid-session `shard_payload` change safe against reorder: a + // straggler of the old geometry can only ever land in ITS OWN frame's buffer. + if frame.shard_bytes != shard_bytes { + drop(stats); + return Ok(None); + } // 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. @@ -883,6 +923,16 @@ impl Reassembler { // jump-to-live, exactly the stale content the flush existed to discard. self.pending_partial = None; } + + /// Test-only: the current in-flight frame-buffer byte commitment (see + /// [`IN_FLIGHT_BUF_FACTOR`]). The mixed-geometry budget tests assert it returns to + /// exactly zero once every frame has terminated — the 0.23.0 lesson: geometry changes + /// breed sizing bugs, and accounting drift here surfaces in the field as a permanent + /// loss storm once the budget wedges. + #[cfg(test)] + pub(crate) fn in_flight(&self) -> usize { + self.in_flight_bytes + } } /// The data shards of a terminating frame that only exist because parity restored them @@ -1024,10 +1074,10 @@ mod reset_tests { #[test] fn reset_drops_a_parked_partial() { let mut r = Reassembler::new(ReassemblerLimits { - shard_bytes: 64, + min_shard_bytes: 64, + max_shard_bytes: 64, max_data_shards: 8, max_total_shards: 16, - max_blocks: 4, max_frame_bytes: 4096, }); r.pending_partial = Some(Frame { diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 2dfc4659..e31a9942 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -7,11 +7,14 @@ use crate::stats::StatsCounters; use zerocopy::{FromBytes, IntoBytes}; fn limits() -> ReassemblerLimits { + // `min == max` pins the whole stream to 16-byte shards — the strictest geometry, so the + // firewall tests below exercise the bounds checks; per-frame-pinning tests build their own + // limits with a real range. Derived per-frame block ceiling: 4096/16 = 256 shards → 32. ReassemblerLimits { - shard_bytes: 16, + min_shard_bytes: 16, + max_shard_bytes: 16, max_data_shards: 8, max_total_shards: 12, - max_blocks: 4, max_frame_bytes: 4096, } } @@ -840,7 +843,7 @@ fn streamed_sentinel_firewall_bounds() { .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 + let h = sentinel(|h| h.block_index = 31); // derived max_blocks == 32 (see `limits()`) assert!(r .push(&packet(h), coder.as_ref(), &stats) .unwrap() @@ -1769,3 +1772,408 @@ fn slice_streamed_in_flight_budget_matches_legacy() { ); } } + +// --------------------------------------------------------------------------- +// Per-frame shard geometry (mid-session shard-payload renegotiation — W0.1, +// design/shard-payload-reneg.md). The 0.23.0 lesson applies in full: geometry +// changes breed sizing bugs, so the slice/sentinel suite re-runs at every +// production shard size and mixed-geometry streams are tortured under reorder. +// --------------------------------------------------------------------------- + +/// The shard sizes the renegotiation actually moves between: the clamp floor (512), a +/// WARP/Tailscale-shaped 1280-MTU path (1216), the 1500-MTU default (1408), and 9000-MTU +/// jumbo (8908 — sealed 8972, inside [`MAX_DATAGRAM_BYTES`]). +const PRODUCTION_SHARDS: [usize; 4] = [512, 1216, 1408, 8908]; + +/// [`prod_slice_config`] at an arbitrary shard payload. +fn geo_config(shard_payload: usize) -> Config { + let mut c = prod_slice_config(); + c.shard_payload = shard_payload; + c.validate().expect("geometry config must be valid"); + c +} + +/// Packetize one legacy AU at the packetizer's CURRENT shard payload with an explicit +/// frame index, returning wire packets + source bytes. +fn legacy_packets_with( + pk: &mut Packetizer, + frame_index: u32, + pts_ns: u64, + len: usize, + coder: &dyn crate::fec::ErasureCoder, +) -> (Vec>, Vec) { + let src: Vec = (0..len) + .map(|i| (i * 131 + frame_index as usize * 7 + 3) as u8) + .collect(); + let mut pkts: Vec> = Vec::new(); + pk.packetize_each(&src, pts_ns, 0, Some(frame_index), coder, |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) +} + +/// The slice-wire regression suite re-run at every production shard size (the design's +/// non-negotiable verification): the exact-multiple sweep (the 0.23.0 filler-shard bug +/// shape), lossy + reversed slice roundtrips, the legacy-streamed sentinel path, and the +/// in-flight budget — each asserting DELIVERED byte-identical frames, never just an +/// absence of errors. +#[test] +fn slice_wire_suite_at_production_shard_sizes() { + let coder = coder_for(FecScheme::Gf16); + for &shard in &PRODUCTION_SHARDS { + let cfg = geo_config(shard); + + // Exact-shard-multiple AUs + the off-by-one sweep around one of them. + for shards in [16usize, 30, 64] { + for extra in 0..3usize { + let n = shards * shard + extra; + let (pkts, src) = streamed_packets_with(&cfg, 1, 1000, true, &[n]); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts) + .unwrap_or_else(|| panic!("shard {shard}: {n}-byte slice AU must complete")); + assert_eq!( + f.data, src, + "shard {shard}: {n}-byte AU must be byte-identical" + ); + assert_eq!( + r.in_flight(), + 0, + "shard {shard}: budget must return to zero" + ); + } + } + + // A multi-slice AU under loss (one data shard of the first flushed block — within + // its ≥ 20% parity) in both delivery orders. Reversed is the critical order: the + // final block's totals arrive first and every sentinel validates against the pin. + for reverse in [false, true] { + let chunks = [20 * shard + 13, 7 * shard + 1, 17 * shard]; + let (pkts, src) = streamed_packets_with(&cfg, 2, 2000, true, &chunks); + let killed = pkts + .iter() + .position(|p| { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + h.shard_index < h.data_shards && h.recovery_shards >= 1 + }) + .expect("suite frame must have a recoverable data shard"); + let mut delivery: Vec> = pkts + .iter() + .enumerate() + .filter(|(i, _)| *i != killed) + .map(|(_, p)| p.clone()) + .collect(); + if reverse { + delivery.reverse(); + } + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &delivery).unwrap_or_else(|| { + panic!("shard {shard} reverse={reverse}: lossy slice AU must complete") + }); + assert_eq!(f.data, src, "shard {shard} reverse={reverse}"); + assert_eq!(r.in_flight(), 0); + } + + // Legacy-streamed (uniform full-K sentinel) path: one AU spanning a sentinel block + // (K = 200) plus a final block. + { + let (pkts, src) = streamed_packets_with(&cfg, 3, 3000, false, &[230 * shard]); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts) + .unwrap_or_else(|| panic!("shard {shard}: legacy-streamed AU must complete")); + assert_eq!(f.data, src); + assert_eq!(r.in_flight(), 0); + } + + // The budget regression at this size: 12 ordinary AUs opened concurrently, no drops. + for slice in [false, true] { + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + for i in 0..12u32 { + let (pkts, _) = + streamed_packets_with(&cfg, i, 1_000_000 * i as u64, slice, &[40_000]); + r.push(&pkts[0], coder.as_ref(), &stats).unwrap(); + } + assert_eq!( + stats + .packets_dropped + .load(std::sync::atomic::Ordering::Relaxed), + 0, + "shard {shard} slice={slice}: 12 AUs in flight must fit the budget" + ); + } + } +} + +/// One packetizer, one reassembler, one continuous stream — the shard payload swapped +/// live between AUs ([`Packetizer::set_shard_payload`], the Phase 1 host seam): every +/// frame across shrink → grow-to-jumbo → shrink-again delivers byte-identically under its +/// own per-frame pin, and the budget returns to zero. +#[test] +fn mid_stream_shard_swap_delivers_every_frame() { + let cfg = geo_config(1408); + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + + // (shard size to swap to, AU length) — swaps happen between AUs, as Phase 1 will. + let schedule = [ + (1408usize, 3 * 1408 + 100), + (1408, 9 * 1408), + (512, 5 * 512 + 17), // shrink (the VPN heal) + (512, 512), + (8908, 12 * 8908 + 1), // grow (jumbo) + (1216, 4 * 1216 + 9), // revert (a mis-proven jumbo hop self-corrects) + ]; + for (i, &(shard, len)) in schedule.iter().enumerate() { + pk.set_shard_payload(shard); + let pts = 1_000_000 * (i as u64 + 1); + let (pkts, src) = legacy_packets_with(&mut pk, i as u32, pts, len, coder.as_ref()); + for p in &pkts { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + assert_eq!( + h.shard_bytes as usize, shard, + "sender must stamp the live size" + ); + } + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts) + .unwrap_or_else(|| panic!("frame {i} at shard {shard} must complete")); + assert_eq!( + f.data, src, + "frame {i} at shard {shard} must be byte-identical" + ); + assert!(f.complete); + } + assert_eq!( + r.in_flight(), + 0, + "budget must be exact across geometry swaps" + ); + assert_eq!(stats.snapshot().frames_dropped, 0); +} + +/// The reorder race the design kills structurally: an old-geometry frame still in flight +/// when new-geometry frames start arriving completes under its OWN pin — its straggler +/// lands in its own buffer, not the new geometry's. +#[test] +fn old_geometry_frame_completes_after_new_geometry_arrived() { + let cfg = geo_config(1408); + let coder = coder_for(FecScheme::Gf16); + let mut pk = Packetizer::new(&cfg); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + + // Frame 0 at 1408: 7 data shards + 2 parity (20% FEC), data-first wire order. Withhold + // THREE data shards — more than parity can bridge — so the frame genuinely stays + // incomplete until a straggler returns (fewer, and FEC would complete it early). + let (pkts0, src0) = legacy_packets_with(&mut pk, 0, 1_000_000, 6 * 1408 + 50, coder.as_ref()); + assert_eq!( + pkts0.len(), + 9, + "expected geometry changed — update the split" + ); + let head: Vec> = pkts0[..4].iter().chain(&pkts0[7..]).cloned().collect(); + let straggler = &pkts0[4]; + assert!( + push_all(&mut r, coder.as_ref(), &stats, &head).is_none(), + "frame 0 must still be incomplete" + ); + + // The stream re-keys to 512: frames 1..=2 arrive whole and deliver. + pk.set_shard_payload(512); + for i in 1..=2u32 { + let pts = 1_000_000 + 1_000_000 * i as u64; + let (pkts, src) = legacy_packets_with(&mut pk, i, pts, 3 * 512 + 7, coder.as_ref()); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts).expect("new-geometry frame"); + assert_eq!(f.data, src); + } + + // Frame 0's old-geometry straggler arrives last — the frame completes byte-identically. + let f = r + .push(straggler, coder.as_ref(), &stats) + .unwrap() + .expect("old-geometry frame must complete under its own pin"); + assert_eq!(f.data, src0); + assert_eq!(f.frame_index, 0); + assert_eq!(r.in_flight(), 0); + assert_eq!(stats.snapshot().frames_dropped, 0); +} + +/// The anti-splice pin: a packet claiming a DIFFERENT (but in-bounds) shard size for an +/// already-pinned frame is dropped — and the frame still completes from its real packets. +#[test] +fn cross_geometry_packet_for_a_pinned_frame_is_dropped() { + let cfg = geo_config(1408); + let coder = coder_for(FecScheme::Gf16); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg)); + let stats = StatsCounters::default(); + + let mut pk_a = Packetizer::new(&geo_config(1408)); + let mut pk_b = Packetizer::new(&geo_config(1216)); + let (pkts, src) = legacy_packets_with(&mut pk_a, 0, 1_000_000, 5 * 1408 + 9, coder.as_ref()); + // The impostor: the same frame index packetized at 1216 — self-consistent (it passes + // the firewall standalone), wrong for THIS frame's pin. + let (impostor, _) = legacy_packets_with(&mut pk_b, 0, 1_000_000, 5 * 1216, coder.as_ref()); + + assert!(r.push(&pkts[0], coder.as_ref(), &stats).unwrap().is_none()); + let before = stats.snapshot().packets_dropped; + assert!(r + .push(&impostor[1], coder.as_ref(), &stats) + .unwrap() + .is_none()); + assert_eq!( + stats.snapshot().packets_dropped, + before + 1, + "cross-geometry packet must be dropped by the frame pin" + ); + let f = push_all(&mut r, coder.as_ref(), &stats, &pkts[1..]) + .expect("the pinned frame must still complete from its real packets"); + assert_eq!(f.data, src, "no impostor bytes may reach the frame"); +} + +/// The firewall bounds on a frame's pinned size: below the floor, above the receive +/// ceiling, or odd ⇒ dropped before any allocation; the exact floor and ceiling are +/// accepted AND deliver (proving the rejections aren't vacuous). +#[test] +fn shard_size_firewall_bounds() { + let cfg = geo_config(1408); + let lim = ReassemblerLimits::from_config(&cfg); + assert_eq!(lim.min_shard_bytes, crate::config::MIN_SHARD_PAYLOAD); + assert_eq!(lim.max_shard_bytes, crate::config::max_shard_payload()); + let coder = coder_for(FecScheme::Gf16); + let mut r = Reassembler::new(lim); + let stats = StatsCounters::default(); + + let single = |shard: usize, frame_index: u32| { + let mut h = base_header(); + h.frame_index = frame_index; + h.shard_bytes = shard as u16; + h.frame_bytes = shard as u32; + h + }; + // Below the floor (even), above the ceiling (even), odd within bounds: all dropped. + for (i, shard) in [510usize, 9154, 1409].into_iter().enumerate() { + let before = stats.snapshot().packets_dropped; + assert!(r + .push(&packet(single(shard, i as u32)), coder.as_ref(), &stats) + .unwrap() + .is_none()); + assert_eq!( + stats.snapshot().packets_dropped, + before + 1, + "shard {shard} must be firewalled" + ); + } + // The exact bounds deliver whole single-shard frames. + for (i, shard) in [ + crate::config::MIN_SHARD_PAYLOAD, + crate::config::max_shard_payload(), + ] + .into_iter() + .enumerate() + { + let f = r + .push( + &packet(single(shard, 10 + i as u32)), + coder.as_ref(), + &stats, + ) + .unwrap() + .unwrap_or_else(|| panic!("boundary shard {shard} must deliver")); + assert_eq!(f.data.len(), shard); + } +} + +mod geometry_proptests { + use super::*; + use proptest::prelude::*; + + /// One generated frame: shard size, slice-vs-legacy wire, size factor, and whether to + /// kill one recoverable data shard. + type GenFrame = (usize, bool, usize, bool); + + fn frame_strategy() -> impl Strategy { + ( + proptest::sample::select(&PRODUCTION_SHARDS[..]), + any::(), + 1usize..30, + any::(), + ) + } + + proptest! { + #![proptest_config(ProptestConfig::with_cases(48))] + + /// Mixed-geometry reorder torture: frames of DIFFERENT shard sizes and wire shapes + /// interleaved into one shuffled delivery, with per-frame recoverable loss — every + /// frame must deliver byte-identically and the in-flight budget must return to + /// exactly zero (the 0.23.0 budget-drift shape, now across geometries). + #[test] + fn mixed_geometry_reorder_torture( + frames in proptest::collection::vec(frame_strategy(), 2..6), + seed in any::(), + ) { + let coder = coder_for(FecScheme::Gf16); + let mut r = Reassembler::new(ReassemblerLimits::from_config(&geo_config(1408))); + let stats = StatsCounters::default(); + + let mut all: Vec<(u64, u32, Vec)> = Vec::new(); // (shuffle key, frame, pkt) + let mut sources: Vec<(u32, Vec)> = Vec::new(); + for (i, &(shard, slice, factor, kill)) in frames.iter().enumerate() { + let cfg = geo_config(shard); + let pts = 1_000_000 * (i as u64 + 1); + let len = factor * shard + (factor % shard.min(7)); + let (mut pkts, src) = if slice { + streamed_packets_with(&cfg, i as u32, pts, true, &[len.max(1)]) + } else { + let mut pk = Packetizer::new(&cfg); + legacy_packets_with(&mut pk, i as u32, pts, len.max(1), coder.as_ref()) + }; + if kill { + if let Some(k) = pkts.iter().position(|p| { + let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap(); + h.shard_index < h.data_shards && h.recovery_shards >= 1 + }) { + pkts.remove(k); + } + } + for (j, p) in pkts.into_iter().enumerate() { + // Deterministic pseudo-shuffle key: interleaves frames and reorders + // within a frame, differently per proptest case. + let key = (seed | 1) + .wrapping_mul(j as u64 + 1) + .wrapping_add((i as u64) << 17) + .rotate_left((j % 61) as u32); + all.push((key, i as u32, p)); + } + sources.push((i as u32, src)); + } + all.sort_by_key(|(k, _, _)| *k); + + let mut delivered: std::collections::HashMap> = + std::collections::HashMap::new(); + for (_, _, p) in &all { + if let Some(f) = r.push(p, coder.as_ref(), &stats).unwrap() { + prop_assert!(f.complete); + prop_assert!(delivered.insert(f.frame_index, f.data).is_none(), + "a frame must deliver exactly once"); + } + } + for (i, src) in &sources { + let got = delivered.get(i); + prop_assert!(got.is_some(), "frame {i} must be DELIVERED, not merely error-free"); + prop_assert_eq!(got.unwrap(), src, "frame {} must be byte-identical", i); + } + prop_assert_eq!(r.in_flight(), 0, "budget must be exact after all frames terminate"); + prop_assert_eq!(stats.snapshot().frames_dropped, 0u64); + } + } +} diff --git a/crates/punktfunk-core/src/quic/control.rs b/crates/punktfunk-core/src/quic/control.rs index 54155cc4..cce568e4 100644 --- a/crates/punktfunk-core/src/quic/control.rs +++ b/crates/punktfunk-core/src/quic/control.rs @@ -55,6 +55,36 @@ pub struct RfiRequest { pub last_frame: u32, } +/// `host → client`, any time after [`Start`]: the video data plane's sealed shard payload +/// changes mid-session (design/shard-payload-reneg.md Phase 1). Sent ONLY to a client whose +/// [`Hello::max_shard_payload`] advertised per-frame geometry (0/absent = legacy — the host +/// must never send this), and never above that advertised ceiling. Asymmetric semantics: +/// +/// - **Shrink** (the mid-session MTU heal): the host may re-key its packetizer at the next +/// AU boundary immediately after sending — per-frame pinning on the client makes the +/// control-vs-datagram reorder race irrelevant and a smaller shard always fits existing +/// buffers. The [`ShardPayloadAck`] is telemetry. +/// - **Grow** (jumbo): the host must not emit a single sealed datagram above the OLD size +/// until the ack arrives — the ack IS the gate, even when the client's buffers would +/// happen to fit (the rule must not erode if the buffer strategy changes later). +/// +/// No `effective_frame_index`: per-frame pinning makes it redundant — every video packet +/// carries its own `shard_bytes` and the receiver follows each frame's pin. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ShardPayloadChanged { + /// The new sealed shard payload in bytes (even, within the client's advertised bounds). + pub shard_payload: u16, +} + +/// `client → host`: answer to [`ShardPayloadChanged`] — echoes the value the client applied. +/// Only sent for an in-bounds request; an out-of-bounds one is dropped WITHOUT an ack (a +/// buggy host must not read silence-then-garbage as a granted grow). The host treats the +/// echoed value as the grant for a pending grow. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ShardPayloadAck { + pub shard_payload: u16, +} + /// `client → host`, periodic: the client's observed data-plane loss, so the host can size FEC to /// the link instead of a flat percentage (adaptive FEC). `loss_ppm` is parts-per-million of shards /// that arrived missing-but-recovered (plus a bump when frames went unrecoverable) over the report @@ -200,6 +230,10 @@ pub const MSG_SET_BITRATE: u8 = 0x05; pub const MSG_BITRATE_CHANGED: u8 = 0x06; /// Type byte of [`RfiRequest`]. pub const MSG_RFI_REQUEST: u8 = 0x07; +/// Type byte of [`ShardPayloadChanged`]. +pub const MSG_SHARD_PAYLOAD_CHANGED: u8 = 0x08; +/// Type byte of [`ShardPayloadAck`]. +pub const MSG_SHARD_PAYLOAD_ACK: u8 = 0x09; /// Type byte of [`ProbeRequest`]. pub const MSG_PROBE_REQUEST: u8 = 0x20; /// Type byte of [`ProbeResult`]. @@ -306,6 +340,46 @@ impl RfiRequest { } } +impl ShardPayloadChanged { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] shard_payload[5..7] + let mut b = Vec::with_capacity(7); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_SHARD_PAYLOAD_CHANGED); + b.extend_from_slice(&self.shard_payload.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 7 || &b[0..4] != CTL_MAGIC || b[4] != MSG_SHARD_PAYLOAD_CHANGED { + return Err(PunktfunkError::InvalidArg("bad ShardPayloadChanged")); + } + Ok(ShardPayloadChanged { + shard_payload: u16::from_le_bytes(b[5..7].try_into().unwrap()), + }) + } +} + +impl ShardPayloadAck { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] shard_payload[5..7] + let mut b = Vec::with_capacity(7); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_SHARD_PAYLOAD_ACK); + b.extend_from_slice(&self.shard_payload.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 7 || &b[0..4] != CTL_MAGIC || b[4] != MSG_SHARD_PAYLOAD_ACK { + return Err(PunktfunkError::InvalidArg("bad ShardPayloadAck")); + } + Ok(ShardPayloadAck { + shard_payload: u16::from_le_bytes(b[5..7].try_into().unwrap()), + }) + } +} + impl LossReport { pub fn encode(&self) -> Vec { // magic[0..4] type[4] loss_ppm[5..9] @@ -1146,6 +1220,33 @@ mod tests { assert!(SetBitrate::decode(&LossReport { loss_ppm: 7 }.encode()).is_err()); } + #[test] + fn shard_payload_messages_roundtrip() { + for shard_payload in [512u16, 1216, 1408, 8908] { + let chg = ShardPayloadChanged { shard_payload }; + assert_eq!(ShardPayloadChanged::decode(&chg.encode()).unwrap(), chg); + let ack = ShardPayloadAck { shard_payload }; + assert_eq!(ShardPayloadAck::decode(&ack.encode()).unwrap(), ack); + // Identical payload shape — the type byte alone must keep the pair disjoint (a + // change echoed back must never re-decode as a change). + assert!(ShardPayloadChanged::decode(&ack.encode()).is_err()); + assert!(ShardPayloadAck::decode(&chg.encode()).is_err()); + } + // Exact length — no trailing bytes, no truncation. + let bytes = ShardPayloadChanged { shard_payload: 512 }.encode(); + assert!(ShardPayloadChanged::decode(&[bytes.as_slice(), &[0]].concat()).is_err()); + assert!(ShardPayloadChanged::decode(&bytes[..bytes.len() - 1]).is_err()); + // Disjoint from the neighboring ids either side (0x07 RfiRequest / 0x20 ProbeRequest). + assert!(ShardPayloadChanged::decode( + &RfiRequest { + first_frame: 1, + last_frame: 2 + } + .encode() + ) + .is_err()); + } + #[test] fn probe_messages_roundtrip() { let req = ProbeRequest { diff --git a/crates/punktfunk-core/src/quic/endpoint.rs b/crates/punktfunk-core/src/quic/endpoint.rs index 8ed25d6b..43a786ff 100644 --- a/crates/punktfunk-core/src/quic/endpoint.rs +++ b/crates/punktfunk-core/src/quic/endpoint.rs @@ -59,7 +59,25 @@ fn stream_transport_idle(idle: std::time::Duration) -> Arc 1500 set, discovery probes up to the sealed JUMBO datagram + // size so a settled connection can PROVE a jumbo path — the actual grow stays + // client-ack-gated (`native/wire_mtu.rs`). The ceiling is per-ENDPOINT, not + // per-connection: with the opt-in set, connections to non-jumbo peers spend a few extra + // failed probes (one PTO each) settling lower; zero cost for anyone who doesn't opt in. + // Derived with the IPv4 overhead — a v6 peer's sealed jumbo target is smaller, so the + // ceiling covers it and discovery settles at the v6 path's own budget. + let probe_ceiling = match crate::config::jumbo_wire_mtu() { + Some(mtu) => { + let shard = crate::config::jumbo_shard_payload_for( + mtu, + std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), + ); + crate::config::sealed_datagram_bytes(shard) as u16 + } + None => crate::config::video_datagram_udp_ceiling() as u16, + }; + mtud.upper_bound(probe_ceiling); t.mtu_discovery_config(Some(mtud)); Arc::new(t) } diff --git a/crates/punktfunk-core/src/quic/handshake.rs b/crates/punktfunk-core/src/quic/handshake.rs index 0f25aa54..1b780de8 100644 --- a/crates/punktfunk-core/src/quic/handshake.rs +++ b/crates/punktfunk-core/src/quic/handshake.rs @@ -90,8 +90,19 @@ pub struct Hello { /// disambiguated by REMAINING LENGTH at decode: fewer than `HDR_META_BODY_LEN` bytes after /// `preferred_codec` ⇒ no HDR block, the tail bytes are the post-HDR fields directly. This /// caps everything after `display_hdr` at `HDR_META_BODY_LEN − 1` bytes total — document any - /// future field here and mind the budget. Omitted when zero and by older clients (→ `0`). + /// future field here and mind the budget (`client_caps` 1 + `max_shard_payload` 2 = 3 of the + /// 27 spent). Omitted when zero and by older clients (→ `0`). pub client_caps: u8, + /// The largest video shard payload this client's receive path accepts — sealed datagrams for + /// shards up to this size fit its transport buffers ([`crate::config::max_shard_payload`]). + /// One field carries BOTH facts the host needs for mid-session shard renegotiation + /// (design/shard-payload-reneg.md W0.3): non-zero ⇒ the client reassembles per-frame + /// geometry (a mid-session `shard_payload` change is safe to send), and the value is the + /// hard ceiling a jumbo grow may never exceed. Appended after `client_caps` as 2 trailing + /// LE bytes (forcing the earlier placeholders). Omitted by older clients (decodes to `0` + /// = legacy: the host must not change the sealed geometry mid-session, and never above + /// the `Welcome` value). + pub max_shard_payload: u16, } /// QUIC application error code a punktfunk/1 client closes the control connection with on a @@ -254,12 +265,14 @@ impl Hello { let pref_present = self.preferred_codec != 0; let hdr_present = self.display_hdr.is_some(); let ccaps_present = self.client_caps != 0; + let msp_present = self.max_shard_payload != 0; let need_placeholders = self.video_caps != 0 || ac_present || vcodecs_present || pref_present || hdr_present - || ccaps_present; + || ccaps_present + || msp_present; match (&self.name, &self.launch) { (None, None) if !need_placeholders => {} (name, _) => { @@ -280,15 +293,21 @@ impl Hello { b.push(self.video_caps); } // audio_channels: emitted when non-stereo OR a later field follows. - if ac_present || vcodecs_present || pref_present || hdr_present || ccaps_present { + if ac_present + || vcodecs_present + || pref_present + || hdr_present + || ccaps_present + || msp_present + { b.push(self.audio_channels); } // video_codecs: emitted when non-zero OR a later field follows. - if vcodecs_present || pref_present || hdr_present || ccaps_present { + if vcodecs_present || pref_present || hdr_present || ccaps_present || msp_present { b.push(self.video_codecs); } // preferred_codec: emitted when non-zero OR a later field follows. - if pref_present || hdr_present || ccaps_present { + if pref_present || hdr_present || ccaps_present || msp_present { b.push(self.preferred_codec); } // display_hdr: fixed HDR_META_BODY_LEN-byte HdrMeta body; omitted when `None` even if @@ -297,10 +316,15 @@ impl Hello { if let Some(m) = &self.display_hdr { super::datagram::write_hdr_meta_body(m, &mut b); } - // client_caps: single byte after the (optional) HDR block. Emitted when non-zero. - if ccaps_present { + // client_caps: single byte after the (optional) HDR block. Emitted when non-zero OR a + // later field follows. + if ccaps_present || msp_present { b.push(self.client_caps); } + // max_shard_payload: 2 trailing LE bytes after client_caps. Emitted when non-zero. + if msp_present { + b.extend_from_slice(&self.max_shard_payload.to_le_bytes()); + } b } @@ -386,6 +410,19 @@ impl Hello { }; b.get(off).copied().unwrap_or(0) }, + // max_shard_payload: 2 LE bytes after client_caps (same post-HDR offset rule). + // Absent on an older client → 0 = no mid-session renegotiation, no jumbo. + max_shard_payload: { + let off = if b.len().saturating_sub(tail + 4) >= super::datagram::HDR_META_BODY_LEN + { + tail + 4 + super::datagram::HDR_META_BODY_LEN + } else { + tail + 4 + }; + b.get(off + 1..off + 3) + .map(|s| u16::from_le_bytes(s.try_into().unwrap())) + .unwrap_or(0) + }, }) } } @@ -867,6 +904,7 @@ mod tests { preferred_codec: CODEC_H264, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; let enc = h.encode(); let dec = Hello::decode(&enc).unwrap(); @@ -944,6 +982,7 @@ mod tests { preferred_codec: CODEC_HEVC, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; assert_eq!(Hello::decode(&h.encode()).unwrap(), h); let s = Start { @@ -975,6 +1014,7 @@ mod tests { preferred_codec: 0, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; let enc = h.encode(); assert_eq!(enc.len(), 26); @@ -1093,6 +1133,7 @@ mod tests { preferred_codec: 0, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; let enc = base.encode(); assert_eq!( @@ -1145,6 +1186,7 @@ mod tests { preferred_codec: 0, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; // launch alone (no name): a zero-length name placeholder keeps the offset deterministic. let with_launch = Hello { @@ -1205,6 +1247,7 @@ mod tests { preferred_codec: 0, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; // A real client-panel volume (P3 primaries, 800-nit peak, 0.05-nit floor, 400-nit FALL). let vol = HdrMeta { @@ -1273,6 +1316,7 @@ mod tests { preferred_codec: 0, display_hdr: None, client_caps: 0, + max_shard_payload: 0, } .encode(); assert!(PairRequest::decode(&h).is_err(), "abi {abi} parsed as pair"); @@ -1306,6 +1350,7 @@ mod tests { preferred_codec: 0, display_hdr: None, client_caps: 0, + max_shard_payload: 0, }; let vol = HdrMeta { display_primaries: [[13250, 34500], [7500, 3000], [34000, 16000]], @@ -1319,6 +1364,7 @@ mod tests { // fixed block length, so the decoder must NOT read it as a truncated HdrMeta). let caps_only = Hello { client_caps: CLIENT_CAP_CURSOR, + max_shard_payload: 0, ..base.clone() }; assert_eq!(Hello::decode(&caps_only.encode()).unwrap(), caps_only); @@ -1326,6 +1372,7 @@ mod tests { let both = Hello { display_hdr: Some(vol), client_caps: CLIENT_CAP_CURSOR, + max_shard_payload: 0, ..base.clone() }; assert_eq!(Hello::decode(&both.encode()).unwrap(), both); @@ -1344,8 +1391,73 @@ mod tests { Hello::decode(&enc[..enc.len() - 1]).unwrap(), Hello { client_caps: 0, + max_shard_payload: 0, ..both.clone() } ); } + + /// `max_shard_payload` (mid-session shard renegotiation, design/shard-payload-reneg.md + /// W0.3): roundtrips, forces the earlier placeholders (deterministic offset), composes + /// with the optional HDR block, and degrades to 0 = legacy in BOTH directions. + #[test] + fn hello_max_shard_payload_roundtrip_and_back_compat() { + let base = Hello { + abi_version: 2, + mode: Mode { + width: 1920, + height: 1080, + refresh_hz: 60, + }, + compositor: CompositorPref::Auto, + gamepad: GamepadPref::Auto, + bitrate_kbps: 0, + name: None, + launch: None, + video_caps: 0, + audio_channels: 2, + video_codecs: 0, + preferred_codec: 0, + display_hdr: None, + client_caps: 0, + max_shard_payload: 0, + }; + // The advertisement alone: every earlier trailing field is emitted as a placeholder + // so the 2 LE bytes land at a deterministic offset — and the whole thing roundtrips. + let adv = Hello { + max_shard_payload: crate::config::max_shard_payload() as u16, + ..base.clone() + }; + assert_eq!(Hello::decode(&adv.encode()).unwrap(), adv); + // Composes with client_caps AND the fixed HDR block (the remaining-length + // disambiguation must still find both fields after it). + let vol = HdrMeta { + display_primaries: [[13250, 34500], [7500, 3000], [34000, 16000]], + white_point: [15635, 16450], + max_display_mastering_luminance: 8_000_000, + min_display_mastering_luminance: 500, + max_cll: 0, + max_fall: 400, + }; + let full = Hello { + display_hdr: Some(vol), + client_caps: CLIENT_CAP_CURSOR, + max_shard_payload: 8908, + ..base.clone() + }; + assert_eq!(Hello::decode(&full.encode()).unwrap(), full); + // An older client (no trailing bytes at all) decodes to 0 = legacy: the host must + // not change the sealed geometry mid-session. + assert_eq!(Hello::decode(&base.encode()).unwrap().max_shard_payload, 0); + // An older HOST reading an advertising Hello never looks past the fields it knows — + // truncating the 2 trailing bytes yields the same Hello minus the advertisement. + let enc = full.encode(); + assert_eq!( + Hello::decode(&enc[..enc.len() - 2]).unwrap(), + Hello { + max_shard_payload: 0, + ..full.clone() + } + ); + } } diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 8ac9033b..a95d6ff5 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -603,6 +603,31 @@ impl Session { self.packetizer.set_fec_percent(pct); } + /// Host: live-swap the wire shard payload between AUs (mid-session shard renegotiation, + /// design/shard-payload-reneg.md). Affects the next sealed AU; call only between AUs + /// (never with a `StreamedAu` in flight — see [`Packetizer::set_shard_payload`]). The new + /// value must satisfy the exact bounds `Config::validate` imposed on the negotiated one + /// (even, > 0, fits a datagram, block count fits the wire) — validated here against a + /// probe of the session config. The PROTOCOL side is the caller's contract: a current + /// client reassembles any in-bounds size per-frame, but a shrink may be sent immediately + /// while a grow must be client-acked and never exceed the client's advertised + /// `Hello::max_shard_payload` ceiling. + pub fn set_shard_payload(&mut self, shard_payload: usize) -> Result<()> { + if self.config.role != Role::Host { + return Err(PunktfunkError::InvalidArg( + "set_shard_payload called on a client session", + )); + } + // Full `Config::validate` parity, zero drift: probe a copy (its key/salt copies are + // zeroized on drop) rather than re-spelling the shard clauses here. + let mut probe = self.config.clone(); + probe.shard_payload = shard_payload; + probe.validate()?; + self.config.shard_payload = shard_payload; + self.packetizer.set_shard_payload(shard_payload); + Ok(()) + } + /// The current FEC recovery percentage (host side). pub fn fec_percent(&self) -> u8 { self.packetizer.fec_percent() @@ -1060,4 +1085,147 @@ mod wire_equivalence_tests { "unflagged AUs must never be delivered partial" ); } + + /// The low-MTU PyroWave guarantee (design/shard-payload-reneg.md): mid-session + /// renegotiation is gated OFF for chunk-aligned sessions, so a constrained path serves + /// them through the leg-1 SESSION-START clamp instead — the learned budget (or + /// `PUNKTFUNK_WIRE_MTU`) sizes `Welcome::shard_payload`, and everything chunk-aligned + /// derives from that ONE number fixed at the handshake: the host packetizes at it, the + /// client's parse window reads it back ([`Session::shard_payload`] → the C-ABI + /// `punktfunk_connection_shard_payload` every embedder walks windows with), and partial + /// delivery zero-fills exact windows of it. Pin that consistency at the clamp shapes a + /// constrained path actually produces: the WARP/Tailscale budget (1216) and the floor + /// (512) — chunk-aligned frames deliver, lose whole windows (never splice), and the + /// window arithmetic matches the session value end to end. + #[test] + fn chunk_aligned_sessions_work_at_clamped_shard_sizes() { + use crate::packet::USER_FLAG_CHUNK_ALIGNED; + for shard in [1216usize, crate::config::MIN_SHARD_PAYLOAD] { + let mk = |role| Config { + role, + phase: ProtocolPhase::P2Punktfunk, + fec: FecConfig { + scheme: FecScheme::Gf16, + fec_percent: 0, // no parity — any drop leaves a hole + max_data_per_block: 64, + }, + shard_payload: shard, + max_frame_bytes: 8 * 1024 * 1024, + encrypt: true, + key: SessionKey::Aes128Gcm([7u8; 16]), + salt: [3, 1, 4, 1], + loopback_drop_period: 0, + }; + let (h, c) = crate::transport::loopback_pair(3, 1); + let mut host = Session::new(mk(Role::Host), Box::new(h)).unwrap(); + let mut client = Session::new(mk(Role::Client), Box::new(c)).unwrap(); + client.set_deliver_partial_frames(true); + // The window every embedder parses with IS the clamped session value. + assert_eq!(client.shard_payload(), shard); + assert_eq!(host.shard_payload(), shard); + + let frame = pattern(8 * shard); + host.submit_frame(&frame, 1_000, USER_FLAG_CHUNK_ALIGNED) + .unwrap(); + let mut got_partial = None; + let mut completes = 0; + for i in 0..80u64 { + host.submit_frame(&pattern(shard), 2_000 + i, USER_FLAG_CHUNK_ALIGNED) + .unwrap(); + loop { + match client.poll_frame() { + Ok(f) if !f.complete => got_partial = Some(f), + Ok(_) => completes += 1, + Err(PunktfunkError::NoFrame) => break, + Err(e) => panic!("shard {shard}: unexpected: {e}"), + } + } + } + let p = got_partial.expect("the lossy frame must be delivered partial"); + assert_eq!(p.data.len(), frame.len(), "shard {shard}"); + // Loss lands on exact `shard`-sized window boundaries: zeroed windows for the + // dropped datagrams, byte-identical survivors — nothing spliced across windows. + let mut zero_windows = 0; + for w in 0..8 { + let win = &p.data[w * shard..(w + 1) * shard]; + if win.iter().all(|&b| b == 0) { + zero_windows += 1; + } else { + assert_eq!( + win, + &frame[w * shard..(w + 1) * shard], + "shard {shard}: window {w} corrupt" + ); + } + } + assert!( + (1..8).contains(&zero_windows), + "shard {shard}: dropped shards zero-filled (got {zero_windows})" + ); + assert!( + completes > 40, + "shard {shard}: surviving filler frames flow normally" + ); + } + } + + /// Mid-session shard renegotiation end to end over the SEALED loopback wire + /// (design/shard-payload-reneg.md): one host session re-keys its packetizer between AUs + /// — shrink, jumbo grow, revert — through one continuous crypto/replay stream, and one + /// client session must DELIVER every frame byte-identically (the vacuous-green lesson: + /// assert delivered frames, never the absence of errors). + #[test] + fn mid_session_shard_swap_delivers_frames_over_the_sealed_wire() { + let mk = |role: Role| { + let mut c = host_cfg(FecScheme::Gf16, 20, true); + c.role = role; + c.shard_payload = 1408; + c.fec.max_data_per_block = 64; + c + }; + let (ht, ct) = loopback_pair(0, 0); + let mut host = Session::new(mk(Role::Host), Box::new(ht)).unwrap(); + let mut client = Session::new(mk(Role::Client), Box::new(ct)).unwrap(); + + let phases: [(usize, &[usize]); 4] = [ + (1408, &[3000, 3 * 1408]), // the negotiated default (incl. exact multiple) + (512, &[2000, 5 * 512 + 17]), // shrink — the mid-session VPN heal + (8908, &[100_000]), // grow — jumbo on a 9000-MTU LAN + (1216, &[2 * 1216 + 9]), // revert — a mis-proven jumbo hop self-corrects + ]; + let mut pts = 0u64; + let mut delivered = 0usize; + for (shard, lens) in phases { + host.set_shard_payload(shard).unwrap(); + assert_eq!(host.shard_payload(), shard); + for &len in lens { + pts += 1_000_000; + let src = pattern(len); + host.submit_frame(&src, pts, 0).unwrap(); + let f = client + .poll_frame() + .unwrap_or_else(|e| panic!("shard {shard}: frame must be DELIVERED ({e})")); + assert_eq!( + f.data, src, + "shard {shard}: {len} B frame must be byte-identical" + ); + assert!(f.complete); + delivered += 1; + } + } + assert_eq!(delivered, 6, "every submitted frame must be delivered"); + // The setter is host-side machinery: a client session must refuse it, and an + // invalid size (odd / oversized) must be rejected without touching the live config. + assert!(client.set_shard_payload(1408).is_err()); + assert!( + host.set_shard_payload(1407).is_err(), + "odd must be rejected" + ); + assert!( + host.set_shard_payload(crate::config::max_shard_payload() + 2) + .is_err(), + "oversized must be rejected" + ); + assert_eq!(host.shard_payload(), 1216, "failed swaps must not stick"); + } } diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 347b9610..46be92c8 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -1091,6 +1091,30 @@ async fn serve_session( // just never fires then. let (cursor_shape_tx, cursor_shape_rx) = tokio::sync::mpsc::unbounded_channel::(); + // Mid-session shard renegotiation (design/shard-payload-reneg.md Phase 2): the wire-MTU + // watcher decides (constrained-path shrink / ack-gated jumbo grow), the control task + // writes the `ShardPayloadChanged` and routes the acks back, and the data-plane loop + // applies `Session::set_shard_payload` between AUs (drained next to `bitrate_rx`). + // Channels are wired unconditionally (they just never fire); the DRIVER exists only for + // a client that advertised `Hello::max_shard_payload` on a non-chunk-aligned session — + // PyroWave clients parse chunk-aligned AUs in windows of the `Welcome` value pinned at + // session start (read once over the C ABI), so those sessions keep the leg-1 + // next-session clamp instead of a mid-stream re-key. + let (shard_change_tx, shard_change_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (shard_ack_tx, shard_ack_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (shard_apply_tx, shard_apply_rx) = std::sync::mpsc::channel::(); + let shard_reneg = (hello.max_shard_payload > 0 && codec != crate::encode::Codec::PyroWave) + .then_some(wire_mtu::ShardReneg { + client_ceiling: hello.max_shard_payload, + change_tx: shard_change_tx, + ack_rx: shard_ack_rx, + apply_tx: shard_apply_tx, + }); + // The session is real: watch this connection's MTU discovery settle and turn it into a + // path verdict (WARN + learned clamp for the next session on a constrained path; clears + // a stale clamp on a healthy one) — and, with the driver above, heal or grow THIS + // session mid-stream. Bounded ~10 s task unless a jumbo grow leaves it as revert guard. + wire_mtu::spawn_watch(conn.clone(), welcome.shard_payload as usize, shard_reneg); // Negotiated cursor forwarding: the HOST_CAP_CURSOR bit the Welcome advertised, read back // rather than recomputed (`handshake::cursor_forward` computed it once, with the encoder // blend-capability gate — re-running it here could drift, and would re-probe). @@ -1146,6 +1170,8 @@ async fn serve_session( probe_result_rx, reconfig_result_rx, retarget_rx, + shard_change_rx, + shard_ack_tx, cursor_shape_rx, cursor_client_draws, clip_enabled, @@ -1588,6 +1614,7 @@ async fn serve_session( keyframe: keyframe_rx, rfi: rfi_rx, bitrate_rx, + shard_rx: shard_apply_rx, compositor, gamescope_route, bitrate_kbps, diff --git a/crates/punktfunk-host/src/native/control.rs b/crates/punktfunk-host/src/native/control.rs index e14934c6..a6bdc9ca 100644 --- a/crates/punktfunk-host/src/native/control.rs +++ b/crates/punktfunk-host/src/native/control.rs @@ -43,6 +43,11 @@ pub(super) async fn run( // Host-initiated bitrate re-target (a rebuild re-resolved an Automatic rate): forwarded to // the client as a `BitrateChanged` so its controller's climb base tracks the real encoder. mut retarget_rx: tokio::sync::mpsc::UnboundedReceiver, + // Mid-session shard renegotiation (design/shard-payload-reneg.md): the wire-MTU watcher + // asks for a `ShardPayloadChanged` here (this task is the control stream's sole writer), + // and the client's `ShardPayloadAck`s flow back on `shard_ack_tx` — the grow gate. + mut shard_change_rx: tokio::sync::mpsc::UnboundedReceiver, + shard_ack_tx: tokio::sync::mpsc::UnboundedSender, mut cursor_shape_rx: tokio::sync::mpsc::UnboundedReceiver, cursor_client_draws: Arc, clip_enabled: Arc, @@ -56,6 +61,9 @@ pub(super) async fn run( // Set once `clip_offer_rx` closes (coordinator gone / inert handle) so its `select!` branch // stops firing on a perpetually-ready `None`. let mut clip_offer_closed = false; + // Same discipline for the wire-MTU watcher's channel — its bounded lifetime ends mid-session + // on every healthy path. + let mut shard_change_closed = false; let mut active = initial_mode; // Host-side switch rate limit (a backstop against a hostile/broken client spamming // Reconfigure into pipeline-rebuild churn — the drain-to-newest in the data plane already @@ -214,6 +222,16 @@ pub(super) async fn run( if bitrate_tx.send(resolved).is_err() { break; // data plane gone } + } else if let Ok(ack) = punktfunk_core::quic::ShardPayloadAck::decode(&msg) { + // Mid-session shard renegotiation: the client applied (or granted) a + // geometry change. Forward to the wire-MTU watcher — for a grow this IS + // the gate that lets the packetizer go above the old size. A dropped + // send just means the watcher already ended (shrink acks are telemetry). + tracing::info!( + shard_payload = ack.shard_payload, + "client acked shard-payload change" + ); + let _ = shard_ack_tx.send(ack.shard_payload); } else if let Ok(req) = ProbeRequest::decode(&msg) { tracing::info!( target_kbps = req.target_kbps, @@ -317,6 +335,19 @@ pub(super) async fn run( break; } } + n = shard_change_rx.recv(), if !shard_change_closed => { + // Mid-session shard renegotiation: the wire-MTU watcher decided (shrink on a + // constrained-path verdict / ack-gated jumbo grow). Only ever fires toward a + // client that advertised `Hello::max_shard_payload` — the watcher owns that + // gate. `None` = the watcher's bounded lifetime ended (normal, NOT a session + // end): disable this branch, exactly the `clip_offer_closed` pattern — a + // closed mpsc yields `None` perpetually and would busy-spin the select. + let Some(n) = n else { shard_change_closed = true; continue }; + let msg = punktfunk_core::quic::ShardPayloadChanged { shard_payload: n }; + if io::write_msg(&mut ctrl_send, &msg.encode()).await.is_err() { + break; + } + } shape = cursor_shape_rx.recv() => { // Cursor-forward bridge (M2): the encode loop diffed a new pointer bitmap. // Rare (shape changes are human-paced); ≤ ~58 KiB fits the u16 frame by diff --git a/crates/punktfunk-host/src/native/handshake.rs b/crates/punktfunk-host/src/native/handshake.rs index c93652df..a018e846 100644 --- a/crates/punktfunk-host/src/native/handshake.rs +++ b/crates/punktfunk-host/src/native/handshake.rs @@ -734,10 +734,9 @@ pub(super) async fn negotiate( let start = Start::decode(&io::read_msg(recv).await?).map_err(|e| anyhow!("Start decode: {e:?}"))?; bringup.mark("start"); - // The session is real: watch this connection's MTU discovery settle and turn it into a - // path verdict (WARN + learned clamp for the next session on a constrained path; clears a - // stale clamp on a healthy one). Bounded ~10 s task, ends by itself. - wire_mtu::spawn_watch(conn.clone(), welcome.shard_payload as usize); + // The wire-MTU watch (`wire_mtu::spawn_watch`) is spawned by `serve_session` after the + // control-task channels exist — it now also DRIVES the mid-session shard renegotiation + // (design/shard-payload-reneg.md), which needs the control stream's writer. Ok::<_, anyhow::Error>(( hello, welcome, diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 6bba243e..6047d088 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -762,6 +762,9 @@ fn send_loop( slice_wire: bool, burst_cap: Option, fec_target: Arc, + // Mid-session shard-payload re-keys from the wire-MTU watcher (validated + ack-gated + // there) — applied between AUs only (design/shard-payload-reneg.md Phase 1). + shard_rx: std::sync::mpsc::Receiver, stats: SendStats, // `Some` = the client advertised VIDEO_CAP_HOST_TIMING: emit one 0xCF datagram per AU right // after its last packet left the socket (capture→sent, the whole host pipeline incl. pacing). @@ -818,6 +821,25 @@ fn send_loop( } // Adaptive FEC: pick up any new recovery target the control task set from client LossReports. apply_fec_target(&mut session, &fec_target); + // Mid-session shard renegotiation: apply a re-key from the wire-MTU watcher — between + // AUs only, NEVER with a streamed AU open (its shard-aligned tiling derives from the + // size it began with; same gate as the probe burst above). Drain to the newest; the + // protocol side (client advertisement, ack-gated grow) was enforced by the watcher. + if streamed.is_none() { + let mut want_shard = None; + while let Ok(s) = shard_rx.try_recv() { + want_shard = Some(s); + } + if let Some(s) = want_shard { + match session.set_shard_payload(s) { + Ok(()) => tracing::info!(shard_payload = s, "wire shard payload re-keyed"), + // Can't fire for a watcher-driven value (it validates the same bounds) — + // belt-and-suspenders for a future driver. + Err(e) => tracing::warn!(shard_payload = s, error = ?e, + "shard re-key refused by session validation"), + } + } + } // 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(send_msg) => { @@ -1171,6 +1193,11 @@ pub(super) struct SessionContext { /// Accepted mid-stream bitrate changes (adaptive bitrate, already clamped) — the encoder /// alone is rebuilt in place at the new rate; capture + virtual output are untouched. pub(super) bitrate_rx: std::sync::mpsc::Receiver, + /// Mid-session shard-payload changes from the wire-MTU watcher (already validated + + /// protocol-gated there; a grow arrives only after the client's ack). Applied between + /// AUs via [`Session::set_shard_payload`] — the packetizer re-keys, capture/encoder/ + /// virtual output are untouched (design/shard-payload-reneg.md Phase 1). + pub(super) shard_rx: std::sync::mpsc::Receiver, /// The resolved compositor backend (moot on Windows — `vdisplay::open` ignores it there). pub(super) compositor: crate::vdisplay::Compositor, /// This session's resolved gamescope sub-mode, or `None` for every other backend. Carried here @@ -1385,6 +1412,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option 0 by construction). + pub client_ceiling: u16, + /// → control task (the control stream's sole writer): send `ShardPayloadChanged{n}`. + pub change_tx: tokio::sync::mpsc::UnboundedSender, + /// ← control task: the client's `ShardPayloadAck`s (the grow gate). + pub ack_rx: tokio::sync::mpsc::UnboundedReceiver, + /// → data plane: apply [`Session::set_shard_payload`] between AUs + /// (drained next to `bitrate_rx` in the encode loop). + pub apply_tx: std::sync::mpsc::Sender, +} + /// Measured UDP-payload budget per peer IP, learned from live control connections whose MTU /// discovery settled below the video-datagram ceiling. In-memory only: a host restart /// re-learns in one session, and entries self-correct (a later ceiling-hit erases, a lower @@ -96,15 +116,26 @@ fn resolve(env_wire_mtu: Option, learned_udp_budget: Option, peer: I } /// Sample the control connection's discovered MTU after the search has settled and turn it -/// into a verdict. Spawned once per negotiated session; the task ends by itself after the -/// final sample (bounded ~10 s lifetime, holding only a cheap `Connection` handle). -pub(super) fn spawn_watch(conn: quinn::Connection, session_shard_payload: usize) { +/// into a verdict — and, with a [`ShardReneg`] driver, act on it MID-SESSION +/// (design/shard-payload-reneg.md Phase 2): a below-ceiling verdict shrinks the live wire at +/// the ~3–10 s mark (session 1 heals instead of staying black), and a settled-at-jumbo +/// verdict grows it, ack-gated, when the operator opted in. Spawned once per negotiated +/// session; without a grow the task ends after the final sample (bounded ~10 s lifetime, +/// holding only a cheap `Connection` handle) — after a grow it stays as the revert guard +/// until the connection closes. +pub(super) fn spawn_watch( + conn: quinn::Connection, + session_shard_payload: usize, + reneg: Option, +) { tokio::spawn(async move { let peer = conn.remote_address().ip(); let ceiling = video_datagram_udp_ceiling() as u16; // Discovery finishes in a handful of RTTs on a LAN (well under the first sample) but // needs a loss timeout per failed probe on a constrained path — the second sample - // covers that with margin. Max, because discovery only ever raises `current_mtu`. + // covers that with margin. Max, because discovery only ever raises `current_mtu` + // (the post-grow revert guard below re-reads it live, where blackhole detection CAN + // lower it again). let mut settled = 0u16; for wait_s in [3u64, 7] { tokio::time::sleep(std::time::Duration::from_secs(wait_s)).await; @@ -113,6 +144,9 @@ pub(super) fn spawn_watch(conn: quinn::Connection, session_shard_payload: usize) break; } } + // The wire this session is CURRENTLY sealed at — moves on a mid-session shrink/grow. + let mut current = session_shard_payload; + let mut reneg = reneg; if settled >= ceiling { // The path carries full-size video datagrams — erase any stale learned clamp so // the next session returns to the default wire. @@ -120,34 +154,116 @@ pub(super) fn spawn_watch(conn: quinn::Connection, session_shard_payload: usize) tracing::info!(peer = %peer, "wire MTU: path re-measured at full size — learned clamp cleared"); } - return; - } - // A closed connection stops discovering, so a session that ended before the final - // sample proves nothing (a healthy high-RTT path could still be mid-search): learn - // only from a connection that stayed alive through the whole window. - if conn.close_reason().is_some() { - return; - } - learned().lock().unwrap().insert(peer, settled); - if sealed_datagram_bytes(session_shard_payload) <= settled as usize { - // This session was already clamped small enough — the path is still constrained - // (keep the record fresh) but video fits, so no alarm. - tracing::info!(peer = %peer, discovered_udp_mtu = settled, - "wire MTU: constrained path re-measured; this session's video is sized to fit"); } else { - tracing::warn!( - peer = %peer, - discovered_udp_mtu = settled, - needed_udp_mtu = ceiling, - "wire MTU: this path CANNOT carry full-size video datagrams — the control \ - plane works but every video packet is oversized for a hop, which streams as \ - an endless black screen with zero reported loss. Typical cause: a VPN/overlay \ - adapter (Tailscale / Cloudflare WARP / ZeroTier) claiming the LAN route, or a \ - lowered NIC MTU — compare `ping -f -l 1450` vs `-l 1200` and check \ - `netsh interface ipv4 show subinterfaces` (Windows) / `ip link` (Linux). The \ - measured budget is recorded: the NEXT session from this client sizes video to \ - fit automatically. To pin it for all sessions set PUNKTFUNK_WIRE_MTU." - ); + // A closed connection stops discovering, so a session that ended before the final + // sample proves nothing (a healthy high-RTT path could still be mid-search): learn + // only from a connection that stayed alive through the whole window. + if conn.close_reason().is_some() { + return; + } + learned().lock().unwrap().insert(peer, settled); + if sealed_datagram_bytes(current) <= settled as usize { + // This session was already clamped small enough — the path is still constrained + // (keep the record fresh) but video fits, so no alarm. + tracing::info!(peer = %peer, discovered_udp_mtu = settled, + "wire MTU: constrained path re-measured; this session's video is sized to fit"); + } else { + tracing::warn!( + peer = %peer, + discovered_udp_mtu = settled, + needed_udp_mtu = ceiling, + "wire MTU: this path CANNOT carry full-size video datagrams — the control \ + plane works but every video packet is oversized for a hop, which streams as \ + an endless black screen with zero reported loss. Typical cause: a VPN/overlay \ + adapter (Tailscale / Cloudflare WARP / ZeroTier) claiming the LAN route, or a \ + lowered NIC MTU — compare `ping -f -l 1450` vs `-l 1200` and check \ + `netsh interface ipv4 show subinterfaces` (Windows) / `ip link` (Linux). The \ + measured budget is recorded: the NEXT session from this client sizes video to \ + fit automatically. To pin it for all sessions set PUNKTFUNK_WIRE_MTU." + ); + // Phase 2 down-leg: heal THIS session at the verdict mark. Shrink is sent + // and applied immediately — per-frame pinning on the client makes ordering + // irrelevant and smaller always fits; the ack is telemetry. The learned + // record above still makes session 2 START right. + if let Some(r) = reneg.as_ref() { + let target = shard_payload_for_udp_budget(settled as usize, peer); + if target < current + && r.change_tx.send(target as u16).is_ok() + && r.apply_tx.send(target).is_ok() + { + tracing::info!( + peer = %peer, + shard_payload = target, + was = current, + "wire MTU: video re-keyed mid-session to fit the constrained path \ + — the stream heals now instead of on the next connect" + ); + current = target; + } + } + } + } + // Phase 2 up-leg: jumbo grow — operator opt-in (PUNKTFUNK_JUMBO / PUNKTFUNK_WIRE_MTU + // > 1500, which also raised the endpoint's probe ceiling so `settled` can even reach + // here), client-advertised headroom, and a settled-at-jumbo proof. The grow is + // ACK-GATED: not one sealed datagram above the old size leaves before the client's + // ack, even though its buffers are statically sized — the rule must not erode. + let (Some(mtu), Some(r)) = (jumbo_wire_mtu(), reneg.as_mut()) else { + return; + }; + let target = jumbo_shard_payload_for(mtu, peer).min(r.client_ceiling as usize); + let target = target - target % 2; + if target <= current || (settled as usize) < sealed_datagram_bytes(target) { + return; + } + if r.change_tx.send(target as u16).is_err() { + return; + } + let acked = tokio::time::timeout(std::time::Duration::from_secs(5), async { + while let Some(v) = r.ack_rx.recv().await { + if v as usize == target { + return true; + } + } + false + }) + .await + .unwrap_or(false); + if !acked { + tracing::warn!(peer = %peer, shard_payload = target, + "wire MTU: jumbo grow not acked — staying at the current wire"); + return; + } + if r.apply_tx.send(target).is_err() { + return; + } + tracing::info!( + peer = %peer, + shard_payload = target, + was = current, + wire_mtu = mtu, + "wire MTU: jumbo grow acked and applied — packets-per-frame cut ~6×" + ); + current = target; + // Revert guard: a mis-proven jumbo hop must self-correct instead of blackholing. + // quinn's PMTU blackhole detection lowers `current_mtu` when the big packets start + // vanishing; sample it and shrink back through the same path the down-leg uses. + loop { + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + if conn.close_reason().is_some() { + return; + } + let mtu_now = conn.stats().path.current_mtu; + if (mtu_now as usize) < sealed_datagram_bytes(current) { + let back = shard_payload_for_udp_budget(mtu_now as usize, peer); + tracing::warn!(peer = %peer, discovered_udp_mtu = mtu_now, + shard_payload = back, was = current, + "wire MTU: jumbo path stopped fitting — reverting the wire to match"); + if r.change_tx.send(back as u16).is_err() || r.apply_tx.send(back).is_err() { + return; + } + current = back; + } } }); } diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 50155777..62d53b32 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -488,7 +488,17 @@ // Largest UDP datagram the core will send or accept. `Config::validate` bounds // `shard_payload` so `HEADER_LEN + shard_payload + CRYPTO_OVERHEAD ≤ MAX_DATAGRAM_BYTES`. -#define MAX_DATAGRAM_BYTES 2048 +// +// Sized for **jumbo frames** (design/shard-payload-reneg.md W0.2): a 9000-MTU LAN carries +// ~8908-byte shards (sealed 8972-byte UDP payloads), and every receive path — the transport +// `RECV_BUF`, the session's `recvmmsg` ring — is sized from this constant, so a deployed +// client can accept a jumbo geometry the moment its host negotiates one. The ring cost is +// 128 × ~9 KiB ≈ 1.1 MiB per **client** session (lazily allocated on first poll; hosts never +// allocate it) — measured against the ~256 KiB it was at 2048, an acceptable static price +// for never having to resize buffers on a mid-session grow. Senders still derive their +// shard payload from the path MTU (`config::mtu1500_shard_payload*`, the wire-MTU clamps); +// this is the acceptance ceiling, not a transmit size. +#define MAX_DATAGRAM_BYTES 9216 // 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 @@ -816,6 +826,16 @@ #define MSG_RFI_REQUEST 7 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`ShardPayloadChanged`]. +#define MSG_SHARD_PAYLOAD_CHANGED 8 +#endif + +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`ShardPayloadAck`]. +#define MSG_SHARD_PAYLOAD_ACK 9 +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // Type byte of [`ProbeRequest`]. #define MSG_PROBE_REQUEST 32