unom/punktfunk
6
0
Fork 1
Code Issues 3 Pull Requests 2 Actions 1 Packages Projects Releases 44 Wiki Activity

fix(core/packet): a slice-streamed frame costs its own size, not the whole frame ceiling

Browse Source 
The 0.23.0 slice wire flushes a block every MIN_STREAM_BLOCK_SHARDS, so every
ordinary access unit is now opened by a SENTINEL — a header with no totals. The
reassembler sized those frames at `max_frame_bytes`, which the QUIC handshake
clamps to 8-64 MiB. That was survivable while sentinels were rare (the streamed
path emitted one only for an AU exceeding a whole FEC block, ~281 KB); it is not
survivable now that every frame is one.

Two consequences, both measured: each access unit allocated and ZEROED a
multi-megabyte buffer, and the in-flight budget (IN_FLIGHT_BUF_FACTOR x
max_frame_bytes) was spent after ~3 concurrent frames — with production geometry,
12 ordinary AUs in flight lost 9 of them outright, every packet dropped before it
could be placed. On a link with normal reorder that is a permanent loss storm:
frames never complete, the re-anchor gate freezes the picture, and the client begs
for keyframes. Only clients advertising VIDEO_CAP_MULTI_SLICE reach this path —
Android and the Linux/Windows session client; Apple and the Windows in-process
client never did, which is why it read as a platform-specific "video pipeline"
fault in the field.

