diff --git a/clients/android/native/src/decode.rs b/clients/android/native/src/decode.rs index 853cddd4..e519d5dd 100644 --- a/clients/android/native/src/decode.rs +++ b/clients/android/native/src/decode.rs @@ -18,7 +18,7 @@ use punktfunk_core::error::PunktfunkError; use punktfunk_core::session::Frame; use std::collections::VecDeque; use std::ffi::c_void; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, Ordering}; use std::sync::{mpsc, Arc, Mutex}; use std::time::{Duration, Instant}; @@ -213,12 +213,12 @@ fn run_sync( // Skew-corrected latency stats (spec: design/stats-unification.md) use the negotiated // host-minus-client clock offset (0 if the host didn't answer the skew handshake — then the // HUD flags it "(same-host clock)"). - let clock_offset = client.clock_offset_ns; + let clock_offset = client.clock_offset_shared(); // Display stage (spec `display` + the capture→displayed headline): frames released with // render = true are parked in the tracker; the OnFrameRendered callback pairs them with // SurfaceFlinger's render timestamp. `render_cb` is the callback's leaked Arc refcount, // reclaimed after the codec is dropped below. - let tracker = DisplayTracker::new(stats.clone(), clock_offset); + let tracker = DisplayTracker::new(stats.clone(), clock_offset.clone()); let render_cb = install_render_callback(&codec, &tracker); // HUD stage split: receipt timestamps keyed by the pts we queue into the codec, so the decoded // point (output-buffer dequeue — MediaCodec round-trips presentationTimeUs) can be paired back @@ -256,6 +256,7 @@ fn run_sync( // the output buffer) for the decoded-point pairing in `drain`. if stats.enabled() { let received_ns = now_realtime_ns(); + let clock_offset = clock_offset.load(Ordering::Relaxed); let lat_ns = received_ns + clock_offset as i128 - frame.pts_ns as i128; let lat_us = (lat_ns > 0 && lat_ns < 10_000_000_000) .then_some((lat_ns / 1000) as u64); @@ -320,7 +321,7 @@ fn run_sync( wait, &stats, &mut in_flight, - clock_offset, + clock_offset.load(Ordering::Relaxed), &tracker, ); rendered += r; @@ -418,8 +419,10 @@ fn now_monotonic_ns() -> i128 { /// endpoint whenever the platform delivers render callbacks). struct DisplayTracker { stats: Arc, - /// Host-minus-client clock offset (ns) for the skew-corrected end-to-end sample. - clock_offset: i64, + /// Live host-minus-client clock offset (ns) for the skew-corrected end-to-end sample — + /// loaded per callback so mid-stream re-syncs apply. Holding the handle (not the client) + /// keeps the leaked render-callback refcount from pinning the whole session alive. + clock_offset: Arc, /// `(pts_us, decoded_real_ns)` of frames released with `render = true`, in release order, /// awaiting their callback. Pushes are HUD-gated by the caller, so this stays empty (and the /// callback early-outs) while the overlay is hidden. @@ -427,7 +430,10 @@ struct DisplayTracker { } impl DisplayTracker { - fn new(stats: Arc, clock_offset: i64) -> Arc { + fn new( + stats: Arc, + clock_offset: Arc, + ) -> Arc { Arc::new(DisplayTracker { stats, clock_offset, @@ -554,7 +560,8 @@ unsafe extern "C" fn on_frame_rendered( } } } - let e2e_ns = displayed_ns + t.clock_offset as i128 - pts_us as i128 * 1000; + let e2e_ns = displayed_ns + t.clock_offset.load(Ordering::Relaxed) as i128 + - pts_us as i128 * 1000; let e2e_us = (e2e_ns > 0 && e2e_ns < 10_000_000_000).then_some((e2e_ns / 1000) as u64); let display_us = decoded_ns.map(|d| ((displayed_ns - d).max(0) / 1000) as u64); t.stats.note_displayed(e2e_us, display_us); @@ -827,13 +834,13 @@ fn run_async( // pts we queue) live in a shared map: the feeder writes them at receipt, this loop pairs decoded // output back to them. Behind a `Mutex` since two threads touch it — only ever locked while the // HUD is visible. - let clock_offset = client.clock_offset_ns; + let clock_offset = client.clock_offset_shared(); let in_flight = Arc::new(Mutex::new(VecDeque::<(u64, i128)>::new())); // Display stage (spec `display` + the capture→displayed headline): the rendered frame is // parked in the tracker at release; the OnFrameRendered callback pairs it with // SurfaceFlinger's render timestamp. `render_cb` is the callback's leaked Arc refcount, // reclaimed after the codec is dropped below. - let tracker = DisplayTracker::new(stats.clone(), clock_offset); + let tracker = DisplayTracker::new(stats.clone(), clock_offset.clone()); let render_cb = install_render_callback(&codec, &tracker); // Feeder thread: block on the network so this loop doesn't (an AU's arrival becomes an event that @@ -842,6 +849,7 @@ fn run_async( let client = client.clone(); let stats = stats.clone(); let in_flight = in_flight.clone(); + let clock_offset = clock_offset.clone(); let shutdown = shutdown.clone(); let ev_tx = ev_tx.clone(); std::thread::Builder::new() @@ -851,7 +859,7 @@ fn run_async( client, stats, in_flight, - clock_offset as i128, + clock_offset, shutdown, ev_tx, ); @@ -929,7 +937,7 @@ fn run_async( &mut ready, &stats, &in_flight, - clock_offset, + clock_offset.load(Ordering::Relaxed), &tracker, &mut rendered, &mut discarded, @@ -999,7 +1007,7 @@ fn feeder_loop( client: Arc, stats: Arc, in_flight: Arc>>, - clock_offset: i128, + clock_offset: Arc, shutdown: Arc, ev_tx: mpsc::Sender, ) { @@ -1010,6 +1018,7 @@ fn feeder_loop( Ok(frame) => { if stats.enabled() { let received_ns = now_realtime_ns(); + let clock_offset = clock_offset.load(Ordering::Relaxed) as i128; let lat_ns = received_ns + clock_offset - frame.pts_ns as i128; let lat_us = (lat_ns > 0 && lat_ns < 10_000_000_000).then_some((lat_ns / 1000) as u64); diff --git a/clients/probe/src/main.rs b/clients/probe/src/main.rs index e790c5da..2a3ddfac 100644 --- a/clients/probe/src/main.rs +++ b/clients/probe/src/main.rs @@ -111,6 +111,11 @@ struct Args { /// `--discover [SECS]` — browse the LAN for native (`_punktfunk._udp`) hosts for `SECS` /// seconds (default 4), print what's found, and exit. No connection is made. discover: Option, + /// `--clock-resync` — after the connect-time skew handshake, immediately run a SECOND + /// handshake on the same control stream and assert both estimates are sane and consistent: + /// the headless validator for the host answering `ClockProbe` at any time (what the native + /// clients' mid-stream re-sync relies on). Aborts the session when the re-probe fails. + clock_resync: bool, } fn parse_mode(m: &str) -> Option { @@ -274,6 +279,7 @@ fn parse_args() -> Args { .iter() .any(|a| a == "--discover") .then(|| get("--discover").and_then(|s| s.parse().ok()).unwrap_or(4)), + clock_resync: argv.iter().any(|a| a == "--clock-resync"), } } @@ -523,7 +529,8 @@ async fn session(args: Args) -> Result<()> { // Wall-clock skew handshake on the still-private control stream (before --remode/--speed-test // take it): align our clock to the host's so the per-frame capture→received latency is valid // across machines. `None` ⇒ an old host that doesn't answer — fall back to a shared clock (0). - let clock_offset_ns = match punktfunk_core::quic::clock_sync(&mut send, &mut recv).await { + let first_skew = punktfunk_core::quic::clock_sync(&mut send, &mut recv).await; + let clock_offset_ns = match &first_skew { Some(skew) => { tracing::info!( offset_ns = skew.offset_ns, @@ -536,6 +543,39 @@ async fn session(args: Args) -> Result<()> { None => None, }; + // `--clock-resync`: prove the host answers `ClockProbe` mid-session, not just at connect — + // the contract the native clients' mid-stream re-sync rests on. Run a full second handshake + // and require a sane, consistent estimate: both batches measure the same physical skew, so + // they must agree to within RTT-scale error (the handshake's own uncertainty is ≈ RTT/2). + if args.clock_resync { + let first = first_skew + .as_ref() + .ok_or_else(|| anyhow!("clock-resync: host never answered the connect-time handshake"))?; + let second = punktfunk_core::quic::clock_sync(&mut send, &mut recv) + .await + .ok_or_else(|| anyhow!("clock-resync: host did not answer the re-probe"))?; + let disagree_ns = (second.offset_ns - first.offset_ns).unsigned_abs(); + let bound_ns = (first.rtt_ns + second.rtt_ns).max(2_000_000); + tracing::info!( + first_offset_ns = first.offset_ns, + second_offset_ns = second.offset_ns, + disagree_us = disagree_ns / 1000, + bound_us = bound_ns / 1000, + second_rtt_us = second.rtt_ns / 1000, + rounds = second.rounds, + "clock re-probe answered" + ); + if second.rounds < 8 || disagree_ns > bound_ns { + return Err(anyhow!( + "clock-resync: re-probe unsound (rounds {}, disagreement {} µs > bound {} µs)", + second.rounds, + disagree_ns / 1000, + bound_ns / 1000 + )); + } + println!("clock-resync OK: offsets {} / {} ns", first.offset_ns, second.offset_ns); + } + // Packet-level receive counters mirrored from `session.stats()` by the data-plane loop. The // speed test reads their delta over the burst window so throughput/loss reflect every delivered // wire packet (graceful past the FEC budget), not just fully-reassembled probe AUs. diff --git a/clients/windows/src/app/stream.rs b/clients/windows/src/app/stream.rs index 810feb14..c64cef51 100644 --- a/clients/windows/src/app/stream.rs +++ b/clients/windows/src/app/stream.rs @@ -52,7 +52,7 @@ impl PartialEq for StreamProps { thread_local! { /// Frames + host clock offset, stashed by the mount effect for `on_mounted` (which fires /// later, once the native panel exists). - static PENDING: RefCell> = const { RefCell::new(None) }; + static PENDING: RefCell)>> = const { RefCell::new(None) }; /// The live render thread; stopped + joined by the unmount cleanup (before panel teardown). static RENDER: RefCell> = const { RefCell::new(None) }; } @@ -88,7 +88,7 @@ pub(crate) fn stream_page(props: &StreamProps, cx: &mut RenderCx) -> Element { move || { if let Some((connector, frames, stop)) = shared.handoff.lock().unwrap().take() { let mode = connector.mode(); - let clock_offset = connector.clock_offset_ns; + let clock_offset = connector.clock_offset_shared(); connector_ref.set(Some(connector.clone())); PENDING.with(|c| *c.borrow_mut() = Some((frames, clock_offset))); crate::input::install(connector, mode, inhibit, show_stats, stop); diff --git a/clients/windows/src/render.rs b/clients/windows/src/render.rs index 92bb341f..bb243bd6 100644 --- a/clients/windows/src/render.rs +++ b/clients/windows/src/render.rs @@ -12,7 +12,7 @@ use crate::present::Presenter; use crate::session::{FrameRx, FrameTimes}; use crossbeam_channel::RecvTimeoutError; -use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -122,12 +122,13 @@ unsafe impl Send for SendPresenter {} /// Spawn the render thread. `frames` carries `(frame, FrameTimes)`; `clock_offset_ns` maps our /// wall clock onto the host's so the end-to-end (capture→on-glass) number is cross-machine valid -/// (same math as the pump's host+network stage). +/// (same math as the pump's host+network stage). A live handle (loaded per present) so +/// mid-stream clock re-syncs keep the number honest after an NTP step / drift. pub fn spawn( presenter: Presenter, frames: FrameRx, shared: Arc, - clock_offset_ns: i64, + clock_offset_ns: Arc, ) -> RenderThread { let boxed = SendPresenter(presenter); let shared_w = shared.clone(); @@ -162,7 +163,12 @@ fn poll_window_dpi() -> Option { } } -fn run(presenter: SendPresenter, frames: FrameRx, shared: Arc, clock_offset_ns: i64) { +fn run( + presenter: SendPresenter, + frames: FrameRx, + shared: Arc, + clock_offset_ns: Arc, +) { let mut p = presenter.0; let mut applied = (0u32, 0u32, 0u32); // last (w, h, dpi) handed to the presenter let mut presented = 0u32; @@ -232,8 +238,9 @@ fn run(presenter: SendPresenter, frames: FrameRx, shared: Arc, clo let displayed_ns = now_ns(); // End-to-end = capture → displayed, host-clock corrected, measured directly // (never the sum of stage percentiles). Clamped (0, 10 s). - let e2e = - (displayed_ns as i128 + clock_offset_ns as i128 - t.pts_ns as i128).max(0) as u64; + let e2e = (displayed_ns as i128 + clock_offset_ns.load(Ordering::Relaxed) as i128 + - t.pts_ns as i128) + .max(0) as u64; if e2e > 0 && e2e < 10_000_000_000 { e2e_us.push(e2e / 1000); } diff --git a/clients/windows/src/session.rs b/clients/windows/src/session.rs index c4b6c204..14faedb3 100644 --- a/clients/windows/src/session.rs +++ b/clients/windows/src/session.rs @@ -330,7 +330,9 @@ fn pump( // "PPS id out of range" (a black screen) until one arrives. let _ = connector.request_keyframe(); - let clock_offset = connector.clock_offset_ns; + // Live host↔client clock offset: loaded per use (Relaxed) so mid-stream re-syncs (an NTP + // step, drift) keep the capture-clock latency stats honest — never cached at session start. + let clock_offset_live = connector.clock_offset_shared(); let mut total_frames = 0u64; let session_start = Instant::now(); let mut window_start = Instant::now(); @@ -363,6 +365,7 @@ fn pump( frames_n += 1; bytes_n += frame.data.len() as u64; // `host+network` stage: capture → received, host-clock corrected. Clamped (0, 10 s). + let clock_offset = clock_offset_live.load(Ordering::Relaxed); let hostnet = (received_ns as i128 + clock_offset as i128 - frame.pts_ns as i128) .max(0) as u64; if hostnet > 0 && hostnet < 10_000_000_000 { @@ -500,7 +503,7 @@ fn pump( host_ms: host_p50 as f32 / 1000.0, net_ms: net_p50 as f32 / 1000.0, split, - same_host: clock_offset == 0, + same_host: clock_offset_live.load(Ordering::Relaxed) == 0, hardware, hdr, codec: connector.codec, diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 08e9aedc..0fc79dea 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -279,7 +279,9 @@ fn pump( }) .flatten(); - let clock_offset = connector.clock_offset_ns; + // Live host↔client clock offset: loaded per frame (Relaxed) so mid-stream re-syncs (an NTP + // step, drift) keep the capture-clock latency stats honest — never cached at session start. + let clock_offset_live = connector.clock_offset_shared(); let mut total_frames = 0u64; let mut window_start = Instant::now(); let mut frames_n = 0u32; @@ -352,6 +354,8 @@ fn pump( let decoded_ns = now_ns(); // `host+network` stage: received expressed in the host's capture // clock, minus the host-stamped capture pts (clamped (0, 10 s)). + let clock_offset = + clock_offset_live.load(std::sync::atomic::Ordering::Relaxed); let hn = (received_ns as i128 + clock_offset as i128 - frame.pts_ns as i128) .max(0) as u64; if hn > 0 && hn < 10_000_000_000 { diff --git a/crates/pf-presenter/src/run.rs b/crates/pf-presenter/src/run.rs index c0f7dcec..d39f741a 100644 --- a/crates/pf-presenter/src/run.rs +++ b/crates/pf-presenter/src/run.rs @@ -154,7 +154,9 @@ struct StreamState { canceled: bool, ready_announced: bool, mode_line: String, - clock_offset_ns: i64, + /// Live host↔client clock offset handle (None until Connected): loaded per present so + /// mid-stream re-syncs keep the end-to-end number honest after an NTP step / drift. + clock_offset: Option>, hdr: bool, // Presenter-side 1 s window (design/stats-unification.md): end-to-end // capture→displayed (host-clock corrected) p50+p95, display = decoded→displayed p50. @@ -205,7 +207,7 @@ impl StreamState { canceled: false, ready_announced: false, mode_line: String::new(), - clock_offset_ns: 0, + clock_offset: None, hdr: false, win_e2e_us: Vec::with_capacity(256), win_disp_us: Vec::with_capacity(256), @@ -657,7 +659,7 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result .set_title(&format!("{} · {}", opts.window_title, st.mode_line)) .ok(); gamepad.attach(c.clone()); - st.clock_offset_ns = c.clock_offset_ns; + st.clock_offset = Some(c.clock_offset_shared()); let mut cap = Capture::new(c.clone()); cap.engage(); // capture engages when the stream starts (ui_stream parity) apply_capture(&mut window, &mouse, true); @@ -960,7 +962,11 @@ fn run_inner(mut opts: SessionOpts, mut mode: ModeCtl) -> Result println!("{{\"ready\":true}}"); } // The `displayed` stamp (same clamp rules as the pump's windows). - let e2e = (displayed_ns as i128 + st.clock_offset_ns as i128 - pts_ns as i128) + let clock_offset_ns = st + .clock_offset + .as_ref() + .map_or(0, |o| o.load(Ordering::Relaxed)); + let e2e = (displayed_ns as i128 + clock_offset_ns as i128 - pts_ns as i128) .max(0) as u64; if e2e > 0 && e2e < 10_000_000_000 { st.win_e2e_us.push(e2e / 1000); diff --git a/crates/punktfunk-core/src/client.rs b/crates/punktfunk-core/src/client.rs index ef6d5226..83f18449 100644 --- a/crates/punktfunk-core/src/client.rs +++ b/crates/punktfunk-core/src/client.rs @@ -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, /// 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>>, + /// 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, worker: Option>, /// 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`, whose drop disconnects). + pub fn clock_offset_shared(&self) -> Arc { + 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, fec_recovered: Arc, hot_tids: Arc>>, + /// 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, } /// 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 = 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!( diff --git a/crates/punktfunk-core/src/quic.rs b/crates/punktfunk-core/src/quic.rs index d87e12d1..5030d0fc 100644 --- a/crates/punktfunk-core/src/quic.rs +++ b/crates/punktfunk-core/src/quic.rs @@ -1778,18 +1778,12 @@ pub async fn clock_sync( send: &mut quinn::SendStream, recv: &mut quinn::RecvStream, ) -> Option { - 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, + 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 diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index ba7174f0..db721f68 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -502,6 +502,11 @@ #define ColorInfo_MC_BT2020_NCL 9 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Rounds per batch — matches the connect-time [`clock_sync`]. +#define ClockResync_ROUNDS 8 +#endif + // Stable C ABI status codes. `Ok` is 0; all errors are negative so callers can // test `rc < 0`. Do not renumber existing variants — only append. enum PunktfunkStatus