diff --git a/clients/android/native/src/decode/async_loop.rs b/clients/android/native/src/decode/async_loop.rs index 7dbfcea0..eede78c1 100644 --- a/clients/android/native/src/decode/async_loop.rs +++ b/clients/android/native/src/decode/async_loop.rs @@ -404,7 +404,14 @@ fn feeder_loop( // stage is consumed: the HUD, or the ABR decode signal (`measure_decode`). The // HUD-only `received` point + host/network split stay gated on the overlay. if stats.enabled() || measure_decode { - let received_ns = now_realtime_ns(); + // Core reassembly-completion stamp (ABI v9), NOT the pull instant: stamping + // here would fold the hand-off queue wait into the network latency figure + // (a client-side standing backlog masquerading as network). 0 = older core. + let received_ns = if frame.received_ns > 0 { + frame.received_ns as i128 + } else { + now_realtime_ns() + }; { let mut g = in_flight .lock() diff --git a/clients/android/native/src/decode/sync_loop.rs b/clients/android/native/src/decode/sync_loop.rs index 6120b13f..fd8e775b 100644 --- a/clients/android/native/src/decode/sync_loop.rs +++ b/clients/android/native/src/decode/sync_loop.rs @@ -221,7 +221,13 @@ pub(super) fn run_sync( // samplers (`received` point, host/network split) stay gated on the overlay so // the hidden steady state adds only a wall-clock read + the receipt push. if stats.enabled() || measure_decode { - let received_ns = now_realtime_ns(); + // Core reassembly-completion stamp (ABI v9), not the pull instant — see + // async_loop: a pull stamp folds hand-off queue wait into "network". + let received_ns = if frame.received_ns > 0 { + frame.received_ns as i128 + } else { + now_realtime_ns() + }; in_flight.push_back((frame.pts_ns / 1000, received_ns)); if in_flight.len() > IN_FLIGHT_CAP { in_flight.pop_front(); // stale — codec never echoed it back diff --git a/clients/apple/Sources/PunktfunkClient/ContentView.swift b/clients/apple/Sources/PunktfunkClient/ContentView.swift index 8f963ac7..043a43b1 100644 --- a/clients/apple/Sources/PunktfunkClient/ContentView.swift +++ b/clients/apple/Sources/PunktfunkClient/ContentView.swift @@ -579,13 +579,21 @@ struct ContentView: View { model?.disconnect() // the captured-state ⌃⌥⇧D combo }, onFrame: { [meter = model.meter, latency = model.latency, - split = model.latencySplit, offset = conn.clockOffsetNs] au in + split = model.latencySplit, queue = model.clientQueue, + offset = conn.clockOffsetNs] au in meter.note(byteCount: au.data.count) latency.record(ptsNs: au.ptsNs, offsetNs: offset) // The same receipt, keyed by pts, awaiting its 0xCF host timing (the - // host/network split — drained by the 1 s stats tick). + // host/network split — drained by the 1 s stats tick). receivedNs is + // the core's reassembly stamp (ABI v9), so the split's network term no + // longer contains the client-queue wait... split.recordReceipt( ptsNs: au.ptsNs, receivedNs: au.receivedNs, offsetNs: offset) + // ...which is measured as its own term instead (receipt→pull, both + // client-local). + queue.record( + ptsNs: UInt64(bitPattern: au.receivedNs), atNs: au.pulledNs, + offsetNs: 0) }, onSessionEnd: { [weak model] in Task { @MainActor in model?.sessionEnded() } diff --git a/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift b/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift index a60c5256..4a4e55f4 100644 --- a/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift +++ b/clients/apple/Sources/PunktfunkClient/Session/SessionModel.swift @@ -102,6 +102,12 @@ final class SessionModel: ObservableObject { @Published var decodeValid = false @Published var displayP50Ms = 0.0 @Published var displayValid = false + /// Client-queue wait: core reassembly receipt → the pump's pull (`AccessUnit.pulledNs − + /// receivedNs`, ABI v9 receipt split — the 2026-07 two-pair investigation). ~0 on a healthy + /// stream; a persistent value is a client-side standing backlog that used to hide inside + /// "network". Shown in the detailed tier only when it says something (≥ ~2 ms). + @Published var clientQueueP50Ms = 0.0 + @Published var clientQueueValid = false /// The measured OS present floor (design/apple-presentation-rebuild.md): the deadline /// engine's vend→glass pipeline depth — an OS property no client can pace under (~2 refresh /// intervals composited; would read ~1 under direct-to-display). The HUD subtracts it from @@ -147,6 +153,9 @@ final class SessionModel: ObservableObject { let endToEnd = LatencyMeter() let decodeStage = LatencyMeter() let displayStage = LatencyMeter() + /// Client-queue sampler (see `clientQueueP50Ms`) — fed per AU by the stream view's onFrame, + /// drained by the same 1 s tick as the stage meters. + let clientQueue = LatencyMeter() /// The OS present floor sampler (see `osFloorP50Ms`) — fed one sample per display-link /// update by the deadline engine, drained by the same 1 s tick as the stage meters. let presentFloor = LatencyMeter() @@ -489,6 +498,7 @@ final class SessionModel: ObservableObject { endToEndValid = false decodeValid = false displayValid = false + clientQueueValid = false osFloorValid = false lostFrames = 0 lostPct = 0 @@ -679,6 +689,12 @@ final class SessionModel: ObservableObject { } else { self.osFloorValid = false } + if let q = self.clientQueue.drain() { + self.clientQueueP50Ms = q.p50Ms + self.clientQueueValid = true + } else { + self.clientQueueValid = false + } // Mirror the window to the unified log (see statsLog) — one line per second, // stages in ms, only while frames actually flowed. `fps` counts RECEIVED AUs; // `presents` counts frames that reached glass (the display meter's sample count) @@ -691,7 +707,7 @@ final class SessionModel: ObservableObject { let line = String( format: "fps=%d presents=%d e2e_p50=%.1f e2e_p95=%.1f hostnet_p50=%.1f " + "decode_p50=%.1f display_p50=%.1f lost=%d " - + "floor_p50=%.1f display_adj=%.1f e2e_adj=%.1f", + + "floor_p50=%.1f display_adj=%.1f e2e_adj=%.1f queue_p50=%.1f", frames, displayWindow?.count ?? 0, self.endToEndValid ? self.endToEndP50Ms : -1, @@ -702,7 +718,8 @@ final class SessionModel: ObservableObject { lost, self.osFloorValid ? self.osFloorP50Ms : -1, self.displayValid ? self.displayAdjP50Ms : -1, - self.endToEndValid ? self.endToEndAdjP50Ms : -1) + self.endToEndValid ? self.endToEndAdjP50Ms : -1, + self.clientQueueValid ? self.clientQueueP50Ms : -1) statsLog.info("\(line, privacy: .public)") } } diff --git a/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift b/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift index b53ccdcb..dc0a3761 100644 --- a/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift +++ b/clients/apple/Sources/PunktfunkClient/Session/StreamHUDView.swift @@ -118,6 +118,16 @@ struct StreamHUDView: View { .font(.system(.caption2, design: .monospaced)) .foregroundStyle(.tertiary) } + // Client-queue wait (reassembly receipt → decode pull, ABI v9 split): ~0 on + // a healthy stream and hidden as noise; shown from 2 ms — a persistent value + // is a client-side standing backlog that pre-split builds displayed as + // "network" (the 2026-07 two-pair plateau). The core's standing-latency + // bleed logs alongside when it acts on the same state. + if model.clientQueueValid && model.clientQueueP50Ms >= 2 { + Text("client queue +\(model.clientQueueP50Ms, specifier: "%.1f") (receive backlog — standing if it persists)") + .font(.system(.caption2, design: .monospaced)) + .foregroundStyle(.tertiary) + } } } else if model.hostNetworkValid { // Stage-1 fallback presenter: the layer decodes + presents internally with no diff --git a/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift b/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift index f3cf9e5e..3a275a7a 100644 --- a/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift +++ b/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift @@ -35,10 +35,31 @@ public struct AccessUnit: Sendable { public let ptsNs: UInt64 public let frameIndex: UInt32 public let flags: UInt32 - /// Client `CLOCK_REALTIME` instant the AU was handed over by the core (post-FEC, decrypted) - /// — the **received** measurement point of design/stats-unification.md. The decode stage is - /// `decodedNs - receivedNs`, both client-local (no skew offset applies). + /// Client `CLOCK_REALTIME` instant the AU finished reassembly in the core (post-FEC, + /// decrypted — `PunktfunkFrame.received_ns`, ABI v9) — the **received** measurement point of + /// design/stats-unification.md. NOT the pull instant: stamping at the pull folded the + /// pre-decode hand-off wait into the network term, which is how the 2026-07 two-pair + /// standing-latency plateau hid as "network". The decode stage is `decodedNs - receivedNs`, + /// both client-local (no skew offset applies). public let receivedNs: Int64 + /// Client `CLOCK_REALTIME` instant this pull returned. `pulledNs - receivedNs` is the + /// client-queue wait (kernel hand-off + FrameChannel dwell) — the term the HUD splits out + /// so a client-side standing backlog can never masquerade as network latency again. + public let pulledNs: Int64 + + /// `pulledNs` defaults to `receivedNs` (zero queue wait) for callers with no pull instant — + /// the synthetic probe AUs and decode tests, where the split is meaningless. + public init( + data: Data, ptsNs: UInt64, frameIndex: UInt32, flags: UInt32, + receivedNs: Int64, pulledNs: Int64? = nil + ) { + self.data = data + self.ptsNs = ptsNs + self.frameIndex = frameIndex + self.flags = flags + self.receivedNs = receivedNs + self.pulledNs = pulledNs ?? receivedNs + } } /// One Opus audio packet (48 kHz stereo, 5 ms frames) — decode with AVAudioConverter @@ -662,11 +683,16 @@ public final class PunktfunkConnection { let data = Data(bytes: base, count: Int(frame.len)) // copy: ptr valid only until next call var ts = timespec() clock_gettime(CLOCK_REALTIME, &ts) - let receivedNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec) + let pulledNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec) + // Receipt = the core's reassembly-completion stamp (ABI v9); the pull instant is + // kept separately so the client-queue wait is its own measured term. 0 would mean a + // pre-v9 core — impossible here (core and Kit ship in one binary), but fall back to + // the pull instant rather than record a 1970 receipt. + let receivedNs = frame.received_ns > 0 ? Int64(frame.received_ns) : pulledNs return AccessUnit( data: data, ptsNs: frame.pts_ns, frameIndex: frame.frame_index, flags: frame.flags, - receivedNs: receivedNs) + receivedNs: receivedNs, pulledNs: pulledNs) case statusNoFrame: return nil case statusClosed: diff --git a/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift b/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift index 352a60b4..cdf56c16 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift @@ -1213,7 +1213,9 @@ public final class Stage2Pipeline { let chunkAligned = au.flags & PunktfunkConnection.userFlagChunkAligned != 0 let ptsNs = au.ptsNs - let receivedNs = au.receivedNs + // Decode stage starts at the PULL (matching the VT path's FrameContext — + // receipt→pull is the HUD's separate client-queue term, ABI v9 split). + let receivedNs = au.pulledNs let flags = au.flags let submitted = decoder.decode( au: au.data, chunkAligned: chunkAligned, windowSize: windowSize diff --git a/clients/apple/Sources/PunktfunkKit/Video/VideoDecoder.swift b/clients/apple/Sources/PunktfunkKit/Video/VideoDecoder.swift index 05e45a36..c8d3be40 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/VideoDecoder.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/VideoDecoder.swift @@ -32,9 +32,12 @@ public enum ReadyImage: @unchecked Sendable { public struct ReadyFrame: @unchecked Sendable { /// Host capture clock (the AU's pts), in nanoseconds. public let ptsNs: UInt64 - /// Client `CLOCK_REALTIME` instant the AU was received (`AccessUnit.receivedNs`, threaded - /// through the decode via the frame refcon), in nanoseconds. 0 when unknown (a caller that - /// didn't stamp receipt) — the decode-stage meter then drops the sample via its sanity guard. + /// Client `CLOCK_REALTIME` instant the AU left `nextAU` (`AccessUnit.pulledNs`, threaded + /// through the decode via the frame refcon), in nanoseconds — the decode stage's start + /// point. (Named for its historical role; since the ABI v9 receipt split the true + /// reassembly receipt lives on `AccessUnit.receivedNs`, and receipt→pull is the HUD's own + /// client-queue term.) 0 when unknown (a caller that didn't stamp) — the decode-stage meter + /// then drops the sample via its sanity guard. public let receivedNs: Int64 /// Client `CLOCK_REALTIME` instant decode completed, in nanoseconds. public let decodedNs: Int64 @@ -167,7 +170,11 @@ public final class VideoDecoder: @unchecked Sendable { var infoOut = VTDecodeInfoFlags() // The AU's receipt instant + wire flags ride through as a retained context; the output // callback reclaims it. Retain immediately before submit so no early return can leak it. - let ctx = FrameContext(receivedNs: au.receivedNs, flags: au.flags) + // The decode stage starts at the PULL (the AU leaving nextAU), not the reassembly + // receipt: both consumers — the decode-stage meter and the ABR decode signal — are + // specified from the pull, and the receipt→pull wait is the HUD's separate client-queue + // term (see AccessUnit.pulledNs). + let ctx = FrameContext(receivedNs: au.pulledNs, flags: au.flags) let refcon = Unmanaged.passRetained(ctx).toOpaque() let status = VTDecompressionSessionDecodeFrame( session, diff --git a/crates/pf-client-core/src/session.rs b/crates/pf-client-core/src/session.rs index 436b277c..5b07a794 100644 --- a/crates/pf-client-core/src/session.rs +++ b/crates/pf-client-core/src/session.rs @@ -447,8 +447,16 @@ fn pump( // every ~8–16 ms at 60–120 Hz anyway, so this rarely times out mid-stream). match connector.next_frame(Duration::from_millis(20)) { Ok(frame) => { - // The `received` point: AU fully reassembled, in hand, before decode. - let received_ns = now_ns(); + // The `received` point: reassembly COMPLETION, stamped by the core session as + // the AU crossed poll_frame (ABI v9). Stamping here at the hand-off pull instead + // would fold the pre-decode queue wait into `host+network` — a client-side + // standing backlog masquerading as network latency (the 2026-07 two-pair + // investigation). 0 = a core predating the stamp; fall back to the pull instant. + let received_ns = if frame.received_ns > 0 { + frame.received_ns + } else { + now_ns() + }; // fps / goodput count every received AU (spec), decoded or not. frames_n += 1; bytes_n += frame.data.len() as u64; diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index a19fccb9..c894107e 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -125,6 +125,12 @@ pub struct PunktfunkFrame { pub frame_index: u32, pub pts_ns: u64, pub flags: u32, + /// Wall-clock reassembly-completion instant (ns since the Unix epoch, CLOCK_REALTIME — the + /// clock `pts_ns` and the skew handshake use). THIS is the receipt stamp for latency math: + /// a stamp the embedder takes itself at the poll return additionally contains the + /// pre-decode hand-off queue wait, so a client-side standing backlog would masquerade as + /// network latency (ABI v9 — the 2026-07 two-pair standing-latency investigation). + pub received_ns: u64, } /// Snapshot of session counters. @@ -391,6 +397,7 @@ pub unsafe extern "C" fn punktfunk_client_poll_frame( frame_index: f.frame_index, pts_ns: f.pts_ns, flags: f.flags, + received_ns: f.received_ns, }; } PunktfunkStatus::Ok @@ -1744,6 +1751,7 @@ pub unsafe extern "C" fn punktfunk_connection_next_au( frame_index: f.frame_index, pts_ns: f.pts_ns, flags: f.flags, + received_ns: f.received_ns, }; } PunktfunkStatus::Ok diff --git a/crates/punktfunk-core/src/client/frame_channel.rs b/crates/punktfunk-core/src/client/frame_channel.rs index 99339bf5..39d7e88d 100644 --- a/crates/punktfunk-core/src/client/frame_channel.rs +++ b/crates/punktfunk-core/src/client/frame_channel.rs @@ -73,6 +73,142 @@ pub(crate) const NOOP_CLOCK_FLUSHES_TO_DISARM: u32 = 2; /// FIRST no-op clock flush — the moment a step is actually suspected. pub(crate) const CLOCK_RESYNC_INTERVAL: Duration = Duration::from_secs(60); +/// Standing-latency bleed (the 2026-07 two-pair investigation): how far above the session's own +/// one-way-delay floor a report window's MINIMUM must sit to count as a standing elevation. The +/// jump-to-live detectors above deliberately ignore anything below ~6 frames / 400 ms, so a +/// small standing state — a sub-frame kernel/reassembly backlog, or a stale clock offset after a +/// wall-clock step — is carried forever and reads as permanent extra "network" latency. 10 ms +/// sits above skew-handshake error + normal LAN jitter, and below a single 60 fps frame period, +/// so the observed one-frame plateau (~17 ms) trips it while a healthy stream cannot. +pub(crate) const STANDING_LAT_THRESH_NS: i128 = 10_000_000; + +/// Consecutive elevated report windows (~750 ms each) before the bleed escalates — ~4.5 s of a +/// continuously standing, loss-free elevation. Windows with any loss reset the run: loss means +/// genuine congestion, which the FEC/ABR machinery owns, not this detector. +pub(crate) const STANDING_LAT_WINDOWS: u32 = 6; + +/// Per-session cap on flush+keyframe bleeds. A standing state that survives a clock re-sync AND +/// this many local flushes is not local and not clock — the path latency itself changed; the +/// detector disarms with a warning instead of paying a recovery keyframe every few seconds. +pub(crate) const STANDING_LAT_MAX_BLEEDS: u32 = 3; + +/// What the standing-latency detector asks the pump to do this window (see [`StandingLatency`]). +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum StandingLatAction { + None, + /// First escalation: ask for a mid-stream clock re-sync — free, and a stale offset from a + /// stepped/slewed wall clock produces exactly this signature (an applied re-sync re-bases + /// the floor via the pump's `clock_gen` watch, clearing the elevation if that was the cause). + Resync { + above_ms: i64, + }, + /// The elevation survived a re-sync attempt: flush the local receive backlog + request a + /// keyframe (the jump-to-live action), draining a real sub-threshold standing queue. The + /// pump reports execution back via [`StandingLatency::bled`]; an unexecuted action simply + /// re-arms next window. + Bleed { + above_ms: i64, + }, + /// Bleed cap reached and the elevation is back: give up and say so. + Disarm { + above_ms: i64, + }, +} + +/// Detector for a small, constant, loss-free one-way-delay elevation — the standing state the +/// jump-to-live thresholds deliberately tolerate. Tracks the session's OWD floor (minimum of +/// report-window minimums since start / last re-base) and escalates when windows sit +/// persistently above it: re-sync first, then a bounded number of flush+keyframe bleeds, then +/// disarm. Pure state machine (no clocks, no I/O) so the escalation ladder is unit-testable. +pub(crate) struct StandingLatency { + /// Lowest window-minimum OWD seen since session start / last [`rebase`](Self::rebase). + floor_ns: Option, + /// Minimum per-frame OWD this report window; `None` = no frames yet. + window_min_ns: Option, + /// Consecutive elevated windows. + run: u32, + /// The current elevation already got its re-sync request — next escalation is a bleed. + resync_tried: bool, + bleeds: u32, + disarmed: bool, +} + +impl StandingLatency { + pub(crate) fn new() -> Self { + StandingLatency { + floor_ns: None, + window_min_ns: None, + run: 0, + resync_tried: false, + bleeds: 0, + disarmed: false, + } + } + + /// Feed one frame's skew-corrected OWD (capture→reassembly-complete, ns). Caller gates on a + /// live clock offset and plausibility (0 < owd < 10 s), like the ABR OWD signal. + pub(crate) fn note_frame(&mut self, owd_ns: i128) { + self.window_min_ns = Some(match self.window_min_ns { + Some(m) => m.min(owd_ns), + None => owd_ns, + }); + } + + /// Close a report window. `loss_free` = the window carried zero loss (loss resets the run — + /// congestion is the FEC/ABR machinery's problem, and queues under loss are not "standing"). + pub(crate) fn on_window(&mut self, loss_free: bool) -> StandingLatAction { + let Some(wmin) = self.window_min_ns.take() else { + return StandingLatAction::None; // no frames this window — no evidence either way + }; + let floor = *self.floor_ns.get_or_insert(wmin); + self.floor_ns = Some(floor.min(wmin)); + let above_ns = wmin - floor; + if self.disarmed { + return StandingLatAction::None; + } + if !loss_free || above_ns < STANDING_LAT_THRESH_NS { + self.run = 0; + if above_ns < STANDING_LAT_THRESH_NS { + self.resync_tried = false; // elevation cleared — a future one re-syncs first again + } + return StandingLatAction::None; + } + self.run += 1; + if self.run < STANDING_LAT_WINDOWS { + return StandingLatAction::None; + } + self.run = 0; // each escalation gets a fresh observation run + let above_ms = (above_ns / 1_000_000) as i64; + if !self.resync_tried { + self.resync_tried = true; + StandingLatAction::Resync { above_ms } + } else if self.bleeds < STANDING_LAT_MAX_BLEEDS { + StandingLatAction::Bleed { above_ms } + } else { + self.disarmed = true; + StandingLatAction::Disarm { above_ms } + } + } + + /// The pump executed a [`StandingLatAction::Bleed`] (flush + keyframe). The floor is KEPT: a + /// successful bleed brings OWD back down to it (elevation clears naturally); an unsuccessful + /// one leaves the elevation visible so the ladder continues toward the cap. + pub(crate) fn bled(&mut self) { + self.bleeds += 1; + self.window_min_ns = None; + } + + /// A mid-stream clock re-sync was APPLIED (the pump's `clock_gen` watch): every OWD reading + /// shifted, so the floor and any elevation measured under the old offset are meaningless — + /// re-learn from scratch. The bleed budget survives (it caps keyframes per session). + pub(crate) fn rebase(&mut self) { + self.floor_ns = None; + self.window_min_ns = None; + self.run = 0; + self.resync_tried = false; + } +} + /// Client decode-stage latency accumulator for the adaptive-bitrate controller's decode signal. /// The embedder adds one sample per decoded frame ([`NativeClient::report_decode_us`], µs from the /// AU leaving [`NativeClient::next_frame`] to its decoded output) and the data-plane pump drains a @@ -191,6 +327,7 @@ mod frame_channel_tests { pts_ns: i as u64, flags: 0, complete: true, + received_ns: 0, } } @@ -258,3 +395,143 @@ mod frame_channel_tests { assert_eq!(popped(&ch), Some(total - FRAME_QUEUE_HARD_CAP as u32)); } } + +#[cfg(test)] +mod standing_latency_tests { + use super::{ + StandingLatAction, StandingLatency, STANDING_LAT_MAX_BLEEDS, STANDING_LAT_THRESH_NS, + STANDING_LAT_WINDOWS, + }; + + const FLOOR: i128 = 2_000_000; // a healthy 2 ms LAN OWD + const ELEVATED: i128 = FLOOR + STANDING_LAT_THRESH_NS + 7_000_000; // ~one 60fps frame above + + /// Run `n` windows at `owd`, asserting every window but the last returns None; returns the + /// last window's action. + fn run_windows(d: &mut StandingLatency, owd: i128, n: u32) -> StandingLatAction { + for i in 0..n { + d.note_frame(owd); + let a = d.on_window(true); + if i + 1 < n { + assert_eq!(a, StandingLatAction::None, "window {i} escalated early"); + } else { + return a; + } + } + unreachable!("n > 0 by construction"); + } + + /// Learn a clean floor: one window at the healthy OWD. + fn learned(d: &mut StandingLatency) { + d.note_frame(FLOOR); + assert_eq!(d.on_window(true), StandingLatAction::None); + } + + #[test] + fn healthy_stream_never_escalates() { + let mut d = StandingLatency::new(); + learned(&mut d); + // Jitter riding above the floor but under the threshold: never a run. + for _ in 0..(STANDING_LAT_WINDOWS * 4) { + d.note_frame(FLOOR + STANDING_LAT_THRESH_NS - 1); + assert_eq!(d.on_window(true), StandingLatAction::None); + } + } + + #[test] + fn escalation_ladder_resync_then_bleeds_then_disarm() { + let mut d = StandingLatency::new(); + learned(&mut d); + // First full elevated run asks for the free fix: a clock re-sync. + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Resync { .. } + )); + // Re-sync didn't help (no rebase came) — each further run is a bleed, up to the cap... + for _ in 0..STANDING_LAT_MAX_BLEEDS { + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Bleed { .. } + )); + d.bled(); + } + // ...then the detector gives up loudly, once, and stays quiet. + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Disarm { .. } + )); + d.note_frame(ELEVATED); + assert_eq!(d.on_window(true), StandingLatAction::None); + } + + #[test] + fn loss_windows_reset_the_run() { + let mut d = StandingLatency::new(); + learned(&mut d); + for _ in 0..(STANDING_LAT_WINDOWS - 1) { + d.note_frame(ELEVATED); + assert_eq!(d.on_window(true), StandingLatAction::None); + } + // A lossy window means congestion, not a standing state: run resets... + d.note_frame(ELEVATED); + assert_eq!(d.on_window(false), StandingLatAction::None); + // ...so the ladder needs the full run again before acting. + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Resync { .. } + )); + } + + #[test] + fn recovery_resets_the_ladder_to_resync_first() { + let mut d = StandingLatency::new(); + learned(&mut d); + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Resync { .. } + )); + // The elevation clears on its own (e.g. the successful bleed case, or transient): the + // next episode starts back at the free escalation, not at a bleed. + d.note_frame(FLOOR); + assert_eq!(d.on_window(true), StandingLatAction::None); + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Resync { .. } + )); + } + + #[test] + fn applied_resync_rebases_and_clears_a_stale_offset_elevation() { + let mut d = StandingLatency::new(); + learned(&mut d); + assert!(matches!( + run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS), + StandingLatAction::Resync { .. } + )); + // The re-sync APPLIES (pump sees clock_gen move) → rebase. The corrected offset brings + // OWD readings back to truth; the floor re-learns and nothing ever escalates to a bleed. + d.rebase(); + for _ in 0..(STANDING_LAT_WINDOWS * 2) { + d.note_frame(FLOOR); + assert_eq!(d.on_window(true), StandingLatAction::None); + } + } + + #[test] + fn empty_windows_are_no_evidence() { + let mut d = StandingLatency::new(); + learned(&mut d); + for _ in 0..(STANDING_LAT_WINDOWS - 1) { + d.note_frame(ELEVATED); + assert_eq!(d.on_window(true), StandingLatAction::None); + } + // A frameless window (paused stream) neither advances nor resets the run... + assert_eq!(d.on_window(true), StandingLatAction::None); + // ...so one more elevated window completes it. + d.note_frame(ELEVATED); + assert!(matches!( + d.on_window(true), + StandingLatAction::Resync { .. } + )); + } +} diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index f70e4b9e..56c4d23c 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -1,8 +1,9 @@ //! The client worker: QUIC handshake + control/input/datagram tasks + the blocking data-plane pump. use super::frame_channel::{ - CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN, FLUSH_LATENCY, - NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_TIME, + StandingLatAction, StandingLatency, CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN, + FLUSH_LATENCY, NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, + STANDING_TIME, }; use super::worker::reject_from_close; use super::*; @@ -514,15 +515,19 @@ pub(super) async fn run_pump(args: WorkerArgs) { // 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!( + // info, not debug: ≤1/min, and it is THE forensic + // trail for a stale-offset (stepped/slewed wall clock) + // latency plateau — the 2026-07 two-pair investigation + // had to reconstruct this blind. + tracing::info!( offset_ns, rtt_us = rtt_ns / 1000, "mid-stream clock re-sync applied" ); + clock_offset.store(offset_ns, Ordering::Relaxed); + clock_gen.fetch_add(1, Ordering::Relaxed); } else { - tracing::debug!( + tracing::info!( rtt_us = rtt_ns / 1000, "clock re-sync batch discarded — RTT above the \ connect-time baseline (congested window)" @@ -729,6 +734,12 @@ pub(super) async fn run_pump(args: WorkerArgs) { let mut clock_detector_armed = true; let mut resync_wanted = false; let mut seen_clock_gen = pump_clock_gen.load(Ordering::Relaxed); + // Standing-latency bleed (see StandingLatency): the third detector, for the small, + // constant, loss-free OWD elevation the two jump-to-live detectors deliberately + // tolerate (< QUEUE_HIGH frames, < FLUSH_LATENCY behind) — a sub-frame standing + // backlog, or a stale clock offset after a wall-clock step, either of which otherwise + // reads as permanent extra "network" latency for the rest of the session. + let mut standing_lat = StandingLatency::new(); 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. @@ -740,6 +751,10 @@ pub(super) async fn run_pump(args: WorkerArgs) { seen_clock_gen = gen; stale_since = None; noop_clock_flushes = 0; + // Every OWD reading shifted with the offset — the standing-latency floor and + // any elevation measured under the old one are meaningless now. If a stale + // offset WAS the elevation, this is also the moment it gets fixed. + standing_lat.rebase(); if !clock_detector_armed { clock_detector_armed = true; tracing::info!( @@ -843,6 +858,51 @@ pub(super) async fn run_pump(args: WorkerArgs) { window_dropped, ); let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm })); + // Standing-latency bleed: close the detector's window with this report's loss + // verdict and run its escalation ladder — re-sync first (free; a stale offset + // from a stepped wall clock produces exactly this signature and the applied + // re-sync rebases the floor), then a bounded flush+keyframe (drains a real + // sub-threshold standing backlog the jump-to-live thresholds tolerate), then a + // loud disarm (the path latency itself changed; nothing local fixes that). + match standing_lat.on_window(loss_ppm == 0 && window_dropped == 0) { + StandingLatAction::None => {} + StandingLatAction::Resync { above_ms } => { + tracing::info!( + above_ms, + "standing latency above the session floor with zero loss — \ + requesting a clock re-sync first (a stale offset reads exactly \ + like this)" + ); + let _ = ctrl_tx.try_send(CtrlRequest::ClockResync); + } + StandingLatAction::Bleed { above_ms } => { + // Shares the jump-to-live cooldown: an unexecuted bleed simply re-arms + // over the next windows (the detector's run rebuilds). + if last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN) { + last_flush = Some(Instant::now()); + flush_in_window = true; + let flushed = session.flush_backlog().unwrap_or(0); + let dropped = frames.clear(); + let _ = ctrl_tx.try_send(CtrlRequest::Keyframe); + standing_lat.bled(); + tracing::warn!( + above_ms, + flushed_datagrams = flushed, + dropped_frames = dropped, + "standing latency survived a clock re-sync — bled the local \ + backlog (flush + keyframe)" + ); + } + } + StandingLatAction::Disarm { above_ms } => { + tracing::warn!( + above_ms, + "standing latency persists after a re-sync and every bleed — not \ + local, not clock; the path latency changed. Leaving it be \ + (reconnect re-baselines)" + ); + } + } // Adaptive bitrate: drain any host ack first (its clamp is authoritative), then // feed the controller this window's congestion signals; a decision becomes a // SetBitrate on the control stream. @@ -981,6 +1041,13 @@ pub(super) async fn run_pump(args: WorkerArgs) { if clock_offset_ns != 0 && lat_ns > 0 { owd_sum_ns += lat_ns; owd_frames += 1; + // The standing-latency detector rides the same signal, but off the + // window MINIMUM (robust against jitter/burst spikes — a standing + // state elevates the floor itself). Same 10 s plausibility clamp as + // the hn stats use. + if lat_ns < 10_000_000_000 { + standing_lat.note_frame(lat_ns); + } } if clock_detector_armed && clock_offset_ns != 0 diff --git a/crates/punktfunk-core/src/lib.rs b/crates/punktfunk-core/src/lib.rs index d98c5adf..45f6936e 100644 --- a/crates/punktfunk-core/src/lib.rs +++ b/crates/punktfunk-core/src/lib.rs @@ -83,7 +83,12 @@ pub use stats::Stats; /// `punktfunk_connection_clipboard_{control,offer,fetch,serve,cancel}` + /// `punktfunk_connection_next_clipboard`. Additive; the wire grows only backward-compatible control /// messages (0x40-0x44) and a new `Welcome::host_caps` bit, so [`WIRE_VERSION`] is unchanged. -pub const ABI_VERSION: u32 = 8; +/// v9: `PunktfunkFrame` grew `received_ns` — the reassembly-completion receipt stamp, so +/// embedders stop stamping receipt at the hand-off pull (which folds the pre-decode queue wait +/// into apparent network latency). Struct-size change on the frame poll surface = a hard ABI +/// break for embedders reading `PunktfunkFrame`; nothing on the wire moved, so [`WIRE_VERSION`] +/// is unchanged. +pub const ABI_VERSION: u32 = 9; /// The punktfunk/1 **wire** version — what `Hello`/`Welcome` carry and hosts equality-check. /// Deliberately its own constant: [`ABI_VERSION`] tracks the embeddable **C surface** diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index 85d55d67..5fb498e5 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -489,6 +489,7 @@ impl Reassembler { pts_ns: done.pts_ns, flags: done.user_flags, complete: true, + received_ns: 0, // stamped by Session::poll_frame at the session boundary })); } Ok(None) @@ -592,6 +593,7 @@ impl ReassemblyWindow { pts_ns: f.pts_ns, flags: f.user_flags, complete: false, + received_ns: 0, // stamped by Session::poll_frame at the session boundary }); } } diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index 42c9af0e..ee7db318 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -30,6 +30,14 @@ pub struct Frame { /// ([`crate::packet::USER_FLAG_CHUNK_ALIGNED`]) are ever delivered partial; missing /// shard ranges are zero-filled at their exact offsets. pub complete: bool, + /// Wall-clock instant (ns since the Unix epoch, CLOCK_REALTIME basis — the same clock the + /// skew handshake compares and the host stamps `pts_ns` with) at which this AU finished + /// reassembly, stamped by [`Session::poll_frame`] as the frame leaves the session. Embedders + /// that previously stamped receipt themselves at the hand-off pull should use this instead: + /// the pull stamp additionally contains the pre-decode queue wait, silently folding any + /// client-side standing backlog into the apparent NETWORK latency. The reassembler itself + /// leaves this 0 (it owns no clock — the stamp is the session boundary's job). + pub received_ns: u64, } /// One end of a stream. Constructed for a single [`Role`]; calling the other role's @@ -82,6 +90,18 @@ pub struct Session { lane_scratch: Vec>, } +/// Stamp [`Frame::received_ns`] as the frame crosses the session boundary in +/// [`Session::poll_frame`] — completed frames return the moment their last shard lands, so +/// stamping at return IS stamping at reassembly completion (µs apart). CLOCK_REALTIME to match +/// `pts_ns` / the skew handshake (deliberately not monotonic — cross-machine latency math). +fn stamp_received(mut f: Frame) -> Frame { + f.received_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0); + f +} + /// Wire-packet count at which a frame's sealing splits across two lanes (plan Phase 1.5): /// below it the channel rendezvous (~µs) isn't worth it; at it the halved AES-GCM span /// (≥ ~125 µs of ~1 µs/packet work) dwarfs the hand-off. ≈300 KB of wire, i.e. ≥150 Mbps @@ -684,7 +704,7 @@ impl Session { // Nothing new on the wire — hand over an aged-out partial if one is // waiting (it can only get staler). if let Some(p) = self.reassembler.take_partial() { - return Ok(p); + return Ok(stamp_received(p)); } return Err(PunktfunkError::NoFrame); } @@ -748,12 +768,12 @@ impl Session { } if let Some(frame) = pushed { StatsCounters::add(&self.stats.frames_completed, 1); - return Ok(frame); + return Ok(stamp_received(frame)); } // A push that completed nothing may still have aged a partial out — deliver it // ahead of further draining (its successors are already arriving). if let Some(p) = self.reassembler.take_partial() { - return Ok(p); + return Ok(stamp_received(p)); } } } diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index bf4c2c69..6f3abaed 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -37,7 +37,12 @@ // `punktfunk_connection_clipboard_{control,offer,fetch,serve,cancel}` + // `punktfunk_connection_next_clipboard`. Additive; the wire grows only backward-compatible control // messages (0x40-0x44) and a new `Welcome::host_caps` bit, so [`WIRE_VERSION`] is unchanged. -#define ABI_VERSION 8 +// v9: `PunktfunkFrame` grew `received_ns` — the reassembly-completion receipt stamp, so +// embedders stop stamping receipt at the hand-off pull (which folds the pre-decode queue wait +// into apparent network latency). Struct-size change on the frame poll surface = a hard ABI +// break for embedders reading `PunktfunkFrame`; nothing on the wire moved, so [`WIRE_VERSION`] +// is unchanged. +#define ABI_VERSION 9 // The punktfunk/1 **wire** version — what `Hello`/`Welcome` carry and hosts equality-check. // Deliberately its own constant: [`ABI_VERSION`] tracks the embeddable **C surface** @@ -1112,6 +1117,12 @@ typedef struct { uint32_t frame_index; uint64_t pts_ns; uint32_t flags; + // Wall-clock reassembly-completion instant (ns since the Unix epoch, CLOCK_REALTIME — the + // clock `pts_ns` and the skew handshake use). THIS is the receipt stamp for latency math: + // a stamp the embedder takes itself at the poll return additionally contains the + // pre-decode hand-off queue wait, so a client-side standing backlog would masquerade as + // network latency (ABI v9 — the 2026-07 two-pair standing-latency investigation). + uint64_t received_ns; } PunktfunkFrame; // A single input event. `#[repr(C)]` — shared verbatim with the C ABI as