feat(core/wire): per-frame shard geometry, jumbo ceiling, Hello advertisement
Phase 0 of mid-session shard-payload renegotiation (planning design/shard-payload-reneg.md), stacked on the leg-1 MTU resilience. All three legs are client-side and forward-compatible: deployed clients that carry them accept a mid-session shard change the moment a future host sends one, and nothing changes on the wire until then. - W0.1 — the reassembler's strict shard_bytes firewall becomes per-frame pinning: a frame's first-arriving packet pins that frame's shard size (bounds-checked to [min_shard_bytes, max_shard_bytes], even), later packets must match the pin, and the per-frame block ceiling derives from the pinned size (a session-level cap would reject legitimate post-shrink frames). The reorder race between an ordered control message and unordered video dies structurally: old-geometry 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. The in-flight budget stays byte-based and exact. - W0.2 — MAX_DATAGRAM_BYTES 2048 → 9216: every receive path (transport RECV_BUF, the recvmmsg ring) now accepts sealed jumbo datagrams (9000-MTU LAN ≈ 8908-byte shards). Static buffers over resize-on-ack: the ring delta is 128 × ~7 KiB ≈ 896 KiB per client session, lazily allocated, hosts unaffected. Grep verdict: no embedder uses the constant directly, so no C ABI bump — the regenerated header rides along (drift gate). - W0.3 — trailing Hello field max_shard_payload: u16 (0/absent = legacy), the append-with-placeholder discipline of video_caps/ client_caps. One field is both the renegotiation capability flag and the jumbo ceiling; core's pump advertises it for all client families, the probe too. - Host seam for Phase 1, dead until wired: Packetizer::set_shard_payload (re-derives the block ceilings; construction delegates to it) + Session::set_shard_payload (host-only, Config::validate parity). Verification (the 0.23.0 lesson — geometry changes breed sizing bugs): the slice-wire suite re-runs at shard 512/1216/1408/8908 (exact-multiple sweep, lossy + reversed roundtrips, sentinel path, in-flight budget); mid-stream shrink→grow→revert delivery; the old-geometry reorder race; cross-geometry splice rejection; firewall bounds non-vacuous both ways; a 48-case mixed-geometry reorder-torture proptest asserting per-frame byte-identical DELIVERY and an exactly-zero final budget; and a sealed loopback session test (continuous crypto/replay) delivering frames across live re-keys — every test asserts delivered frames, never the absence of errors. core: 294/294 --features quic + clippy -D warnings (macOS), fmt.
This commit is contained in:
@@ -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(),
|
||||
)
|
||||
|
||||
@@ -156,6 +156,12 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result<Handshake
|
||||
// stop compositing the pointer, so only an embedder that actually renders the
|
||||
// cursor locally may set it (the embedder decides, we pass through).
|
||||
client_caps: args.client_caps,
|
||||
// Unconditional like STREAMED_AU: the shared reassembler pins geometry
|
||||
// per-frame and every receive buffer is sized from MAX_DATAGRAM_BYTES, so
|
||||
// every embedder accepts a mid-session shard change up to this ceiling
|
||||
// (design/shard-payload-reneg.md W0.3 — the host only renegotiates, and only
|
||||
// grows to jumbo, when this advertises it).
|
||||
max_shard_payload: crate::config::max_shard_payload() as u16,
|
||||
}
|
||||
.encode(),
|
||||
)
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<u8>>, Vec<u8>) {
|
||||
let src: Vec<u8> = (0..len)
|
||||
.map(|i| (i * 131 + frame_index as usize * 7 + 3) as u8)
|
||||
.collect();
|
||||
let mut pkts: Vec<Vec<u8>> = 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<Vec<u8>> = 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<Vec<u8>> = 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<Value = GenFrame> {
|
||||
(
|
||||
proptest::sample::select(&PRODUCTION_SHARDS[..]),
|
||||
any::<bool>(),
|
||||
1usize..30,
|
||||
any::<bool>(),
|
||||
)
|
||||
}
|
||||
|
||||
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::<u64>(),
|
||||
) {
|
||||
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<u8>)> = Vec::new(); // (shuffle key, frame, pkt)
|
||||
let mut sources: Vec<(u32, Vec<u8>)> = 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<u32, Vec<u8>> =
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,64 @@ mod wire_equivalence_tests {
|
||||
"unflagged AUs must never be delivered partial"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -481,7 +481,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
|
||||
|
||||
Reference in New Issue
Block a user