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 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}"); 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/abr.rs b/crates/punktfunk-core/src/abr.rs index 1c8c4631..5b542c3a 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: @@ -205,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 @@ -250,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. @@ -356,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), @@ -393,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). /// @@ -572,13 +658,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 +816,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 — @@ -1160,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); @@ -2204,6 +2393,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) diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index 1dae3421..4718eaf6 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); @@ -154,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())); @@ -174,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(), @@ -249,9 +266,11 @@ pub(super) async fn run_pump(args: WorkerArgs) { fec_recovered, bitrate_ack, recovery_kf, + pipeline_gap, 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/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 6457b23c..a844b8d8 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -31,11 +31,19 @@ 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). 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 { @@ -56,9 +64,11 @@ impl DataPump { fec_recovered, bitrate_ack, recovery_kf: pump_recovery_kf, + pipeline_gap: pump_pipeline_gap, 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 +110,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 @@ -133,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); @@ -190,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 @@ -356,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 })); @@ -694,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 f16fad56..bc516d70 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