refactor(core): peel session.rs anti-replay, perf telemetry, and seal lane into submodules

session.rs (1203) sharpens down to the two hot-path state machines + lifecycle (887): ReplayWindow + seq_of + their tests -> session/replay.rs; the PUNKTFUNK_PERF trio PumpPerf/SealPerf/TimedCoder -> session/perf.rs; the Phase-1.5 lane machinery SealLane/SealJob/seal_wire_slice/ TWO_LANE_MIN_PACKETS -> session/seal.rs. Facade pattern (session.rs stays the parent file); pub use keeps session::{PumpPerf,SealPerf} stable and lib.rs re-exports are untouched. Pure code motion + pub(super) bumps — seal_frame_inner/poll_frame/poll_input bodies unchanged; the wire-equivalence tests stay co-located with the seal path they pin. 196 lib tests pass, clippy --features quic --all-targets clean, fmt clean. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-20 17:56:26 +02:00
parent e0a1edb9d4
commit eacaaa5cd8
4 changed files with 354 additions and 323 deletions
@@ -102,160 +102,15 @@ fn stamp_received(mut f: Frame) -> Frame {
    f
 }
-/// Wire-packet count at which a frame's sealing splits across two lanes (plan Phase 1.5):
+mod perf;
-/// below it the channel rendezvous (~µs) isn't worth it; at it the halved AES-GCM span
+mod replay;
-/// (≥ ~125 µs of ~1 µs/packet work) dwarfs the hand-off. ≈300 KB of wire, i.e. ≥150 Mbps
+mod seal;
 /// at 60 fps — small frames and the probe's ~17-packet AUs stay strictly single-lane.
 const TWO_LANE_MIN_PACKETS: usize = 256;
-/// One two-lane seal hand-off: the frame's back-half wire buffers, sealed by the worker with
+pub use perf::{PumpPerf, SealPerf};
 /// nonces `seq_base + i` (the nonce order is deterministic per shard index, which is what
 /// makes the split sound). Round-trips through the channels so the buffers return to the pool.
 struct SealJob {
    bufs: Vec<Vec<u8>>,
    seq_base: u64,
    timed: bool,
    /// Worker-lane CPU ns (when `timed`) and the seal outcome, filled in by the worker.
    ns: u64,
    result: Result<()>,
 }
