fix(core): five sweep lows — seal-lane fallback, flush partial, codec echo, idle clamp, pair-name UTF-8
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 <noreply@anthropic.com>
This commit is contained in:
@@ -518,6 +518,10 @@ impl Reassembler {
|
|||||||
// The dropped frames' buffers (and their parity bufs) go back to the allocator, not the
|
// 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.
|
// pool — a flush is the rare path. The budget resets with them.
|
||||||
self.in_flight_bytes = 0;
|
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()"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -91,8 +91,13 @@ pub fn resolve_codec(client_codecs: u8, host_capable: u8, preferred: u8) -> Opti
|
|||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
// Honor the client's preference when the host can also emit it; else fall back to precedence.
|
// 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 {
|
if preferred != 0 && shared & preferred != 0 {
|
||||||
return Some(preferred);
|
let want = shared & preferred;
|
||||||
|
return Some(want & want.wrapping_neg());
|
||||||
}
|
}
|
||||||
// Precedence: HEVC > AV1 > H.264.
|
// Precedence: HEVC > AV1 > H.264.
|
||||||
[CODEC_HEVC, CODEC_AV1, CODEC_H264]
|
[CODEC_HEVC, CODEC_AV1, CODEC_H264]
|
||||||
@@ -222,4 +227,24 @@ mod tests {
|
|||||||
0
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -26,10 +26,12 @@ fn stream_transport() -> Arc<quinn::TransportConfig> {
|
|||||||
/// path is PINGed at least twice per window and a single lost PING (wifi roam / brief blip) won't
|
/// 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
|
/// 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
|
/// 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<quinn::TransportConfig> {
|
fn stream_transport_idle(idle: std::time::Duration) -> Arc<quinn::TransportConfig> {
|
||||||
use std::time::Duration;
|
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 keep_alive = (idle / 2).min(Duration::from_secs(4));
|
||||||
let mut t = quinn::TransportConfig::default();
|
let mut t = quinn::TransportConfig::default();
|
||||||
t.max_idle_timeout(Some(
|
t.max_idle_timeout(Some(
|
||||||
@@ -305,4 +307,13 @@ mod tests {
|
|||||||
assert_eq!(a, endpoint::cert_fingerprint(b"cert-a"));
|
assert_eq!(a, endpoint::cert_fingerprint(b"cert-a"));
|
||||||
assert_ne!(a, endpoint::cert_fingerprint(b"cert-b"));
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
/// 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.
|
/// the cap is dropped whole, never split). Shared by Hello's length-prefixed name/launch fields
|
||||||
fn truncate_to(s: &str, max: usize) -> &str {
|
/// 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 {
|
if s.len() <= max {
|
||||||
return s;
|
return s;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -78,13 +78,16 @@ fn get_bytes(b: &[u8], off: usize) -> Result<(&[u8], usize)> {
|
|||||||
|
|
||||||
impl PairRequest {
|
impl PairRequest {
|
||||||
pub fn encode(&self) -> Vec<u8> {
|
pub fn encode(&self) -> Vec<u8> {
|
||||||
let name = self.name.as_bytes();
|
// Same cap, same rule as Hello's copy of this field: truncate on a char boundary —
|
||||||
let n = name.len().min(64);
|
// 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());
|
let mut b = Vec::with_capacity(8 + n + self.spake_a.len());
|
||||||
b.extend_from_slice(CTL_MAGIC);
|
b.extend_from_slice(CTL_MAGIC);
|
||||||
b.push(MSG_PAIR_REQUEST);
|
b.push(MSG_PAIR_REQUEST);
|
||||||
b.push(n as u8);
|
b.push(n as u8);
|
||||||
b.extend_from_slice(&name[..n]);
|
b.extend_from_slice(name);
|
||||||
put_bytes(&mut b, &self.spake_a);
|
put_bytes(&mut b, &self.spake_a);
|
||||||
b
|
b
|
||||||
}
|
}
|
||||||
@@ -201,4 +204,20 @@ mod tests {
|
|||||||
bad.push(0);
|
bad.push(0);
|
||||||
assert!(PairProof::decode(&bad).is_err());
|
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"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -358,7 +358,10 @@ impl Session {
|
|||||||
}
|
}
|
||||||
let mut split_done = false;
|
let mut split_done = false;
|
||||||
if two_lane && used >= TWO_LANE_MIN_PACKETS {
|
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 half = used / 2;
|
||||||
let mut tail = std::mem::take(lane_scratch);
|
let mut tail = std::mem::take(lane_scratch);
|
||||||
tail.extend(wires.drain(half..));
|
tail.extend(wires.drain(half..));
|
||||||
@@ -369,26 +372,42 @@ impl Session {
|
|||||||
ns: 0,
|
ns: 0,
|
||||||
result: Ok(()),
|
result: Ok(()),
|
||||||
};
|
};
|
||||||
if lane.to_worker.send(job).is_ok() {
|
match lane.to_worker.send(job) {
|
||||||
// Seal the front half while the worker runs; collect BOTH results
|
Ok(()) => {
|
||||||
// before erroring so the lane is always drained and reusable.
|
// Seal the front half while the worker runs; collect BOTH results
|
||||||
let t0 = perf_armed.then(std::time::Instant::now);
|
// before erroring so the lane is always drained and reusable.
|
||||||
let front = seal_wire_slice(c, &mut wires, seq_base);
|
let t0 = perf_armed.then(std::time::Instant::now);
|
||||||
if let Some(t0) = t0 {
|
let front = seal_wire_slice(c, &mut wires, seq_base);
|
||||||
seal_ns += t0.elapsed().as_nanos() as u64;
|
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 {
|
if !split_done {
|
||||||
@@ -790,6 +809,42 @@ mod wire_equivalence_tests {
|
|||||||
(0..len).map(|i| (i * 31 + 7) as u8).collect()
|
(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::<SealJob>(1);
|
||||||
|
let (done_tx, from_worker) = std::sync::mpsc::sync_channel::<SealJob>(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
|
/// 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
|
/// reach is DELIVERED once it ages out — `complete: false`, received shards at their
|
||||||
/// exact offsets, missing ranges zero-filled — instead of silently dropping. Plain
|
/// exact offsets, missing ranges zero-filled — instead of silently dropping. Plain
|
||||||
|
|||||||
Reference in New Issue
Block a user