From 76ac3cf8673e335ba6bda14ed6de1dd686557363 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 20 Jul 2026 18:16:02 +0200 Subject: [PATCH] =?UTF-8?q?fix(core):=20five=20sweep=20lows=20=E2=80=94=20?= =?UTF-8?q?seal-lane=20fallback,=20flush=20partial,=20codec=20echo,=20idle?= =?UTF-8?q?=20clamp,=20pair-name=20UTF-8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From the 2026-07-20 punktfunk-core quality sweep's adjudicated lows (~/punktfunk-sweeps/punktfunk-core-2026-07-20.json), each with a regression test: - session: the two-lane seal's dead-worker fallback dropped the frame's back half and returned Ok with half an access unit; the corpse lane was also never respawned. The send-failure arm now reclaims the tail (single-lane seals the WHOLE frame) and both failure arms drop the lane so the next large frame respawns it. A recv-side death now surfaces as an error instead of silent truncation. - reassemble: Reassembler::reset() left pending_partial parked, so a pre-flush stale partial survived Session::flush_backlog and was delivered as the first "frame" after a jump-to-live. - caps: resolve_codec echoed a non-conformant multi-bit preferred byte verbatim (downstream from_wire folds it to HEVC — possibly outside the shared set). It now isolates one bit of the intersection. - endpoint: stream_transport_idle only floored the value; an absurd operator-supplied idle timeout blew past QUIC's VarInt ms ceiling and panicked host startup through the expect. Clamped to 1s..1h. - pairing: PairRequest::encode cut the device name at a raw byte-64 boundary, splitting multi-byte UTF-8 (host showed U+FFFD forever); it now shares Hello's char-boundary truncate_to. The frozen-FEC-ceiling finding (reassemble.rs:109) was already fixed on main by the sweep's high-severity commit — skipped as stale. Co-Authored-By: Claude Fable 5 --- .../punktfunk-core/src/packet/reassemble.rs | 36 +++++++ crates/punktfunk-core/src/quic/caps.rs | 27 +++++- crates/punktfunk-core/src/quic/endpoint.rs | 15 ++- crates/punktfunk-core/src/quic/handshake.rs | 5 +- crates/punktfunk-core/src/quic/pairing.rs | 25 ++++- crates/punktfunk-core/src/session.rs | 93 +++++++++++++++---- 6 files changed, 174 insertions(+), 27 deletions(-) diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index b91f6b14..fda0de27 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -518,6 +518,10 @@ impl Reassembler { // The dropped frames' buffers (and their parity bufs) go back to the allocator, not the // pool — a flush is the rare path. The budget resets with them. self.in_flight_bytes = 0; + // An aged-out partial parked for delivery is from the discarded past too — without this + // it survives `flush_backlog` and gets handed up as the first "frame" after the + // jump-to-live, exactly the stale content the flush existed to discard. + self.pending_partial = None; } } @@ -645,3 +649,35 @@ impl ReassemblyWindow { } } } + +#[cfg(test)] +mod reset_tests { + use super::*; + + /// `flush_backlog` discards the past wholesale — an aged-out partial parked for delivery is + /// part of that past and must not survive [`Reassembler::reset`] to be handed up as the + /// first "frame" after a jump-to-live. + #[test] + fn reset_drops_a_parked_partial() { + let mut r = Reassembler::new(ReassemblerLimits { + shard_bytes: 64, + max_data_shards: 8, + max_total_shards: 16, + max_blocks: 4, + max_frame_bytes: 4096, + }); + r.pending_partial = Some(Frame { + data: vec![0u8; 64], + frame_index: 7, + pts_ns: 1, + flags: 0, + complete: false, + received_ns: 0, + }); + r.reset(); + assert!( + r.take_partial().is_none(), + "a pre-flush partial must not survive reset()" + ); + } +} diff --git a/crates/punktfunk-core/src/quic/caps.rs b/crates/punktfunk-core/src/quic/caps.rs index 5c3d3956..46bce69d 100644 --- a/crates/punktfunk-core/src/quic/caps.rs +++ b/crates/punktfunk-core/src/quic/caps.rs @@ -91,8 +91,13 @@ pub fn resolve_codec(client_codecs: u8, host_capable: u8, preferred: u8) -> Opti return None; } // Honor the client's preference when the host can also emit it; else fall back to precedence. + // `preferred` is a single-bit field by contract but arrives as a raw wire byte — isolate ONE + // bit of the intersection instead of echoing the request, so a non-conformant multi-bit + // value can never escape as a codec id (downstream `from_wire` folds unknown values to HEVC, + // which may not even be in the shared set). if preferred != 0 && shared & preferred != 0 { - return Some(preferred); + let want = shared & preferred; + return Some(want & want.wrapping_neg()); } // Precedence: HEVC > AV1 > H.264. [CODEC_HEVC, CODEC_AV1, CODEC_H264] @@ -222,4 +227,24 @@ mod tests { 0 ); } + + #[test] + fn resolve_codec_canonicalizes_a_multi_bit_preference() { + // A non-conformant peer may stuff its capability MASK into `preferred` — the result + // must still be a single bit of the shared set, never the raw multi-bit echo (which + // folds to HEVC downstream and can select a codec the client can't decode). + assert_eq!( + resolve_codec(CODEC_H264, CODEC_H264 | CODEC_AV1, CODEC_H264 | CODEC_AV1), + Some(CODEC_H264) + ); + // Several shared preferred bits: still exactly one bit, and one of the preferred ones. + let got = resolve_codec( + CODEC_H264 | CODEC_HEVC | CODEC_AV1, + CODEC_H264 | CODEC_HEVC | CODEC_AV1, + CODEC_AV1 | CODEC_HEVC, + ) + .unwrap(); + assert_eq!(got.count_ones(), 1); + assert_ne!(got & (CODEC_AV1 | CODEC_HEVC), 0); + } } diff --git a/crates/punktfunk-core/src/quic/endpoint.rs b/crates/punktfunk-core/src/quic/endpoint.rs index 321cff7e..ed7813e5 100644 --- a/crates/punktfunk-core/src/quic/endpoint.rs +++ b/crates/punktfunk-core/src/quic/endpoint.rs @@ -26,10 +26,12 @@ fn stream_transport() -> Arc { /// path is PINGed at least twice per window and a single lost PING (wifi roam / brief blip) won't /// false-close. `idle` is clamped to a ≥1s floor so a misconfigured tiny value can't tear live /// sessions down. Active sessions are unaffected either way: video keeps the connection live and -/// the keep-alive holds it open through quiet control periods. +/// the keep-alive holds it open through quiet control periods. Clamped to a 1 s..1 h window: +/// the ceiling keeps an absurd operator-supplied value inside QUIC's VarInt millisecond range, +/// so the conversion below genuinely cannot fail (it used to panic host startup instead). fn stream_transport_idle(idle: std::time::Duration) -> Arc { use std::time::Duration; - let idle = idle.max(Duration::from_secs(1)); + let idle = idle.clamp(Duration::from_secs(1), Duration::from_secs(3600)); let keep_alive = (idle / 2).min(Duration::from_secs(4)); let mut t = quinn::TransportConfig::default(); t.max_idle_timeout(Some( @@ -305,4 +307,13 @@ mod tests { assert_eq!(a, endpoint::cert_fingerprint(b"cert-a")); assert_ne!(a, endpoint::cert_fingerprint(b"cert-b")); } + + #[test] + fn absurd_idle_timeout_is_clamped_not_a_panic() { + // The conversion to quinn's IdleTimeout fails past the QUIC VarInt millisecond + // ceiling — an operator-supplied huge PUNKTFUNK_IDLE_TIMEOUT_MS used to panic host + // startup through the `expect`. Both extremes must construct. + let _ = super::stream_transport_idle(std::time::Duration::MAX); + let _ = super::stream_transport_idle(std::time::Duration::ZERO); + } } diff --git a/crates/punktfunk-core/src/quic/handshake.rs b/crates/punktfunk-core/src/quic/handshake.rs index ca2a3c0c..076177b5 100644 --- a/crates/punktfunk-core/src/quic/handshake.rs +++ b/crates/punktfunk-core/src/quic/handshake.rs @@ -182,8 +182,9 @@ pub struct Start { } /// Truncate `s` to at most `max` bytes on a UTF-8 char boundary (so a multi-byte char straddling -/// the cap is dropped whole, never split). Shared by Hello's length-prefixed name/launch fields. -fn truncate_to(s: &str, max: usize) -> &str { +/// the cap is dropped whole, never split). Shared by Hello's length-prefixed name/launch fields +/// and [`PairRequest`](super::PairRequest)'s copy of the same name cap. +pub(super) fn truncate_to(s: &str, max: usize) -> &str { if s.len() <= max { return s; } diff --git a/crates/punktfunk-core/src/quic/pairing.rs b/crates/punktfunk-core/src/quic/pairing.rs index 013250c7..1f1ede26 100644 --- a/crates/punktfunk-core/src/quic/pairing.rs +++ b/crates/punktfunk-core/src/quic/pairing.rs @@ -78,13 +78,16 @@ fn get_bytes(b: &[u8], off: usize) -> Result<(&[u8], usize)> { impl PairRequest { pub fn encode(&self) -> Vec { - let name = self.name.as_bytes(); - let n = name.len().min(64); + // Same cap, same rule as Hello's copy of this field: truncate on a char boundary — + // a raw byte cut mid-sequence put invalid UTF-8 on the wire, and the host showed the + // name with a permanent replacement char in its paired-clients list. + let name = super::handshake::truncate_to(&self.name, HELLO_NAME_MAX).as_bytes(); + let n = name.len(); let mut b = Vec::with_capacity(8 + n + self.spake_a.len()); b.extend_from_slice(CTL_MAGIC); b.push(MSG_PAIR_REQUEST); b.push(n as u8); - b.extend_from_slice(&name[..n]); + b.extend_from_slice(name); put_bytes(&mut b, &self.spake_a); b } @@ -201,4 +204,20 @@ mod tests { bad.push(0); assert!(PairProof::decode(&bad).is_err()); } + + #[test] + fn pair_request_name_cap_respects_char_boundaries() { + // A multi-byte char straddling the 64-byte cap must be dropped whole (Hello's rule), + // not split mid-sequence into invalid UTF-8 the host then renders as U+FFFD forever. + let pr = PairRequest { + name: format!("{}\u{00fc}", "x".repeat(HELLO_NAME_MAX - 1)), + spake_a: vec![1, 2, 3], + }; + let dec = PairRequest::decode(&pr.encode()).unwrap(); + assert!(dec.name.len() <= HELLO_NAME_MAX && dec.name.starts_with('x')); + assert!( + !dec.name.contains('\u{FFFD}'), + "name must never be split mid-char on the wire" + ); + } } diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index b674ba6e..6195fa5b 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -358,7 +358,10 @@ impl Session { } let mut split_done = false; if two_lane && used >= TWO_LANE_MIN_PACKETS { - if let Some(lane) = seal_lane.as_ref() { + // Take the lane for the frame: a healthy round-trip puts it back; either + // failure arm drops the corpse so the next large frame respawns a fresh one + // instead of retrying a dead channel forever. + if let Some(lane) = seal_lane.take() { let half = used / 2; let mut tail = std::mem::take(lane_scratch); tail.extend(wires.drain(half..)); @@ -369,26 +372,42 @@ impl Session { ns: 0, result: Ok(()), }; - if lane.to_worker.send(job).is_ok() { - // Seal the front half while the worker runs; collect BOTH results - // before erroring so the lane is always drained and reusable. - let t0 = perf_armed.then(std::time::Instant::now); - let front = seal_wire_slice(c, &mut wires, seq_base); - if let Some(t0) = t0 { - seal_ns += t0.elapsed().as_nanos() as u64; + match lane.to_worker.send(job) { + Ok(()) => { + // Seal the front half while the worker runs; collect BOTH results + // before erroring so the lane is always drained and reusable. + let t0 = perf_armed.then(std::time::Instant::now); + let front = seal_wire_slice(c, &mut wires, seq_base); + if let Some(t0) = t0 { + seal_ns += t0.elapsed().as_nanos() as u64; + } + match lane.from_worker.recv() { + Ok(mut done) => { + *seal_lane = Some(lane); + seal_ns += done.ns; + wires.append(&mut done.bufs); + *lane_scratch = done.bufs; + front?; + done.result?; + split_done = true; + } + Err(_) => { + // The worker died holding the back half — the frame is + // unrecoverable (its packets are gone), but the error now + // SURFACES instead of `Ok` with half an access unit. + front?; + return Err(PunktfunkError::Unsupported("seal lane died")); + } + } + } + Err(std::sync::mpsc::SendError(job)) => { + // The worker is gone but the channel hands the job back: reclaim + // the back half so the single-lane pass below seals the WHOLE + // frame — previously this fall-through sealed and returned only + // the front half, silently, as `Ok`. + wires.extend(job.bufs); } - let mut done = lane - .from_worker - .recv() - .map_err(|_| PunktfunkError::Unsupported("seal lane died"))?; - seal_ns += done.ns; - wires.append(&mut done.bufs); - *lane_scratch = done.bufs; - front?; - done.result?; - split_done = true; } - // A failed send means the worker is gone — fall through to single-lane. } } if !split_done { @@ -790,6 +809,42 @@ mod wire_equivalence_tests { (0..len).map(|i| (i * 31 + 7) as u8).collect() } + /// A dead seal lane (worker gone, channels dangling) must degrade to a single-lane seal of + /// the WHOLE frame — the old fall-through sealed and returned only the front half as `Ok` — + /// and the corpse must be dropped so the next large frame respawns a fresh lane. + #[test] + fn dead_seal_lane_falls_back_to_single_lane_whole_frame() { + let mut opt = host_session(host_cfg(FecScheme::Gf16, 20, true)); + let mut refr = host_session(host_cfg(FecScheme::Gf16, 20, true)); + // A lane whose worker has already exited: both far ends dropped, so `send` fails + // immediately and hands the job (with the frame's back half) back. + let (to_worker, jobs) = std::sync::mpsc::sync_channel::(1); + let (done_tx, from_worker) = std::sync::mpsc::sync_channel::(1); + drop(jobs); + drop(done_tx); + opt.seal_lane = Some(SealLane { + to_worker, + from_worker, + }); + let frame = pattern(20000); // > TWO_LANE_MIN_PACKETS wire packets → takes the split path + let got = opt.seal_frame(&frame, 7, 0).unwrap(); + let want = seal_via_wrapper(&mut refr, &frame, 7, 0); + assert_eq!(got, want, "fallback must seal the whole frame, not half"); + assert!( + opt.seal_lane.is_none(), + "the dead lane must be dropped, not retried forever" + ); + // The next large frame respawns a fresh, working lane. + opt.reclaim_wires(got); + let got2 = opt.seal_frame(&frame, 8, 1).unwrap(); + let want2 = seal_via_wrapper(&mut refr, &frame, 8, 1); + assert_eq!(got2, want2); + assert!( + opt.seal_lane.is_some(), + "a fresh lane respawns on the next large frame" + ); + } + /// Partial delivery (plan §4.4): a chunk-aligned frame that loses shards past FEC's /// reach is DELIVERED once it ages out — `complete: false`, received shards at their /// exact offsets, missing ranges zero-filled — instead of silently dropping. Plain