-/// The persistent second seal lane: a worker thread that AES-GCM-seals the back half of a
+use perf::TimedCoder;
-/// large frame's packets while the send thread seals the front half. Rendezvous channels
+use replay::{seq_of, ReplayWindow};
-/// (bound 1) — the send thread submits, seals its half, then waits; no per-frame spawn.
+use seal::{seal_wire_slice, SealJob, SealLane, TWO_LANE_MIN_PACKETS};
 /// Dropping the struct closes the channel and the worker exits.
 struct SealLane {
    to_worker: std::sync::mpsc::SyncSender<SealJob>,
    from_worker: std::sync::mpsc::Receiver<SealJob>,
 }
 impl SealLane {
    fn spawn(crypto: std::sync::Arc<SessionCrypto>) -> Option<SealLane> {
        let (to_worker, jobs) = std::sync::mpsc::sync_channel::<SealJob>(1);
        let (done_tx, from_worker) = std::sync::mpsc::sync_channel::<SealJob>(1);
        std::thread::Builder::new()
            .name("punktfunk-seal2".into())
            .spawn(move || {
                while let Ok(mut job) = jobs.recv() {
                    let t0 = job.timed.then(std::time::Instant::now);
                    job.result = seal_wire_slice(&crypto, &mut job.bufs, job.seq_base);
                    if let Some(t0) = t0 {
                        job.ns = t0.elapsed().as_nanos() as u64;
                    }
                    if done_tx.send(job).is_err() {
                        break; // session gone mid-frame — nothing left to seal for
                    }
                }
            })
            .ok()?;
        Some(SealLane {
            to_worker,
            from_worker,
        })
    }
 }
 /// Seal a run of pre-written wire buffers in place: buffer `i` is `seq(8) ‖ plaintext ‖ tag
 /// scratch` and seals over `[8..]` with sequence `seq_base + i` — the exact per-packet layout
 /// and nonce order of the fused single-lane path. Shared by both lanes.
 fn seal_wire_slice(c: &SessionCrypto, wires: &mut [Vec<u8>], seq_base: u64) -> Result<()> {
    for (i, wire) in wires.iter_mut().enumerate() {
        c.seal_in_place(seq_base.wrapping_add(i as u64), &mut wire[8..])?;
    }
    Ok(())
 }
 /// 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,
 }
 /// Accumulated host send-path stage timings since the last [`Session::take_seal_perf`] (plan
 /// Phase 0.4, host half). Answers "where does the send thread go" at rate: FEC parity
 /// generation (`fec_ns`, inside [`ErasureCoder::encode_into`]) vs AES-GCM (`seal_ns`,
 /// per-packet `seal_in_place`) vs the socket handoff (`sock_ns` — `send_gso`/`sendmmsg`
 /// syscalls; the internal submit paths time it here, the paced video path folds its chunk
 /// sends in via [`Session::note_sock_ns`]). The Phase 1.5 gate reads off this split: build
 /// two-lane seal only if `seal_ns` exceeds ~15% of the send thread at 2 Gbps.
 #[derive(Debug, Default, Clone, Copy)]
 pub struct SealPerf {
    /// ns inside `ErasureCoder::encode_into` (parity generation).
    pub fec_ns: u64,
    /// ns inside `seal_in_place` across all wire packets (AES-128-GCM).
    pub seal_ns: u64,
    /// ns inside `send_sealed` (socket syscalls), where the session can see it.
    pub sock_ns: u64,
    /// Frames sealed and wire packets sealed over the accumulation window.
    pub frames: u64,
    pub packets: u64,
 }
 /// [`ErasureCoder`] shim accumulating the time spent in `encode_into` (the send-path FEC
 /// stage) — only constructed when `PUNKTFUNK_PERF` armed the session's [`SealPerf`]. The
 /// counter is atomic purely to satisfy the trait's `Sync` bound; it lives on one thread.
 struct TimedCoder<'a> {
    inner: &'a dyn ErasureCoder,
    ns: &'a std::sync::atomic::AtomicU64,
 }
 impl ErasureCoder for TimedCoder<'_> {
    fn scheme(&self) -> crate::config::FecScheme {
        self.inner.scheme()
    }
    fn encode(
        &self,
        data: &[&[u8]],
        recovery_count: usize,
    ) -> std::result::Result<Vec<Vec<u8>>, crate::fec::FecError> {
        self.inner.encode(data, recovery_count)
    }
    fn encode_into(
        &self,
        data: &[&[u8]],
        recovery_count: usize,
        out: &mut Vec<Vec<u8>>,
    ) -> std::result::Result<(), crate::fec::FecError> {
        let t0 = std::time::Instant::now();
        let r = self.inner.encode_into(data, recovery_count, out);
        self.ns.fetch_add(
            t0.elapsed().as_nanos() as u64,
            std::sync::atomic::Ordering::Relaxed,
        );
        r
    }
    fn reconstruct(
        &self,
        data_count: usize,
        recovery_count: usize,
        received: &mut [Option<Vec<u8>>],
    ) -> std::result::Result<Vec<Vec<u8>>, crate::fec::FecError> {
        self.inner.reconstruct(data_count, recovery_count, received)
    }
    fn reconstruct_into(
        &self,
        recovery_count: usize,
        data: &mut [&mut [u8]],
        have: &[bool],
        recovery: &[(usize, &[u8])],
    ) -> std::result::Result<(), crate::fec::FecError> {
        self.inner
            .reconstruct_into(recovery_count, data, have, recovery)
    }
 }
 /// Datagrams drained per `recvmmsg` syscall on the client (the reused ring's size). 128 keeps
 /// the syscall rate ≤ ~3.4k/s even at the ~430k pkt/s the post-2026-07-14 receive path delivers
