fix(core/client): fix the four ways a speed-test probe corrupts ABR and the report tick

There is one `ProbeState` per session and no correlation id, and two independent
requesters share it: the pump's startup link-capacity probe and the embedder's
`NativeClient::request_probe` (the shipped "Test connection" in the Windows GUI
and the Linux GTK app). The startup path had a busy check, a watchdog and a
window rebase; the embedder path had none of them, and the two collide by
default — both shipped speed tests call `request_probe` on the statement right
after `connect()`, and the pump's probe fires 2 s later, inside that burst.

Four defects, one cluster:

- The pump overwrote the shared slot unconditionally. Mid-burst it drops
  `base_packets`/`base_bytes`, the pump re-snapshots them against the host's
  full-burst denominator, and the user's speed test reports roughly an order of
  magnitude low; if the result lands first instead, the reset wipes `done` and
  the embedder's poll loop never sees its own measurement. Now it defers and
  retries rather than stealing the slot.

- An unanswered embedder probe never timed out. `probe_active` gates the entire
  report tick — LossReport, the ABR window feed, the standing-latency ladder and
  a pending ClockResync all live inside it. A host that ignores ProbeRequest is
  an anticipated configuration (the startup path was given a 6 s timeout for
  exactly that) so the embedder path could latch `active` forever and silently
  switch off every adaptation mechanism for the rest of the session. Now a
  watchdog covers a probe of either origin.

- `request_probe` latched `active` before `try_send`, so a full or closed ctrl
  channel returned `Closed` to the caller while leaving the session wedged in
  the state above. It now rolls back, mirroring the startup path.

- Only the startup path rebased the ABR window past the burst. Probe filler is
  counted into `bytes_received` for every accepted datagram but never reaches
  the decoder, and the tick is suppressed for the whole burst, so the first
  post-burst window read the burst rate as `actual_kbps`. That poisons
  `proven_kbps`, a monotone high-water mark that is never decayed, which
  disables the x1.5 evidence-gated climb guard for the session and authorizes a
  climb into a rate the decoder has never carried. The rebase now happens on the
  falling edge of any probe, and covers the loss/packet anchors too so the first
  post-burst LossReport isn't divided by a filler-inflated packet count.

Also: `wants_decode_latency` advertised on two of the three terms the pump
actually requires to arm ABR, omitting `resolved_bitrate_kbps > 0`, so against
an old host that reports no rate an embedder fed decode latency to a controller
that never runs.

No wire or ABI change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

crates/punktfunk-core/src/client/mod.rs

@@ -456,9 +456,14 @@ impl NativeClient {
             hot_tids,
             clock_offset,
             decode_lat,
-            // The controller arms exactly when the pump does (see `abr::BitrateController::new`
+            // The controller arms exactly when the pump does — all three terms, not two: Automatic
-            // below): Automatic (the user asked for bitrate 0) and not a rate-pinned PyroWave stream.
+            // (the user asked for bitrate 0), not a rate-pinned PyroWave stream, AND the host
-            wants_decode: bitrate_kbps == 0 && negotiated.codec != crate::quic::CODEC_PYROWAVE,
+            // echoed the rate it actually configured. Dropping the last term made this
+            // over-advertise against an old host that reports no rate, so an embedder fed decode
+            // latency to a controller that never runs.
+            wants_decode: bitrate_kbps == 0
+                && negotiated.codec != crate::quic::CODEC_PYROWAVE
+                && negotiated.bitrate_kbps > 0,
             mode: mode_slot,
             host_fingerprint: negotiated.host_fingerprint,
             resolved_compositor: negotiated.compositor,
@@ -703,14 +708,23 @@ impl NativeClient {
         // Reset the accumulator so a fresh run doesn't blend into the previous one.
         *self.probe.lock().unwrap() = ProbeState {
             active: true,
+            duration_ms,
             ..Default::default()
         };
-        self.ctrl_tx
+        let sent = self
+            .ctrl_tx
             .try_send(CtrlRequest::Probe(ProbeRequest {
                 target_kbps,
                 duration_ms,
             }))
-            .map_err(|_| PunktfunkError::Closed)
+            .map_err(|_| PunktfunkError::Closed);
+        if sent.is_err() {
+            // Nothing was asked of the host, so nothing will ever answer. Leaving `active` latched
+            // would suppress the pump's entire report tick for the rest of the session (the pump
+            // mirrors the startup path's rollback at the same point).
+            self.probe.lock().unwrap().active = false;
+        }
+        sent
     }
 
     /// Read the current speed-test measurement (partial until `done`, final once the host's

crates/punktfunk-core/src/client/probe.rs

@@ -32,6 +32,11 @@ pub(crate) struct ProbeState {
     pub(crate) host_duration_ms: u32,
     /// The host's `ProbeResult` arrived → the measurement is final.
     pub(crate) done: bool,
+    /// The requested burst length, so the pump can arm a watchdog for a host that never answers.
+    /// Without one, an ignored `ProbeRequest` latches `active` forever and the pump's whole report
+    /// tick — loss reports, the ABR window feed, the standing-latency ladder and pending clock
+    /// re-syncs — stays suppressed for the rest of the session.
+    pub(crate) duration_ms: u32,
 }
 
 /// A finished/partial speed-test measurement, returned by [`NativeClient::probe_result`].

crates/punktfunk-core/src/client/pump.rs

@@ -718,6 +718,12 @@ pub(super) async fn run_pump(args: WorkerArgs) {
             && std::env::var("PUNKTFUNK_ABR_PROBE").map_or(true, |v| v != "0"))
         .then(|| Instant::now() + CAPACITY_PROBE_DELAY);
         let mut capacity_probe_deadline: Option<Instant> = None;
+        // Edge detector + watchdog for a probe of EITHER origin (the startup capacity probe or an
+        // embedder speed test via `NativeClient::request_probe`). The startup path had both built
+        // 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;
+        let mut probe_watchdog: Option<Instant> = None;
         let (mut owd_sum_ns, mut owd_frames) = (0i128, 0u32);
         let mut flush_in_window = false;
         // Jump-to-live state (see the guard in the loop below): when the clock-based over-bound
@@ -783,31 +789,70 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                 }
                 p.active && !p.done
             };
