fix(core): remaining 15 sweep lows — client, clipboard, C ABI (v10), FEC, GSO, GTK

Second and final batch from the 2026-07-20 core-sweep lows: client: - connect() timeout now sets `quit` before shutdown, so a handshake that completes after the deadline closes with QUIT_CLOSE_CODE instead of leaving the host lingering (virtual display up) for a reconnect that never comes. - probe_result: saturating_add on the wire-supplied wire_packets + send_dropped counters (debug-build overflow panic / release wrap). - the standing-latency bleed no longer sets flush_in_window: that flag is the ABR's SEVERE (×0.7) verdict, and the bleed fires only on provably loss-free windows the controller itself scores as fine. clipboard: - fetch_cancels pruned on every new fetch (was: one dead oneshot per paste for the session). - serve chunks gated on a parked waiter + capped at CLIP_FETCH_CAP with an Error event (was: unbounded silent accumulation under any req_id). - serve_inbound park bounded by FETCH_STALL_SECS + send.stopped() (was: an unanswered FetchRequest parked the task, waiter, and bi-stream forever; ~100 of them exhaust the connection's bidi budget). C ABI (ABI_VERSION 9 → 10, header regenerated, additive only): - new punktfunk_connection_clock_offset_now_ns — the LIVE re-synced offset (Swift/Kotlin latency math read the frozen connect-time value ~40ms wrong after a wall-clock step). - to_config: checked u64→usize narrowing of max_frame_bytes (32-bit armeabi truncated >4GiB to a plausible residue). - host_poll_input: no &mut held across the embedder callback (re-entry aliased it — UB under noalias); mid-drain callback clears now stick. - next_audio_pcm: DTX (empty) payloads skipped — decode synthesized 120ms of concealment per 5ms slot and grew the playout ring forever. - next_clipboard releases the parked payload on an empty poll (a one-off 50MiB paste stayed resident all session). - frames_dropped / wants_decode_latency write their documented 0/false defaults before the NULL-handle check. - gamepad constant docs match pick_gamepad() reality (DualSenseEdge/ SwitchPro landed; DualSense/DS4 honored on Windows UMDF too). FEC: - gf8 reconstruct/reconstruct_into reuse the (k,m) codec cache like encode_into (was: fresh 230×200 generator + decode inversion per lossy block on the pump thread). - vendored fec-rs: the 8 safe wrap_mul_slice shims assert equal lengths (x86 SIMD callees bound stores on input.len() — safe-code OOB write); ReconstructShard's safety contract gains the len()==get().len() clause and reconstruct_internal sizes raw slices from the slice. transport/GTK: - Linux GSO super-buffer capped at the real UDP payload ceiling (65487) not 65535 — seg sizes ≥1024 could EMSGSIZE and latch GSO off process-wide, blamed on the network. - GTK settings dialog no longer rewrites an unlisted-but-valid stored gamepad preference to "auto" on close. Already fixed on main (verified stale, skipped): the reassembler FEC ceiling, wants_decode_latency's third term, request_probe rollback + the generalized probe watchdog. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-20 18:26:43 +02:00
parent 76ac3cf867
commit de7b54d4e0
12 changed files with 337 additions and 50 deletions
@@ -78,6 +78,11 @@ impl PunktfunkConfig {
            u8::try_from(self.fec_percent).map_err(|_| PunktfunkStatus::InvalidArg)?;
        let max_data_per_block =
            u16::try_from(self.max_data_per_block).map_err(|_| PunktfunkStatus::InvalidArg)?;
+        // The one narrowing here that differs by target width: on 32-bit (armeabi-v7a) an
+        // `as usize` silently truncates a >4 GiB value to a plausible-looking residue that
+        // passes validate() — reject it instead, like every narrowing above.
+        let max_frame_bytes =
+            usize::try_from(self.max_frame_bytes).map_err(|_| PunktfunkStatus::InvalidArg)?;
        let cfg = Config {
            role,
            phase,
@@ -87,7 +92,7 @@ impl PunktfunkConfig {
                max_data_per_block,
            },
            shard_payload: self.shard_payload as usize,
-            max_frame_bytes: self.max_frame_bytes as usize,
+            max_frame_bytes,
            encrypt: self.encrypt != 0,
            key: self.key,
            salt: self.salt,
@@ -463,23 +468,31 @@ pub unsafe extern "C" fn punktfunk_set_input_callback(
 #[no_mangle]
 pub unsafe extern "C" fn punktfunk_host_poll_input(s: *mut PunktfunkSession) -> i32 {
    let r = std::panic::catch_unwind(AssertUnwindSafe(|| {
-        let s = match unsafe { s.as_mut() } {
-            Some(s) => s,
-            None => return PunktfunkStatus::NullPointer as i32,
-        };
-        let cb = s.input_cb;
        let mut count = 0i32;
        loop {
-            match s.inner.poll_input() {
-                Ok(Some(ev)) => {
-                    if let Some((cb, user)) = cb {
-                        cb(&ev as *const InputEvent, user);
-                    }
-                    count += 1;
+            // Narrow scope: re-derive the handle and pull ONE event, then drop the borrow
+            // before dispatching. The callback may legally re-enter `punktfunk_*` on this
+            // handle (get_stats, send_input, clearing the callback) — with a `&mut` held
+            // across the call that re-entry aliased it (UB under noalias). Re-reading
+            // `input_cb` per iteration also makes a mid-drain
+            // `punktfunk_set_input_callback(s, NULL, NULL)` take effect immediately instead
+            // of firing the cleared callback for the queued remainder. (Freeing the session
+            // from inside the callback remains forbidden, as on every entry point.)
+            let (ev, cb) = {
+                let s = match unsafe { s.as_mut() } {
+                    Some(s) => s,
+                    None => return PunktfunkStatus::NullPointer as i32,
+                };
+                match s.inner.poll_input() {
+                    Ok(Some(ev)) => (ev, s.input_cb),
+                    Ok(None) => break,
+                    Err(e) => return e.status() as i32,
                }
-                Ok(None) => break,
-                Err(e) => return e.status() as i32,
+            };
+            if let Some((cb, user)) = cb {
+                cb(&ev as *const InputEvent, user);
            }
+            count += 1;
        }
        count
    }));
@@ -885,8 +898,8 @@ pub const PUNKTFUNK_GAMEPAD_AUTO: u32 = 0;
 /// uinput X-Box 360 pad (the universal default — every game speaks XInput).
 pub const PUNKTFUNK_GAMEPAD_XBOX360: u32 = 1;
 /// UHID DualSense (kernel `hid-playstation`): adaptive triggers, lightbar, touchpad, motion —
-/// feedback arrives on the HID-output plane ([`punktfunk_connection_next_hidout`]). Honored
-/// only where available (Linux hosts); otherwise the host falls back to X-Box 360.
+/// feedback arrives on the HID-output plane ([`punktfunk_connection_next_hidout`]). Honored on
+/// Linux (UHID) and Windows (UMDF minidriver) hosts; otherwise the host falls back to X-Box 360.
 pub const PUNKTFUNK_GAMEPAD_DUALSENSE: u32 = 2;
 /// uinput X-Box One / Series pad — the X-Box 360 backend with the One/Series USB identity, so
 /// games show One/Series glyphs. XInput-identical to `XBOX360` otherwise (no game-visible gain;
@@ -895,8 +908,8 @@ pub const PUNKTFUNK_GAMEPAD_DUALSENSE: u32 = 2;
 pub const PUNKTFUNK_GAMEPAD_XBOXONE: u32 = 3;
 /// UHID DualShock 4 (kernel `hid-playstation` ≥ 6.2): lightbar, touchpad, motion, rumble — the
 /// touchpad/motion arrive over the rich-input plane and lightbar over the HID-output plane, like
-/// DualSense (minus adaptive triggers / player LEDs / mute). Honored only where available (Linux
-/// hosts); otherwise the host falls back to X-Box 360.
+/// DualSense (minus adaptive triggers / player LEDs / mute). Honored on Linux (UHID) and Windows
+/// (UMDF minidriver) hosts; otherwise the host falls back to X-Box 360.
 pub const PUNKTFUNK_GAMEPAD_DUALSHOCK4: u32 = 4;
 /// UHID classic Steam Controller (Valve `28DE:1102`, kernel `hid-steam`): one stick + dual
 /// trackpads + two grip paddles. Honored only where available (Linux hosts); else Xbox 360.
@@ -906,10 +919,12 @@ pub const PUNKTFUNK_GAMEPAD_STEAMCONTROLLER: u32 = 5;
 /// host. Honored on Linux AND Windows hosts; else folds to X-Box 360.
 pub const PUNKTFUNK_GAMEPAD_STEAMDECK: u32 = 6;
 /// DualSense Edge (Sony `054C:0DF2`): the DualSense plus two back buttons + two Fn buttons, so a
-/// client's back paddles land on native slots. Folds to `DUALSENSE` until its backend lands.
+/// client's back paddles land on native slots. Honored on Linux (UHID `hid-playstation`) and
+/// Windows (UMDF) hosts; otherwise the host falls back to X-Box 360.
 pub const PUNKTFUNK_GAMEPAD_DUALSENSEEDGE: u32 = 7;
 /// Nintendo Switch Pro Controller (Nintendo `057E:2009`, kernel `hid-nintendo`): Nintendo glyphs +
-/// positional layout, gyro/accel, HD rumble. Folds to `XBOX360` until its backend lands.
+/// positional layout, gyro/accel, HD rumble. Honored only where available (Linux hosts, UHID
+/// `hid-nintendo`); otherwise the host falls back to X-Box 360.
 pub const PUNKTFUNK_GAMEPAD_SWITCHPRO: u32 = 8;
 /// New Steam Controller (2026, Valve `28DE:1302`) passed through AS-IS: the host mirrors the
 /// client's raw Triton input reports out of a virtual SC2 with the real identity, and Steam's
@@ -1914,6 +1929,13 @@ pub unsafe extern "C" fn punktfunk_connection_next_audio_pcm(
        }
        let AudioPcmState { decoder, pcm } = &mut *state;
        let dec = decoder.as_mut().unwrap();
+        // A header-only datagram (DTX silence — a legal wire form) must be SKIPPED, not
+        // decoded: `decode_float` treats an empty payload as a loss and synthesizes a full
+        // 120 ms of concealment for a ~5 ms slot, growing the playout ring without bound.
+        // Mirrors the host mic pump's guard; the sink underruns to silence on its own.
+        if pkt.data.is_empty() {
+            return PunktfunkStatus::NoFrame;
+        }
        // `decode_float` divides the output buffer length by the channel count to get the
        // per-channel capacity; an empty payload requests packet-loss concealment.
        match dec.decode_float(&pkt.data, pcm, false) {
@@ -2964,7 +2986,14 @@ pub unsafe extern "C" fn punktfunk_connection_next_clipboard(
                unsafe { *out = out_ev };
                PunktfunkStatus::Ok
            }
-            Err(e) => e.status(),
+            Err(e) => {
+                // Release the parked payload once the embedder polls past it: clipboard
+                // traffic is sporadic, so without this a one-off 50 MiB paste stays resident
+                // for the rest of the session (there is no other release entry point). The
+                // borrow contract already says `out` data is valid only until the next call.
+                *c.last_clip.lock().unwrap() = None;
+                e.status()
+            }
        }
    })
 }
@@ -3054,6 +3083,35 @@ pub unsafe extern "C" fn punktfunk_connection_clock_offset_ns(
    })
 }

+/// The **live** host↔client wall-clock offset (nanoseconds, host minus client): the
+/// connect-time estimate of [`punktfunk_connection_clock_offset_ns`], updated by every applied
+/// mid-stream clock re-sync. Ongoing latency math (per-frame `received − pts` splits, the
+/// glass-to-glass meter) must use this one — after a wall-clock step/slew the frozen
+/// connect-time value reads tens of milliseconds wrong for the rest of the session, while the
+/// core itself has already re-synced. Same clock contract as the connect-time getter.
+///
+/// # Safety
+/// `c` is a valid connection handle; `offset_ns` is writable (NULL is skipped).
+#[cfg(feature = "quic")]
+#[no_mangle]
+pub unsafe extern "C" fn punktfunk_connection_clock_offset_now_ns(
+    c: *const PunktfunkConnection,
+    offset_ns: *mut i64,
+) -> PunktfunkStatus {
+    guard(|| {
+        let c = match unsafe { c.as_ref() } {
+            Some(c) => c,
+            None => return PunktfunkStatus::NullPointer,
+        };
+        unsafe {
+            if !offset_ns.is_null() {
+                *offset_ns = c.inner.clock_offset_now_ns();
+            }
+        }
+        PunktfunkStatus::Ok
+    })
+}
+
 /// Ask the host to switch the live session to `width`x`height`@`refresh_hz` without
 /// reconnecting (window resized, refresh changed). Non-blocking enqueue: on acceptance the
 /// stream continues at the new mode — the first new-mode access unit is an IDR with
@@ -3193,6 +3251,11 @@ pub unsafe extern "C" fn punktfunk_connection_frames_dropped(
    out: *mut u64,
 ) -> PunktfunkStatus {
    guard(|| {
+        // The header promises "writes 0 on a NULL connection" — honor it BEFORE the handle
+        // check, so an embedder that skips the status never reads an uninitialized slot.
+        if !out.is_null() {
+            unsafe { *out = 0 };
+        }
        let c = match unsafe { c.as_ref() } {
            Some(c) => c,
            None => return PunktfunkStatus::NullPointer,
@@ -3250,6 +3313,11 @@ pub unsafe extern "C" fn punktfunk_connection_wants_decode_latency(
    out: *mut bool,
 ) -> PunktfunkStatus {
    guard(|| {
+        // The header promises "writes 0 on a NULL connection" — honor it BEFORE the handle
+        // check: an uninitialized byte is not even a valid C++/Swift bool to read.
+        if !out.is_null() {
+            unsafe { *out = false };
+        }
        let c = match unsafe { c.as_ref() } {
            Some(c) => c,
            None => return PunktfunkStatus::NullPointer,
@@ -425,6 +425,12 @@ impl NativeClient {
            Ok(Ok(t)) => t,
            Ok(Err(e)) => return Err(e),
            Err(_) => {
+                // A connect we already reported as failed must not leave a lingering host
+                // session if the handshake lands late: mark it a deliberate QUIT (not a plain
+                // drop / close code 0) so the worker's close tells the host to tear down now
+                // instead of holding the session (and its virtual display) for a reconnect
+                // that will never come.
+                quit.store(true, Ordering::SeqCst);
                shutdown.store(true, Ordering::SeqCst);
                return Err(PunktfunkError::Timeout);
            }
@@ -759,7 +765,9 @@ impl NativeClient {
            0.0
        } as f32;
        // Host-side drop: what the send buffer couldn't even accept (the host-side ceiling).
-        let offered_wire = p.host_wire_packets + p.host_send_dropped;
+        // Saturating: both counters arrive verbatim off the wire (same discipline as the
+        // saturating_sub/mul above — a hostile sum must not overflow-panic a debug build).
+        let offered_wire = p.host_wire_packets.saturating_add(p.host_send_dropped);
        let host_drop_pct = if offered_wire > 0 {
            p.host_send_dropped as f64 / offered_wire as f64 * 100.0
        } else {
@@ -310,7 +310,12 @@ impl DataPump {
                        // 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;
+                            // Deliberately NOT `flush_in_window = true`: that flag is the ABR's
+                            // SEVERE verdict (an immediate ×0.7 back-off), and the bleed fires
+                            // only after ~6 provably loss-free windows with a sub-25ms elevation
+                            // the controller itself scores as fine. The bleed's effect reaches
+                            // the ABR through the window's own honest signals (OWD/loss/decode);
+                            // the flag stays exclusive to the jump-to-live path below.
                            let flushed = session.flush_backlog().unwrap_or(0);
                            let dropped = frames.clear();
                            let _ = ctrl_tx.try_send(CtrlRequest::Keyframe);
@@ -145,6 +145,11 @@ pub async fn run(
                let Some(cmd) = cmd else { break }; // NativeClient dropped
                match cmd {
                    ClipCommand::Fetch { xfer_id, seq, file_index, mime } => {
+                        // Prune finished fetches first: a completed/failed/timed-out fetch task
+                        // drops its cancel receiver, so its sender reads closed. Without this
+                        // the map grew by one dead sender per paste for the whole session (only
+                        // an explicit Cancel ever removed entries).
+                        fetch_cancels.retain(|_, tx| !tx.is_closed());
                        let (cancel_tx, cancel_rx) = oneshot::channel();
                        fetch_cancels.insert(xfer_id, cancel_tx);
                        let conn = conn.clone();
@@ -153,11 +158,36 @@ pub async fn run(
                        tokio::spawn(run_outbound_fetch(conn, xfer_id, req, events, cancel_rx));
                    }
                    ClipCommand::Serve { req_id, bytes, last } => {
-                        serve_bufs.entry(req_id).or_default().extend_from_slice(&bytes);
-                        if last {
-                            let full = serve_bufs.remove(&req_id).unwrap_or_default();
-                            if let Some(tx) = serve_waiters.lock().unwrap().remove(&req_id) {
-                                let _ = tx.send(Some(full));
+                        // Gate on a genuinely parked fetch (the waiter registers before the
+                        // FetchRequest event is emitted, so a live serve always finds it):
+                        // bytes served under a stale/unknown/cancelled req_id would otherwise
+                        // pool here for the whole session with Ok returned for every chunk.
+                        if !serve_waiters.lock().unwrap().contains_key(&req_id) {
+                            serve_bufs.remove(&req_id);
+                        } else {
+                            let buf = serve_bufs.entry(req_id).or_default();
+                            if buf.len().saturating_add(bytes.len()) > CLIP_FETCH_CAP {
+                                // The requester bounds its read at CLIP_FETCH_CAP — anything
+                                // larger is memory burned toward a guaranteed peer rejection.
+                                // Fail the transfer NOW: peer reads UNAVAILABLE, embedder gets
+                                // an Error instead of silent Ok-per-chunk.
+                                serve_bufs.remove(&req_id);
+                                if let Some(tx) = serve_waiters.lock().unwrap().remove(&req_id) {
+                                    let _ = tx.send(None);
+                                }
+                                let _ = events.try_send(ClipEventCore::Error {
+                                    id: req_id,
+                                    code: PunktfunkStatus::InvalidArg as i32,
+                                });
+                            } else {
+                                buf.extend_from_slice(&bytes);
+                                if last {
+                                    let full = serve_bufs.remove(&req_id).unwrap_or_default();
+                                    if let Some(tx) = serve_waiters.lock().unwrap().remove(&req_id)
+                                    {
+                                        let _ = tx.send(Some(full));
+                                    }
+                                }
                            }
                        }
                    }
@@ -224,8 +254,27 @@ async fn serve_inbound(
        return;
    }

-    match rx.await {
-        Ok(Some(bytes)) => {
+    // Overall stall bound (§3.4) — the inbound mirror of `run_outbound_fetch`'s: an embedder
+    // that never answers must not hold the waiter, this task, and the accepted bi-stream open
+    // for the rest of the session (~100 unanswered pastes exhaust the connection's bidi-stream
+    // budget and every later host paste stalls). `send.stopped()` additionally wakes us the
+    // moment the peer gives up on its side of the fetch.
+    let answer = tokio::select! {
+        r = rx => r.ok().flatten(),
+        _ = send.stopped() => {
+            let _ = events.try_send(ClipEventCore::Cancelled { id: req_id });
+            None
+        }
+        _ = tokio::time::sleep(std::time::Duration::from_secs(FETCH_STALL_SECS)) => {
+            let _ = events.try_send(ClipEventCore::Cancelled { id: req_id });
+            None
+        }
+    };
+    // Idempotent — the answering/denying paths already removed it; the stall paths must.
+    waiters.lock().unwrap().remove(&req_id);
+
+    match answer {
+        Some(bytes) => {
            if clipstream::write_fetch_hdr(
                &mut send,
                &ClipFetchHdr {
@@ -302,3 +351,65 @@ async fn run_outbound_fetch(
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::quic::test_util::connect_pair;
+    use crate::quic::CLIP_FILE_INDEX_NONE;
+
+    /// A serve chunk that alone breaches the requester-side CLIP_FETCH_CAP must fail the
+    /// transfer immediately — Error to the embedder, UNAVAILABLE to the peer — instead of
+    /// accumulating Ok-per-chunk toward a guaranteed peer-side rejection. Also pins the
+    /// waiter-membership gate: the serve only accumulated because the fetch was parked.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn oversized_serve_fails_the_transfer_instead_of_buffering() {
+        let (_s, _c, host_conn, client_conn) = connect_pair().await;
+        let (ev_tx, ev_rx) = std::sync::mpsc::sync_channel(16);
+        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
+        tokio::spawn(run(client_conn, ev_tx, cmd_rx));
+
+        // Host pastes: it opens a fetch bi-stream toward the client.
+        let req = ClipFetch {
+            seq: 1,
+            file_index: CLIP_FILE_INDEX_NONE,
+            mime: "text/plain;charset=utf-8".into(),
+        };
+        let (_send, mut recv) = clipstream::open_fetch(&host_conn, &req).await.unwrap();
+
+        // The client core surfaces the FetchRequest (the waiter is parked now).
+        let (req_id, ev_rx) = tokio::task::spawn_blocking(move || {
+            match ev_rx
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .unwrap()
+            {
+                ClipEventCore::FetchRequest { req_id, .. } => (req_id, ev_rx),
+                other => panic!("expected FetchRequest, got {other:?}"),
+            }
+        })
+        .await
+        .unwrap();
+
+        cmd_tx
+            .send(ClipCommand::Serve {
+                req_id,
+                bytes: vec![0u8; CLIP_FETCH_CAP + 1],
+                last: false,
+            })
+            .unwrap();
+        let ev = tokio::task::spawn_blocking(move || {
+            ev_rx
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .unwrap()
+        })
+        .await
+        .unwrap();
+        match ev {
+            ClipEventCore::Error { id, .. } => assert_eq!(id, req_id),
+            other => panic!("expected Error, got {other:?}"),
+        }
+        // The peer's read side sees the transfer refused, not a hang.
+        let hdr = clipstream::read_fetch_hdr(&mut recv).await.unwrap();
+        assert_eq!(hdr.status, CLIP_FETCH_UNAVAILABLE);
+    }
+}
@@ -86,8 +86,18 @@ impl ErasureCoder for Gf8Coder {
            // No FEC: every original must already be present.
            return collect_originals(received, data_count);
        }
-        let rs = ReedSolomon::new(data_count, recovery_count)
-            .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
+        // Same (k, m)-keyed cache as `encode_into`: a fresh ReedSolomon per lossy block costs a
+        // full generator build (k×2k Gauss-Jordan + total×k×k multiply) on the real-time pump
+        // thread AND forfeits the instance's decode-matrix cache for stable loss patterns.
+        let mut guard = self.rs.lock().unwrap_or_else(|p| p.into_inner());
+        let cached =
+            matches!(&*guard, Some((ck, cm, _)) if *ck == data_count && *cm == recovery_count);
+        if !cached {
+            let rs = ReedSolomon::new(data_count, recovery_count)
+                .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
+            *guard = Some((data_count, recovery_count, rs));
+        }
+        let rs = &guard.as_ref().expect("cache populated above").2;
        rs.reconstruct_data(received)
            .map_err(|_| FecError::Backend("gf8 reconstruct"))?;
        collect_originals(received, data_count)
@@ -116,8 +126,17 @@ impl ErasureCoder for Gf8Coder {
        for &(j, bytes) in recovery {
            received[data_count + j] = Some(bytes.to_vec());
        }
-        let rs = ReedSolomon::new(data_count, recovery_count)
-            .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
+        // Cache the codec by (k, m) exactly as `encode_into`/`reconstruct` do (see the note
+        // there) — this path runs per lossy block on the pump thread.
+        let mut guard = self.rs.lock().unwrap_or_else(|p| p.into_inner());
+        let cached =
+            matches!(&*guard, Some((ck, cm, _)) if *ck == data_count && *cm == recovery_count);
+        if !cached {
+            let rs = ReedSolomon::new(data_count, recovery_count)
+                .map_err(|_| FecError::Config("invalid GF(2^8) shard counts"))?;
+            *guard = Some((data_count, recovery_count, rs));
+        }
+        let rs = &guard.as_ref().expect("cache populated above").2;
        rs.reconstruct_data(&mut received)
            .map_err(|_| FecError::Backend("gf8 reconstruct"))?;
        for (i, h) in have.iter().enumerate() {
@@ -99,7 +99,10 @@ pub use stats::Stats;
 /// 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;
+/// v10: added `punktfunk_connection_clock_offset_now_ns` — the LIVE (mid-stream re-synced)
+/// clock offset ongoing latency math must use; the connect-time getter stays frozen by
+/// contract. Additive, client-local — no wire change, so [`WIRE_VERSION`] is unchanged.
+pub const ABI_VERSION: u32 = 10;

 /// The punktfunk/1 **wire** version — what `Hello`/`Welcome` carry and hosts equality-check.
 /// Deliberately its own constant: [`ABI_VERSION`] tracks the embeddable **C surface**
@@ -79,4 +79,4 @@ pub use pairing::*;
 pub use crate::reject::*;

 #[cfg(test)]
-mod test_util;
+pub(crate) mod test_util;
@@ -198,8 +198,14 @@ pub(super) fn send_gso(t: &UdpTransport, packets: &[&[u8]]) -> std::io::Result<u
        return send_batch(t, packets);
    }
    let fd = t.socket.as_raw_fd();
-    // A GSO super-buffer is capped at 64 segments AND 65535 payload bytes (kernel limits).
-    let max_seg = (65535 / seg).clamp(1, 64);
+    // A GSO super-buffer is capped at 64 segments AND, in bytes, by the kernel's UDP payload
+    // ceiling. 65535 is the IP-datagram cap — the payload is that minus the IP + UDP headers
+    // the cork accounts for (IPv4: 65507, IPv6: 65487). Use the tighter v6 figure: it costs at
+    // most one segment per train, while a super-buffer over the ceiling is bounced with
+    // EMSGSIZE — which gso_unsupported() reads as "no GSO on this path" and latches GSO off
+    // process-wide, silently forfeiting the multi-Gbps lever over a local arithmetic slip.
+    const GSO_MAX_PAYLOAD: usize = 65535 - 40 - 8;
+    let max_seg = (GSO_MAX_PAYLOAD / seg).clamp(1, 64);
    let mut scratch: Vec<u8> = Vec::with_capacity(seg * max_seg);
    let mut sent = 0usize;
    for chunk in packets.chunks(max_seg) {
@@ -187,34 +187,58 @@ pub fn detect_mul_slice() -> (MulSliceFn, MulSliceFn) {
 // Safe wrappers for SIMD functions (used as function pointer targets)
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_gfni_avx2(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_gfni_avx2(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_xor_gfni_avx2(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_xor_gfni_avx2(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_avx2(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_avx2(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_xor_avx2(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_xor_avx2(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_gfni_sse(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_gfni_sse(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_xor_gfni_sse(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_xor_gfni_sse(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_ssse3(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_ssse3(c, input, out) }
 }
 #[cfg(target_arch = "x86_64")]
 fn wrap_mul_slice_xor_ssse3(c: u8, input: &[u8], out: &mut [u8]) {
+    // The unsafe callee bounds every load AND store on input.len(); unequal
+    // lengths would write past `out`. Same invariant mul_slice/mul_slice_xor enforce.
+    assert_eq!(input.len(), out.len());
    unsafe { mul_slice_xor_ssse3(c, input, out) }
 }

@@ -624,8 +624,15 @@ impl ReedSolomon {
            for (i_input, &valid_idx) in valid_indices.iter().enumerate() {
                // SAFETY: valid_idx and missing indices are disjoint sets,
                // so we can safely read from valid_idx while writing to missing indices.
-                let input_ptr = shards[valid_idx].get().unwrap().as_ptr();
-                let input_slice = unsafe { std::slice::from_raw_parts(input_ptr, shard_len) };
+                // Pointer AND length from the same `get()` — an impl whose `len()`
+                // disagrees with its slice then panics in `mul_slice` instead of this
+                // fabricating an out-of-bounds slice from the trait-reported length.
+                let (input_ptr, input_len) = {
+                    let input = shards[valid_idx].get().unwrap();
+                    debug_assert_eq!(input.len(), shard_len);
+                    (input.as_ptr(), input.len())
+                };
+                let input_slice = unsafe { std::slice::from_raw_parts(input_ptr, input_len) };

                for (i_out, &missing_idx) in missing_data_indices.iter().enumerate() {
                    let c = matrix_rows[i_out][i_input];
@@ -659,8 +666,13 @@ impl ReedSolomon {

            for i_input in 0..self.data_shard_count {
                // SAFETY: data shards (0..data_shard_count) are disjoint from parity shards.
-                let input_ptr = shards[i_input].get().unwrap().as_ptr();
-                let input_slice = unsafe { std::slice::from_raw_parts(input_ptr, shard_len) };
+                // Same discipline as the data-shard loop above: length from the slice.
+                let (input_ptr, input_len) = {
+                    let input = shards[i_input].get().unwrap();
+                    debug_assert_eq!(input.len(), shard_len);
+                    (input.as_ptr(), input.len())
+                };
+                let input_slice = unsafe { std::slice::from_raw_parts(input_ptr, input_len) };

                for (i_out, &missing_idx) in missing_parity_indices.iter().enumerate() {
                    let c = matrix_rows[i_out][i_input];
@@ -702,6 +714,10 @@ impl ReedSolomon {
 /// yield non-overlapping memory regions from `get()`/`get_mut()`. Specifically:
 /// - `get()` on element `i` must not alias `get_mut()` on element `j` when `i != j`.
 /// - The returned slices must remain valid and not be moved/reallocated while borrows are active.
+/// - `len()` must return `Some(n)` exactly when `get()`/`get_mut()` return `Some(s)`, and then
+///   `s.len() == n`; after `initialize(n)`, `get_mut()` must return a slice of length `n`.
+///   `reconstruct_internal` sizes its raw-pointer reads from `len()`, so a disagreement would
+///   otherwise read past the allocation.
 ///
 /// This is required because `reconstruct_internal` uses raw pointers to read from
 /// some shard indices while writing to others simultaneously.