@@ -836,102 +691,6 @@ impl Session {
    }
 }
 /// Extract the AEAD-authenticated 8-byte big-endian sequence prefix from a sealed wire datagram.
 /// Only called on the encrypted receive path, where a preceding successful open has already
 /// established `wire.len() >= 8`.
 fn seq_of(wire: &[u8]) -> u64 {
    u64::from_be_bytes(wire[..8].try_into().unwrap())
 }
 /// Depth of the anti-replay window, in sequences. The sender advances its sequence once per
 /// datagram, so this must cover the reassembler's 120 ms loss window
 /// ([`LOSS_WINDOW_NS`](crate::packet)) at line-rate packet rates — otherwise the replay filter
 /// silently re-tightens the "late ≠ lost" fix: a Wi-Fi-retry-delayed shard the reassembler would
 /// still use gets dropped here as "older than the window" first (4096 was only ~33 ms at the
 /// ~125k pkt/s of a 1 Gbps stream; 32768 topped out around ~2 Gbps — which the client now
 /// exceeds: the 2026-07-14 zero-copy + hardware-AES work measured ~4.8 Gbps wire ≈ 430k pkt/s
 /// delivered). 131072 covers 120 ms up to ~1.09M pkt/s (≈12 Gbps wire) and is effectively
 /// unbounded for the sparse input stream, while still bounding how far back a replay could
 /// hide; the bitmap costs 16 KiB per session.
 const REPLAY_WINDOW: u64 = 131072;
 const REPLAY_WORDS: usize = (REPLAY_WINDOW / 64) as usize;
 /// Sliding-window anti-replay filter over the AEAD-authenticated wire sequence. The sender counts
 /// its datagrams from 0, and the protocol never legitimately re-sends a sequence (FEC recovery
 /// shards get fresh ones), so a sequence seen twice is a replay. The AEAD tag already authenticates
 /// the sequence — a forged one can't open — so this only has to reject *duplicates* of validly
 /// sealed datagrams (and anything older than the window, which we can no longer prove is fresh).
 /// Genuine reordering within the window is accepted. Bitmap-per-sequence, indexed `seq % WINDOW`.
 struct ReplayWindow {
    /// Highest sequence accepted so far; `seen` stays false until the first datagram.
    highest: u64,
    seen: bool,
    /// One bit per in-window sequence in `(highest - WINDOW, highest]`.
    bits: [u64; REPLAY_WORDS],
 }
 impl ReplayWindow {
    fn new() -> ReplayWindow {
        ReplayWindow {
            highest: 0,
            seen: false,
            bits: [0; REPLAY_WORDS],
        }
    }
    #[inline]
    fn word_bit(seq: u64) -> (usize, u64) {
        let idx = (seq % REPLAY_WINDOW) as usize;
        (idx / 64, 1u64 << (idx % 64))
    }
    fn is_set(&self, seq: u64) -> bool {
        let (w, b) = Self::word_bit(seq);
        self.bits[w] & b != 0
    }
    fn set(&mut self, seq: u64) {
        let (w, b) = Self::word_bit(seq);
        self.bits[w] |= b;
    }
    fn unset(&mut self, seq: u64) {
        let (w, b) = Self::word_bit(seq);
        self.bits[w] &= !b;
    }
    /// Record `seq`, returning `true` if it's fresh (accept) or `false` if it's a replay / too old.
    fn accept(&mut self, seq: u64) -> bool {
        if !self.seen {
            self.seen = true;
            self.highest = seq;
            self.set(seq);
            return true;
        }
        if seq > self.highest {
            // Advance the window. Sequences between the old and new high slide in unseen, so clear
            // their (possibly stale, from a full window ago) slots — unless we jumped an entire
            // window, in which case wipe the bitmap wholesale.
            if seq - self.highest >= REPLAY_WINDOW {
                self.bits = [0; REPLAY_WORDS];
            } else {
                let mut s = self.highest + 1;
                while s < seq {
                    self.unset(s);
                    s += 1;
                }
            }
            self.highest = seq;
            self.set(seq);
            true
        } else if self.highest - seq >= REPLAY_WINDOW || self.is_set(seq) {
            // Older than the window (can't prove it isn't a replay) or already seen (a duplicate) —
            // either way, drop it.
            false
        } else {
            self.set(seq); // in-window and not yet seen — a genuine reorder
            true
        }
    }
 }
 #[cfg(test)]
 mod wire_equivalence_tests {
    use super::*;
@@ -1126,78 +885,3 @@ mod wire_equivalence_tests {
        );
    }
 }
 #[cfg(test)]
 mod replay_tests {
    use super::*;
    #[test]
    fn accepts_in_order_and_rejects_duplicates() {
        let mut w = ReplayWindow::new();
        for seq in 0..1000 {
            assert!(w.accept(seq), "fresh in-order seq {seq} must be accepted");
        }
        // Every one of those is now a replay.
        for seq in 0..1000 {
            assert!(!w.accept(seq), "replayed seq {seq} must be rejected");
        }
    }
    #[test]
    fn accepts_reorder_within_window_once() {
        let mut w = ReplayWindow::new();
        assert!(w.accept(100));
        // Earlier-but-in-window sequences (a genuine reorder) are accepted exactly once.
        assert!(w.accept(80));
        assert!(!w.accept(80), "second copy of a reordered seq is a replay");
        assert!(w.accept(99));
        assert!(
            !w.accept(100),
            "the high-water seq itself can't be replayed"
        );
    }
    #[test]
    fn rejects_older_than_window() {
        let mut w = ReplayWindow::new();
        assert!(w.accept(REPLAY_WINDOW * 2));
        // Anything a full window or more behind the high-water mark is dropped (can't prove fresh).
        assert!(!w.accept(REPLAY_WINDOW * 2 - REPLAY_WINDOW));
        assert!(!w.accept(0));
        // But just inside the window is still accepted.
        assert!(w.accept(REPLAY_WINDOW * 2 - (REPLAY_WINDOW - 1)));
    }
    #[test]
    fn large_forward_jump_wipes_stale_bits() {
        let mut w = ReplayWindow::new();
        assert!(w.accept(5));
        // Jump far forward (more than a window). The slot for an old seq that aliases 5 mod WINDOW
        // must read as unseen afterward, i.e. the jump cleared it — so a NEW seq there is accepted.
        let far = 10 * REPLAY_WINDOW + 5;
        assert!(w.accept(far));
        assert!(
            !w.accept(5),
            "the pre-jump seq is now far older than the window"
        );
        // A fresh seq aliasing 5 (mod WINDOW) but inside the new window is accepted, proving the
        // stale bit was cleared rather than mistaken for a replay.
        assert!(w.accept(far - REPLAY_WINDOW + 1));
    }
    #[test]
    fn first_seq_need_not_be_zero() {
        // Startup loss can mean the first datagram we ever open isn't seq 0.
        let mut w = ReplayWindow::new();
        assert!(w.accept(42));
        assert!(!w.accept(42));
        assert!(w.accept(43));
    }
    #[test]
    fn seq_of_reads_the_big_endian_prefix() {
        let mut wire = 0x0102_0304_0506_0708u64.to_be_bytes().to_vec();
        wire.extend_from_slice(b"ciphertext-and-tag");
        assert_eq!(seq_of(&wire), 0x0102_0304_0506_0708);
    }
 }
