feat(core): receive-path stage timing + frame-jitter observability (PUNKTFUNK_PERF)

Session::poll_frame accumulates per-stage ns (recv_batch syscall, AES-GCM
open, Reassembler::push incl. FEC) into a PumpPerf drained via
take_pump_perf(); the client pump logs the split plus completed-AU
inter-arrival jitter (p50/p95/max + late count) every report window.
Gated on PUNKTFUNK_PERF — one branch per stage when off.

Smoothness previously had no metric at all (jump-to-live counters fire
seconds late), and the receive core had no attribution. First live use
pinned the 1.57 Gbps client wall on software AES-GCM (7 µs/pkt) vs
0.4 µs reassembly — see punktfunk-planning/design/throughput-beyond-1gbps.md.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/punktfunk-core/src/client.rs

@@ -1935,6 +1935,13 @@ async fn worker_main(args: WorkerArgs) {
         const ADAPT_REPORT_INTERVAL: Duration = Duration::from_millis(750);
         let mut last_report = Instant::now();
         let (mut last_recovered, mut last_received, mut last_dropped) = (0u64, 0u64, 0u64);
+        // PUNKTFUNK_PERF: per-window pump observability — the Session's receive stage split
+        // (recv / decrypt / reassemble+FEC, see `Session::take_pump_perf`) and completed-AU
+        // inter-arrival jitter. Smoothness has no metric otherwise: jump-to-live counters only
+        // fire after the stream is already seconds behind.
+        let pump_perf_on = std::env::var("PUNKTFUNK_PERF").is_ok_and(|v| v != "0");
+        let mut arrivals_us: Vec<u32> = Vec::new();
+        let mut last_arrival: Option<Instant> = None;
         // Adaptive bitrate (see `crate::abr`): armed only when the embedder asked for Automatic
         // (`bitrate_kbps == 0`) and the host echoed the rate it actually configured (an old host
         // echoes 0 → controller stays permanently off). Fed once per report window with the same
@@ -2037,12 +2044,55 @@ async fn worker_main(args: WorkerArgs) {
                 last_recovered = st.fec_recovered_shards;
                 last_received = st.packets_received;
                 last_dropped = st.frames_dropped;
+                if pump_perf_on {
+                    if let Some(p) = session.take_pump_perf() {
+                        let per_pkt_ns = |ns: u64| ns.checked_div(p.packets).unwrap_or(0);
+                        tracing::info!(
+                            recv_ms = p.recv_ns / 1_000_000,
+                            decrypt_ms = p.decrypt_ns / 1_000_000,
+                            reasm_ms = p.reasm_ns / 1_000_000,
+                            packets = p.packets,
+                            batches = p.batches,
+                            pkts_per_batch = p.packets.checked_div(p.batches).unwrap_or(0),
+                            decrypt_ns_pkt = per_pkt_ns(p.decrypt_ns),
+                            reasm_ns_pkt = per_pkt_ns(p.reasm_ns),
+                            "pump stage split (window)"
+                        );
+                    }
+                    // Inter-arrival jitter over the window's completed AUs. `late` counts gaps
+                    // over 2× the window median — the "a frame arrived visibly off-beat" tally.
+                    if arrivals_us.len() >= 8 {
+                        arrivals_us.sort_unstable();
+                        let pct = |q: usize| arrivals_us[(arrivals_us.len() - 1) * q / 100];
+                        let (p50, p95) = (pct(50), pct(95));
+                        let late = arrivals_us.iter().filter(|&&d| d > p50 * 2).count();
+                        tracing::info!(
+                            frames = arrivals_us.len() + 1,
+                            arrival_p50_us = p50,
+                            arrival_p95_us = p95,
+                            arrival_max_us = arrivals_us.last().copied().unwrap_or(0),
+                            late,
+                            "frame inter-arrival jitter (window)"
+                        );
+                    }
+                    arrivals_us.clear();
+                }
             }
             match session.poll_frame() {
                 Ok(frame) => {
                     if frame.flags & FLAG_PROBE as u32 != 0 {
                         continue; // speed-test filler, not video — measured via the counters above
                     }
+                    if pump_perf_on {
+                        let now = Instant::now();
+                        if let Some(prev) = last_arrival.replace(now) {
+                            // 4096 ≈ 17 s at 240 fps — a stuck window can't grow it unbounded.
+                            if arrivals_us.len() < 4096 {
+                                arrivals_us.push((now - prev).as_micros().min(u32::MAX as u128)
+                                    as u32);
+                            }
+                        }
+                    }
                     // Jump-to-live guard. A standing receive/hand-off queue never drains by itself —
                     // the pump consumes strictly in order at the arrival rate, so once behind, the
                     // stream stays behind for good (observed live: stuck 6–7 s). Pre-decode AUs are

crates/punktfunk-core/src/session.rs

@@ -59,6 +59,28 @@ pub struct Session {
     /// then returns them via [`reclaim_wires`](Self::reclaim_wires)). After warmup each buffer keeps
     /// its capacity, so the per-packet ciphertext + wire `Vec` allocations vanish from the hot path.
     wire_pool: Vec<Vec<u8>>,
+    /// Receive-path stage timing (`PUNKTFUNK_PERF`), read+reset via [`take_pump_perf`]
+    /// (Self::take_pump_perf). `None` when disabled — the hot path then pays one branch per stage.
+    perf: Option<PumpPerf>,
+}
+
+/// Accumulated client receive-path stage timings since the last [`Session::take_pump_perf`].
+/// Answers "where does the pump core go" at line rate: kernel drain (`recv_ns`) vs AES-GCM
+/// (`decrypt_ns`) vs reassembly+FEC (`reasm_ns`, the `Reassembler::push` round-trip including
+/// shard copies and block reconstruction). 2026-07-14 sweep context: the pump pegs one core at
+/// ~1.5 Gbps wire, ~85% of it userspace — this split is what Phase 2.1 (pooled reassembly) is
+/// validated against.
+#[derive(Debug, Default, Clone, Copy)]
+pub struct PumpPerf {
+    /// ns inside `recv_batch` (recvmmsg / recvmsg_x), i.e. syscall + kernel copy.
+    pub recv_ns: u64,
+    /// ns inside `open_in_place` across all datagrams (AES-128-GCM + replay-window upkeep).
+    pub decrypt_ns: u64,
+    /// ns inside `Reassembler::push` (header parse, shard copy, FEC reconstruct, AU assembly).
+    pub reasm_ns: u64,
+    /// recv_batch calls (batches) and datagrams processed over the accumulation window.
+    pub batches: u64,
+    pub packets: u64,
 }
 
 /// Datagrams drained per `recvmmsg` syscall on the client (the reused ring's size). At ~125k
@@ -91,10 +113,20 @@ impl Session {
             recv_count: 0,
             recv_idx: 0,
             wire_pool: Vec::new(),
+            // Same opt-in the host's stage logs use; read once — set it before connecting.
+            perf: std::env::var("PUNKTFUNK_PERF")
+                .is_ok_and(|v| v != "0")
+                .then(PumpPerf::default),
             config,
         })
     }
 
+    /// Drain the receive-path stage timings accumulated since the last call (window semantics —
+    /// the pump reads this once per report interval). `None` when `PUNKTFUNK_PERF` is off.
+    pub fn take_pump_perf(&mut self) -> Option<PumpPerf> {
+        self.perf.as_mut().map(std::mem::take)
+    }
+
     pub fn role(&self) -> Role {
         self.config.role
     }
@@ -362,9 +394,14 @@ impl Session {
         loop {
             // Refill the ring with one `recvmmsg` batch when the current one is drained.
             if self.recv_idx >= self.recv_count {
+                let t0 = self.perf.is_some().then(std::time::Instant::now);
                 self.recv_count = self
                     .transport
                     .recv_batch(&mut self.recv_scratch, &mut self.recv_lens)?;
+                if let (Some(p), Some(t0)) = (self.perf.as_mut(), t0) {
+                    p.recv_ns += t0.elapsed().as_nanos() as u64;
+                    p.batches += 1;
+                }
                 self.recv_idx = 0;
                 if self.recv_count == 0 {
                     return Err(PunktfunkError::NoFrame);
@@ -384,6 +421,9 @@ impl Session {
             // one). The plaintext lands at [8..8+n] of the sealed wire (behind the seq prefix); an
             // unencrypted (probe) datagram IS the packet. Field-precise borrows keep the slice into
             // `recv_scratch` alive across the replay/reassembler calls below.
+            // Perf note: the two `continue`s below (short / undecryptable noise) skip the decrypt
+            // accounting — they are the exception path, not line-rate traffic.
+            let t_dec = self.perf.is_some().then(std::time::Instant::now);
             let (pkt_range, seq) = match &self.crypto {
                 Some(c) => {
                     // A sealed datagram is at least seq prefix + tag; anything shorter is noise.
@@ -398,6 +438,9 @@ impl Session {
                 }
                 None => (0..len, None),
             };
+            if let (Some(p), Some(t)) = (self.perf.as_mut(), t_dec) {
+                p.decrypt_ns += t.elapsed().as_nanos() as u64;
+            }
             // Anti-replay (same rationale as poll_input): reject a datagram whose authenticated
             // sequence was already seen. Video also dedups per-frame downstream, but filtering here
             // is uniform and cheap.
@@ -412,10 +455,16 @@ impl Session {
             StatsCounters::add(&self.stats.bytes_received, pkt.len() as u64);
             // The reassembler validates the packet via its parsed header (`magic`),
             // ignoring anything that isn't a well-formed video packet.
-            if let Some(frame) = self
+            let t_push = self.perf.is_some().then(std::time::Instant::now);
+            let pushed = self
                 .reassembler
-                .push(pkt, self.coder.as_ref(), &self.stats)?
+                .push(pkt, self.coder.as_ref(), &self.stats)?;
-            {
+            if let (Some(p), Some(t)) = (self.perf.as_mut(), t_push) {
+                p.reasm_ns += t.elapsed().as_nanos() as u64;
+                // Counts datagrams that reached the reassembler (replay-rejected ones don't).
+                p.packets += 1;
+            }
+            if let Some(frame) = pushed {
                 StatsCounters::add(&self.stats.frames_completed, 1);
                 return Ok(frame);
             }