feat(core): mid-stream clock re-sync — live offset survives wall-clock steps and drift

Networking-audit deferred plan §2. The host↔client offset was measured once at connect; an NTP step or slow drift silently corrupted the clock-based jump-to-live signal, the ABR one-way-delay signal, and every latency stat — 4a3b1ae2's disarm backstop stopped the IDR storm but lost the detector for the session. Now the client re-estimates mid-stream and recovers it. - quic: ClockResync — the connect-time 8-round probe/echo estimate as a select!-driven state machine (rounds matched by echoed t1, stale batches ignored), plus accept_resync (batch min-RTT ≤ max(2 ms, 1.5× connect RTT) so a congested window can never bias the offset). No wire change: the host has always answered ClockProbe at any time on the control stream. - client: the offset lives in an Arc<AtomicI64> seeded at connect; the control task re-probes every 60 s and immediately after the pump's FIRST no-op clock flush (the "clock stepped under me" signal, sent on the next report tick). On apply: store, reset stale_frames/noop_clock_flushes, re-arm the clock detector if a step had disarmed it. The disarm heuristic stays as the final backstop. Public NativeClient::clock_offset_ns keeps the connect-time value (ABI untouched); new clock_offset_now_ns() / clock_offset_shared() expose the live value. - consumers migrated to the live offset: pf-client-core session stats, the pf-presenter e2e stamp, Windows session/render, Android feeder/drain/ DisplayTracker (the tracker holds the shared handle, not the client, so the leaked render-callback refcount can't pin the session). - probe: --clock-resync runs a second full handshake mid-connection and asserts a sane, consistent estimate. Live against the local canary host: offsets 8646/2139 ns, disagreement 6 µs, 8/8 rounds — OK. Unit tests cover the round collection, stale-echo rejection, batch restart, min-RTT selection, and the acceptance guard. cargo ndk check green. Remaining manual validation: `sudo date -s "+2 sec"` on a live streaming client → expect one no-op flush, a re-sync, re-armed detector, no IDR pulse. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-10 15:43:43 +02:00
parent 68a863866a
commit d4467a44e2
10 changed files with 436 additions and 59 deletions
@@ -17,14 +17,14 @@ use crate::error::{PunktfunkError, Result};
 use crate::input::InputEvent;
 use crate::packet::FLAG_PROBE;
 use crate::quic::{
-    endpoint, io, window_loss_ppm, BitrateChanged, ColorInfo, HdrMeta, Hello, HidOutput,
-    LossReport, ProbeRequest, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, RichInput,
-    SetBitrate, Start, Welcome,
+    accept_resync, endpoint, io, wall_clock_ns, window_loss_ppm, BitrateChanged, ClockEcho,
+    ClockResync, ColorInfo, HdrMeta, Hello, HidOutput, LossReport, ProbeRequest, ProbeResult,
+    Reconfigure, Reconfigured, RequestKeyframe, ResyncStep, RichInput, SetBitrate, Start, Welcome,
 };
 use crate::session::{Frame, Session};
 use crate::transport::UdpTransport;
 use std::collections::VecDeque;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender};
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::{Duration, Instant};
@@ -53,6 +53,10 @@ enum CtrlRequest {
    /// Adaptive bitrate: ask the host to re-target its encoder (kbps). Sent by the pump's
    /// [`BitrateController`] when the user's bitrate setting is Automatic.
    SetBitrate(u32),
+    /// Start a mid-stream clock re-sync batch now (see [`ClockResync`]). Sent by the pump on
+    /// its report tick after the first no-op clock flush — the "the clock stepped under me"
+    /// signal; the control task also self-triggers one every [`CLOCK_RESYNC_INTERVAL`].
+    ClockResync,
 }

 /// What the worker reports to [`NativeClient::connect`] once the handshake lands: the