@@ -0,0 +1,98 @@
 //! `PUNKTFUNK_PERF` stage-timing telemetry for the two hot paths: where the client pump
 //! and the host send thread actually spend their time, accumulated per report window and
 //! drained via [`Session::take_pump_perf`](super::Session::take_pump_perf) /
 //! [`Session::take_seal_perf`](super::Session::take_seal_perf).
 use crate::fec::ErasureCoder;
 /// Accumulated client receive-path stage timings since the last [`Session::take_pump_perf`](super::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,
 }
 /// Accumulated host send-path stage timings since the last [`Session::take_seal_perf`](super::Session::take_seal_perf) (plan
 /// Phase 0.4, host half). Answers "where does the send thread go" at rate: FEC parity
 /// generation (`fec_ns`, inside [`ErasureCoder::encode_into`]) vs AES-GCM (`seal_ns`,
 /// per-packet `seal_in_place`) vs the socket handoff (`sock_ns` — `send_gso`/`sendmmsg`
 /// syscalls; the internal submit paths time it here, the paced video path folds its chunk
 /// sends in via [`Session::note_sock_ns`](super::Session::note_sock_ns)). The Phase 1.5 gate reads off this split: build
 /// two-lane seal only if `seal_ns` exceeds ~15% of the send thread at 2 Gbps.
 #[derive(Debug, Default, Clone, Copy)]
 pub struct SealPerf {
    /// ns inside `ErasureCoder::encode_into` (parity generation).
    pub fec_ns: u64,
    /// ns inside `seal_in_place` across all wire packets (AES-128-GCM).
    pub seal_ns: u64,
    /// ns inside `send_sealed` (socket syscalls), where the session can see it.
    pub sock_ns: u64,
    /// Frames sealed and wire packets sealed over the accumulation window.
    pub frames: u64,
    pub packets: u64,
 }
 /// [`ErasureCoder`] shim accumulating the time spent in `encode_into` (the send-path FEC
 /// stage) — only constructed when `PUNKTFUNK_PERF` armed the session's [`SealPerf`]. The
 /// counter is atomic purely to satisfy the trait's `Sync` bound; it lives on one thread.
 pub(super) struct TimedCoder<'a> {
    pub(super) inner: &'a dyn ErasureCoder,
    pub(super) ns: &'a std::sync::atomic::AtomicU64,
 }
 impl ErasureCoder for TimedCoder<'_> {
    fn scheme(&self) -> crate::config::FecScheme {
        self.inner.scheme()
    }
    fn encode(
        &self,
        data: &[&[u8]],
        recovery_count: usize,
    ) -> std::result::Result<Vec<Vec<u8>>, crate::fec::FecError> {
        self.inner.encode(data, recovery_count)
    }
    fn encode_into(
        &self,
        data: &[&[u8]],
        recovery_count: usize,
        out: &mut Vec<Vec<u8>>,
    ) -> std::result::Result<(), crate::fec::FecError> {
        let t0 = std::time::Instant::now();
        let r = self.inner.encode_into(data, recovery_count, out);
        self.ns.fetch_add(
            t0.elapsed().as_nanos() as u64,
            std::sync::atomic::Ordering::Relaxed,
        );
        r
    }
    fn reconstruct(
        &self,
        data_count: usize,
        recovery_count: usize,
        received: &mut [Option<Vec<u8>>],
    ) -> std::result::Result<Vec<Vec<u8>>, crate::fec::FecError> {
        self.inner.reconstruct(data_count, recovery_count, received)
    }
    fn reconstruct_into(
        &self,
        recovery_count: usize,
        data: &mut [&mut [u8]],
        have: &[bool],
        recovery: &[(usize, &[u8])],
    ) -> std::result::Result<(), crate::fec::FecError> {
        self.inner
            .reconstruct_into(recovery_count, data, have, recovery)
    }
 }
