diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index 3e0692b5..b2db9524 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -4116,7 +4116,11 @@ pub struct PunktfunkProbeResult { /// Application goodput bytes / access units the host offered. pub host_bytes: u64, pub host_packets: u32, - /// The host's measured burst duration, milliseconds (the throughput denominator). + /// The throughput denominator, milliseconds: the client-measured burst receive interval + /// (first → last probe-packet arrival) once `done`; the host's measured send-window + /// duration when fewer than two probe packets arrived (no interval to measure from). The + /// host duration alone overstates throughput — its window closes while the bottleneck + /// queue is still draining toward the client. pub elapsed_ms: u32, /// Delivered wire throughput = `recv_bytes * 8 / elapsed_ms` (kilobits/second). pub throughput_kbps: u32, @@ -4130,7 +4134,7 @@ pub struct PunktfunkProbeResult { } /// Start a bandwidth speed test: ask the host to burst filler over the data plane at -/// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 3 Gbps / ≤ 5 s), +/// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 10 Gbps / ≤ 5 s), /// *briefly pausing video*. Non-blocking — poll [`punktfunk_connection_probe_result`] until its /// `done` field is 1. Starting a probe resets any prior measurement. /// diff --git a/crates/punktfunk-core/src/abr.rs b/crates/punktfunk-core/src/abr.rs index ffcae169..f194d7e0 100644 --- a/crates/punktfunk-core/src/abr.rs +++ b/crates/punktfunk-core/src/abr.rs @@ -31,6 +31,10 @@ //! after ~4.5 s clean, ceilinged). Changes are rate-limited (each one costs the IDR the host's //! rebuilt encoder opens with) and the whole controller disables itself against a host that never //! answers [`crate::quic::BitrateChanged`] (an older build that ignores unknown control messages). +//! Standing limits are LEARNED rather than re-poked: two identical short host acks latch the +//! encoder's ceiling (`host_cap_kbps`), two consecutive decode-severe backoffs at a similar rate +//! latch the client decoder's knee (`decode_cap_kbps`) — and both re-probe slowly +//! ([`CAP_REPROBE_WINDOWS`]) so neither latch outlives the condition that taught it. //! //! Climbs are additionally **evidence-gated**. The target is only a *promise* to the encoder — //! how many bits it actually emits depends on the content — so on calm content (a menu, an idle @@ -129,7 +133,16 @@ const ENCODE_SEVERE_US: i64 = 12_000; /// evidence, not a spec limit — without a re-probe, one heavy scene would cap the whole /// session. A still-standing limit just re-teaches itself in two short acks, which the host /// pre-clamps without touching the encoder — the re-probe costs no rebuild, no IDR. +/// The [`decode cap`](BitrateController::decode_cap_kbps) re-probes on the same clock for the +/// same reason: the decoder's knee moves with content and thermals, so its latch must not be +/// permanent either. const CAP_REPROBE_WINDOWS: u32 = 80; +/// Two consecutive decode-driven backoffs latch the +/// [`decode cap`](BitrateController::decode_cap_kbps) only when their pre-backoff rates agree +/// within ±1/8: the decoder's knee is a RATE, so repeated chokes at the same rate are its +/// signature — two unrelated events (a Wi-Fi flush at 300 Mbps, a decode spike at 500) share +/// no knee and must not teach one. +const DECODE_CAP_SIMILAR_DIV: u32 = 8; /// Rolling window (in 750 ms report windows, ~30 s) whose minimum mean is the OWD baseline. /// Long enough to remember the uncongested floor, short enough to follow genuine path changes. const BASELINE_WINDOWS: usize = 40; @@ -137,6 +150,23 @@ const BASELINE_WINDOWS: usize = 40; /// predates bitrate renegotiation and going quiet for the rest of the session. const MAX_UNACKED: u32 = 3; +/// Operator escape hatch: `PUNKTFUNK_ABR_MAX_MBPS` (megabits/second, the +/// `PUNKTFUNK_PYROWAVE_MAX_MBPS` convention) caps the climb ceiling however it is learned. +/// The startup link-capacity probe MEASURES the ceiling, and +/// [`set_ceiling`](BitrateController::set_ceiling)'s deliberate monotonicity makes an inflated +/// measurement permanent for the session — a link that mis-measures (a bursty middlebox, a +/// queue-flattered interval) needs a knob that binds regardless of what any probe claims. +/// `PUNKTFUNK_ABR_PROBE_KBPS` is NOT that knob: it only shrinks the burst target, not what the +/// measurement may conclude. Unset/0/garbage → no cap. Read once per controller, at +/// construction. +fn ceiling_cap_from_env() -> Option { + std::env::var("PUNKTFUNK_ABR_MAX_MBPS") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|&m| m > 0) + .map(|m| m.saturating_mul(1_000)) +} + /// One decision per report window; `Some(kbps)` = send a [`crate::quic::SetBitrate`]. pub(crate) struct BitrateController { /// `false` = permanently off (explicit user bitrate, an old host, or ack silence). @@ -147,6 +177,10 @@ pub(crate) struct BitrateController { /// raises it via [`set_ceiling`](Self::set_ceiling) — that measurement is what lets an /// Automatic session scale past its conservative start. ceiling_kbps: u32, + /// The `PUNKTFUNK_ABR_MAX_MBPS` cap in kbps (see [`ceiling_cap_from_env`]), injected at + /// construction so tests exercise the clamp without touching the process environment. + /// `None` = no cap. + ceiling_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. @@ -178,6 +212,24 @@ pub(crate) struct BitrateController { short_acks: u32, /// Clean windows spent parked at the learned cap (the re-probe clock). cap_probe_windows: u32, + /// The client-decoder rate cap, mirroring [`host_cap_kbps`](Self::host_cap_kbps) for the + /// OTHER end of the pipe: latched when two CONSECUTIVE backoffs carried decode-severe + /// evidence (a deep decode-latency excursion, or a jump-to-live flush — in the + /// decoder-saturation regime the flushed backlog formed BEHIND a decoder that stopped + /// keeping up) at a similar pre-backoff rate. Without it a decoder knee below the link + /// ceiling is a permanent 30–60 s sawtooth: every ×0.7 backoff re-climbs toward a ceiling + /// the decoder can't hold, and each cycle costs a flush plus a dropped-frame burst (the + /// 1440p120 HEVC field case: knee ~490 Mbps under a ~658 Mbps ceiling). Slowly re-probed + /// on the [`CAP_REPROBE_WINDOWS`] clock, exactly like the host cap, so a decoder that + /// recovers (lighter content, thermal headroom) climbs again — the latch is never + /// permanent. + decode_cap_kbps: Option, + /// The previous decode-driven backoff's pre-backoff rate (0 = the last backoff wasn't + /// decode-driven): the reference the next one must land near ([`DECODE_CAP_SIMILAR_DIV`]) + /// to latch the cap — one spurious flush teaches nothing. + decode_backoff_kbps: u32, + /// Clean windows spent parked at the learned decode cap (its re-probe clock). + decode_cap_probe_windows: u32, /// Proven throughput: the session's highest windowed ACTUAL delivered rate seen with flat /// decode latency — the known-good high-water mark climbs are bounded against. Never decays; /// shrinking capacity (thermals, a heavier scene) is the reactive decode signal's job. On @@ -196,10 +248,17 @@ impl BitrateController { /// to build a permanently-disabled controller (explicit bitrate / an old host that didn't /// echo one — no known ceiling to work against). pub(crate) fn new(start_kbps: u32) -> Self { + Self::with_ceiling_cap(start_kbps, ceiling_cap_from_env()) + } + + /// [`new`](Self::new) with the `PUNKTFUNK_ABR_MAX_MBPS` cap injected — the seam the unit + /// tests use so the clamp's behavior never depends on the test process's environment. + fn with_ceiling_cap(start_kbps: u32, ceiling_cap_kbps: Option) -> Self { BitrateController { enabled: start_kbps > 0, current_kbps: start_kbps, ceiling_kbps: start_kbps, + ceiling_cap_kbps, floor_kbps: FLOOR_KBPS.min(start_kbps.max(1)), probing: true, owd_means: VecDeque::with_capacity(BASELINE_WINDOWS), @@ -210,6 +269,9 @@ impl BitrateController { short_ack_kbps: 0, short_acks: 0, cap_probe_windows: 0, + decode_cap_kbps: None, + decode_backoff_kbps: 0, + decode_cap_probe_windows: 0, proven_kbps: 0, bad_windows: 0, clean_windows: 0, @@ -222,8 +284,12 @@ impl BitrateController { /// delivered throughput with headroom already subtracted by the caller). Without this call /// the ceiling stays the negotiated start rate — exactly the old behavior. Never lowers: /// a congested-moment measurement must not shrink authority below what was negotiated - /// (descent is the congestion signals' job). + /// (descent is the congestion signals' job). The `PUNKTFUNK_ABR_MAX_MBPS` cap clamps HERE + /// — the one funnel every learned ceiling passes through — so it binds no matter how the + /// 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)); if self.enabled && kbps > self.ceiling_kbps { self.ceiling_kbps = kbps; } @@ -274,11 +340,16 @@ impl BitrateController { /// An accepted mode switch: the encoder's ceiling and compute knee are properties of the /// MODE (4K120 caps where 1080p60 never would) — drop the mode-scoped learned state. The - /// probe-measured `ceiling_kbps` (a LINK property) survives. + /// decoder's knee is just as mode-scoped (pixel rate drives both ends of the codec), so + /// the decode cap goes with it. The probe-measured `ceiling_kbps` (a LINK property) + /// survives. pub(crate) fn on_mode_switch(&mut self) { self.host_cap_kbps = None; self.short_acks = 0; self.cap_probe_windows = 0; + self.decode_cap_kbps = None; + self.decode_backoff_kbps = 0; + self.decode_cap_probe_windows = 0; self.encode_means.clear(); } @@ -427,6 +498,30 @@ impl BitrateController { } } } + // The decode cap re-probes on the same clock and for the same reason: the knee is + // content- and thermals-dependent evidence, not a spec limit — a decoder that recovers + // must get its headroom back, so the latch clears UPWARD through here rather than ever + // being permanent. A still-standing knee re-latches from the next pair of + // decode-driven backoffs. + if let Some(cap) = self.decode_cap_kbps { + if bad { + self.decode_cap_probe_windows = 0; + } else if self.current_kbps >= cap.saturating_sub(cap / 16) { + self.decode_cap_probe_windows += 1; + if self.decode_cap_probe_windows >= CAP_REPROBE_WINDOWS { + self.decode_cap_probe_windows = 0; + let lifted = cap.saturating_add(cap / 8).min(self.ceiling_kbps); + if lifted > cap { + tracing::debug!( + from_kbps = cap, + to_kbps = lifted, + "adaptive bitrate: re-probing above the learned decode cap" + ); + self.decode_cap_kbps = Some(lifted); + } + } + } + } let cooled = self .last_change .is_none_or(|t| now.duration_since(t) >= CHANGE_COOLDOWN); @@ -436,6 +531,31 @@ impl BitrateController { if (self.bad_windows >= BAD_WINDOWS_TO_DECREASE || (severe && self.bad_windows >= 1)) && self.current_kbps > self.floor_kbps { + // Decode-cap learning (see [`decode_cap_kbps`](Self::decode_cap_kbps)): a backoff + // with decode-severe evidence — the deep decode excursion, or the flush that + // drained the queue behind a stalled decoder — remembers its pre-backoff rate; the + // SECOND consecutive one at a similar rate latches that rate as the decoder's + // knee. One event never latches (a spurious flush must stay a one-off), and a + // backoff without decode evidence in between breaks the streak — whatever it saw, + // it wasn't the same knee. + if decode_severe || flushed { + let rate = self.current_kbps; + let similar = self.decode_backoff_kbps > 0 + && rate.abs_diff(self.decode_backoff_kbps) + <= self.decode_backoff_kbps / DECODE_CAP_SIMILAR_DIV; + if similar && self.decode_cap_kbps.is_none_or(|c| rate < c) { + tracing::info!( + cap_kbps = rate, + "adaptive bitrate: decode cap learned (decoder knee) — climbs stop \ + here until it lifts" + ); + self.decode_cap_kbps = Some(rate.max(self.floor_kbps)); + self.decode_cap_probe_windows = 0; + } + self.decode_backoff_kbps = rate; + } else { + self.decode_backoff_kbps = 0; + } let next = ((self.current_kbps as u64 * 7 / 10) as u32).max(self.floor_kbps); self.bad_windows = 0; return self.request(next, now); @@ -447,11 +567,13 @@ impl BitrateController { // utilized window after a long-enough clean run climbs immediately. let utilized = actual_kbps as u64 * UTILIZATION_DEN >= self.current_kbps as u64 * UTILIZATION_NUM; - // The effective ceiling folds in the host-taught cap: the probe measured the LINK, but - // the host's short acks measured the ENCODER — whichever binds first is the limit. + // The effective ceiling folds in both learned caps: the probe measured the LINK, the + // host's short acks measured the ENCODER, and the decode cap measured the CLIENT + // DECODER — whichever binds first is the limit. let eff_ceiling = self .ceiling_kbps - .min(self.host_cap_kbps.unwrap_or(u32::MAX)); + .min(self.host_cap_kbps.unwrap_or(u32::MAX)) + .min(self.decode_cap_kbps.unwrap_or(u32::MAX)); let cap = eff_ceiling .min(self.proven_kbps.saturating_mul(PROVEN_HEADROOM_NUM) / PROVEN_HEADROOM_DEN); if self.current_kbps < eff_ceiling && utilized && cap > self.current_kbps { @@ -1447,6 +1569,243 @@ mod tests { } } + #[test] + fn env_max_mbps_caps_every_learned_ceiling() { + // PUNKTFUNK_ABR_MAX_MBPS=50 (injected — `new` reads the env exactly once, at + // construction): a probe "measuring" 886 Mbps (the divisor bug's field figure) must + // not out-rank the user's cap… + let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000)); + c.set_ceiling(886_312); + assert_eq!(c.ceiling_kbps, 50_000); + // …while a measurement under the cap stands untouched. + let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000)); + c.set_ceiling(40_000); + assert_eq!(c.ceiling_kbps, 40_000); + // And the climb honors it: slow start doubles 20→40, the capped ceiling truncates the + // next step to 50, then quiet — never a request past the user's limit. + let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000)); + c.set_ceiling(886_312); + let start = Instant::now(); + assert_eq!(run_clean(&mut c, start, 0, 1), Some(40_000)); + c.on_ack(40_000); + assert_eq!(run_clean(&mut c, start, 2, 1), Some(50_000)); + c.on_ack(50_000); + assert_eq!(run_clean(&mut c, start, 4, 20), None); + } + + #[test] + fn decode_cap_latches_after_two_consecutive_decode_severe_backoffs() { + // The 1440p120 field sawtooth: a decoder knee (~500 Mbps) well under the (inflated) + // link ceiling — nothing ever LEARNED the knee, so every re-climb ended in a flush + + // dropped-frame burst. Establish a decode baseline on calm windows, choke twice at the + // same rate, and the second decode-severe backoff must latch the knee. + let mut c = BitrateController::new(500_000); + c.set_ceiling(900_000); + let start = Instant::now(); + // Calm baseline windows (2 Mb/s actual: unutilized, so no climb interferes). + for i in 0..4 { + assert_eq!( + c.on_window( + ticks(start, i), + 0, + 0, + Some(10_000), + Some(8_000), + None, + 2_000, + false, + 0 + ), + None + ); + } + // First deep decode excursion → immediate ×0.7, but ONE event must not latch. + assert_eq!( + c.on_window( + ticks(start, 4), + 0, + 0, + Some(10_000), + Some(60_000), + None, + 490_000, + false, + 0 + ), + Some(350_000) + ); + assert!(c.decode_cap_kbps.is_none()); + // Second consecutive decode-severe backoff at the same pre-backoff rate: latch. + assert_eq!( + c.on_window( + ticks(start, 6), + 0, + 0, + Some(10_000), + Some(60_000), + None, + 490_000, + false, + 0 + ), + Some(350_000) + ); + assert_eq!(c.decode_cap_kbps, Some(500_000)); + // The backoff applies; from here every climb must stop AT the knee — not the 900 Mbps + // link ceiling the old sawtooth kept re-poking. + c.on_ack(350_000); + let mut max_req = 0; + for i in 8..70 { + if let Some(k) = c.on_window( + ticks(start, i), + 0, + 0, + Some(10_000), + Some(8_000), + None, + 1_000_000, + false, + 0, + ) { + assert!(k <= 500_000, "climb past the decode cap: {k}"); + max_req = max_req.max(k); + c.on_ack(k); + } + } + assert_eq!(max_req, 500_000); + assert_eq!(c.current_kbps, 500_000); + assert_eq!(c.decode_cap_kbps, Some(500_000)); + } + + #[test] + fn a_single_flush_or_dissimilar_backoffs_never_latch_a_decode_cap() { + // The latch's false-positive guards. A lone jump-to-live flush (a Wi-Fi clump can + // flush once at ANY rate) backs off but teaches nothing… + let mut c = BitrateController::new(500_000); + c.set_ceiling(900_000); + let start = Instant::now(); + assert_eq!( + c.on_window(ticks(start, 0), 0, 0, None, None, None, 490_000, true, 0), + Some(350_000) + ); + assert!(c.decode_cap_kbps.is_none()); + c.on_ack(350_000); + // …a LOSS-driven backoff in between breaks the streak… + assert_eq!( + c.on_window(ticks(start, 2), 1, 0, None, None, None, 340_000, false, 0), + Some(245_000) + ); + assert!(c.decode_cap_kbps.is_none()); + c.on_ack(245_000); + // …so the next flush counts as a FIRST decode event again — still no latch… + assert_eq!( + c.on_window(ticks(start, 4), 0, 0, None, None, None, 240_000, true, 0), + Some(171_500) + ); + assert!(c.decode_cap_kbps.is_none()); + c.on_ack(171_500); + // …and two consecutive decode events at DISSIMILAR rates (245 vs 171.5 Mbps — no + // common knee) must not latch either. + assert_eq!( + c.on_window(ticks(start, 6), 0, 0, None, None, None, 170_000, true, 0), + Some(120_050) + ); + assert!(c.decode_cap_kbps.is_none()); + } + + #[test] + fn decode_cap_reprobes_after_a_sustained_clean_run() { + // The knee is content/thermals evidence, not a spec limit: after ~60 s parked clean at + // the latched cap, it lifts one step (+12.5 %, ceiling-bounded) — the re-probe path is + // how the latch clears (never permanent), and a still-standing knee just re-latches + // from the next pair of decode-driven backoffs. + let mut c = BitrateController::new(500_000); + c.set_ceiling(900_000); + let start = Instant::now(); + for i in 0..4 { + let _ = c.on_window( + ticks(start, i), + 0, + 0, + Some(10_000), + Some(8_000), + None, + 2_000, + false, + 0, + ); + } + for i in [4, 6] { + let _ = c.on_window( + ticks(start, i), + 0, + 0, + Some(10_000), + Some(60_000), + None, + 490_000, + false, + 0, + ); + } + assert_eq!(c.decode_cap_kbps, Some(500_000)); + // The host's ack parks the session at the knee (its clamp is authoritative). + c.on_ack(500_000); + for i in 0..CAP_REPROBE_WINDOWS { + let _ = c.on_window( + ticks(start, 8 + i), + 0, + 0, + Some(10_000), + Some(8_000), + None, + 490_000, + false, + 0, + ); + } + assert_eq!(c.decode_cap_kbps, Some(500_000 + 500_000 / 8)); + } + + #[test] + fn mode_switch_clears_the_decode_cap() { + // A 1440p120 knee means nothing at the new mode's pixel rate — the decode cap must + // not survive the switch (the probe-measured link ceiling does). + let mut c = BitrateController::new(500_000); + c.set_ceiling(900_000); + let start = Instant::now(); + for i in 0..4 { + let _ = c.on_window( + ticks(start, i), + 0, + 0, + Some(10_000), + Some(8_000), + None, + 2_000, + false, + 0, + ); + } + for i in [4, 6] { + let _ = c.on_window( + ticks(start, i), + 0, + 0, + Some(10_000), + Some(60_000), + None, + 490_000, + false, + 0, + ); + } + assert_eq!(c.decode_cap_kbps, Some(500_000)); + c.on_mode_switch(); + assert!(c.decode_cap_kbps.is_none()); + assert_eq!(c.ceiling_kbps, 900_000); + } + #[test] fn ack_silence_disables_the_controller() { let mut c = BitrateController::new(20_000); diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index eb9234b4..eaece606 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -883,7 +883,7 @@ impl NativeClient { /// `target_kbps` of goodput for `duration_ms`, *briefly pausing video*. Non-blocking — the /// measurement accumulates in the background; poll [`NativeClient::probe_result`] until its /// `done` flag is set. Starting a probe resets any prior measurement. The host clamps both - /// fields (≤ 3 Gbps, ≤ 5 s). + /// fields (≤ 10 Gbps, ≤ 5 s). pub fn request_probe(&self, target_kbps: u32, duration_ms: u32) -> Result<()> { // Reset the accumulator so a fresh run doesn't blend into the previous one. *self.probe.lock().unwrap() = ProbeState { @@ -922,8 +922,12 @@ impl NativeClient { p.rx_bytes_now.saturating_sub(base_b), ) }; - // The host's burst duration is the throughput denominator. bytes × 8 / ms = kilobits/second. - let window_ms = p.host_duration_ms; + // The throughput denominator: the client-measured receive interval once the report + // froze one, the host's send-window duration as the fallback (see + // `ProbeState::measured_interval_ms` for why the host window alone overstates the + // link). Both are 0 until the report lands, so a partial read reports 0 throughput — + // unchanged. bytes × 8 / ms = kilobits/second. + let window_ms = p.throughput_window_ms(); let throughput_kbps = if window_ms > 0 { (delivered_bytes.saturating_mul(8) / window_ms as u64) as u32 } else { diff --git a/crates/punktfunk-core/src/client/probe.rs b/crates/punktfunk-core/src/client/probe.rs index c0a6d4b9..5dae0d8c 100644 --- a/crates/punktfunk-core/src/client/probe.rs +++ b/crates/punktfunk-core/src/client/probe.rs @@ -1,34 +1,50 @@ //! Speed-test probe state (`ProbeState`, pump-mirrored) and the public `ProbeOutcome`. /// Accumulated state of an in-flight / finished speed test. The data-plane pump mirrors the -/// session's packet-level receive counters here; the control task finalizes the delivered figure +/// session's probe-scoped receive counters here; the control task finalizes the delivered figure /// and folds in the host's [`ProbeResult`] when it lands. Read by [`NativeClient::probe_result`]. /// /// Counting at the *packet* level (every delivered wire packet) — not whole reassembled probe AUs — /// is what makes the measurement degrade gracefully: once loss exceeds the FEC budget no AU /// completes, so the old AU-based count cliffed to zero even though most bytes still arrived. +/// Counting *probe* packets only (the reassembler stamps dedicated counters at its FLAG_PROBE +/// routing) keeps video out of the numerator: the burst pauses video, but frames already in +/// flight land during its head, and resumed video lands between the last probe packet and the +/// host's report — both used to inflate the all-datagram byte delta this mirrored before. #[derive(Default)] pub(crate) struct ProbeState { /// A probe is in progress: set by `request_probe`, cleared when the host's [`ProbeResult`] /// lands (a re-probe just overwrites the whole state — the latest one wins). pub(crate) active: bool, - /// `session.stats()` receive counters at the burst's start (snapshotted by the pump on its first - /// tick while active) and latest, mirrored every pump iteration. + /// Probe-scoped receive counters (`Stats::probe_*`) at the burst's start (snapshotted by the + /// pump on its first tick while active) and latest, mirrored every pump iteration. pub(crate) base_packets: Option, pub(crate) base_bytes: Option, pub(crate) rx_packets_now: u64, pub(crate) rx_bytes_now: u64, + /// First / last probe-packet arrival stamps (monotonic ns, 0 = none yet), mirrored from the + /// probe-scoped session counters. Their difference is the interval the delivered bytes + /// actually arrived in — the honest throughput denominator (see + /// [`measured_interval_ms`](Self::measured_interval_ms)). + pub(crate) first_arrival_ns: u64, + pub(crate) last_arrival_ns: u64, /// Delivered wire packets / plaintext bytes (header + shard), frozen when the host's report lands /// (so resumed video after the burst can't inflate them). pub(crate) delivered_packets: u64, pub(crate) delivered_bytes: u64, + /// The client-measured receive interval (ms), frozen alongside the delivered figures; 0 = no + /// usable interval (the burst delivered fewer than two probe packets) — consumers fall back + /// to [`host_duration_ms`](Self::host_duration_ms) via + /// [`throughput_window_ms`](Self::throughput_window_ms). + pub(crate) client_interval_ms: u32, /// The host's end-of-burst report. pub(crate) host_goodput_bytes: u64, pub(crate) host_au: u32, /// Wire packets the host actually put on the link, and the ones its send buffer dropped. pub(crate) host_wire_packets: u32, pub(crate) host_send_dropped: u32, - /// The host's measured burst duration (the throughput denominator). + /// The host's measured burst duration (the throughput denominator's FALLBACK — see + /// [`throughput_window_ms`](Self::throughput_window_ms)). pub(crate) host_duration_ms: u32, /// The host's `ProbeResult` arrived → the measurement is final. pub(crate) done: bool, @@ -39,6 +55,40 @@ pub(crate) struct ProbeState { pub(crate) duration_ms: u32, } +impl ProbeState { + /// The client-measured receive interval of a finished burst, in ms: first → last + /// probe-packet arrival, floored at 1 (a sub-ms burst divided by 0 ms would read as + /// infinite throughput). `None` — the caller falls back to the host's duration — when + /// fewer than two probe packets arrived or the stamps are degenerate (unset / identical / + /// reversed): a single arrival spans no interval. + /// + /// Why not the host's `duration_ms`: it measures the SEND window, which closes while the + /// bottleneck (switch/kernel) queue is still draining toward the client — the tail of the + /// bytes lands *after* it. Dividing client-side bytes by the host-side window therefore + /// overstates the link: a 1 GbE link under a 2 Gbps burst target "measured" 1266 Mbps and + /// handed the ABR an 886 Mbps ceiling it could never deliver — and + /// [`set_ceiling`](crate::abr::BitrateController::set_ceiling) never lowers, so the lie + /// was permanent for the session. + pub(crate) fn measured_interval_ms(first_ns: u64, last_ns: u64, packets: u64) -> Option { + if packets < 2 || first_ns == 0 || last_ns <= first_ns { + return None; + } + let ms = ((last_ns - first_ns) / 1_000_000).max(1); + Some(u32::try_from(ms).unwrap_or(u32::MAX)) + } + + /// The throughput denominator, in ms: the client-measured receive interval when the burst + /// produced one, else the host's send-window duration (an old measurement is better than + /// none — and strictly conservative territory only when packets were too few to matter). + pub(crate) fn throughput_window_ms(&self) -> u32 { + if self.client_interval_ms > 0 { + self.client_interval_ms + } else { + self.host_duration_ms + } + } +} + /// A finished/partial speed-test measurement, returned by [`NativeClient::probe_result`]. #[derive(Clone, Copy, Debug, Default)] pub struct ProbeOutcome { @@ -50,7 +100,11 @@ pub struct ProbeOutcome { /// Application goodput bytes / access units the host offered. pub host_bytes: u64, pub host_packets: u32, - /// The burst duration the host measured, in milliseconds (the throughput denominator). + /// The throughput denominator, in milliseconds: the client-measured receive interval + /// (first → last probe-packet arrival) once `done`; the host's measured send-window + /// duration when the burst delivered fewer than two probe packets (no interval to measure + /// from). The host duration alone overstates throughput — its window closes while the + /// bottleneck queue is still draining toward the client. pub elapsed_ms: u32, /// Delivered wire throughput = `recv_bytes * 8 / elapsed_ms` (kilobits/second). The figure to /// drive a [`Hello::bitrate_kbps`] choice from (allow headroom for the FEC overhead + loss). @@ -66,3 +120,63 @@ pub struct ProbeOutcome { pub wire_packets_sent: u32, pub send_dropped: u32, } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn interval_needs_two_packets_and_a_nonzero_span() { + // <2 packets: no interval exists — the caller must fall back to the host duration. + assert_eq!(ProbeState::measured_interval_ms(0, 0, 0), None); + assert_eq!( + ProbeState::measured_interval_ms(5_000_000, 5_000_000, 1), + None + ); + // Two packets in the same ns / a reversed pair / an unset first stamp: same fallback. + assert_eq!( + ProbeState::measured_interval_ms(5_000_000, 5_000_000, 2), + None + ); + assert_eq!( + ProbeState::measured_interval_ms(9_000_000, 5_000_000, 2), + None + ); + assert_eq!(ProbeState::measured_interval_ms(0, 5_000_000, 2), None); + } + + #[test] + fn interval_is_floored_at_one_ms() { + // Two packets 0.4 ms apart truncate to 0 ms — the floor keeps the division honest + // instead of infinite. + assert_eq!(ProbeState::measured_interval_ms(1_000, 401_000, 2), Some(1)); + } + + #[test] + fn interval_measures_first_to_last_arrival() { + assert_eq!( + ProbeState::measured_interval_ms(1_000_000, 801_000_000, 1_000), + Some(800) + ); + } + + #[test] + fn throughput_window_falls_back_to_the_host_duration() { + // No client interval frozen (a <2-packet burst) → the host's send window is the + // denominator, exactly the old behavior. + let p = ProbeState { + host_duration_ms: 800, + ..Default::default() + }; + assert_eq!(p.throughput_window_ms(), 800); + // With an interval, the client measurement wins — the 1 GbE field case: the same + // bytes over 1010 ms instead of the host's 800 ms is the difference between an + // honest ~940 Mbps and an impossible 1266 Mbps. + let p = ProbeState { + client_interval_ms: 1_010, + host_duration_ms: 800, + ..Default::default() + }; + assert_eq!(p.throughput_window_ms(), 1_010); + } +} diff --git a/crates/punktfunk-core/src/client/pump/control_task.rs b/crates/punktfunk-core/src/client/pump/control_task.rs index 50765441..291f47a1 100644 --- a/crates/punktfunk-core/src/client/pump/control_task.rs +++ b/crates/punktfunk-core/src/client/pump/control_task.rs @@ -132,12 +132,24 @@ impl ControlTask { } } else if let Ok(result) = ProbeResult::decode(&msg) { let mut p = probe.lock().unwrap(); - // Freeze the delivered figures now (the burst is done), before resumed - // video can inflate the packet counters. + // Freeze the delivered figures now (the burst is done). The mirrored + // counters are probe-scoped (stamped at the reassembler's FLAG_PROBE + // routing), so video around the burst inflates nothing; the client's + // first→last arrival interval is frozen with them — the denominator + // that measures when the bytes actually ARRIVED, not when the host + // stopped sending (its window closes while the bottleneck queue is + // still draining this way, which is how a 1 GbE link once "measured" + // 1266 Mbps). let base_p = p.base_packets.unwrap_or(p.rx_packets_now); let base_b = p.base_bytes.unwrap_or(p.rx_bytes_now); p.delivered_packets = p.rx_packets_now.saturating_sub(base_p); p.delivered_bytes = p.rx_bytes_now.saturating_sub(base_b); + p.client_interval_ms = ProbeState::measured_interval_ms( + p.first_arrival_ns, + p.last_arrival_ns, + p.delivered_packets, + ) + .unwrap_or(0); p.host_goodput_bytes = result.bytes_sent; p.host_au = result.packets_sent; p.host_wire_packets = result.wire_packets_sent; @@ -151,6 +163,7 @@ impl ControlTask { send_dropped = result.send_dropped, duration_ms = result.duration_ms, delivered_packets = p.delivered_packets, + client_interval_ms = p.client_interval_ms, "speed-test probe result" ); } else if let Ok(ack) = BitrateChanged::decode(&msg) { diff --git a/crates/punktfunk-core/src/client/pump/data.rs b/crates/punktfunk-core/src/client/pump/data.rs index 6b5527dc..a83339b8 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -200,10 +200,23 @@ impl DataPump { let probe_active = { let mut p = pump_probe.lock().unwrap(); if p.active && !p.done { - p.rx_packets_now = st.packets_received; - p.rx_bytes_now = st.bytes_received; - p.base_packets.get_or_insert(st.packets_received); - p.base_bytes.get_or_insert(st.bytes_received); + // Arm edge (first mirror tick): zero the arrival stamps before the burst can + // claim them — the ProbeRequest is still queued locally (the burst starts a + // round trip later), so the reset cannot race a probe packet. `st` predates + // the reset, so the stamps mirror 0 on this tick and live values after. + let arming = p.base_bytes.is_none(); + if arming { + session.reset_probe_arrivals(); + } + p.rx_packets_now = st.probe_packets_received; + p.rx_bytes_now = st.probe_bytes_received; + (p.first_arrival_ns, p.last_arrival_ns) = if arming { + (0, 0) + } else { + (st.probe_first_arrival_ns, st.probe_last_arrival_ns) + }; + p.base_packets.get_or_insert(st.probe_packets_received); + p.base_bytes.get_or_insert(st.probe_bytes_received); } p.active && !p.done }; @@ -280,16 +293,23 @@ impl DataPump { if p.done { capacity_probe_deadline = None; // An all-zero reply is a decline (old host / probe-less build) — keep the - // negotiated ceiling. Otherwise: delivered wire kbps × 0.7. + // negotiated ceiling. Otherwise: delivered wire kbps × 0.7, over the + // CLIENT-measured receive interval (the host's send window closes while the + // bottleneck queue is still draining toward us, so dividing by ITS duration + // overstates the link — a 1 GbE link "measured" 1266 Mbps, and the inflated + // ceiling is permanent because set_ceiling never lowers); the host duration + // is the fallback when the burst delivered too few packets for an interval. if p.host_duration_ms > 0 && p.delivered_bytes > 0 { - let delivered_kbps = (p.delivered_bytes.saturating_mul(8) - / p.host_duration_ms.max(1) as u64) - as u32; + let window_ms = p.throughput_window_ms(); + let delivered_kbps = + (p.delivered_bytes.saturating_mul(8) / window_ms.max(1) as u64) as u32; let ceiling = delivered_kbps.saturating_mul(7) / 10; abr.set_ceiling(ceiling); tracing::info!( delivered_kbps, ceiling_kbps = ceiling, + client_interval_ms = p.client_interval_ms, + host_duration_ms = p.host_duration_ms, "adaptive bitrate: link-capacity probe done — climb ceiling set" ); } else { diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index 26507e62..144cfa37 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -409,6 +409,27 @@ impl Reassembler { // can neither advance the video anchor nor be dropped as stale against it (and its aged-out // frames never count as `frames_dropped`, which would fire video loss recovery). let is_probe = hdr.user_flags & (FLAG_PROBE as u32) != 0; + if is_probe { + // Probe-scoped receive accounting (the speed-test numerator + denominator, see + // `Stats::probe_first_arrival_ns`), stamped at the routing decision so video in + // flight around the burst contaminates neither the byte count nor the arrival + // stamps. Byte unit mirrors `bytes_received` (whole plaintext packet). The first + // probe packet since the pump armed the probe claims the first-arrival slot (the + // pump zeroes it before the burst can reach the host); every probe packet + // refreshes the last-arrival stamp. + let now_ns = crate::stats::now_monotonic_ns(); + StatsCounters::add(&stats.probe_packets_received, 1); + StatsCounters::add(&stats.probe_bytes_received, pkt.len() as u64); + let _ = stats.probe_first_arrival_ns.compare_exchange( + 0, + now_ns, + std::sync::atomic::Ordering::Relaxed, + std::sync::atomic::Ordering::Relaxed, + ); + stats + .probe_last_arrival_ns + .store(now_ns, std::sync::atomic::Ordering::Relaxed); + } let win = if is_probe { probe } else { video }; win.advance_window( hdr.frame_index, diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index fd15e672..8ac9033b 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -218,6 +218,18 @@ impl Session { self.stats.snapshot() } + /// Re-arm the probe-scoped arrival stamps (see [`Stats::probe_first_arrival_ns`]): zero + /// them so the NEXT burst's first packet claims the first-arrival slot. Called by the + /// client pump when it arms a probe, strictly before the burst can have reached the host + /// (the `ProbeRequest` is still queued locally) — so the reset cannot race a probe packet. + /// The cumulative probe byte/packet counters are left alone: per-burst deltas come from + /// base snapshots, the same pattern the total counters use. + pub fn reset_probe_arrivals(&self) { + let l = std::sync::atomic::Ordering::Relaxed; + self.stats.probe_first_arrival_ns.store(0, l); + self.stats.probe_last_arrival_ns.store(0, l); + } + /// Wrap a packet for the wire: when encrypting, prepend the 8-byte big-endian /// sequence (the receiver derives the GCM nonce from it) then the ciphertext. /// Seal one plaintext packet into the reused `wire` buffer in place (no allocation): the wire is diff --git a/crates/punktfunk-core/src/stats.rs b/crates/punktfunk-core/src/stats.rs index 44f4ec7c..d293afd5 100644 --- a/crates/punktfunk-core/src/stats.rs +++ b/crates/punktfunk-core/src/stats.rs @@ -1,6 +1,19 @@ //! Live counters for the frame-pacing / quality logic and the web UI. use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::OnceLock; +use std::time::Instant; + +/// Monotonic now, in ns since an arbitrary process-wide epoch — the basis for the probe +/// arrival stamps below. Monotonic on purpose: the stamps are only ever DIFFERENCED on this +/// machine (the burst's receive interval), and a wall-clock step mid-burst — the exact event +/// the clock re-sync machinery exists for — must not corrupt the one measurement the ABR +/// ceiling is built from, so the CLOCK_REALTIME basis `pts_ns` uses is wrong here. Floored +/// at 1 so a stamp can never collide with the 0 = "unset" sentinel. +pub(crate) fn now_monotonic_ns() -> u64 { + static EPOCH: OnceLock = OnceLock::new(); + (EPOCH.get_or_init(Instant::now).elapsed().as_nanos() as u64).max(1) +} /// Immutable snapshot, copied across the C ABI as `PunktfunkStats`. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -26,11 +39,29 @@ pub struct Stats { pub fec_late_shards: u64, pub bytes_sent: u64, pub bytes_received: u64, + /// Probe-scoped receive counters: wire packets / plaintext bytes carrying + /// [`FLAG_PROBE`](crate::packet::FLAG_PROBE) (speed-test filler), counted at the + /// reassembler's probe routing decision. `bytes_received` counts EVERY accepted datagram, + /// so a speed-test numerator built from it inherits whatever video was in flight around + /// the burst — these keep video out of the probe math. Deliberately NOT mirrored into the + /// C-ABI `PunktfunkStats` (probe measurements surface via `ProbeOutcome`). + pub probe_packets_received: u64, + pub probe_bytes_received: u64, + /// First / last probe-packet arrival (monotonic ns, see [`now_monotonic_ns`]; 0 = none + /// since the last probe arm). Their difference is the burst's client-side receive + /// interval — the honest speed-test denominator: the host's send window closes while the + /// switch/kernel queue toward the client is still draining, so dividing client bytes by + /// the HOST duration overstates the link (a 1 GbE link "measured" 1266 Mbps). The client + /// pump zeroes both when it arms a probe (`Session::reset_probe_arrivals`). + pub probe_first_arrival_ns: u64, + pub probe_last_arrival_ns: u64, } /// Atomic accumulators owned by a [`Session`](crate::session::Session). Snapshot to /// [`Stats`] for readers. `Relaxed` ordering is fine: these are monotonic counters -/// read for display, never used to synchronize other memory. +/// read for display, never used to synchronize other memory. (The two probe arrival +/// stamps are the exception — slots, not counters — but they carry no synchronization +/// duty either: they are read hundreds of ms after the last write.) #[derive(Default)] pub struct StatsCounters { pub frames_submitted: AtomicU64, @@ -44,6 +75,10 @@ pub struct StatsCounters { pub fec_late_shards: AtomicU64, pub bytes_sent: AtomicU64, pub bytes_received: AtomicU64, + pub probe_packets_received: AtomicU64, + pub probe_bytes_received: AtomicU64, + pub probe_first_arrival_ns: AtomicU64, + pub probe_last_arrival_ns: AtomicU64, } impl StatsCounters { @@ -66,6 +101,10 @@ impl StatsCounters { fec_late_shards: self.fec_late_shards.load(l), bytes_sent: self.bytes_sent.load(l), bytes_received: self.bytes_received.load(l), + probe_packets_received: self.probe_packets_received.load(l), + probe_bytes_received: self.probe_bytes_received.load(l), + probe_first_arrival_ns: self.probe_first_arrival_ns.load(l), + probe_last_arrival_ns: self.probe_last_arrival_ns.load(l), } } } diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 4974b382..5905efd6 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -71,7 +71,7 @@ pub(super) fn synthetic_stream( Ok(()) } -/// Bounds a speed-test [`ProbeRequest`] before bursting: a 3 Gbps / 5 s ceiling keeps a probe from +/// Bounds a speed-test [`ProbeRequest`] before bursting: a 10 Gbps / 5 s ceiling keeps a probe from /// monopolizing the link or stalling the stream for too long. The ceiling is set ABOVE the session /// bitrate cap ([`MAX_BITRATE_KBPS`], 2 Gbps) on purpose — a probe should be able to demonstrate /// headroom past the rate a session will actually be configured to use, so the client can pick a diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 2bc5d07e..27706ff3 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -1767,7 +1767,11 @@ typedef struct { // Application goodput bytes / access units the host offered. uint64_t host_bytes; uint32_t host_packets; - // The host's measured burst duration, milliseconds (the throughput denominator). + // The throughput denominator, milliseconds: the client-measured burst receive interval + // (first → last probe-packet arrival) once `done`; the host's measured send-window + // duration when fewer than two probe packets arrived (no interval to measure from). The + // host duration alone overstates throughput — its window closes while the bottleneck + // queue is still draining toward the client. uint32_t elapsed_ms; // Delivered wire throughput = `recv_bytes * 8 / elapsed_ms` (kilobits/second). uint32_t throughput_kbps; @@ -2846,7 +2850,7 @@ PunktfunkStatus punktfunk_connection_wants_decode_latency(const PunktfunkConnect #if defined(PUNKTFUNK_FEATURE_QUIC) // Start a bandwidth speed test: ask the host to burst filler over the data plane at -// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 3 Gbps / ≤ 5 s), +// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 10 Gbps / ≤ 5 s), // *briefly pausing video*. Non-blocking — poll [`punktfunk_connection_probe_result`] until its // `done` field is 1. Starting a probe resets any prior measurement. //