@@ -70,6 +74,10 @@ struct Negotiated {
    bitrate_kbps: u32,
    /// Host clock minus client clock (ns); `0` = no skew handshake (old host / synced clocks).
    clock_offset_ns: i64,
+    /// Min RTT of the connect-time skew handshake (ns); `None` = the host never answered —
+    /// mid-stream re-syncs are pointless then and stay off. The re-sync acceptance guard
+    /// compares each batch against this baseline ([`accept_resync`]).
+    clock_rtt_ns: Option<u64>,
    /// Resolved encode bit depth: `8`, or `10` for a Main10 / HDR session.
    bit_depth: u8,
    /// Resolved CICP colour signalling.
@@ -195,10 +203,18 @@ const FLUSH_COOLDOWN: Duration = Duration::from_secs(2);
 const NOOP_FLUSH_DATAGRAMS: u64 = 64;

 /// Consecutive no-op clock-triggered flushes (see [`NOOP_FLUSH_DATAGRAMS`]) before the clock-based
-/// detector is disarmed for the rest of the session. The clock-free standing-queue detector stays
-/// armed — it measures the local queue directly and can't be fooled by a clock step.
+/// detector is disarmed. The clock-free standing-queue detector stays armed — it measures the
+/// local queue directly and can't be fooled by a clock step. No longer for the rest of the
+/// session: an applied mid-stream clock re-sync re-arms the detector (the disarm stays as the
+/// final backstop between re-syncs).
 const NOOP_CLOCK_FLUSHES_TO_DISARM: u32 = 2;

+/// Cadence of the control task's periodic mid-stream clock re-sync (see [`ClockResync`]): often
+/// enough to bound slow drift and pick up an NTP step within a minute, rare enough to be free
+/// (8 tiny control messages per batch). The pump additionally fires one immediately after the
+/// FIRST no-op clock flush — the moment a step is actually suspected.
+const CLOCK_RESYNC_INTERVAL: Duration = Duration::from_secs(60);
+
 /// The pre-decode video hand-off from the data-plane pump to the embedder. Unlike the side planes
 /// (self-contained samples that drop the newest on overflow), video AUs are reference-chained under the
 /// host's infinite GOP: dropping ANY frame mid-stream corrupts every dependent frame until the next
@@ -365,6 +381,11 @@ pub struct NativeClient {
    /// so the CPU governor keeps the whole video pipeline on fast cores. Empty on platforms without
    /// `gettid` (see [`current_hot_tid`]).
    hot_tids: Arc<Mutex<Vec<i32>>>,
+    /// The LIVE host↔client clock offset (ns): seeded with the connect-time estimate, then kept
+    /// fresh by the control task's mid-stream re-syncs (every [`CLOCK_RESYNC_INTERVAL`], plus on
+    /// the pump's first no-op clock flush). Shared with the pump and, via
+    /// [`clock_offset_shared`](Self::clock_offset_shared), with embedder latency-math threads.
+    clock_offset: Arc<AtomicI64>,
    worker: Option<std::thread::JoinHandle<()>>,
    /// The currently active session mode (the Welcome's, then updated by every accepted
    /// [`NativeClient::request_mode`]).
@@ -386,7 +407,9 @@ pub struct NativeClient {
    /// Host clock minus client clock (ns), from the connect-time skew handshake. Add it to a local
    /// receive/present timestamp to express it in the host's capture clock (the AU `pts_ns`), making
    /// glass-to-glass latency valid across machines. `0` = no correction (an old host that didn't
-    /// answer, or genuinely synced clocks).
+    /// answer, or genuinely synced clocks). This is the CONNECT-TIME estimate, kept for ABI/compat;
+    /// ongoing latency math should read [`clock_offset_now_ns`](Self::clock_offset_now_ns), which
+    /// follows mid-stream re-syncs after a wall-clock step or drift.
    pub clock_offset_ns: i64,
    /// The encode bit depth the host resolved for this session ([`Welcome::bit_depth`]): `8`, or
    /// `10` for a Main10 / HDR session. `8` for an older host that didn't report it.
@@ -537,6 +560,7 @@ impl NativeClient {
        let frames_dropped = Arc::new(AtomicU64::new(0));
        let fec_recovered = Arc::new(AtomicU64::new(0));
        let hot_tids = Arc::new(Mutex::new(Vec::new()));
+        let clock_offset = Arc::new(AtomicI64::new(0));

        let host = host.to_string();
        let frame_chan_w = frame_chan.clone();
@@ -547,6 +571,7 @@ impl NativeClient {
        let frames_dropped_w = frames_dropped.clone();
        let fec_recovered_w = fec_recovered.clone();
        let hot_tids_w = hot_tids.clone();
+        let clock_offset_w = clock_offset.clone();
        let ctrl_tx_pump = ctrl_tx.clone(); // the data-plane pump sends adaptive-FEC LossReports
        let worker = std::thread::Builder::new()
            .name("punktfunk-client".into())
@@ -599,6 +624,7 @@ impl NativeClient {
                    frames_dropped: frames_dropped_w,
                    fec_recovered: fec_recovered_w,
                    hot_tids: hot_tids_w,
+                    clock_offset: clock_offset_w,
                }));
            })
            .map_err(PunktfunkError::Io)?;
@@ -630,6 +656,7 @@ impl NativeClient {
            frames_dropped,
            fec_recovered,
            hot_tids,
+            clock_offset,
            mode: mode_slot,
            host_fingerprint: negotiated.host_fingerprint,
            resolved_compositor: negotiated.compositor,
@@ -859,6 +886,23 @@ impl NativeClient {
        self.hot_tids.lock().map(|v| v.clone()).unwrap_or_default()
    }

+    /// The LIVE host↔client clock offset (ns): the connect-time skew estimate, kept fresh by
+    /// mid-stream re-syncs (every 60 s, plus immediately when a wall-clock step is suspected).
+    /// Prefer this over the connect-time [`clock_offset_ns`](Self::clock_offset_ns) field for any
+    /// ongoing latency math — after an NTP step or slow drift the connect-time value silently
+    /// corrupts every capture-clock comparison. `0` = no skew handshake (old host / synced clocks).
+    pub fn clock_offset_now_ns(&self) -> i64 {
+        self.clock_offset.load(Ordering::Relaxed)
+    }
+
+    /// Shared handle to the live clock offset for plane threads that outlive a `&self` borrow
+    /// (render/display trackers). Read with [`AtomicI64::load`]`(Ordering::Relaxed)` at each use —
+    /// never cache the value across frames. Holding this does NOT keep the session alive (unlike
+    /// an `Arc<NativeClient>`, whose drop disconnects).
+    pub fn clock_offset_shared(&self) -> Arc<AtomicI64> {
+        self.clock_offset.clone()
+    }
+
    /// Start a bandwidth speed test: ask the host to burst filler over the data plane at
    /// `target_kbps` of goodput for `duration_ms`, *briefly pausing video*. Non-blocking — the
    /// measurement accumulates in the background; poll [`NativeClient::probe_result`] until its
@@ -1084,6 +1128,9 @@ struct WorkerArgs {
    frames_dropped: Arc<AtomicU64>,
    fec_recovered: Arc<AtomicU64>,
    hot_tids: Arc<Mutex<Vec<i32>>>,
+    /// The live clock offset (see [`NativeClient::clock_offset`]): the worker seeds it with the
+    /// connect-time estimate; the control task's mid-stream re-syncs update it.
+    clock_offset: Arc<AtomicI64>,
 }

 /// The worker: QUIC handshake, then the input/datagram/control tasks + the blocking
@@ -1122,6 +1169,7 @@ async fn worker_main(args: WorkerArgs) {
        frames_dropped,
        fec_recovered,
        hot_tids,
+        clock_offset,
    } = args;
    let setup = async {
        let remote: std::net::SocketAddr = join_host_port(&host, port)
@@ -1213,18 +1261,19 @@ async fn worker_main(args: WorkerArgs) {
        // it): align our clock to the host's so the embedder can express receive/present instants in
        // the host's capture clock (the AU `pts_ns`). 0 ⇒ an old host that didn't answer (shared-clock
        // assumption, as before). This is the substrate for glass-to-glass present-time measurement.
-        let clock_offset_ns = match crate::quic::clock_sync(&mut send, &mut recv).await {
-            Some(skew) => {
-                tracing::info!(
-                    offset_ns = skew.offset_ns,
-                    rtt_us = skew.rtt_ns / 1000,
-                    rounds = skew.rounds,
-                    "clock skew estimated (host-client)"
-                );
-                skew.offset_ns
-            }
-            None => 0,
-        };
+        let (clock_offset_ns, clock_rtt_ns) =
+            match crate::quic::clock_sync(&mut send, &mut recv).await {
+                Some(skew) => {
+                    tracing::info!(
+                        offset_ns = skew.offset_ns,
+                        rtt_us = skew.rtt_ns / 1000,
+                        rounds = skew.rounds,
+                        "clock skew estimated (host-client)"
+                    );
+                    (skew.offset_ns, Some(skew.rtt_ns))
+                }
+                None => (0, None),
+            };

        let host_udp = std::net::SocketAddr::new(remote.ip(), welcome.udp_port);
        let transport =
@@ -1248,6 +1297,7 @@ async fn worker_main(args: WorkerArgs) {
                host_fingerprint: fingerprint,
                bitrate_kbps: welcome.bitrate_kbps,
                clock_offset_ns,
+                clock_rtt_ns,
                bit_depth: welcome.bit_depth,
                color: welcome.color,
                chroma_format: welcome.chroma_format,
@@ -1267,8 +1317,14 @@ async fn worker_main(args: WorkerArgs) {
        }
    };
    // Copies the pump needs after `negotiated` is handed over to `connect`.
-    let clock_offset_ns = negotiated.clock_offset_ns;
+    let clock_rtt_ns = negotiated.clock_rtt_ns;
    let resolved_bitrate_kbps = negotiated.bitrate_kbps;
+    // 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);
+    // Bumped by the control task each time a re-sync batch is APPLIED; the pump watches it to
+    // reset its staleness counters and re-arm the clock-based jump-to-live detector.
+    let clock_gen = Arc::new(AtomicU32::new(0));
    let _ = ready_tx.send(Ok(negotiated));

    // Input task: embedder events → QUIC datagrams. Toward a host that advertised
@@ -1352,7 +1408,20 @@ async fn worker_main(args: WorkerArgs) {
        let mode_slot = mode_slot.clone();
        let probe = probe.clone();
        let bitrate_ack = bitrate_ack.clone();
+        let clock_offset = clock_offset.clone();
+        let clock_gen = clock_gen.clone();
        tokio::spawn(async move {
+            // Mid-stream clock re-sync (see [`ClockResync`]): a batch runs every
+            // CLOCK_RESYNC_INTERVAL and whenever the pump asks (CtrlRequest::ClockResync after
+            // its first no-op clock flush). Echoes interleave with the other control replies in
+            // the read arm below; only when the host answered the connect-time handshake — an
+            // old host would just eat the probes.
+            let mut resync = ClockResync::new();
+            let mut resync_tick = tokio::time::interval_at(
+                tokio::time::Instant::now() + CLOCK_RESYNC_INTERVAL,
+                CLOCK_RESYNC_INTERVAL,
+            );
+            resync_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
            loop {
                tokio::select! {
                    req = ctrl_rx.recv() => {
@@ -1363,11 +1432,23 @@ async fn worker_main(args: WorkerArgs) {
                            CtrlRequest::Keyframe => RequestKeyframe.encode(),
                            CtrlRequest::Loss(r) => r.encode(),
                            CtrlRequest::SetBitrate(k) => SetBitrate { bitrate_kbps: k }.encode(),
+                            CtrlRequest::ClockResync => {
+                                if clock_rtt_ns.is_none() {
+                                    continue; // no connect-time handshake — host can't answer
+                                }
+                                resync.begin(wall_clock_ns()).encode()
+                            }
                        };
                        if io::write_msg(&mut ctrl_send, &bytes).await.is_err() {
                            break;
                        }
                    }
+                    _ = resync_tick.tick(), if clock_rtt_ns.is_some() => {
+                        let probe = resync.begin(wall_clock_ns());
+                        if io::write_msg(&mut ctrl_send, &probe.encode()).await.is_err() {
+                            break;
+                        }
+                    }
                    msg = io::read_msg(&mut ctrl_recv) => {
                        let Ok(msg) = msg else { break }; // stream closed
                        if let Ok(ack) = Reconfigured::decode(&msg) {
@@ -1408,6 +1489,35 @@ async fn worker_main(args: WorkerArgs) {
                                "host re-targeted encoder bitrate"
                            );
                            *bitrate_ack.lock().unwrap() = Some(ack.bitrate_kbps);
+                        } else if let Ok(echo) = ClockEcho::decode(&msg) {
+                            match resync.on_echo(&echo, wall_clock_ns()) {
+                                ResyncStep::Probe(p) => {
+                                    if io::write_msg(&mut ctrl_send, &p.encode()).await.is_err() {
+                                        break;
+                                    }
+                                }
+                                ResyncStep::Done { offset_ns, rtt_ns } => {
+                                    // Never let a congested window bias the offset (frames read
+                                    // late exactly then) — keep the old estimate and let the next
+                                    // periodic batch try again.
+                                    if accept_resync(rtt_ns, clock_rtt_ns.unwrap_or(0)) {
+                                        clock_offset.store(offset_ns, Ordering::Relaxed);
+                                        clock_gen.fetch_add(1, Ordering::Relaxed);
+                                        tracing::debug!(
+                                            offset_ns,
+                                            rtt_us = rtt_ns / 1000,
+                                            "mid-stream clock re-sync applied"
+                                        );
+                                    } else {
+                                        tracing::debug!(
+                                            rtt_us = rtt_ns / 1000,
+                                            "clock re-sync batch discarded — RTT above the \
+                                             connect-time baseline (congested window)"
+                                        );
+                                    }
+                                }
+                                ResyncStep::Idle => {}
+                            }
                        } else {
                            tracing::warn!("unknown control message — ignoring");
                        }
@@ -1474,6 +1584,8 @@ async fn worker_main(args: WorkerArgs) {
    let pump_shutdown = shutdown.clone();
    let pump_probe = probe.clone();
    let pump_hot_tids = hot_tids.clone();
+    let pump_clock_offset = clock_offset.clone();
+    let pump_clock_gen = clock_gen.clone();
    let _ = tokio::task::spawn_blocking(move || {
        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
@@ -1504,10 +1616,32 @@ async fn worker_main(args: WorkerArgs) {
        let mut last_flush: Option<Instant> = None;
        // Clock-detector health: consecutive clock-triggered flushes that found no local backlog
        // (see NOOP_FLUSH_DATAGRAMS). Reaching NOOP_CLOCK_FLUSHES_TO_DISARM turns the clock-based
-        // detector off for the session (a clock step / upstream queue it can't fix).
+        // detector off (a clock step / upstream queue it can't fix) — until a mid-stream clock
+        // re-sync lands and re-arms it (`pump_clock_gen` below). The FIRST no-op flush also asks
+        // the control task for an immediate re-sync (via the report tick): the flush finding no
+        // local backlog IS the "the wall clock stepped under me" signal.
        let mut noop_clock_flushes: u32 = 0;
        let mut clock_detector_armed = true;
+        let mut resync_wanted = false;
+        let mut seen_clock_gen = pump_clock_gen.load(Ordering::Relaxed);
        while !pump_shutdown.load(Ordering::SeqCst) {
+            // The live host↔client offset: re-loaded every iteration so an applied mid-stream
+            // re-sync takes effect on the very next frame's latency math.
+            let clock_offset_ns = pump_clock_offset.load(Ordering::Relaxed);
+            // An applied re-sync invalidates the staleness run measured under the OLD offset:
+            // reset the counters and re-arm the clock-based detector if a step had disarmed it.
+            let gen = pump_clock_gen.load(Ordering::Relaxed);
+            if gen != seen_clock_gen {
+                seen_clock_gen = gen;
+                stale_frames = 0;
+                noop_clock_flushes = 0;
+                if !clock_detector_armed {
+                    clock_detector_armed = true;
+                    tracing::info!(
+                        "clock re-sync applied — clock-based jump-to-live re-armed"
+                    );
+                }
+            }
            // 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
@@ -1526,6 +1660,12 @@ async fn worker_main(args: WorkerArgs) {
                p.active && !p.done
            };
            if !probe_active && last_report.elapsed() >= ADAPT_REPORT_INTERVAL {
+                // A no-op clock flush earlier in this window suspected a wall-clock step: fire
+                // the mid-stream re-sync now (once — the 60 s periodic covers everything else).
+                if resync_wanted {
+                    resync_wanted = false;
+                    let _ = ctrl_tx.send(CtrlRequest::ClockResync);
+                }
                let window_dropped = st.frames_dropped.wrapping_sub(last_dropped);
                let loss_ppm = window_loss_ppm(
                    st.fec_recovered_shards.wrapping_sub(last_recovered),
@@ -1640,6 +1780,13 @@ async fn worker_main(args: WorkerArgs) {
                                && dropped == 0
                            {
                                noop_clock_flushes += 1;
+                                if noop_clock_flushes == 1 {
+                                    // First no-op flush = a wall-clock step is the prime
+                                    // suspect: ask for an immediate re-sync (sent on the next
+                                    // report tick). Applied, it resets these counters and
+                                    // re-arms the detector before the disarm below triggers.
+                                    resync_wanted = true;
+                                }
                                if noop_clock_flushes >= NOOP_CLOCK_FLUSHES_TO_DISARM {
                                    clock_detector_armed = false;
                                    tracing::warn!(
@@ -1778,18 +1778,12 @@ pub async fn clock_sync(
    send: &mut quinn::SendStream,
    recv: &mut quinn::RecvStream,
 ) -> Option<ClockSkew> {
-    use std::time::{Duration, SystemTime, UNIX_EPOCH};
-    fn now_ns() -> u64 {
-        SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .map(|d| d.as_nanos() as u64)
-            .unwrap_or(0)
-    }
+    use std::time::Duration;
    const ROUNDS: usize = 8;
    let read_timeout = Duration::from_secs(2);
    let mut samples: Vec<(u64, u64, u64, u64)> = Vec::with_capacity(ROUNDS);
    for _ in 0..ROUNDS {
-        let t1 = now_ns();
+        let t1 = wall_clock_ns();
        let probe = ClockProbe { t1_ns: t1 }.encode();
        if io::write_msg(send, &probe).await.is_err() {
            break;
@@ -1802,7 +1796,7 @@ pub async fn clock_sync(
            },
            _ => break, // timeout or stream error -> old host / no skew support
        };
-        samples.push((echo.t1_ns, echo.t2_ns, echo.t3_ns, now_ns()));
+        samples.push((echo.t1_ns, echo.t2_ns, echo.t3_ns, wall_clock_ns()));
    }
    clock_offset_ns(&samples).map(|(offset_ns, rtt_ns)| ClockSkew {
        offset_ns,
@@ -1811,6 +1805,93 @@ pub async fn clock_sync(
    })
 }

+/// Wall-clock now (ns since the Unix epoch) — the clock the skew handshake stamps and the host
+/// stamps AU `pts_ns` with (CLOCK_REALTIME basis, deliberately NOT monotonic: steps/slew are
+/// exactly what the handshake measures across machines).
+pub fn wall_clock_ns() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0)
+}
+
+/// What [`ClockResync::on_echo`] asks the driver to do next.
+#[derive(Debug, PartialEq, Eq)]
+pub enum ResyncStep {
+    /// Nothing — the echo was stale (a previous batch) or no batch is in flight.
+    Idle,
+    /// Send this next-round probe and keep feeding echoes.
+    Probe(ClockProbe),
+    /// The batch is complete: the min-RTT estimate over its rounds, per [`clock_offset_ns`].
+    Done { offset_ns: i64, rtt_ns: u64 },
+}
+
+/// Mid-stream wall-clock re-sync (networking-audit deferred plan §2): the same 8-round
+/// probe/echo estimate as the connect-time [`clock_sync`], restructured as a state machine so
+/// the client's control task can drive it from its `select!` loop without blocking the stream —
+/// echoes interleave with other control traffic; rounds are matched by the echoed `t1`.
+///
+/// A step or slow drift of either wall clock after connect silently corrupts the clock-based
+/// jump-to-live signal, the ABR one-way-delay signal, and every latency stat. Re-syncing
+/// restores them; the disarm heuristic stays as the final backstop.
+pub struct ClockResync {
+    /// `t1_ns` of the probe in flight; `None` = no batch active. An echo whose `t1` doesn't
+    /// match is stale (an abandoned batch) and ignored.
+    pending_t1: Option<u64>,
+    samples: Vec<(u64, u64, u64, u64)>,
+}
+
+impl ClockResync {
+    /// Rounds per batch — matches the connect-time [`clock_sync`].
+    pub const ROUNDS: usize = 8;
+
+    pub fn new() -> ClockResync {
+        ClockResync {
+            pending_t1: None,
+            samples: Vec::with_capacity(Self::ROUNDS),
+        }
+    }
+
+    /// Start a (new) batch, abandoning any batch still in flight — its late echoes won't match
+    /// `pending_t1` and get ignored. Returns the first probe to send, stamped `now_ns`.
+    pub fn begin(&mut self, now_ns: u64) -> ClockProbe {
+        self.samples.clear();
+        self.pending_t1 = Some(now_ns);
+        ClockProbe { t1_ns: now_ns }
+    }
+
+    /// Feed an inbound [`ClockEcho`] received at `now_ns` (the round's `t4`).
+    pub fn on_echo(&mut self, echo: &ClockEcho, now_ns: u64) -> ResyncStep {
+        if self.pending_t1 != Some(echo.t1_ns) {
+            return ResyncStep::Idle; // stale (abandoned batch) or unsolicited
+        }
+        self.samples.push((echo.t1_ns, echo.t2_ns, echo.t3_ns, now_ns));
+        if self.samples.len() < Self::ROUNDS {
+            self.pending_t1 = Some(now_ns);
+            return ResyncStep::Probe(ClockProbe { t1_ns: now_ns });
+        }
+        self.pending_t1 = None;
+        match clock_offset_ns(&self.samples) {
+            Some((offset_ns, rtt_ns)) => ResyncStep::Done { offset_ns, rtt_ns },
+            None => ResyncStep::Idle, // unreachable: ROUNDS > 0 samples were just collected
+        }
+    }
+}
+
+impl Default for ClockResync {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Acceptance guard for a re-sync batch: apply the new offset only when its min RTT is
+/// comparable to the connect-time RTT — `≤ max(2 ms, 1.5 × connect RTT)`. A congested window
+/// biases the offset by its queueing delay, and frames already read late exactly then; better
+/// to keep the old estimate and let the next batch try again.
+pub fn accept_resync(batch_rtt_ns: u64, connect_rtt_ns: u64) -> bool {
+    batch_rtt_ns <= (connect_rtt_ns + connect_rtt_ns / 2).max(2_000_000)
+}
+
 /// quinn endpoint constructors. Host: self-signed identity (fresh, or persisted PEMs via
 /// [`endpoint::server_with_identity`]). Client: fingerprint pinning / TOFU via
 /// [`endpoint::client_pinned`] ([`endpoint::client_insecure`] is the no-pin special case).
@@ -2846,6 +2927,81 @@ mod tests {
        assert!(clock_offset_ns(&[]).is_none());
    }

+    /// The mid-stream re-sync state machine: 8 rounds collected via matched echoes, stale
+    /// echoes ignored, a restarted batch abandons the old one, and the batch result is the
+    /// min-RTT estimate — the exact behavior the connect-time `clock_sync` loop has.
+    #[test]
+    fn clock_resync_collects_rounds_and_ignores_stale_echoes() {
+        // Host clock +1 ms ahead; symmetric 100 µs one-way paths except one congested round.
+        const OFF: i64 = 1_000_000;
+        let echo_for = |t1: u64, one_way: u64| ClockEcho {
+            t1_ns: t1,
+            t2_ns: (t1 as i64 + one_way as i64 + OFF) as u64,
+            t3_ns: (t1 as i64 + one_way as i64 + OFF) as u64 + 10_000,
+        };
+        let t4_for = |e: &ClockEcho, one_way: u64| (e.t3_ns as i64 - OFF + one_way as i64) as u64;
+
+        let mut rs = ClockResync::new();
+        // An unsolicited echo before any batch is ignored.
+        assert_eq!(rs.on_echo(&echo_for(42, 100_000), 500_000), ResyncStep::Idle);
+
+        let mut probe = rs.begin(1_000_000);
+        // A stale echo (wrong t1: the abandoned pre-begin probe) is ignored mid-batch.
+        assert_eq!(rs.on_echo(&echo_for(42, 100_000), 500_000), ResyncStep::Idle);
+        for round in 0..ClockResync::ROUNDS {
+            // Round 3 is congested (5 ms one-way) — it must lose the min-RTT selection.
+            let one_way = if round == 3 { 5_000_000 } else { 100_000 };
+            let echo = echo_for(probe.t1_ns, one_way);
+            let t4 = t4_for(&echo, one_way);
+            match rs.on_echo(&echo, t4) {
+                ResyncStep::Probe(p) => {
+                    assert!(round < ClockResync::ROUNDS - 1, "batch overran its rounds");
+                    probe = p;
+                }
+                ResyncStep::Done { offset_ns, rtt_ns } => {
+                    assert_eq!(round, ClockResync::ROUNDS - 1, "batch ended early");
+                    assert_eq!(offset_ns, OFF, "min-RTT round recovers the offset exactly");
+                    assert_eq!(rtt_ns, 200_000); // 2×100 µs; host processing (t3−t2) excluded
+                }
+                ResyncStep::Idle => panic!("matched echo must advance the batch"),
+            }
+        }
+        // The batch is done: even a matching-t1 replay no longer advances anything.
+        assert_eq!(
+            rs.on_echo(&echo_for(probe.t1_ns, 100_000), probe.t1_ns + 300_000),
+            ResyncStep::Idle
+        );
+
+        // begin() mid-batch abandons the in-flight batch: its echo is stale afterwards.
+        let old = rs.begin(2_000_000);
+        let fresh = rs.begin(3_000_000);
+        assert_eq!(
+            rs.on_echo(&echo_for(old.t1_ns, 100_000), 2_300_000),
+            ResyncStep::Idle
+        );
+        assert!(matches!(
+            rs.on_echo(&echo_for(fresh.t1_ns, 100_000), 3_300_000),
+            ResyncStep::Probe(_)
+        ));
+    }
+
+    /// The acceptance guard: a batch measured through a congested window (fat RTT) must not
+    /// replace the offset — its queueing delay biases the estimate exactly when frames
+    /// already read late. Floor of 2 ms so a near-zero connect RTT (same-host/LAN) doesn't
+    /// reject every later batch over normal jitter.
+    #[test]
+    fn clock_resync_acceptance_guard() {
+        // Generous connect RTT (10 ms): accept up to 1.5×.
+        assert!(accept_resync(14_000_000, 10_000_000));
+        assert!(!accept_resync(16_000_000, 10_000_000));
+        // Tiny connect RTT (200 µs, wired LAN): the 2 ms floor governs.
+        assert!(accept_resync(1_900_000, 200_000));
+        assert!(!accept_resync(2_100_000, 200_000));
+        // Boundary: exactly at the bound is accepted.
+        assert!(accept_resync(2_000_000, 0));
+        assert!(accept_resync(15_000_000, 10_000_000));
+    }
+
    #[test]
    fn control_messages_disjoint_from_hello() {
        // A Hello uses MAGIC (PKF1); control messages use CTL_MAGIC (PKFc). No Hello — at