@@ -0,0 +1,175 @@
 //! Sliding-window anti-replay filter over the AEAD-authenticated wire sequence
 //! (plan §1). Applied on both encrypted receive paths —
 //! [`Session::poll_frame`](super::Session::poll_frame) and
 //! [`Session::poll_input`](super::Session::poll_input).
 /// Extract the AEAD-authenticated 8-byte big-endian sequence prefix from a sealed wire datagram.
 /// Only called on the encrypted receive path, where a preceding successful open has already
 /// established `wire.len() >= 8`.
 pub(super) fn seq_of(wire: &[u8]) -> u64 {
    u64::from_be_bytes(wire[..8].try_into().unwrap())
 }
 /// Depth of the anti-replay window, in sequences. The sender advances its sequence once per
 /// datagram, so this must cover the reassembler's 120 ms loss window
 /// ([`LOSS_WINDOW_NS`](crate::packet)) at line-rate packet rates — otherwise the replay filter
 /// silently re-tightens the "late ≠ lost" fix: a Wi-Fi-retry-delayed shard the reassembler would
 /// still use gets dropped here as "older than the window" first (4096 was only ~33 ms at the
 /// ~125k pkt/s of a 1 Gbps stream; 32768 topped out around ~2 Gbps — which the client now
 /// exceeds: the 2026-07-14 zero-copy + hardware-AES work measured ~4.8 Gbps wire ≈ 430k pkt/s
 /// delivered). 131072 covers 120 ms up to ~1.09M pkt/s (≈12 Gbps wire) and is effectively
 /// unbounded for the sparse input stream, while still bounding how far back a replay could
 /// hide; the bitmap costs 16 KiB per session.
 const REPLAY_WINDOW: u64 = 131072;
 const REPLAY_WORDS: usize = (REPLAY_WINDOW / 64) as usize;
 /// Sliding-window anti-replay filter over the AEAD-authenticated wire sequence. The sender counts
 /// its datagrams from 0, and the protocol never legitimately re-sends a sequence (FEC recovery
 /// shards get fresh ones), so a sequence seen twice is a replay. The AEAD tag already authenticates
 /// the sequence — a forged one can't open — so this only has to reject *duplicates* of validly
 /// sealed datagrams (and anything older than the window, which we can no longer prove is fresh).
 /// Genuine reordering within the window is accepted. Bitmap-per-sequence, indexed `seq % WINDOW`.
 pub(super) struct ReplayWindow {
    /// Highest sequence accepted so far; `seen` stays false until the first datagram.
    highest: u64,
    seen: bool,
    /// One bit per in-window sequence in `(highest - WINDOW, highest]`.
    bits: [u64; REPLAY_WORDS],
 }
 impl ReplayWindow {
    pub(super) fn new() -> ReplayWindow {
        ReplayWindow {
            highest: 0,
            seen: false,
            bits: [0; REPLAY_WORDS],
        }
    }
    #[inline]
    fn word_bit(seq: u64) -> (usize, u64) {
        let idx = (seq % REPLAY_WINDOW) as usize;
        (idx / 64, 1u64 << (idx % 64))
    }
    fn is_set(&self, seq: u64) -> bool {
        let (w, b) = Self::word_bit(seq);
        self.bits[w] & b != 0
    }
    fn set(&mut self, seq: u64) {
        let (w, b) = Self::word_bit(seq);
        self.bits[w] |= b;
    }
    fn unset(&mut self, seq: u64) {
        let (w, b) = Self::word_bit(seq);
        self.bits[w] &= !b;
    }
    /// Record `seq`, returning `true` if it's fresh (accept) or `false` if it's a replay / too old.
    pub(super) fn accept(&mut self, seq: u64) -> bool {
        if !self.seen {
            self.seen = true;
            self.highest = seq;
            self.set(seq);
            return true;
        }
        if seq > self.highest {
            // Advance the window. Sequences between the old and new high slide in unseen, so clear
            // their (possibly stale, from a full window ago) slots — unless we jumped an entire
            // window, in which case wipe the bitmap wholesale.
            if seq - self.highest >= REPLAY_WINDOW {
                self.bits = [0; REPLAY_WORDS];
            } else {
                let mut s = self.highest + 1;
                while s < seq {
                    self.unset(s);
                    s += 1;
                }
            }
            self.highest = seq;
            self.set(seq);
            true
        } else if self.highest - seq >= REPLAY_WINDOW || self.is_set(seq) {
            // Older than the window (can't prove it isn't a replay) or already seen (a duplicate) —
            // either way, drop it.
            false
        } else {
            self.set(seq); // in-window and not yet seen — a genuine reorder
            true
        }
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn accepts_in_order_and_rejects_duplicates() {
        let mut w = ReplayWindow::new();
        for seq in 0..1000 {
            assert!(w.accept(seq), "fresh in-order seq {seq} must be accepted");
        }
        // Every one of those is now a replay.
        for seq in 0..1000 {
            assert!(!w.accept(seq), "replayed seq {seq} must be rejected");
        }
    }
    #[test]
    fn accepts_reorder_within_window_once() {
        let mut w = ReplayWindow::new();
        assert!(w.accept(100));
        // Earlier-but-in-window sequences (a genuine reorder) are accepted exactly once.
        assert!(w.accept(80));
        assert!(!w.accept(80), "second copy of a reordered seq is a replay");
        assert!(w.accept(99));
        assert!(
            !w.accept(100),
            "the high-water seq itself can't be replayed"
        );
    }
    #[test]
    fn rejects_older_than_window() {
        let mut w = ReplayWindow::new();
        assert!(w.accept(REPLAY_WINDOW * 2));
        // Anything a full window or more behind the high-water mark is dropped (can't prove fresh).
        assert!(!w.accept(REPLAY_WINDOW * 2 - REPLAY_WINDOW));
        assert!(!w.accept(0));
        // But just inside the window is still accepted.
        assert!(w.accept(REPLAY_WINDOW * 2 - (REPLAY_WINDOW - 1)));
    }
    #[test]
    fn large_forward_jump_wipes_stale_bits() {
        let mut w = ReplayWindow::new();
        assert!(w.accept(5));
        // Jump far forward (more than a window). The slot for an old seq that aliases 5 mod WINDOW
        // must read as unseen afterward, i.e. the jump cleared it — so a NEW seq there is accepted.
        let far = 10 * REPLAY_WINDOW + 5;
        assert!(w.accept(far));
        assert!(
            !w.accept(5),
            "the pre-jump seq is now far older than the window"
        );
        // A fresh seq aliasing 5 (mod WINDOW) but inside the new window is accepted, proving the
        // stale bit was cleared rather than mistaken for a replay.
        assert!(w.accept(far - REPLAY_WINDOW + 1));
    }
    #[test]
    fn first_seq_need_not_be_zero() {
        // Startup loss can mean the first datagram we ever open isn't seq 0.
        let mut w = ReplayWindow::new();
        assert!(w.accept(42));
        assert!(!w.accept(42));
        assert!(w.accept(43));
    }
    #[test]
    fn seq_of_reads_the_big_endian_prefix() {
        let mut wire = 0x0102_0304_0506_0708u64.to_be_bytes().to_vec();
        wire.extend_from_slice(b"ciphertext-and-tag");
        assert_eq!(seq_of(&wire), 0x0102_0304_0506_0708);
    }
 }
@@ -0,0 +1,74 @@
 //! The second seal lane (plan Phase 1.5): a persistent worker thread that AES-GCM-seals
 //! the back half of a large frame's wire packets while the send thread seals the front.
 //! [`Session`](super::Session)`::seal_frame_inner` owns the split policy; this module owns
 //! the lane machinery and the shared per-slice seal loop.
 use crate::crypto::SessionCrypto;
 use crate::error::Result;
 /// 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
 /// at 60 fps — small frames and the probe's ~17-packet AUs stay strictly single-lane.
 pub(super) const TWO_LANE_MIN_PACKETS: usize = 256;
 /// One two-lane seal hand-off: the frame's back-half wire buffers, sealed by the worker with
 /// nonces `seq_base + i` (the nonce order is deterministic per shard index, which is what
 /// makes the split sound). Round-trips through the channels so the buffers return to the pool.
 pub(super) struct SealJob {
    pub(super) bufs: Vec<Vec<u8>>,
    pub(super) seq_base: u64,
    pub(super) timed: bool,
    /// Worker-lane CPU ns (when `timed`) and the seal outcome, filled in by the worker.
    pub(super) ns: u64,
    pub(super) result: Result<()>,
 }
 /// The persistent second seal lane: a worker thread that AES-GCM-seals the back half of a
 /// large frame's packets while the send thread seals the front half. Rendezvous channels
 /// (bound 1) — the send thread submits, seals its half, then waits; no per-frame spawn.
 /// Dropping the struct closes the channel and the worker exits.
 pub(super) struct SealLane {
    pub(super) to_worker: std::sync::mpsc::SyncSender<SealJob>,
    pub(super) from_worker: std::sync::mpsc::Receiver<SealJob>,
 }
 impl SealLane {
    pub(super) fn spawn(crypto: std::sync::Arc<SessionCrypto>) -> Option<SealLane> {
        let (to_worker, jobs) = std::sync::mpsc::sync_channel::<SealJob>(1);
        let (done_tx, from_worker) = std::sync::mpsc::sync_channel::<SealJob>(1);
        std::thread::Builder::new()
            .name("punktfunk-seal2".into())
            .spawn(move || {
                while let Ok(mut job) = jobs.recv() {
                    let t0 = job.timed.then(std::time::Instant::now);
                    job.result = seal_wire_slice(&crypto, &mut job.bufs, job.seq_base);
                    if let Some(t0) = t0 {
                        job.ns = t0.elapsed().as_nanos() as u64;
                    }
                    if done_tx.send(job).is_err() {
                        break; // session gone mid-frame — nothing left to seal for
                    }
                }
            })
            .ok()?;
        Some(SealLane {
            to_worker,
            from_worker,
        })
    }
 }
 /// Seal a run of pre-written wire buffers in place: buffer `i` is `seq(8) ‖ plaintext ‖ tag
 /// scratch` and seals over `[8..]` with sequence `seq_base + i` — the exact per-packet layout
 /// and nonce order of the fused single-lane path. Shared by both lanes.
 pub(super) fn seal_wire_slice(
    c: &SessionCrypto,
    wires: &mut [Vec<u8>],
    seq_base: u64,
 ) -> Result<()> {
    for (i, wire) in wires.iter_mut().enumerate() {
        c.seal_in_place(seq_base.wrapping_add(i as u64), &mut wire[8..])?;
    }
    Ok(())
 }