A sentinel carries no total but does pin its own block's extent: a slice sentinel
by its wire base, a legacy one by its full-K position. Size the buffer to that and
grow as later blocks (or the final block's totals) reveal more. The budget is
re-checked on growth for the same reason it is checked at open.

The same flush also drained `pending` to empty whenever the AU's length was an
exact multiple of the shard payload, leaving `finish_streamed` to seal a final
block of one zero-padded FILLER shard. Its derived base overlapped the block
flushed a moment earlier, retro-validation correctly read that as a lying header,
and the whole AU died — one frame in every 1408 on a 1500-MTU link, ~12 s apart at
120 fps, each costing a freeze and a recovery keyframe. A flush now keeps one
whole shard back, restoring the invariant `StreamedAu::pending` already documented.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
enricobuehler
2026-08-02 20:15:45 +02:00
co-authored by Claude Opus 5
parent 29473d6280
commit 97b2c01ac1
3 changed files with 223 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/punktfunk-core/src/packet/packetize.rs
+15 -4
View File
@@ -73,9 +73,9 @@ pub struct StreamedAu {
pts_ns: u64,
user_flags: u32,
/// Bytes not yet sealed into a block: the sub-shard remainder plus anything below the
/// slice-flush threshold. The final block always has ≥ 1 byte (flushes emit only whole
/// shards and never drain to empty on a slice that ends the AU — `finish_streamed` seals
/// whatever remains).
/// slice-flush threshold. The final block always has ≥ 1 byte — flushes emit only whole
/// shards, and a flush that WOULD empty this keeps one shard back (see `push_streamed`),
/// so `finish_streamed` always has something real to seal.
pending: Vec<u8>,
/// Sentinel blocks already emitted.
blocks_out: u16,
@@ -418,7 +418,18 @@ impl Packetizer {
"streamed AU exceeds the negotiated max_frame_bytes",
));
}
let k = whole.min(self.fec.max_data_per_block as usize);
// Never drain `pending` to EMPTY. [`finish_streamed`] must have bytes left to seal,
// or the final block degenerates to a single zero-padded filler shard whose derived
// base (`total_data − 1`) overlaps the block flushed just now — which the receiver's
// retro-validation correctly reads as a lying header and kills the whole AU. It bites
// exactly when the AU's length is a multiple of `shard_payload` (~1 in 1408 frames on
// a 1500-MTU link), and only on the slice arm: the legacy `must_flush` is a strict
// `>`, so its remainder is never empty. Keeping one whole shard back costs nothing —
// it rides out in the final block, which has to exist regardless.
let mut k = whole.min(self.fec.max_data_per_block as usize);
if k > 1 && k == whole && au.pending.len() == whole * payload {
k -= 1;
}
let sof = !au.opened;
let (bi, pts, uf) = (au.blocks_out, au.pts_ns, au.user_flags);
let fi = au.frame_index;
crates/punktfunk-core/src/packet/reassemble.rs
+41 -7
View File
@@ -467,14 +467,33 @@ impl Reassembler {
return Ok(None);
}
// First packet of a frame allocates its whole (zeroed) buffer, budget-gated; later
// packets must agree with its geometry. A sentinel-opened (streamed) frame allocates at
// the limits' maximum — its real size doesn't exist yet.
let buf_len = if sentinel {
total_data_max * shard_bytes
// How many shards of frame buffer THIS packet proves the frame needs. A sentinel carries
// no total, but it does pin its own block's extent — a slice sentinel by its wire base,
// a legacy one by its full-K position — and that is what the buffer must cover to place
// the shard. The frame grows as later blocks reveal more, and the final (non-sentinel)
// block's totals settle it.
//
// ⚠ NOT `total_data_max` (= the negotiated `max_frame_bytes`, 8-64 MiB): that shape
// shipped in 0.23.0 and was survivable only while sentinels were rare — the streamed
// path emitted one solely for an AU exceeding a whole FEC block (~281 KB). The slice
// wire flushes at `MIN_STREAM_BLOCK_SHARDS`, so EVERY ordinary AU became sentinel-opened
// and every one of them committed the full ceiling: a multi-megabyte zeroed allocation
// per access unit, and an in-flight budget (`IN_FLIGHT_BUF_FACTOR × max_frame_bytes`)
// exhausted after ~3 concurrent frames — beyond which every packet of every further
// frame was dropped outright. On a jittery link that is a permanent loss storm.
let need_shards = if sentinel && slice_stream {
frame_bytes / shard_bytes + data_shards
} else if sentinel {
// Legacy sentinels are full-K uniform blocks (firewall-enforced), so the block's
// index alone gives its end.
(block_idx + 1).saturating_mul(lim.max_data_shards)
} else {
total_data * shard_bytes
};
total_data
}
.min(total_data_max);
// First packet of a frame allocates its (zeroed) buffer, budget-gated; later packets must
// agree with its geometry.
let buf_len = need_shards * shard_bytes;
let frame = match win.frames.entry(hdr.frame_index) {
std::collections::hash_map::Entry::Occupied(e) => e.into_mut(),
std::collections::hash_map::Entry::Vacant(e) => {
@@ -602,6 +621,21 @@ impl Reassembler {
drop(stats);
return Ok(None);
}
// Grow to this packet's proven extent. A streamed frame opens at whichever block arrived
// first and learns its real size from the final block's totals (or a later, higher
// sentinel base) — reorder means either can come first, so the buffer is sized by
// whatever the frame has proven so far. Never shrinks: the totals only settle the frame's
// END, and completion truncates to `frame_bytes` anyway. The budget is re-checked here
// for exactly the reason it is checked at open — growth commits memory too.
if buf_len > frame.buf.len() {
let delta = buf_len - frame.buf.len();
if *in_flight_bytes + delta > IN_FLIGHT_BUF_FACTOR * lim.max_frame_bytes {
drop(stats);
return Ok(None);
}
*in_flight_bytes += delta;
frame.buf.resize(buf_len, 0);
}
let FrameBuf {
buf,
blocks,
crates/punktfunk-core/src/packet/tests.rs
+167 -10
View File
@@ -941,8 +941,9 @@ fn slice_config() -> Config {
/// Slice chunks chosen to exercise every packetizer path: an exact-shard slice, a slice with
/// a sub-shard remainder, a slice below [`MIN_STREAM_BLOCK_SHARDS`] that must accumulate,
/// and a finish tail. 1023 B total → blocks (K, base-shard): (20, 0), (25, 20), (18, 45),
/// final (1, 63) with block_count 4.
/// and a finish tail. 1023 B total → blocks (K, base-shard): (19, 0), (26, 19), (18, 45),
/// final (1, 63) with block_count 4. Chunk 0 is an exact 20-shard multiple and flushes 19:
/// a flush never drains `pending` to empty, so `finish_streamed` always seals real bytes.
fn slice_chunks() -> Vec<Vec<u8>> {
[320usize, 403, 100, 200]
.iter()
@@ -1007,7 +1008,8 @@ fn slice_streamed_wire_shape_and_roundtrip() {
assert_eq!(src.len(), 1023);
// (block_index, K, base bytes) — chunk 2 (100 B) accumulated instead of flushing (6
// whole shards < MIN_STREAM_BLOCK_SHARDS) and rode into block 2 with chunk 3's bytes.
let expect = [(0u16, 20u16, 0u32), (1, 25, 320), (2, 18, 720)];
// Block 0 keeps one shard back (chunk 0 is an exact multiple), which rides into block 1.
let expect = [(0u16, 19u16, 0u32), (1, 26, 304), (2, 18, 720)];
for p in &pkts {
let h = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap();
assert_ne!(
@@ -1478,15 +1480,24 @@ fn parts_flow_for_legacy_streamed_frames() {
assert!(got.last().unwrap().complete);
}
/// A sentinel first-packet commits a MAX-sized frame buffer, so the in-flight budget must
/// bite after IN_FLIGHT_BUF_FACTOR frames — the amplification bound for one-datagram opens.
/// A one-datagram open commits only the buffer its OWN header proves it needs, and the
/// in-flight budget still bounds the ones that claim a lot.
///
/// Both halves matter. A sentinel that claims little must cost little: sizing every
/// sentinel-opened frame at `max_frame_bytes` (the 0.23.0 shape) was survivable only while
/// sentinels were rare, and the slice wire made every ordinary AU one — after which the budget
/// was spent on ~3 frames and everything else on the link was dropped. A sentinel that claims a
/// lot must still be bounded: its wire base can point near the frame ceiling, which is the
/// amplification this budget exists for.
#[test]
fn streamed_open_amplification_is_budget_bounded() {
let mut r = Reassembler::new(limits());
fn streamed_open_commits_its_own_extent_and_stays_bounded() {
let coder = coder_for(FecScheme::Gf8);
// limits(): shard 16 B, max_data_shards 8, max_frame_bytes 4096 → budget = 4 × 4096.
// Modest legacy sentinels (block 0, full K = 8 → 128 B each): far more than
// IN_FLIGHT_BUF_FACTOR of them must fit, because none of them claims the ceiling.
let mut r = Reassembler::new(limits());
let stats = StatsCounters::default();
// limits(): max_frame_bytes 4096 → each sentinel open commits 4096 B; budget = 4×4096.
for fi in 0..5u32 {
for fi in 0..32u32 {
let mut h = base_header();
h.block_count = 0;
h.frame_bytes = 0;
@@ -1498,10 +1509,35 @@ fn streamed_open_amplification_is_budget_bounded() {
.unwrap()
.is_none());
}
assert_eq!(
stats.snapshot().packets_dropped,
0,
"ordinary one-datagram opens must not exhaust the in-flight budget"
);
// A SLICE sentinel whose wire base sits just under the ceiling really does commit a
// max-sized frame (base 3968 B + K 8 = 256 shards = 4096 B) — four fit the budget, the
// fifth must be refused.
let mut r = Reassembler::new(limits());
let stats = StatsCounters::default();
for fi in 0..5u32 {
let mut h = base_header();
h.user_flags = USER_FLAG_SLICE_STREAM;
h.block_count = 0;
h.frame_bytes = 4096 - 8 * 16;
h.block_index = 1;
h.data_shards = 8;
h.recovery_shards = 0;
h.frame_index = fi;
assert!(r
.push(&packet(h), coder.as_ref(), &stats)
.unwrap()
.is_none());
}
assert_eq!(
stats.snapshot().packets_dropped,
1,
"the fifth max-sized open must be refused by the in-flight budget"
"the fifth ceiling-claiming open must be refused by the in-flight budget"
);
}
@@ -1612,3 +1648,124 @@ fn streamed_second_final_with_different_totals_is_rejected() {
.expect("frame completes under the first pinned totals");
assert_eq!(got.data.len(), 160);
}
/// Production-shaped slice geometry: a 1500-MTU shard payload and the smallest frame ceiling
/// the QUIC handshake ever negotiates (`max_frame_bytes` is clamped to ≥ 8 MiB there).
fn prod_slice_config() -> Config {
use crate::config::{FecConfig, ProtocolPhase, Role};
Config {
role: Role::Host,
phase: ProtocolPhase::P2Punktfunk,
fec: FecConfig {
scheme: FecScheme::Gf16,
fec_percent: 20,
max_data_per_block: 200,
},
shard_payload: crate::config::mtu1500_shard_payload(),
max_frame_bytes: 8 << 20,
encrypt: false,
key: SessionKey::Aes128Gcm([0u8; 16]),
salt: [0u8; 4],
loopback_drop_period: 0,
}
}
/// Packetize one streamed AU of `chunks`, each chunk an encoder slice boundary.
fn streamed_packets_with(
cfg: &Config,
frame_index: u32,
pts_ns: u64,
slice: bool,
chunks: &[usize],
) -> (Vec<Vec<u8>>, Vec<u8>) {
let coder = coder_for(cfg.fec.scheme);
let mut pk = Packetizer::new(cfg);
let uf = if slice { USER_FLAG_SLICE_STREAM } else { 0 };
let mut au = pk.begin_streamed(pts_ns, uf, Some(frame_index));
let (mut pkts, mut src) = (Vec::new(), Vec::new());
let sink = |pkts: &mut Vec<Vec<u8>>, h: &PacketHeader, b: &[u8]| {
let mut p = Vec::with_capacity(HEADER_LEN + b.len());
p.extend_from_slice(h.as_bytes());
p.extend_from_slice(b);
pkts.push(p);
};
for (c, &n) in chunks.iter().enumerate() {
let data: Vec<u8> = (0..n).map(|i| (c * 57 + i * 131 + 7) as u8).collect();
src.extend_from_slice(&data);
pk.push_streamed(&mut au, &data, true, coder.as_ref(), |h, b| {
sink(&mut pkts, h, b);
Ok(())
})
.unwrap();
}
pk.finish_streamed(au, coder.as_ref(), |h, b| {
sink(&mut pkts, h, b);
Ok(())
})
.unwrap();
(pkts, src)
}
/// An AU whose length is an exact multiple of the shard payload must still reassemble.
///
/// Regression: the slice flush drained `pending` to empty, so `finish_streamed` sealed a final
/// block of one zero-padded FILLER shard. Its derived base (`total_data − 1`) overlapped the
/// sentinel block flushed a moment earlier, the receiver's retro-validation read that as a lying
/// header, and the whole AU was destroyed — one frame in every `shard_payload` (~12 s at 120 fps),
/// each costing a re-anchor freeze and a recovery keyframe.
#[test]
fn slice_streamed_exact_shard_multiple_completes() {
let cfg = prod_slice_config();
let coder = coder_for(FecScheme::Gf16);
let payload = cfg.shard_payload;
for shards in [16usize, 29, 30, 64] {
let (pkts, src) = streamed_packets_with(&cfg, 1, 1000, true, &[shards * payload]);
// Whatever the block split, the final block must carry real bytes — never a lone
// zero-pad shard sitting on top of the previous block's range.
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!("{shards}-shard AU (exact multiple) must complete"));
assert_eq!(f.data, src, "{shards}-shard AU must be byte-identical");
}
// ...and the sweep around one of them, so an off-by-one in the keep-back can't hide.
for extra in 0..3usize {
let n = 30 * payload + extra;
let (pkts, src) = streamed_packets_with(&cfg, 2, 2000, 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!("{n}-byte AU must complete"));
assert_eq!(f.data, src);
}
}
/// A slice-streamed frame must cost the reassembler its OWN size, not the negotiated ceiling.
///
/// Regression: sentinel-opened frames allocated `max_frame_bytes` (8-64 MiB) each. Since the
/// slice wire makes every ordinary AU sentinel-opened, the in-flight budget
/// (`IN_FLIGHT_BUF_FACTOR × max_frame_bytes`) was spent after ~3 concurrent frames and every
/// packet of every further frame was dropped outright — a permanent loss storm on any link with
/// normal reorder, plus a multi-megabyte zeroing per access unit.
#[test]
fn slice_streamed_in_flight_budget_matches_legacy() {
let cfg = prod_slice_config();
let coder = coder_for(FecScheme::Gf16);
// A normal 40 KB access unit, opened but not completed — the shape a link with reorder
// holds several of at once.
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,
"slice={slice}: 12 ordinary AUs in flight must fit the in-flight budget"
);
}
}