From a602e7cf9195cf7992e018d24e522ab0552516b7 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Sun, 16 Aug 2026 23:55:46 +0200 Subject: [PATCH 1/5] fix(pf-bitstream): AV1 needs 32-bit reads, and the 31-bit cap belonged to the signed path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every AV1 session on an AMD host died after ~287 frames and silently fell back to H.265. The client log named it on the first access unit — "AV1 parse: more than 31 (32) bits were requested" — and then "No sequence header parsed yet" for every AU after, because the sequence header never parsed and each new keyframe re-hit the same wall. The vendored cros-codecs BitReader refused any read wider than 31 bits, "because that would break the read_bits_signed() function". True of the signed path's i32 accumulator, and misplaced: AV1 needs 32 bits in five places — timing_info's num_units_in_display_tick and time_scale, decoder_model_info's num_units_in_decoding_tick, and the variable-width buffer-delay and buffer_removal_time fields, whose lengths come from the stream and reach 32. AMF sets timing_info_present_flag; NVENC does not, which is why the rung's own evidence string ("one vendor, no soak") described a codec that had never once decoded on AMD. Upstream's BitWriter already accepted 32 bits, so the crate could emit a header it could not read back. Relaxing the guard alone would have been worse than the bug — three edits are required together: - the trailing mask is u32::MAX at 32. `1u32 << 32` overflows: a debug panic, and in release a mask of zero, i.e. a silent 0 return; - the byte cursor is advanced before the accumulation loop when it sits at zero remaining bits, which otherwise shifts by the full width and ORs the spent byte in. At <=31 bits the mask discarded those bits, so it was invisible; at 32 the mask is all-ones and cannot; - read_bits_signed carries its own > 31 guard, so widening the unsigned path does not silently widen the signed one into an overflow. This is the limit the original comment was actually protecting. That last guard made a latent panic reachable by test: the sign extension `-1 ^ ((1 << num_bits) - 1)` overflows at num_bits == 31, where 1i32 << 31 is i32::MIN and subtracting one from it panics in debug — a width the guard admits and upstream considered safe. Rewritten as `-1i32 << num_bits`, equal for every accepted width. Blast radius is provably AV1-only: neither H.264 nor H.265 has a read wider than 31 bits, literal or variable — every dynamic-width call site in the vendored tree is in the AV1 parser. The 52 upstream conformance tests (H.264/H.265/AV1/VP9) still pass unchanged. Tests: 32-bit reads byte-aligned, mid-byte, and entered on a spent cursor; 33 bits still refused; the signed path stops at 31 and still sign-extends; every width 1..=31 checked against an independent extraction across a spent-byte boundary; zero-width reads still consume nothing. End to end, an AV1 sequence header carrying timing_info now survives a synthesize/parse round trip — and reproduces the field error string exactly when the guard is reverted. Recorded as PROVENANCE deviation 8; owed upstream as a cros-codecs issue. --- .../vendor/cros-codecs/PROVENANCE.md | 37 +++++ .../vendor/cros-codecs/src/bitstream_utils.rs | 136 +++++++++++++++++- .../cros-codecs/src/codec/av1/synthesizer.rs | 105 ++++++++++++++ 3 files changed, 271 insertions(+), 7 deletions(-) diff --git a/crates/pf-bitstream/vendor/cros-codecs/PROVENANCE.md b/crates/pf-bitstream/vendor/cros-codecs/PROVENANCE.md index d72fb701..24d0ec45 100644 --- a/crates/pf-bitstream/vendor/cros-codecs/PROVENANCE.md +++ b/crates/pf-bitstream/vendor/cros-codecs/PROVENANCE.md @@ -68,5 +68,42 @@ in the future." (`a_hostile_long_term_count_is_a_parse_error_not_a_panic`). **Reported upstream 2026-08-06: .** +8. `src/bitstream_utils.rs` — `BitReader::read_bits` accepts 32 bits, and the 31-bit + limit moves to `read_bits_signed` where its reason lives. Upstream capped the unsigned + read at 31 "because that would break the read_bits_signed() function" — true of the + signed path's `i32` accumulator, but it denies the unsigned path a width AV1 requires + in five places: `timing_info`'s `num_units_in_display_tick` and `time_scale`, + `decoder_model_info`'s `num_units_in_decoding_tick` (all `f(32)`), and the + variable-width buffer-delay and `buffer_removal_time` fields, whose lengths are read + from the stream and reach 32. A sequence header with `timing_info_present_flag` set + was therefore unparseable — a legal stream that AMD's AMF encoder emits and NVENC does + not, so **every AV1 session on an AMD host failed**: `AV1 parse: more than 31 (32) bits + were requested` on the first access unit, then `No sequence header parsed yet` for + every one after. Upstream's own `BitWriter::write_f` already accepts 32, so the crate + could emit a header it could not read back. + + Three edits, all required together — the guard alone is not the fix: + - the trailing mask is `u32::MAX` at 32 (`1u32 << 32` overflows: a debug panic, and in + release a mask of zero, i.e. a silent `0` return); + - the byte cursor is advanced before the accumulation loop when it sits at zero + remaining bits, which otherwise shifts by the full width and ORs the spent byte in. + At ≤31 bits the mask discarded those bits, so it was invisible; at 32 it cannot; + - `read_bits_signed` carries its own `> 31` guard, so widening the unsigned path does + not silently widen the signed one into an overflow. + + Also fixed in the same function: the sign extension `-1 ^ ((1 << num_bits) - 1)` + overflows at `num_bits == 31` (`1i32 << 31` is `i32::MIN`; subtracting one from it + panics in debug) — a latent panic at a width the guard admits and upstream's comment + considered safe. Rewritten as `-1i32 << num_bits`, equal for every accepted width. + Regression-tested here (`read_thirty_two_bits_*`, `signed_reads_stop_at_thirty_one_bits`, + `widths_below_thirty_two_are_unchanged_across_a_spent_byte`) and end-to-end as an AV1 + synthesize/parse round trip (`sequence_header_obu_round_trips_timing_info`), which + reproduces the field error string exactly when the guard is reverted. **Report upstream + — not yet filed.** + + Not changed: `read_bits_signed(0)` still underflows on `num_bits - 1`. Unreachable — + AV1 has its own `read_su`, and the H.26x `se(v)` callers all pass positive widths — so + it is left to upstream rather than widened into this deviation. + Re-sync procedure: fetch the AOSP tree, re-apply this trim, diff `codec/` + `bitstream_utils.rs` (expect near-zero conflicts), update the commit pin above. diff --git a/crates/pf-bitstream/vendor/cros-codecs/src/bitstream_utils.rs b/crates/pf-bitstream/vendor/cros-codecs/src/bitstream_utils.rs index 07d59dfe..cfb1ce49 100644 --- a/crates/pf-bitstream/vendor/cros-codecs/src/bitstream_utils.rs +++ b/crates/pf-bitstream/vendor/cros-codecs/src/bitstream_utils.rs @@ -96,15 +96,36 @@ impl<'a> BitReader<'a> { } } - /// Read up to 31 bits from the stream. Note that we don't want to read 32 - /// bits even though we're returning a u32 because that would break the - /// read_bits_signed() function. 31 bits should be overkill for compressed - /// header parsing anyway. + /// Read up to 32 bits from the stream. + /// + /// Upstream capped this at 31 "because that would break the read_bits_signed() + /// function". The reasoning was sound and the placement was not: the i32 accumulator + /// that cannot hold 32 bits belongs to [`Self::read_bits_signed`], which now carries + /// that limit itself, while the unsigned path gets the width the AV1 spec actually + /// asks for. AV1 requests 32 bits in five places — `timing_info`'s + /// `num_units_in_display_tick` and `time_scale`, `decoder_model_info`'s + /// `num_units_in_decoding_tick`, and the variable-width buffer-delay and + /// `buffer_removal_time` fields, whose lengths are read from the stream and reach 32. + /// A sequence header with `timing_info_present_flag` set was therefore unparseable, + /// which is a legal stream every AMF encoder emits. pub fn read_bits>(&mut self, num_bits: usize) -> Result { - if num_bits > 31 { + if num_bits > 32 { return Err(ReadBitsError::TooManyBitsRequested(num_bits).to_string()); } + // Normalise the cursor before accumulating. A read that consumed a byte exactly + // leaves `num_remaining_bits_in_curr_byte` at 0 with `curr_byte` still holding the + // spent byte, and the loop below would then shift by the whole of `bits_left` and + // OR that spent byte in. At <= 31 bits the trailing mask discarded the result (the + // stale bits all land at or above bit `num_bits`), so it was invisible; at 32 the + // mask is all-ones and cannot discard anything, and the shift itself overflows. + // Advancing first keeps `bits_left - num_remaining_bits_in_curr_byte` <= 31 for + // every width this function accepts. Zero-width reads are left alone: they consume + // nothing today and must keep consuming nothing. + if num_bits > 0 && self.num_remaining_bits_in_curr_byte == 0 { + self.move_to_next_byte().map_err(|err| err.to_string())?; + } + let mut bits_left = num_bits; let mut out = 0u32; @@ -115,7 +136,13 @@ impl<'a> BitReader<'a> { } out |= (self.curr_byte >> (self.num_remaining_bits_in_curr_byte - bits_left)) as u32; - out &= (1 << num_bits) - 1; + // `1u32 << 32` overflows — and at 32 bits every bit read is wanted, so the mask is + // the identity. Left as a shift for every narrower width, unchanged. + out &= if num_bits == 32 { + u32::MAX + } else { + (1 << num_bits) - 1 + }; self.num_remaining_bits_in_curr_byte -= bits_left; self.position += num_bits as u64; @@ -124,12 +151,28 @@ impl<'a> BitReader<'a> { /// Reads a two's complement signed integer of length |num_bits|. pub fn read_bits_signed>(&mut self, num_bits: usize) -> Result { + // The 31-bit limit lives here, where its reason is: the accumulator below is an + // i32, so a 32-bit read cannot round-trip through it — the `u32 -> i32` conversion + // fails for any value with the top bit set, and `1 << num_bits` in the + // sign-extension overflows. This used to be enforced indirectly by + // [`Self::read_bits`], which meant widening that function silently widened this + // one too. + if num_bits > 31 { + return Err(ReadBitsError::TooManyBitsRequested(num_bits).to_string()); + } let mut out: i32 = self .read_bits::(num_bits)? .try_into() .map_err(|_| ReadBitsError::ConversionFailed.to_string())?; if out >> (num_bits - 1) != 0 { - out |= -1i32 ^ ((1 << num_bits) - 1); + // Sign-extend by setting every bit at or above `num_bits`. Written as a shift + // of -1 rather than upstream's `-1 ^ ((1 << num_bits) - 1)`: the two are equal + // for every width this function accepts, but the original overflows at + // `num_bits == 31`, where `1i32 << 31` is `i32::MIN` and subtracting one from + // it panics in a debug build. 31 is a width the guard above admits and the + // upstream comment explicitly considered safe, so this was a latent panic on + // legal input rather than an unreachable edge. + out |= -1i32 << num_bits; } U::try_from(out).map_err(|_| ReadBitsError::ConversionFailed.to_string()) @@ -785,4 +828,83 @@ mod tests { let mut reader = BitReader::new(&[0b1111_0000], false); assert_eq!(reader.read_bits_signed::(4).unwrap(), -1); } + + /// AV1's `timing_info` reads `f(32)`, and the top bit is routinely set (`time_scale` + /// carries values like 1_000_000_000). Byte-aligned from a fresh reader. + #[test] + fn read_thirty_two_bits_aligned() { + let mut reader = BitReader::new(&[0xDE, 0xAD, 0xBE, 0xEF], false); + assert_eq!(reader.read_bits::(32).unwrap(), 0xDEAD_BEEF); + } + + /// The same width entered mid-byte: the accumulation loop's first shift is then + /// `32 - num_remaining_bits_in_curr_byte`, which is the widest shift this function + /// ever performs and must stay under 32. + #[test] + fn read_thirty_two_bits_unaligned() { + // 1 bit, then 32 bits, out of a 5-byte run: 0b1 then 0xBD5B7DDE. + let mut reader = BitReader::new(&[0xDE, 0xAD, 0xBE, 0xEF, 0x00], false); + assert_eq!(reader.read_bits::(1).unwrap(), 1); + assert_eq!(reader.read_bits::(32).unwrap(), 0xBD5B_7DDE); + } + + /// The regression the mask fix exists for: entering a 32-bit read with the byte + /// cursor exactly spent (`num_remaining_bits_in_curr_byte == 0`) used to shift by 32 + /// and OR the already-consumed byte into a result the mask could no longer clean. + #[test] + fn read_thirty_two_bits_on_a_spent_byte() { + let mut reader = BitReader::new(&[0xFF, 0xDE, 0xAD, 0xBE, 0xEF], false); + // Consume the first byte exactly, leaving the cursor at zero remaining bits. + assert_eq!(reader.read_bits::(8).unwrap(), 0xFF); + assert_eq!(reader.read_bits::(32).unwrap(), 0xDEAD_BEEF); + } + + /// 33 bits is still refused — the widening is to exactly the width AV1 needs. + #[test] + fn more_than_thirty_two_bits_is_still_refused() { + let mut reader = BitReader::new(&[0xDE, 0xAD, 0xBE, 0xEF, 0x00], false); + assert!(reader.read_bits::(33).is_err()); + } + + /// The signed path keeps the 31-bit limit on its own account: its accumulator is an + /// i32. Before this it was enforced by `read_bits`, so widening that function would + /// have widened this one into an overflow. + #[test] + fn signed_reads_stop_at_thirty_one_bits() { + let mut reader = BitReader::new(&[0xFF, 0xFF, 0xFF, 0xFF, 0x00], false); + assert!(reader.read_bits_signed::(32).is_err()); + // 31 still works, and still sign-extends. + let mut reader = BitReader::new(&[0xFF, 0xFF, 0xFF, 0xFF, 0x00], false); + assert_eq!(reader.read_bits_signed::(31).unwrap(), -1); + } + + /// The cursor normalisation must not change any narrower read. Walk every width from + /// 1 to 31 across a byte-spent boundary and check the value against an independent + /// big-endian bit extraction of the same stream. + #[test] + fn widths_below_thirty_two_are_unchanged_across_a_spent_byte() { + const DATA: [u8; 8] = [0xA5, 0x3C, 0x91, 0x7E, 0xDB, 0x42, 0x68, 0xF1]; + // Independent reference: bit `i` of the stream, MSB-first. + let bit = |i: usize| (DATA[i / 8] >> (7 - (i % 8))) & 1; + for width in 1..=31usize { + let mut reader = BitReader::new(&DATA, false); + // Land the cursor exactly on a byte boundary with zero remaining bits. + assert_eq!(reader.read_bits::(8).unwrap(), DATA[0] as u32); + let got = reader.read_bits::(width).unwrap(); + let want = (8..8 + width).fold(0u32, |acc, i| (acc << 1) | bit(i) as u32); + assert_eq!(got, want, "width {width}"); + } + } + + /// A zero-width read consumes nothing and returns zero, on a spent cursor as well as + /// a fresh one — `read_ue` reaches this whenever its first bit is set. + #[test] + fn zero_width_reads_consume_nothing() { + let mut reader = BitReader::new(&[0xAB, 0xCD], false); + assert_eq!(reader.read_bits::(0).unwrap(), 0); + assert_eq!(reader.read_bits::(8).unwrap(), 0xAB); + // Cursor now spent; a zero-width read must not pull the next byte in. + assert_eq!(reader.read_bits::(0).unwrap(), 0); + assert_eq!(reader.read_bits::(8).unwrap(), 0xCD); + } } diff --git a/crates/pf-bitstream/vendor/cros-codecs/src/codec/av1/synthesizer.rs b/crates/pf-bitstream/vendor/cros-codecs/src/codec/av1/synthesizer.rs index 4faf7346..ffcd1a2a 100644 --- a/crates/pf-bitstream/vendor/cros-codecs/src/codec/av1/synthesizer.rs +++ b/crates/pf-bitstream/vendor/cros-codecs/src/codec/av1/synthesizer.rs @@ -1628,6 +1628,111 @@ mod tests { assert_eq!(buf, SEQ_HDR_RAW); } + /// A sequence header carrying `timing_info` must survive a write/read round trip. + /// + /// This is the field failure: AMF emits `timing_info_present_flag = 1`, whose + /// `num_units_in_display_tick` and `time_scale` are `f(32)`, and the reader refused + /// any width above 31 — so every AMD AV1 session died with "AV1 parse: more than 31 + /// (32) bits were requested" on its first access unit and then "No sequence header + /// parsed yet" forever after. The writer had always accepted 32 bits, so this crate + /// could emit a header it could not read back. + /// + /// Both values deliberately have their top bit set: a mask computed as + /// `(1 << 32) - 1` truncates to zero rather than to all-ones, so a wrong mask shows up + /// here as a zero, not as a near-miss. + #[test] + fn sequence_header_obu_round_trips_timing_info() { + use crate::codec::av1::parser::ObuAction; + use crate::codec::av1::parser::ParsedObu; + use crate::codec::av1::parser::Parser; + use crate::codec::av1::parser::TimingInfo; + + const TICK: u32 = 0xDEAD_BEEF; + const SCALE: u32 = 0xFFFF_FFFF; + + let seq_hdr = SequenceHeaderObu { + obu_header: ObuHeader { + obu_type: ObuType::SequenceHeader, + extension_flag: false, + has_size_field: true, + temporal_id: 0, + spatial_id: 0, + }, + + seq_profile: Profile::Profile0, + num_planes: 3, + still_picture: false, + reduced_still_picture_header: false, + timing_info_present_flag: true, + timing_info: TimingInfo { + num_units_in_display_tick: TICK, + time_scale: SCALE, + equal_picture_interval: false, + num_ticks_per_picture_minus_1: 0, + }, + decoder_model_info_present_flag: false, + initial_display_delay_present_flag: false, + operating_points_cnt_minus_1: 0, + frame_width_bits_minus_1: 8, + frame_height_bits_minus_1: 7, + max_frame_width_minus_1: 319, + max_frame_height_minus_1: 239, + frame_id_numbers_present_flag: false, + use_128x128_superblock: true, + enable_filter_intra: true, + enable_intra_edge_filter: true, + enable_interintra_compound: true, + enable_masked_compound: true, + enable_warped_motion: true, + enable_dual_filter: true, + enable_order_hint: true, + enable_jnt_comp: true, + enable_ref_frame_mvs: true, + seq_choose_screen_content_tools: true, + seq_force_screen_content_tools: SELECT_SCREEN_CONTENT_TOOLS as u32, + seq_choose_integer_mv: true, + seq_force_integer_mv: SELECT_INTEGER_MV as u32, + order_hint_bits_minus_1: 6, + order_hint_bits: 7, + enable_superres: false, + enable_cdef: true, + enable_restoration: true, + color_config: ColorConfig { + high_bitdepth: false, + mono_chrome: false, + color_description_present_flag: false, + color_range: false, + subsampling_x: true, + subsampling_y: true, + chroma_sample_position: ChromaSamplePosition::Unknown, + separate_uv_delta_q: false, + ..Default::default() + }, + film_grain_params_present: false, + + ..Default::default() + }; + + let mut buf = Vec::::new(); + Synthesizer::<'_, SequenceHeaderObu, _>::synthesize(&seq_hdr, &mut buf).unwrap(); + + let mut parser = Parser::default(); + let obu = match parser.read_obu(&buf).expect("the OBU header must read") { + ObuAction::Process(obu) => obu, + ObuAction::Drop(_) => panic!("a sequence header must not be dropped"), + }; + let parsed = parser + .parse_obu(obu) + .expect("the sequence header must parse"); + let ParsedObu::SequenceHeader(seq) = parsed else { + panic!("expected a sequence header back"); + }; + + assert!(seq.timing_info_present_flag); + assert_eq!(seq.timing_info.num_units_in_display_tick, TICK); + assert_eq!(seq.timing_info.time_scale, SCALE); + } + #[test] fn sequence_header_obu_av1_annexb() { // Extraced from: ./src/codec/av1/test_data/av1-annexb.ivf.av1 -- 2.54.0 From 88d071c03327afdaf63b3cf1e42495390305e2dd Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 17 Aug 2026 00:00:59 +0200 Subject: [PATCH 2/5] fix(client/abr): a window that carried nothing cannot report the encoder as slow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An exclusive-topology eviction on a Windows host rebuilds the capture ring and the encoder in place — 401 ms, entirely host-local, no packet lost. The client's 750 ms report window straddled one and recorded 390 kbps delivered against a 20 000 target, loss_ppm=0, no flush, and encode_mean_us=15063 against a ~2800 baseline. That cleared ENCODE_SEVERE_US, took the one-window path, and cost a x0.7 plus slow start for the rest of the session. Recovery is then +6 % per six clean windows, so the field session sat at ~15 Mbps for the three minutes it had left. encode_us is a per-AU host measurement averaged over the window. When almost no AUs flowed, the mean is taken over the handful that straddled whatever interrupted them, and their encode time carries that interruption rather than the cost of encoding at this rate. It is not a measurement, so it is now withheld entirely when the window is STARVED — the predicate that already existed for exactly this shape of window, hoisted above the signal scoring. Passed as absent rather than ignored, so it cannot teach the rolling-minimum baseline either. Deliberately narrow. The first attempt took the one-window shortcut away from every severe signal in a starved window and broke two tests that turned out to be right: loss, a flush and a dropped frame describe what reached the CLIENT and mean the same thing however little flowed, so the periodic-capture-stall case still backs off on one window as STARVED_DELIVERY_DIV's own comment requires. Only the host-encode signal is withheld, because only it is measured over AUs that did not exist. Slow start is likewise left alone: a starved window that is bad for a legitimate reason still ends it. Tests: the field window verbatim decides nothing and leaves slow start armed, and the same encode excursion in a window that actually carried its rate still backs off on one window — which is what proves the withheld sample never entered the baseline. 459 core tests green. --- crates/punktfunk-core/src/abr.rs | 122 +++++++++++++++++++++++++++++-- 1 file changed, 115 insertions(+), 7 deletions(-) diff --git a/crates/punktfunk-core/src/abr.rs b/crates/punktfunk-core/src/abr.rs index 1c8c4631..a74e76ff 100644 --- a/crates/punktfunk-core/src/abr.rs +++ b/crates/punktfunk-core/src/abr.rs @@ -163,8 +163,13 @@ const DECODE_CAP_SIMILAR_DIV: u32 = 8; /// stream barely flowed (a host-side capture stall, an outage, a mid-window pause), so whatever /// distress the window carries — a flush, a keyframe-ask burst — is starvation-shaped, not /// rate-shaped, and the decoder decoded almost nothing at the nominal rate. Such a window may -/// still back off (real damage deserves the safe response) but must never be a decode-knee -/// sample: latching `current_kbps` off a starved window teaches a phantom decoder cap at +/// still back off on what the CLIENT saw — loss, a flush, a dropped frame mean the same thing +/// however little flowed, and real damage deserves the safe response — but two things it must +/// never do. It must never be a decode-knee sample, and it must never carry the HOST-ENCODE +/// signal: `encode_us` is averaged over the AUs of the window, so when almost none flowed the +/// mean describes whatever interrupted them rather than the cost of encoding at this rate (see +/// the withholding in [`BitrateController::on_window`]). Latching `current_kbps` off a starved +/// window teaches a phantom decoder cap at /// whatever rate the stall interrupted (the periodic-capture-stall field case: every 5 s cycle /// offers another pair of "backoffs" at the same rate — a bogus latch that then fights the /// re-probe ladder for minutes). Deliberately far below the ×¾ utilization bar climbs require: @@ -572,13 +577,35 @@ impl BitrateController { DECODE_RISE_US, DECODE_SEVERE_US, ); + // STARVED (see [`STARVED_DELIVERY_DIV`]): the window carried under a quarter of the rate + // it was allowed. Hoisted above the signal scoring because the encode signal below is not + // merely inconvenient in such a window, it is not a measurement — see there. `current_kbps` + // does not move inside this function, so this is the same value the backoff block reads. + let starved = + (actual_kbps as u64) * (STARVED_DELIVERY_DIV as u64) < self.current_kbps as u64; // Host-encode latency: the same rolling-min-baseline treatment, measuring the HOST'S // encoder — the compute-knee down-driver (see [`ENCODE_RISE_US`]). This is the only // signal that can push an already-too-high rate back under the knee: the host refuses // further climbs while behind cadence, but nothing else ever DESCENDS on a clean LAN. + // + // Withheld entirely in a STARVED window. `encode_us` is a per-AU host measurement averaged + // over the window, so when almost no AUs flowed the mean is taken over the handful that + // straddled whatever interrupted them — and their encode time carries that interruption, + // not the cost of encoding at this rate. The field case: a 401 ms capture-ring and encoder + // rebuild (an exclusive-topology eviction, entirely host-local) produced one window with + // `encode_mean_us=15063` against a ~2800 baseline, `actual_kbps=390` against a 20 000 + // target, and `loss_ppm=0`. That cleared [`ENCODE_SEVERE_US`], took the one-window path, + // and cost a ×0.7 plus slow start for the rest of the session — on a link that never + // dropped a packet. Passed as absent rather than ignored so it cannot teach the rolling + // baseline either: a sample that measures a stall is not evidence about anything. + // + // The other signals keep their full power here on purpose. Loss, a flush and a dropped + // frame describe what reached the CLIENT, and they mean the same thing however little + // flowed — the periodic-capture-stall case (see [`STARVED_DELIVERY_DIV`]) still backs off + // on one window, as its tests require. let (encode_bad, encode_severe) = score_baseline( &mut self.encode_means, - encode_mean_us, + encode_mean_us.filter(|_| !starved), ENCODE_RISE_US, ENCODE_SEVERE_US, ); @@ -708,10 +735,9 @@ impl BitrateController { || self.streak_decode_windows >= BAD_WINDOWS_TO_DECREASE || (recovery_kf >= RECOVERY_KF_BAD && loss_ppm < HEAVY_LOSS_PPM) || (flushed && (decode_bad || decode_mean_us.is_none())); - // Starved deciding window (see [`STARVED_DELIVERY_DIV`]): the stream barely flowed, - // so the window says nothing about what the decoder can hold at this rate. - let starved = - (actual_kbps as u64) * (STARVED_DELIVERY_DIV as u64) < self.current_kbps as u64; + // `starved` (the deciding window barely flowed, so it says nothing about what the + // decoder can hold at this rate) is now computed once at the top of the window — the + // same predicate also governs the severe tier and slow start. if !self.climb_since_backoff { // Still draining the previous backoff: the host acks a ×0.7 request in ~100 ms, // so this window's rate is one the decoder never choked at while keeping up — @@ -2204,6 +2230,88 @@ mod tests { ); } + /// The host-rebuild field window, verbatim from the 0.29 log: an exclusive-topology eviction + /// rebuilt the capture ring and the encoder in place (401 ms, entirely host-local), and the + /// client's report window straddled it — 390 kbps delivered against a 20 000 target, zero + /// loss, no flush, and an encode mean of 15 063 µs against a ~2 800 baseline. + /// + /// That used to clear the severe encode tier and take the one-window path, costing a ×0.7 and + /// slow start for the rest of the session on a link that never dropped a packet. The encode + /// mean over a window in which almost nothing flowed is not a measurement of encode cost, so + /// the signal is withheld and the window decides nothing. + #[test] + fn a_starved_window_cannot_back_off_on_host_encode_time_alone() { + let mut c = BitrateController::new(20_000); + c.set_ceiling(657_000); + let start = Instant::now(); + let mut t = 0; + // Seed the latency baselines. Half-utilized on purpose: above the starved bar (a quarter + // of target) so the encode samples count, below the climb bar (three quarters) so no step + // fires and `current_kbps` stays put. + for _ in 0..BASELINE_MIN_WINDOWS { + assert_eq!( + c.on_window( + ticks(start, t), + 0, + 0, + Some(3_500), + Some(200), + Some(2_800), + 10_000, + false, + 0 + ), + None + ); + t += 1; + } + assert!( + c.probing, + "slow start is still armed going into the rebuild" + ); + + let verdict = c.on_window( + ticks(start, t), + 0, + 0, + Some(15_711), + Some(129), + Some(15_063), + 390, + false, + 0, + ); + t += 1; + assert_eq!( + verdict, None, + "a host-local rebuild must not move the rate: nothing was lost and nothing was slow" + ); + assert_eq!(c.current_kbps, 20_000, "and the rate is untouched"); + assert!( + c.probing, + "nor may it retire slow start — recovery would crawl at +6 % per six windows" + ); + + // The signal itself must still work: the starved sample was withheld rather than folded + // into the rolling minimum, so the SAME encode excursion in a window that actually + // carried its rate is still severe, and still backs off on one window. + let verdict = c.on_window( + ticks(start, t), + 0, + 0, + Some(3_600), + Some(210), + Some(15_063), + 20_000, + false, + 0, + ); + assert!( + verdict.is_some_and(|k| k < 20_000), + "a real encode excursion at full delivery still backs off, got {verdict:?}" + ); + } + #[test] fn decode_cap_latches_when_the_reclimb_chokes_at_the_same_knee() { // The 1440p120 field sawtooth: a decoder knee (~500 Mbps) well under the (inflated) -- 2.54.0 From abaac704d72d99c9accf9a2a23cc40759bee7cab Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 17 Aug 2026 00:04:56 +0200 Subject: [PATCH 3/5] feat(client/abr): bound the learned climb ceiling by what the stream can use MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The climb ceiling is pure link capacity — `delivered_kbps * 0.7`, with no term for resolution, frame rate, codec or bit depth. The utilization gate cannot supply one either: a hardware encoder in CBR mode genuinely fills whatever target it is handed, and the field log shows utilization at 99 % the whole way up, so "the encoder could not use the rate" never fires. On a gigabit LAN the probe measured 939 Mbps and the session walked to 657 Mbps for 1440p120 — 1.49 bits per pixel — in 37 seconds. Getting there took the client's decode latency from 0.78 ms to 10 ms. `stream_ceiling_kbps` computes what the shape could plausibly use from pixel rate and a bits-per-pixel allowance that varies by codec generation, bit depth and chroma, and `set_ceiling` holds the measured link ceiling to it — the same funnel PUNKTFUNK_ABR_MAX_MBPS already passes through, and it logs both numbers whenever it binds, because a cap that silently trims what the link offered is the kind of thing nobody reports. Deliberately generous: this is a bound on the absurd, not a quality opinion. 1440p120 HEVC Main10 lands at ~414 Mbps — under the decode knee this session actually found (flat at ~396 Mbps delivered, 10 ms by ~461) — while 1080p60 HEVC keeps ~93 Mbps, above anything people run there. Tested from both directions, because a cap that trims a happy user is a regression nobody files. Binds only what the probe LEARNS. A negotiated start rate is a number the host resolved on purpose and is left alone; an explicit bitrate and every PyroWave session are outside the controller entirely and never reach here. Sessions that never call set_stream_cap behave exactly as before. This is a backstop, not the answer to "how much is enough" — that is content-dependent and only the encoder knows it, at minimum QP. It is the part that works without new host telemetry. --- crates/punktfunk-core/src/abr.rs | 165 +++++++++++++++++- crates/punktfunk-core/src/client/pump.rs | 12 ++ crates/punktfunk-core/src/client/pump/data.rs | 10 ++ 3 files changed, 186 insertions(+), 1 deletion(-) diff --git a/crates/punktfunk-core/src/abr.rs b/crates/punktfunk-core/src/abr.rs index a74e76ff..5b542c3a 100644 --- a/crates/punktfunk-core/src/abr.rs +++ b/crates/punktfunk-core/src/abr.rs @@ -210,6 +210,61 @@ fn ceiling_cap_from_env() -> Option { .map(|m| m.saturating_mul(1_000)) } +/// The most bitrate this stream's SHAPE could plausibly use, in kbps — the backstop the +/// probe-measured link ceiling has never had. +/// +/// The measured ceiling is pure link capacity (`delivered × 0.7`) with no term for what is being +/// carried, and the utilization gate cannot supply one: a hardware encoder in CBR mode genuinely +/// fills whatever target it is handed, so "the encoder could not use the rate" never fires. The +/// field session climbed to 657 Mbps for 1440p120 — 1.49 bits per pixel, some 3× beyond any rate +/// an inter-coded stream benefits from — and reaching for it drove the client's decode latency +/// from 0.8 ms to 10 ms. +/// +/// Deliberately generous. This is a bound on the absurd, not a quality opinion: it is set well +/// above what anyone actually runs, so it should never bind on a real session, and where it does +/// bind [`BitrateController::set_ceiling`] says so in the log. A session with an explicit bitrate, +/// and every PyroWave session, is outside the controller entirely and never reaches here. +/// +/// It is NOT the answer to "how much is enough" — that is content-dependent and only the encoder +/// knows it (at minimum QP more bits buy nothing). This is the part that works without new +/// telemetry. +pub(crate) fn stream_ceiling_kbps( + width: u32, + height: u32, + refresh_hz: u32, + codec: u8, + bit_depth: u8, + chroma_format: u8, +) -> u32 { + let pixel_rate = (width as u64) + .saturating_mul(height as u64) + .saturating_mul(refresh_hz.max(1) as u64); + if pixel_rate == 0 { + return u32::MAX; + } + // Milli-bits per pixel, so the whole computation stays in integers. H.264 is the least + // efficient of the three and is allowed correspondingly more. + let milli_bpp: u64 = match codec { + crate::quic::CODEC_H264 => 1_000, + _ => 750, + }; + // 10-bit carries 25 % more sample depth; 4:4:4 carries twice the chroma of 4:2:0, which is + // half again as many samples overall. + let milli_bpp = if bit_depth >= 10 { + milli_bpp * 5 / 4 + } else { + milli_bpp + }; + let milli_bpp = if chroma_format == crate::quic::CHROMA_IDC_444 { + milli_bpp * 3 / 2 + } else { + milli_bpp + }; + // bits/s = pixel_rate × bpp; kbps = that / 1000. The milli- factor and the kbps divisor + // cancel, so this is just pixel_rate × milli_bpp / 1_000_000. + u32::try_from(pixel_rate.saturating_mul(milli_bpp) / 1_000_000).unwrap_or(u32::MAX) +} + /// Score one window's latency sample against its rolling-min baseline, then record it. /// /// Shared by all three latency signals (OWD, client decode, host encode) — same shape, different @@ -255,6 +310,11 @@ pub(crate) struct BitrateController { /// construction so tests exercise the clamp without touching the process environment. /// `None` = no cap. ceiling_cap_kbps: Option, + /// What this stream's SHAPE could plausibly use (see [`stream_ceiling_kbps`]), set once the + /// session's mode and codec are known. `None` = never set, i.e. exactly the old behavior. + /// Bounds only what [`set_ceiling`](Self::set_ceiling) LEARNS: the negotiated start rate is a + /// number the host resolved on purpose and is left alone. + stream_cap_kbps: Option, floor_kbps: u32, /// Slow start: true until the first congestion signal — clean windows DOUBLE the rate /// (cooldown-paced) instead of the +6 % additive step. @@ -361,6 +421,7 @@ impl BitrateController { // [`on_window`](Self::on_window). ceiling_kbps: start_kbps.min(ceiling_cap_kbps.unwrap_or(u32::MAX)), ceiling_cap_kbps, + stream_cap_kbps: None, floor_kbps: FLOOR_KBPS.min(start_kbps.max(1)), probing: true, owd_means: VecDeque::with_capacity(BASELINE_WINDOWS), @@ -398,12 +459,32 @@ impl BitrateController { /// ceiling was learned; monotonicity is precisely why the user needs it (one inflated /// measurement is otherwise permanent for the session). pub(crate) fn set_ceiling(&mut self, kbps: u32) { - let kbps = kbps.min(self.ceiling_cap_kbps.unwrap_or(u32::MAX)); + let measured = kbps; + let kbps = kbps + .min(self.ceiling_cap_kbps.unwrap_or(u32::MAX)) + .min(self.stream_cap_kbps.unwrap_or(u32::MAX)); + if self.enabled && kbps < measured { + // Say so when it binds. A cap that silently trims what the link offered is exactly + // the kind of thing nobody reports and everybody wonders about, so a field log should + // carry both numbers. + tracing::info!( + measured_kbps = measured, + bounded_kbps = kbps, + "adaptive bitrate: link ceiling bounded by what this stream can use" + ); + } if self.enabled && kbps > self.ceiling_kbps { self.ceiling_kbps = kbps; } } + /// Teach the controller what this session's mode and codec could plausibly use (see + /// [`stream_ceiling_kbps`]). Applied to LEARNED ceilings only, at the same funnel as the + /// operator's env cap. + pub(crate) fn set_stream_cap(&mut self, kbps: u32) { + self.stream_cap_kbps = Some(kbps); + } + /// The host's [`crate::quic::BitrateChanged`] ack: its clamp is authoritative for what the /// encoder now targets, and any ack proves the host renegotiates (resets the silence counter). /// @@ -1186,6 +1267,88 @@ mod tests { assert_eq!(c.ceiling_kbps, 20_000); } + /// The stream bound must cut the field runaway and must NOT touch a session anyone + /// actually runs. Both halves matter: a cap that silently trims a happy user is a + /// regression nobody reports. + #[test] + fn the_stream_bound_cuts_the_absurd_and_spares_the_ordinary() { + use crate::quic::{CHROMA_IDC_420, CHROMA_IDC_444, CODEC_H264, CODEC_HEVC}; + + // The field session: 1440p120 HEVC Main10 4:2:0. The probe measured 939 Mbps and the + // ceiling became 657 Mbps — 1.49 bits/pixel. The client's decode latency was still flat + // (0.78 ms) at ~396 Mbps delivered and blew up to 10 ms by ~461 Mbps, so the bound has to + // land below that knee to have helped. + let field = stream_ceiling_kbps(2560, 1440, 120, CODEC_HEVC, 10, CHROMA_IDC_420); + assert!( + field < 657_000, + "the bound must actually bind on the field case, got {field}" + ); + assert!( + field < 460_000, + "and land under the decode knee this session found, got {field}" + ); + + // 1080p60 HEVC 8-bit: people do run 80-100 Mbps here and must not be trimmed. + let ordinary = stream_ceiling_kbps(1920, 1080, 60, CODEC_HEVC, 8, CHROMA_IDC_420); + assert!( + ordinary >= 90_000, + "an ordinary 1080p60 session must keep its headroom, got {ordinary}" + ); + + // H.264 needs more bits for the same picture, and 4:4:4 / 10-bit carry more samples. + assert!( + stream_ceiling_kbps(1920, 1080, 60, CODEC_H264, 8, CHROMA_IDC_420) > ordinary, + "H.264 is allowed more than HEVC" + ); + assert!( + stream_ceiling_kbps(1920, 1080, 60, CODEC_HEVC, 10, CHROMA_IDC_420) > ordinary, + "10-bit is allowed more than 8-bit" + ); + assert!( + stream_ceiling_kbps(1920, 1080, 60, CODEC_HEVC, 8, CHROMA_IDC_444) > ordinary, + "4:4:4 is allowed more than 4:2:0" + ); + // A degenerate mode must not produce a bound of zero and strangle the session. + assert_eq!( + stream_ceiling_kbps(0, 0, 0, CODEC_HEVC, 8, CHROMA_IDC_420), + u32::MAX + ); + } + + /// The bound rides the same funnel as the operator's env cap, and binds only what the probe + /// LEARNS — a rate the host resolved on purpose is left alone. + #[test] + fn the_stream_bound_clamps_a_learned_ceiling_only() { + let mut c = BitrateController::new(20_000); + c.set_stream_cap(100_000); + c.set_ceiling(657_000); + assert_eq!(c.ceiling_kbps, 100_000, "a learned ceiling is bounded"); + + // Never set → exactly the old behaviour. + let mut c = BitrateController::new(20_000); + c.set_ceiling(657_000); + assert_eq!(c.ceiling_kbps, 657_000); + + // A negotiated start rate above the bound stands: the host resolved that number. + let mut c = BitrateController::new(300_000); + c.set_stream_cap(100_000); + assert_eq!(c.ceiling_kbps, 300_000); + c.set_ceiling(657_000); + assert_eq!( + c.ceiling_kbps, 300_000, + "and a learned ceiling under it never lowers what was negotiated" + ); + + // The tighter of the two caps wins. + let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000)); + c.set_stream_cap(100_000); + c.set_ceiling(657_000); + assert_eq!( + c.ceiling_kbps, 50_000, + "the env cap still binds when it is tighter" + ); + } + #[test] fn owd_rise_alone_is_a_congestion_signal() { let mut c = BitrateController::new(20_000); diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index 1dae3421..9008c91f 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -86,6 +86,17 @@ pub(super) async fn run_pump(args: WorkerArgs) { let clock_rtt_ns = negotiated.clock_rtt_ns; let resolved_bitrate_kbps = negotiated.bitrate_kbps; let negotiated_codec = negotiated.codec; + // What this session's mode + codec could plausibly use — the bound the ABR holds its + // probe-measured link ceiling to. Computed here because this is where the Welcome-resolved + // geometry lives; the data pump stays codec-agnostic. + let stream_cap_kbps = crate::abr::stream_ceiling_kbps( + negotiated.mode.width, + negotiated.mode.height, + negotiated.mode.refresh_hz, + negotiated.codec, + negotiated.bit_depth, + negotiated.chroma_format, + ); // Seed the live offset with the connect-time estimate BEFORE the embedder can observe the // client (ready_tx): clock_offset_now_ns() never reads a pre-handshake 0 on a skewed pair. clock_offset.store(negotiated.clock_offset_ns, Ordering::Relaxed); @@ -252,6 +263,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { bitrate_kbps, resolved_bitrate_kbps, negotiated_codec, + stream_cap_kbps, }; let _ = tokio::task::spawn_blocking(move || pump.run()).await; diff --git a/crates/punktfunk-core/src/client/pump/data.rs b/crates/punktfunk-core/src/client/pump/data.rs index 6457b23c..925d3e90 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -36,6 +36,10 @@ pub(super) struct DataPump { /// The rate the host actually configured (echoed in Welcome). pub(super) resolved_bitrate_kbps: u32, pub(super) negotiated_codec: u8, + /// What this session's mode + codec could plausibly use (see + /// [`crate::abr::stream_ceiling_kbps`]) — the bound the probe-measured link ceiling is held + /// to. Computed where the negotiated geometry lives, so this module stays codec-agnostic. + pub(super) stream_cap_kbps: u32, } impl DataPump { @@ -59,6 +63,7 @@ impl DataPump { bitrate_kbps, resolved_bitrate_kbps, negotiated_codec, + stream_cap_kbps, } = self; pin_thread_user_interactive(); // feeds the frame channel → the user-interactive video pump register_hot_tid(&pump_hot_tids); // this thread does UDP receive + FEC reassembly — hint it @@ -100,6 +105,11 @@ impl DataPump { } else { 0 }); + // Bound whatever the capacity probe measures by what this stream's shape could plausibly + // use. Without it the climb ceiling is pure link capacity, and a fat LAN authorizes rates + // no inter-coded stream benefits from — the field session walked to 657 Mbps for 1440p120 + // and drove the client's decode latency from 0.8 ms to 10 ms getting there. + abr.set_stream_cap(stream_cap_kbps); // Startup link-capacity probe (Automatic sessions): the controller's ceiling is the // negotiated start rate — the conservative 20 Mbps default, historically a box Automatic // could NEVER climb out of. One speed-test burst shortly after the stream settles -- 2.54.0 From e50941c8394eff467edbd98c2a236dda7d57598c Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 17 Aug 2026 00:08:12 +0200 Subject: [PATCH 4/5] fix(win-display): the EDID unlock's expected no-op is not a warning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every host start on an AMD box logged four of these: WARN edid_lock: adl-unlock-mode-off adapter5.connector0[DP] ok=false rc=-8(ADL_ERR_NOT_SUPPORTED) The unlock is deliberately idempotent and runs over every connector, including the ones that were never pinned — and over all of them on a host recovering from an unclean exit. Some drivers answer NOT_SUPPORTED to "turn emulation off" where there is no emulation to turn off, so a perfectly healthy start emitted one warning per connector, saying nothing. Four standing warnings are how a log stops being read: these were the first thing to catch the eye in a field bundle whose actual fault was elsewhere. Scoped to the mode-off call on purpose. adl-unlock-remove is the call that actually clears a pin, so its rc is the one that means something, and it keeps its warning. No behaviour change — the unlock did and does the same thing; only its severity when it no-ops is now honest. --- crates/pf-win-display/src/adl_emul.rs | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/crates/pf-win-display/src/adl_emul.rs b/crates/pf-win-display/src/adl_emul.rs index 98ad51a8..36fc78fb 100644 --- a/crates/pf-win-display/src/adl_emul.rs +++ b/crates/pf-win-display/src/adl_emul.rs @@ -617,6 +617,21 @@ fn journal_path() -> std::path::PathBuf { pf_paths::config_dir().join("edid-lock-active.json") } +/// A non-`ADL_OK` rc that is this call's documented no-op rather than a failure. +/// +/// The unlock is deliberately idempotent and runs over EVERY connector — including the ones that +/// were never pinned, and every connector at all on a host recovering from an unclean exit. Some +/// drivers answer `ADL_ERR_NOT_SUPPORTED` to "turn emulation off" where there is no emulation to +/// turn off, so a clean host start emitted one WARN per connector, every time, saying nothing. +/// Four standing warnings are how a log stops being read. +/// +/// Scoped to the mode-off call on purpose: `adl-unlock-remove` is the call that actually clears a +/// pin, so its rc is the one that means something, and it keeps its warning. +fn is_expected_noop(r: &OpRecord) -> bool { + const ADL_ERR_NOT_SUPPORTED: i32 = -8; + r.op == "adl-unlock-mode-off" && r.rc == ADL_ERR_NOT_SUPPORTED +} + fn tracing_log(prefix: &str, outcome: &RunOutcome) { match outcome { RunOutcome::NoAdl => tracing::info!( @@ -625,7 +640,7 @@ fn tracing_log(prefix: &str, outcome: &RunOutcome) { ), RunOutcome::InitFailed(recs) | RunOutcome::Done(recs) => { for r in recs { - if r.ok() { + if r.ok() || is_expected_noop(r) { tracing::info!("{prefix}: {r}"); } else { tracing::warn!("{prefix}: {r}"); -- 2.54.0 From 44692a0dc1b3a7def1cd88267cb2d208a5b376bc Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 17 Aug 2026 01:36:49 +0200 Subject: [PATCH 5/5] feat(host,client): the host tells the client when it stopped the stream itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Windows exclusive-topology eviction makes the host rebuild its capture ring and encoder in place. It takes a few hundred milliseconds and is entirely host-local: no packet is lost, the link never changes. But the client's bitrate controller decides on 750 ms report windows, and a window straddling that rebuild sees almost no stream. The 0.29 field log: 401 ms of rebuild produced a window reporting actual_kbps=390 against a 20 000 target with loss_ppm=0, which the controller read as congestion — x0.7 and slow start retired, three minutes at ~15 Mbps on a link that never dropped a packet. The client already knows how to throw a window away. `discard_abr_window` feeds the controller nothing, sends no LossReport (so a bogus window cannot spike the host's adaptive FEC) and closes the standing-latency detector as not-loss-free. It had exactly one cause: the tail of the client's own speed test. This adds the second, and it is the one party that actually knows — the host. `PipelineGap` (0x0A, host->client, 9 bytes) carries the rebuild's measured span. A DURATION, never an instant: host and client clocks are not in the same domain — 14.7 s apart in that same log — so an instant would need skew-correcting before it meant anything. The client anchors the gap to its own receive time and gap_ms is evidence for the log rather than an input to arithmetic. The span is read from the transition trace's own total, so the number the client logs is the total_ms on the host's trace line: one number, checkable from either end of a field report. Announced on both in-place rebuild paths, including the mode switch. The mode switch is NOT already covered by the client's `on_mode_switch` reset: that clears the learned caps, the three latency baselines and the proven mark, which does mute OWD/decode/encode for a few windows — but it touches neither `current_kbps`, `probing` nor `bad_windows`, and every signal that needs no baseline (an unrecoverable frame, a flush, heavy loss over a near-empty denominator, a keyframe-ask storm) still scores the straddling window, where one severe verdict costs the same x0.7 plus slow start. Deliberate limitation: only the window in flight is discarded. A rebuild long enough to straddle a window boundary damaged the previous window too, and that one is already decided. Retracting it would mean holding every window back by a window in case a gap follows — trading a rare over-reaction for a permanent one. Additive: no ABI bump, no wire break. 0x0A extends the contiguous 0x01-0x09 rate-control block its only consumer already lives in, and is not in the 0x30 clock block precisely because no clock domain is involved. A client that predates it hits its "unknown control message" arm and keeps today's behaviour; a host that predates it never sends one. Tests: wire round trip including cross-decode against the three identically-shaped rate-control messages either side of it (the type byte is the only thing keeping a gap from re-decoding as a SetBitrate), and an end-to-end pump test driving a real ControlTask over a real QUIC control stream into a real DataPump, asserting the straddling window produces no request and the next one reports normally. 464 core tests green, fmt and clippy clean. Not verified here: the Windows-gated eviction call site compiles on no platform available locally (aws-lc-sys needs windows.h to cross-compile). Type-checked by reading; owed a Windows runner build. Known gaps, stated rather than papered over: a mode-switch rebuild that FAILS keeps streaming the old mode and leaves its stall unannounced, and three other in-place rebuild sites are still silent — the Gaming/Desktop session switch, the ABR re-target's fallback open_video rebuild (~0.6 s, and self-inflicted: the controller causing the stall its next window reads as congestion), and reset_stalled_encoder. --- crates/punktfunk-core/cbindgen.toml | 1 + crates/punktfunk-core/src/client/pump.rs | 7 + .../src/client/pump/control_task.rs | 26 ++ crates/punktfunk-core/src/client/pump/data.rs | 265 +++++++++++++++++- crates/punktfunk-core/src/quic/control.rs | 82 ++++++ crates/punktfunk-host/src/native.rs | 14 +- crates/punktfunk-host/src/native/control.rs | 23 ++ crates/punktfunk-host/src/native/stream.rs | 54 ++++ include/punktfunk_core.h | 8 + 9 files changed, 467 insertions(+), 13 deletions(-) diff --git a/crates/punktfunk-core/cbindgen.toml b/crates/punktfunk-core/cbindgen.toml index 54c97663..2a181f83 100644 --- a/crates/punktfunk-core/cbindgen.toml +++ b/crates/punktfunk-core/cbindgen.toml @@ -221,6 +221,7 @@ include = ["PunktfunkEndReason"] "MSG_PAIR_REQUEST" = "PUNKTFUNK_MSG_PAIR_REQUEST" "MSG_PAIR_RESULT" = "PUNKTFUNK_MSG_PAIR_RESULT" "MSG_PHASE_REPORT" = "PUNKTFUNK_MSG_PHASE_REPORT" +"MSG_PIPELINE_GAP" = "PUNKTFUNK_MSG_PIPELINE_GAP" "MSG_PROBE_REQUEST" = "PUNKTFUNK_MSG_PROBE_REQUEST" "MSG_PROBE_RESULT" = "PUNKTFUNK_MSG_PROBE_RESULT" "MSG_RECONFIGURE" = "PUNKTFUNK_MSG_RECONFIGURE" diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index 9008c91f..4718eaf6 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -165,6 +165,11 @@ pub(super) async fn run_pump(args: WorkerArgs) { // outbound `CtrlRequest::Keyframe` — the one choke point all emitters funnel through — and // the pump drains the count per report window. let recovery_kf = Arc::new(AtomicU32::new(0)); + // Host-announced capture/encode pipeline rebuilds (`PipelineGap`): the control task parks the + // gap's length here and the pump drains it every iteration, discarding the report window in + // flight. A host-local rebuild starves a window of stream without the link doing anything + // wrong, and the controller cannot tell that apart from congestion on its own. + let pipeline_gap = Arc::new(AtomicU32::new(0)); // Host-encode-latency accumulator (the ABR encode signal, see [`EncodeLatAcc`]): the // datagram task adds one sample per 0xCF; the pump drains a window mean per report tick. let encode_lat = Arc::new(Mutex::new(super::frame_channel::EncodeLatAcc::default())); @@ -185,6 +190,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { bitrate_ack: bitrate_ack.clone(), live_bitrate, recovery_kf: recovery_kf.clone(), + pipeline_gap: pipeline_gap.clone(), clock_offset: clock_offset.clone(), clock_gen: clock_gen.clone(), clip_event_tx: clip_event_tx.clone(), @@ -260,6 +266,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { fec_recovered, bitrate_ack, recovery_kf, + pipeline_gap, bitrate_kbps, resolved_bitrate_kbps, negotiated_codec, diff --git a/crates/punktfunk-core/src/client/pump/control_task.rs b/crates/punktfunk-core/src/client/pump/control_task.rs index 1ff1792f..b4c7a0b8 100644 --- a/crates/punktfunk-core/src/client/pump/control_task.rs +++ b/crates/punktfunk-core/src/client/pump/control_task.rs @@ -23,6 +23,11 @@ pub(super) struct ControlTask { /// every emitter funnels through (embedder, `note_frame_index`, the pump's own asks) — the /// pump drains the count per report window as the ABR's recovery signal. pub(super) recovery_kf: Arc, + /// The last host-announced pipeline gap in ms ([`crate::quic::PipelineGap`]), `0` = none + /// pending. Written here on arrival, drained by the pump, which discards the report window in + /// flight — a host-local capture/encoder rebuild is not congestion (the `bitrate_ack` pattern, + /// as an atomic because the value is a plain number the pump only ever swaps out). + pub(super) pipeline_gap: Arc, pub(super) clock_offset: Arc, pub(super) clock_gen: Arc, /// Clipboard metadata events (ClipState/ClipOffer) feed the same event plane the @@ -60,6 +65,7 @@ impl ControlTask { bitrate_ack, live_bitrate, recovery_kf, + pipeline_gap, clock_offset, clock_gen, clip_event_tx, @@ -194,6 +200,26 @@ impl ControlTask { live_bitrate.store(ack.bitrate_kbps, Ordering::Relaxed); } *bitrate_ack.lock().unwrap() = Some(ack.bitrate_kbps); + } else if let Ok(gap) = crate::quic::PipelineGap::decode(&msg) { + // The host rebuilt its capture ring + encoder in place and nothing flowed + // while it did. Park it for the pump, which discards the report window in + // flight: that window carries almost no stream through no fault of the + // link, and one such "congestion" verdict ends slow start for the session + // (the 0.29 field log: 401 ms of rebuild cost three minutes at ~15 Mbps). + // Latest-wins — a second gap inside one window is still one window to + // discard, and the newer number is the one worth logging. Floored at 1 + // because 0 is the slot's "nothing pending": the ANNOUNCEMENT is what + // arms the discard, so a host that rounds its measurement down to zero + // must not silently disarm it. + // + // info, not debug: this is the forensic trail that separates a host-local + // stall from a link event in a field log, and it is rare by construction. + tracing::info!( + gap_ms = gap.gap_ms, + "host rebuilt its capture/encode pipeline — discarding the report \ + window in flight" + ); + pipeline_gap.store(gap.gap_ms.max(1), Ordering::Relaxed); } else if let Ok(echo) = ClockEcho::decode(&msg) { match resync.on_echo(&echo, wall_clock_ns()) { ResyncStep::MoreRounds => { diff --git a/crates/punktfunk-core/src/client/pump/data.rs b/crates/punktfunk-core/src/client/pump/data.rs index 925d3e90..a844b8d8 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -31,6 +31,10 @@ pub(super) struct DataPump { /// Outbound decode-recovery keyframe asks, counted by the control task at its send choke /// point; drained per report window as the ABR's recovery signal. pub(super) recovery_kf: Arc, + /// The host announced a capture/encode pipeline rebuild ([`crate::quic::PipelineGap`]): the + /// gap's length in ms, `0` = none pending. Drained every iteration — see + /// [`take_pipeline_gap`]. + pub(super) pipeline_gap: Arc, /// The embedder's REQUESTED rate (0 = Automatic — the only case the ABR arms). pub(super) bitrate_kbps: u32, /// The rate the host actually configured (echoed in Welcome). @@ -60,6 +64,7 @@ impl DataPump { fec_recovered, bitrate_ack, recovery_kf: pump_recovery_kf, + pipeline_gap: pump_pipeline_gap, bitrate_kbps, resolved_bitrate_kbps, negotiated_codec, @@ -143,14 +148,22 @@ impl DataPump { // in; the embedder path had neither, so an unanswered request wedged the report tick and a // finished one left the ABR window anchored before the burst. let mut was_probing = false; - // Set when a probe ends: the FIRST post-probe report window is discarded outright (no - // LossReport, no standing-latency close, no ABR feed). The `last_*` rebase below cannot - // fully clean it — probe frames still pending in the reassembler age out as - // `frames_dropped` for another LOSS_WINDOW (~120 ms) AFTER the rebase, and the burst may - // have latched `flush_in_window` — and either reads as SEVERE congestion. The 2026-07 - // field report's Automatic session backed off 20→14 Mb/s one second in (exactly one - // report tick after its capacity probe) and, with slow start dead from that first - // "congestion", crawled additively for the entire match. + // The window this closes is discarded outright: no LossReport, no standing-latency close, + // no ABR feed. Two causes, both of them "this window's signals describe something other + // than the link, and one bogus congestion verdict here ends slow start for good": + // + // * a probe just ended. The `last_*` rebase below cannot fully clean the tail — probe + // frames still pending in the reassembler age out as `frames_dropped` for another + // LOSS_WINDOW (~120 ms) AFTER the rebase, and the burst may have latched + // `flush_in_window` — and either reads as SEVERE congestion. The 2026-07 field + // report's Automatic session backed off 20→14 Mb/s one second in (exactly one report + // tick after its capacity probe) and, with slow start dead from that first + // "congestion", crawled additively for the entire match. + // * the HOST announced that it rebuilt its capture ring and encoder in place + // ([`crate::quic::PipelineGap`], drained just below). Nothing flowed while it did, so + // the straddling window carries a fraction of its target with zero loss — the 0.29 + // field log's 401 ms exclusive-topology eviction, which cost that session three + // minutes at ~15 Mbps. let mut discard_abr_window = false; let mut probe_watchdog: Option = None; let (mut owd_sum_ns, mut owd_frames) = (0i128, 0u32); @@ -200,6 +213,30 @@ impl DataPump { tracing::info!("clock re-sync applied — clock-based jump-to-live re-armed"); } } + // A host-announced capture/encode rebuild (see `discard_abr_window` above). Drained + // here rather than at the report tick so the flag is set before the tick that closes + // the window the gap landed in — that is the window whose signals the rebuild + // corrupted, and it is the one we can still do something about. + // + // A rebuild long enough to straddle a window boundary damaged the PREVIOUS window + // too, and that one is already decided: it was fed to the controller and its + // LossReport is on the wire. Retracting it would mean holding every window back by a + // window in case a gap follows, which trades a rare over-reaction for a permanent one. + // So the limitation is deliberate: only the window in flight is discarded. The host + // sends this the moment the rebuild completes, so the announcement lands inside the + // damaged window whenever the rebuild is shorter than a window — the 401 ms field case + // against 750 ms windows, and every case observed so far. A rebuild that outlasts a + // window still leaks its first one: the controller's two-window confirmation holds the + // RATE unless that window also cleared a severe tier, but ANY bad window retires slow + // start, so a leak still costs the doubling climb. + if let Some(gap_ms) = take_pipeline_gap(&pump_pipeline_gap) { + discard_abr_window = true; + tracing::debug!( + gap_ms, + window_ms = last_report.elapsed().as_millis() as u64, + "host pipeline gap — the report window in flight is discarded" + ); + } // Mirror the reassembler's unrecoverable-drop count for the client's keyframe-recovery // loop, and (during a speed test) the packet-level receive counters for the throughput // measurement. Updated every iteration (not just on a produced frame) so they stay current @@ -366,12 +403,15 @@ impl DataPump { window_dropped, ); if discard { - // Probe-tail residue (see `discard_abr_window`): a LossReport from this - // window would also spike the host's adaptive FEC off deliberate overload. + // See `discard_abr_window` for the two causes. The LossReport goes with it + // either way: from a probe tail it would spike the host's adaptive FEC off + // deliberate overload, and across a host rebuild `loss_ppm` is computed over a + // window that received almost nothing — a denominator near zero, where one + // aged-out shard reads as several percent (see `window_loss_ppm`'s own tests). tracing::debug!( loss_ppm, window_dropped, - "discarding the first post-probe ABR window (probe-tail residue)" + "discarding this ABR window (probe tail or a host pipeline gap)" ); } else { let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm })); @@ -704,3 +744,206 @@ impl DataPump { frames.close(); } } + +/// Take the host's pending pipeline gap, if one landed since the last call: `Some(gap_ms)` = the +/// host finished rebuilding its capture ring and encoder, so the report window in flight must be +/// discarded. Drains the slot (the control task writes it; `0` = nothing pending), which is what +/// makes the discard cover exactly ONE window — a rebuild announced once must not go on poisoning +/// windows that were never near it. +fn take_pipeline_gap(slot: &AtomicU32) -> Option { + match slot.swap(0, Ordering::Relaxed) { + 0 => None, + gap_ms => Some(gap_ms), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_pipeline_gap_is_taken_exactly_once() { + let slot = AtomicU32::new(0); + assert_eq!( + take_pipeline_gap(&slot), + None, + "an idle session announces nothing" + ); + slot.store(401, Ordering::Relaxed); + assert_eq!(take_pipeline_gap(&slot), Some(401)); + // The drain is what bounds the damage to ONE window: a rebuild announced once must not + // keep discarding windows that were nowhere near it (each discarded window is also a + // LossReport the host never gets, and its adaptive FEC reads that silence as a clean link). + assert_eq!(take_pipeline_gap(&slot), None); + } + + /// A client-role session over a loopback that carries nothing — the pump under test is not + /// being asked about frames, only about what it does at its report tick. + fn idle_client_session() -> (crate::transport::LoopbackTransport, Session) { + let (host_tp, client_tp) = crate::transport::loopback_pair(0, 0); + let cfg = crate::config::Config { + role: crate::config::Role::Client, + phase: crate::config::ProtocolPhase::P2Punktfunk, + fec: crate::config::FecConfig { + scheme: crate::config::FecScheme::Gf16, + fec_percent: 25, + max_data_per_block: 32, + }, + shard_payload: 1024, + max_frame_bytes: 1 << 20, + encrypt: false, + key: crate::crypto::SessionKey::Aes128Gcm([7u8; 16]), + salt: [1, 2, 3, 4], + loopback_drop_period: 0, + }; + // The host end is returned rather than dropped so the link stays whole for the pump's + // whole run — a half-torn transport is a different test than this one. + (host_tp, Session::new(cfg, Box::new(client_tp)).unwrap()) + } + + /// The client half of the host-rebuild repair, end to end: a real [`PipelineGap`] arrives on a + /// real control stream, the real control task parks it, and the real pump throws away the + /// report window it landed in. + /// + /// What the assertions watch is the window's LOSS REPORT, because that is the discarded + /// window's only externally visible product on an idle session — the ABR feed it also + /// suppresses is the very next branch off the same `discard`, and the controller can't be + /// coaxed into a visible verdict without traffic to decide about. The suppression matters in + /// its own right too: across a gap the window's `loss_ppm` is computed over a denominator of + /// nearly nothing, and reporting that figure would have the host raise FEC against a link that + /// never dropped anything. + /// + /// The second window is asserted too, and is half the point: the discard must cover the window + /// the gap landed in and then get out of the way. + #[tokio::test(flavor = "multi_thread", worker_threads = 3)] + async fn a_host_pipeline_gap_discards_the_report_window_in_flight() { + let server = crate::quic::endpoint::server("127.0.0.1:0".parse().unwrap()).unwrap(); + let addr = server.local_addr().unwrap(); + let client = crate::quic::endpoint::client_insecure().unwrap(); + let accept = tokio::spawn(async move { + let incoming = server.accept().await.expect("incoming"); + (server, incoming.await.expect("host side connects")) + }); + let client_conn = client.connect(addr, "punktfunk").unwrap().await.unwrap(); + let (_server_ep, host_conn) = accept.await.unwrap(); + // The host opens the control stream here (in a session the client opens it during the + // handshake) purely because this test's host end only ever WRITES: a stream the client + // opened would stay invisible to a peer that never sends. + let accept_ctrl = tokio::spawn(async move { client_conn.accept_bi().await.unwrap() }); + let (mut host_send, _host_recv) = host_conn.open_bi().await.unwrap(); + io::write_msg(&mut host_send, &crate::quic::RequestKeyframe.encode()) + .await + .expect("open the stream with a message the client ignores"); + let (ctrl_send, ctrl_recv) = accept_ctrl.await.unwrap(); + + // The slot the control task writes and the pump drains — the whole subject of the test. + let pipeline_gap = Arc::new(AtomicU32::new(0)); + // The control task's own outbound channel: its sender is held to the end of the test so + // the task doesn't exit on a closed request channel mid-run. + let (_task_ctrl_tx, task_ctrl_rx) = tokio::sync::mpsc::channel::(8); + let (clip_event_tx, _clip_event_rx) = std::sync::mpsc::sync_channel(8); + let (cursor_shape_tx, _cursor_shape_rx) = std::sync::mpsc::sync_channel(8); + let (access_tx, _access_rx) = std::sync::mpsc::sync_channel(8); + tokio::spawn( + super::super::control_task::ControlTask { + ctrl_rx: task_ctrl_rx, + ctrl_send, + ctrl_recv: io::MsgReader::new(ctrl_recv), + clock_rtt_ns: None, // no connect handshake ⇒ no re-sync batches to interleave + mode_slot: Arc::new(Mutex::new(crate::config::Mode { + width: 1920, + height: 1080, + refresh_hz: 60, + })), + probe: Arc::new(Mutex::new(ProbeState::default())), + bitrate_ack: Arc::new(Mutex::new(None)), + live_bitrate: Arc::new(AtomicU32::new(0)), + recovery_kf: Arc::new(AtomicU32::new(0)), + pipeline_gap: pipeline_gap.clone(), + clock_offset: Arc::new(std::sync::atomic::AtomicI64::new(0)), + clock_gen: Arc::new(AtomicU32::new(0)), + clip_event_tx, + cursor_shape_tx, + mode_gen: Arc::new(AtomicU32::new(0)), + access_grants: Arc::new(AtomicU32::new(0)), + access_deadline_unix: Arc::new(std::sync::atomic::AtomicU64::new(0)), + access_tx, + } + .run(), + ); + + // The pump. An EXPLICIT bitrate (not Automatic) keeps both the controller and the startup + // capacity probe out of this: the probe would fire at 2 s and discard a window of its own, + // which is the other cause of the very flag under test. + let (pump_ctrl_tx, mut pump_ctrl_rx) = tokio::sync::mpsc::channel::(8); + let shutdown = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let (_host_tp, session) = idle_client_session(); + let pump = DataPump { + session, + frames: Arc::new(FrameChannel::new()), + ctrl_tx: pump_ctrl_tx, + shutdown: shutdown.clone(), + probe: Arc::new(Mutex::new(ProbeState::default())), + hot_tids: Arc::new(Mutex::new(Vec::new())), + clock_offset: Arc::new(std::sync::atomic::AtomicI64::new(0)), + clock_gen: Arc::new(AtomicU32::new(0)), + decode_lat: Arc::new(Mutex::new(DecodeLatAcc::default())), + encode_lat: Arc::new(Mutex::new(Default::default())), + mode_gen: Arc::new(AtomicU32::new(0)), + frames_dropped: Arc::new(std::sync::atomic::AtomicU64::new(0)), + fec_recovered: Arc::new(std::sync::atomic::AtomicU64::new(0)), + bitrate_ack: Arc::new(Mutex::new(None)), + recovery_kf: Arc::new(AtomicU32::new(0)), + pipeline_gap: pipeline_gap.clone(), + bitrate_kbps: 20_000, + resolved_bitrate_kbps: 20_000, + negotiated_codec: crate::quic::CODEC_HEVC, + stream_cap_kbps: 100_000, + }; + let started = Instant::now(); + let pump_thread = std::thread::spawn(move || pump.run()); + + // Mid-window, the way a rebuild actually lands: 200 ms into a 750 ms window. + tokio::time::sleep(Duration::from_millis(200)).await; + io::write_msg( + &mut host_send, + &crate::quic::PipelineGap { gap_ms: 401 }.encode(), + ) + .await + .unwrap(); + + // Past the first report tick (750 ms), nowhere near the second (1500 ms): the window the + // gap landed in must have produced NOTHING. A pump that ignored the gap reports here. + tokio::time::sleep_until( + tokio::time::Instant::from_std(started) + Duration::from_millis(1_150), + ) + .await; + assert!( + pump_ctrl_rx.try_recv().is_err(), + "the window the host's rebuild landed in must be discarded, not reported" + ); + assert_eq!( + pipeline_gap.load(Ordering::Relaxed), + 0, + "and the announcement must be drained, so it can't discard a second window" + ); + + // The NEXT window is clean and must report normally — the discard is one window wide, and + // a pump that had wedged instead of discarding would fail here rather than pass above. + let reported = tokio::time::timeout(Duration::from_millis(1_500), pump_ctrl_rx.recv()) + .await + .expect("the window after the gap reports on schedule"); + assert!( + matches!(reported, Some(CtrlRequest::Loss(_))), + "the window after the gap must produce a loss report — an idle session's only \ + outbound request" + ); + assert!( + started.elapsed() >= Duration::from_millis(1_400), + "and it must be the SECOND window's report, not a late first" + ); + + shutdown.store(true, std::sync::atomic::Ordering::SeqCst); + pump_thread.join().unwrap(); + } +} diff --git a/crates/punktfunk-core/src/quic/control.rs b/crates/punktfunk-core/src/quic/control.rs index e62224c3..9c978d72 100644 --- a/crates/punktfunk-core/src/quic/control.rs +++ b/crates/punktfunk-core/src/quic/control.rs @@ -124,6 +124,37 @@ pub struct BitrateChanged { pub bitrate_kbps: u32, } +/// `host → client`, unsolicited: the host tore its own capture ring and encoder down and rebuilt +/// them in place, and nothing flowed for `gap_ms`. Entirely host-local — no packet was lost, the +/// link never changed — but the client's adaptive-bitrate controller decides on 750 ms report +/// windows, and a window straddling the rebuild sees almost no stream. +/// +/// The 0.29 field log is the case this exists for: an exclusive-topology eviction on a Windows +/// host rebuilt the pipeline for 401 ms, and the straddling window reported `actual_kbps=390` +/// against a 20 000 target with `loss_ppm=0` and a host encode mean of 15 063 µs against a ~2 800 +/// baseline. The controller read that as congestion, backed off ×0.7 and retired slow start, and +/// the session spent the next three minutes at ~15 Mbps on a link that never dropped a packet. +/// The client already knows how to throw a window away — it does exactly that for the tail of its +/// own speed-test probe — so the host announcing the rebuild is all that was missing. +/// +/// A DURATION, never an instant, on purpose: host and client clocks are not in the same domain +/// (14.7 s apart in that same log), so an instant stamped in the host's clock would need +/// skew-correcting before it meant anything on the client. The control stream is reliable and +/// sub-millisecond on a LAN, so the client anchors the gap to its OWN receive time — "the rebuild +/// just ended" — and `gap_ms` is evidence for the log rather than an input to the arithmetic. +/// +/// Fire-and-forget, and sent only after a rebuild that SUCCEEDED. The eviction recovery's failure +/// arm ends the session outright, and the reconnect re-baselines everything the controller had +/// learned; a mode-switch rebuild that fails keeps streaming the old mode, and that one does leave +/// its stall unannounced today — a known gap, not a claim that no such gap exists. +/// +/// A client that predates this hits its "unknown control message" arm and keeps the old behavior. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct PipelineGap { + /// How long the host's pipeline was down, in milliseconds (the rebuild's measured span). + pub gap_ms: u32, +} + /// `client → host`, any time after [`Start`]: run a bandwidth speed test. The host bursts /// filler access units (flagged [`crate::packet::FLAG_PROBE`]) over the data plane at /// `target_kbps` of application goodput for `duration_ms`, *pausing video for the duration*, then @@ -234,6 +265,11 @@ pub const MSG_RFI_REQUEST: u8 = 0x07; pub const MSG_SHARD_PAYLOAD_CHANGED: u8 = 0x08; /// Type byte of [`ShardPayloadAck`]. pub const MSG_SHARD_PAYLOAD_ACK: u8 = 0x09; +/// Type byte of [`PipelineGap`]. 0x0A extends the video/rate-control block (0x01-0x09) it belongs +/// to: its only consumer is the same adaptive-bitrate controller [`LossReport`], [`SetBitrate`] +/// and [`BitrateChanged`] already feed. Deliberately NOT in the 0x30 clock block — it carries a +/// duration precisely so that no clock domain is involved. +pub const MSG_PIPELINE_GAP: u8 = 0x0A; /// Type byte of [`ProbeRequest`]. pub const MSG_PROBE_REQUEST: u8 = 0x20; /// Type byte of [`ProbeResult`]. @@ -440,6 +476,26 @@ impl BitrateChanged { } } +impl PipelineGap { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] gap_ms[5..9] + let mut b = Vec::with_capacity(9); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_PIPELINE_GAP); + b.extend_from_slice(&self.gap_ms.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 9 || &b[0..4] != CTL_MAGIC || b[4] != MSG_PIPELINE_GAP { + return Err(PunktfunkError::InvalidArg("bad PipelineGap")); + } + Ok(PipelineGap { + gap_ms: u32::from_le_bytes(b[5..9].try_into().unwrap()), + }) + } +} + /// Compute a [`LossReport`] `loss_ppm` from one window's session-stat deltas: shards FEC recovered /// (the loss it absorbed), recovered-but-then-arrived shards (`late` — reordered delivery lets a /// block reconstruct early, so those were never lost; netting them out keeps plain reordering from @@ -1273,6 +1329,32 @@ mod tests { assert!(SetBitrate::decode(&LossReport { loss_ppm: 7 }.encode()).is_err()); } + #[test] + fn pipeline_gap_roundtrips() { + // 401 ms is the 0.29 field rebuild verbatim; the rest are the boundaries a duration can + // legitimately take (an instant rebuild, a whole minute of it). + for gap_ms in [1u32, 401, 60_000, u32::MAX] { + let m = PipelineGap { gap_ms }; + assert_eq!(PipelineGap::decode(&m.encode()).unwrap(), m); + } + // 0x0A shares its 9-byte shape with the three rate-control messages either side of it, so + // the type byte is the ONLY thing keeping them apart — a gap that re-decoded as a + // `SetBitrate` would retarget the encoder to 401 kbps. + let gap = PipelineGap { gap_ms: 401 }.encode(); + assert_eq!(gap[4], MSG_PIPELINE_GAP); + assert!(LossReport::decode(&gap).is_err()); + assert!(SetBitrate::decode(&gap).is_err()); + assert!(BitrateChanged::decode(&gap).is_err()); + assert!(PipelineGap::decode(&LossReport { loss_ppm: 401 }.encode()).is_err()); + assert!(PipelineGap::decode(&SetBitrate { bitrate_kbps: 401 }.encode()).is_err()); + assert!(PipelineGap::decode(&BitrateChanged { bitrate_kbps: 401 }.encode()).is_err()); + // …and the neighbouring id (0x09) an old peer would have to fall past to reach its + // "unknown control message" arm. Length is exact — no trailing bytes, no truncation. + assert!(ShardPayloadAck::decode(&gap).is_err()); + assert!(PipelineGap::decode(&[gap.as_slice(), &[0]].concat()).is_err()); + assert!(PipelineGap::decode(&gap[..gap.len() - 1]).is_err()); + } + #[test] fn shard_payload_messages_roundtrip() { for shard_payload in [512u16, 1216, 1408, 8908] { diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index fde17e45..e9424583 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -28,8 +28,8 @@ use punktfunk_core::input::{InputEvent, InputKind}; use punktfunk_core::packet::{FLAG_PIC, FLAG_PROBE, FLAG_SOF}; use punktfunk_core::quic::{ classify, endpoint, io, AccessUpdate, BitrateChanged, ClockEcho, ClockProbe, ColorInfo, - GrantClass, Hello, LossReport, PairRequest, ProbeRequest, ProbeResult, Reconfigure, - Reconfigured, RequestKeyframe, RfiRequest, SetBitrate, Start, Welcome, GRANT_ALL, + GrantClass, Hello, LossReport, PairRequest, PipelineGap, ProbeRequest, ProbeResult, + Reconfigure, Reconfigured, RequestKeyframe, RfiRequest, SetBitrate, Start, Welcome, GRANT_ALL, GRANT_CLIPBOARD, GRANT_GAMEPAD, GRANT_LAUNCH, GRANT_MIC, GRANT_POINTER, }; use punktfunk_core::transport::UdpTransport; @@ -1422,6 +1422,14 @@ async fn serve_session( // downward, with the rebuild it costs. Tell the client instead; `BitrateChanged` already // means exactly this and old clients already handle one arriving unprompted. let (retarget_tx, retarget_rx) = tokio::sync::mpsc::unbounded_channel::(); + // Pipeline-gap announcements, data plane → control task (the same bridge pattern, for the same + // reason: the control task is the control stream's sole writer). A rebuild that keeps the + // session up — a mode switch, or the Windows exclusive-topology eviction recovery — still + // stops the stream dead for a few hundred milliseconds, and the client's adaptive-bitrate + // controller reads the report window that straddles it as congestion. We are the only party + // that knows it was us, so we say so: the channel carries the rebuild's length in ms, and the + // control task turns it into a `PipelineGap` the client answers by discarding that window. + let (gap_tx, gap_rx) = tokio::sync::mpsc::unbounded_channel::(); // Cursor-forward bridge (M2): the encode loop diffs each frame's cursor serial and hands // changed SHAPES here; the control task (the control stream's sole writer) sends them. // Same shape as `probe_result_tx`. Wired even when the channel wasn't negotiated — it @@ -1535,6 +1543,7 @@ async fn serve_session( probe_result_rx, reconfig_result_rx, retarget_rx, + gap_rx, shard_change_rx, shard_ack_tx, cursor_shape_rx, @@ -2124,6 +2133,7 @@ async fn serve_session( probe_result_tx, reconfig_result_tx, retarget_tx, + gap_tx, fec_target: fec_target_dp, phase: phase_ctl, conn: conn_stream, diff --git a/crates/punktfunk-host/src/native/control.rs b/crates/punktfunk-host/src/native/control.rs index 5ad596dd..055558fb 100644 --- a/crates/punktfunk-host/src/native/control.rs +++ b/crates/punktfunk-host/src/native/control.rs @@ -44,6 +44,10 @@ pub(super) async fn run( // Host-initiated bitrate re-target (a rebuild re-resolved an Automatic rate): forwarded to // the client as a `BitrateChanged` so its controller's climb base tracks the real encoder. mut retarget_rx: tokio::sync::mpsc::UnboundedReceiver, + // Pipeline-gap announcements (see `gap_tx`): a rebuild that kept the session up stopped the + // stream for this many ms, forwarded to the client as a `PipelineGap` so its bitrate + // controller discards the report window that straddled our own stall. + mut gap_rx: tokio::sync::mpsc::UnboundedReceiver, // Mid-session shard renegotiation (design/shard-payload-reneg.md): the wire-MTU watcher // asks for a `ShardPayloadChanged` here (this task is the control stream's sole writer), // and the client's `ShardPayloadAck`s flow back on `shard_ack_tx` — the grow gate. @@ -428,6 +432,25 @@ pub(super) async fn run( break; } } + gap = gap_rx.recv() => { + // A rebuild that kept the session up (a mode switch, or the Windows + // exclusive-topology eviction recovery) just finished. Tell the client how long + // its stream was stopped so its bitrate controller can throw the straddling + // report window away instead of reading our own stall as congestion — see + // `PipelineGap`. Sent here because this task is the control stream's sole writer, + // and sent AFTER the fact so `gap_ms` is a measurement rather than a promise. + let Some(gap_ms) = gap else { break }; // data plane gone + tracing::info!( + gap_ms, + "pipeline rebuilt in place — telling the client the stream had a gap" + ); + if io::write_msg(&mut ctrl_send, &PipelineGap { gap_ms }.encode()) + .await + .is_err() + { + break; + } + } correction = reconfig_result_rx.recv() => { // H2 rollback/correction ack: the data plane reports the mode ACTUALLY live // after a rebuild that failed (stayed at the old mode) or that the backend diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 98a1a62a..8b345639 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -1343,6 +1343,10 @@ pub(super) struct SessionContext { /// Host-initiated bitrate re-target → control task → the client's `BitrateChanged`. Fired /// by [`adopt_built_bitrate`] when a rebuild lands on a rate the client wasn't told about. pub(super) retarget_tx: tokio::sync::mpsc::UnboundedSender, + /// Pipeline-gap announcement → control task → the client's + /// [`punktfunk_core::quic::PipelineGap`]. Fired by [`announce_pipeline_gap`] after a rebuild + /// that kept the session up, carrying how long the stream was stopped. + pub(super) gap_tx: tokio::sync::mpsc::UnboundedSender, /// Adaptive-FEC target the control task updates from the client's loss reports. pub(super) fec_target: Arc, /// The QUIC control connection (carries host→client 0xCE source-HDR metadata mid-stream). @@ -1604,6 +1608,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option, gap_ms: u32) { + if gap_ms == 0 { + return; + } + let _ = gap.send(gap_ms); // control task gone ⇒ the session is ending anyway +} + /// Encode-stall recovery: rebuild the encoder in place (keeping capture + the session up) and /// discard the owed in-flight frame records — their AUs died with the old encoder instance. /// Returns `false` when the backend has no in-place rebuild ([`crate::encode::Encoder::reset`]'s diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 16e90303..de053f96 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -1206,6 +1206,14 @@ #define PUNKTFUNK_MSG_SHARD_PAYLOAD_ACK 9 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`PipelineGap`]. 0x0A extends the video/rate-control block (0x01-0x09) it belongs +// to: its only consumer is the same adaptive-bitrate controller [`LossReport`], [`SetBitrate`] +// and [`BitrateChanged`] already feed. Deliberately NOT in the 0x30 clock block — it carries a +// duration precisely so that no clock domain is involved. +#define PUNKTFUNK_MSG_PIPELINE_GAP 10 +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // Type byte of [`ProbeRequest`]. #define PUNKTFUNK_MSG_PROBE_REQUEST 32 -- 2.54.0