+            // A probe just ended (either kind): rebase EVERY window anchor past the burst. Its
+            // FLAG_PROBE filler landed in `bytes_received`/`packets_received` (session.rs counts
+            // every accepted datagram) but never reached the decoder, and the report tick was
+            // suppressed for the whole burst, so `last_*` still points before it. Without this the
+            // first post-burst window reads the burst rate as `actual_kbps` and poisons the ABR's
+            // monotone proven-throughput high-water mark — which never decays — and divides the
+            // window's loss by a packet count inflated with filler.
+            if was_probing && !probe_active {
+                last_recovered = st.fec_recovered_shards;
+                last_late = st.fec_late_shards;
+                last_received = st.packets_received;
+                last_dropped = st.frames_dropped;
+                last_bytes = st.bytes_received;
+                last_report = Instant::now();
+            }
+            // Arm a watchdog on the leading edge of ANY probe, so a host that silently ignores
+            // `ProbeRequest` (an old build — anticipated, see the capacity-probe timeout below)
+            // cannot latch `active` forever and suppress the report tick for the whole session.
+            if !was_probing && probe_active {
+                let burst = Duration::from_millis(pump_probe.lock().unwrap().duration_ms as u64);
+                probe_watchdog = Some(Instant::now() + burst + CAPACITY_PROBE_TIMEOUT);
+            }
+            if !probe_active {
+                probe_watchdog = None;
+            } else if let Some(deadline) = probe_watchdog {
+                if Instant::now() >= deadline {
+                    probe_watchdog = None;
+                    pump_probe.lock().unwrap().active = false;
+                    tracing::warn!(
+                        "speed-test probe unanswered — clearing it so loss reports and ABR resume"
+                    );
+                }
+            }
+            was_probing = probe_active;
             // Fire the startup link-capacity probe once the stream has settled (see the constants
             // above), and fold its measurement into the ABR ceiling when the result lands.
-            if let Some(at) = capacity_probe_at {
+            // Never steal the slot from an embedder speed test in flight: there is one `ProbeState`
-                if Instant::now() >= at {
+            // and no correlation id, so a clobber both wrecks the user's "Test connection" figure
-                    capacity_probe_at = None;
+            // (its base counters get re-snapshotted mid-burst against the full-burst denominator)
-                    *pump_probe.lock().unwrap() = ProbeState {
+            // and mis-scales our own ceiling. Retry once it finishes.
-                        active: true,
+            if capacity_probe_at.is_some_and(|at| Instant::now() >= at) && probe_active {
-                        ..Default::default()
+                capacity_probe_at = Some(Instant::now() + CAPACITY_PROBE_DELAY);
-                    };
+            } else if capacity_probe_at.is_some_and(|at| Instant::now() >= at) {
-                    if ctrl_tx
+                capacity_probe_at = None;
-                        .try_send(CtrlRequest::Probe(ProbeRequest {
+                *pump_probe.lock().unwrap() = ProbeState {
-                            target_kbps: CAPACITY_PROBE_KBPS,
+                    active: true,
-                            duration_ms: CAPACITY_PROBE_MS,
+                    duration_ms: CAPACITY_PROBE_MS,
-                        }))
+                    ..Default::default()
-                        .is_ok()
+                };
-                    {
+                if ctrl_tx
-                        capacity_probe_deadline = Some(Instant::now() + CAPACITY_PROBE_TIMEOUT);
+                    .try_send(CtrlRequest::Probe(ProbeRequest {
-                        tracing::info!(
+                        target_kbps: CAPACITY_PROBE_KBPS,
-                            target_kbps = CAPACITY_PROBE_KBPS,
+                        duration_ms: CAPACITY_PROBE_MS,
-                            duration_ms = CAPACITY_PROBE_MS,
+                    }))
-                            "adaptive bitrate: startup link-capacity probe"
+                    .is_ok()
-                        );
+                {
-                    } else {
+                    capacity_probe_deadline = Some(Instant::now() + CAPACITY_PROBE_TIMEOUT);
-                        pump_probe.lock().unwrap().active = false; // ctrl queue full — skip
+                    tracing::info!(
-                    }
+                        target_kbps = CAPACITY_PROBE_KBPS,
+                        duration_ms = CAPACITY_PROBE_MS,
+                        "adaptive bitrate: startup link-capacity probe"
+                    );
+                } else {
+                    pump_probe.lock().unwrap().active = false; // ctrl queue full — skip
                 }
             }
             if let Some(deadline) = capacity_probe_deadline {