From eacaaa5cd8c1f830866024c5c7ea3e1979b16215 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Mon, 20 Jul 2026 17:56:26 +0200 Subject: [PATCH] refactor(core): peel session.rs anti-replay, perf telemetry, and seal lane into submodules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/punktfunk-core/src/session.rs | 330 +------------------- crates/punktfunk-core/src/session/perf.rs | 98 ++++++ crates/punktfunk-core/src/session/replay.rs | 175 +++++++++++ crates/punktfunk-core/src/session/seal.rs | 74 +++++ 4 files changed, 354 insertions(+), 323 deletions(-) create mode 100644 crates/punktfunk-core/src/session/perf.rs create mode 100644 crates/punktfunk-core/src/session/replay.rs create mode 100644 crates/punktfunk-core/src/session/seal.rs diff --git a/crates/punktfunk-core/src/session.rs b/crates/punktfunk-core/src/session.rs index ee7db318..b674ba6e 100644 --- a/crates/punktfunk-core/src/session.rs +++ b/crates/punktfunk-core/src/session.rs @@ -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): -/// 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. -const TWO_LANE_MIN_PACKETS: usize = 256; +mod perf; +mod replay; +mod seal; -/// 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. -struct SealJob { - bufs: Vec>, - seq_base: u64, - timed: bool, - /// Worker-lane CPU ns (when `timed`) and the seal outcome, filled in by the worker. - ns: u64, - result: Result<()>, -} +pub use perf::{PumpPerf, SealPerf}; -/// 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. -struct SealLane { - to_worker: std::sync::mpsc::SyncSender, - from_worker: std::sync::mpsc::Receiver, -} - -impl SealLane { - fn spawn(crypto: std::sync::Arc) -> Option { - let (to_worker, jobs) = std::sync::mpsc::sync_channel::(1); - let (done_tx, from_worker) = std::sync::mpsc::sync_channel::(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], 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>, crate::fec::FecError> { - self.inner.encode(data, recovery_count) - } - fn encode_into( - &self, - data: &[&[u8]], - recovery_count: usize, - out: &mut Vec>, - ) -> 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>], - ) -> std::result::Result>, 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) - } -} +use perf::TimedCoder; +use replay::{seq_of, ReplayWindow}; +use seal::{seal_wire_slice, SealJob, SealLane, TWO_LANE_MIN_PACKETS}; /// 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); - } -} diff --git a/crates/punktfunk-core/src/session/perf.rs b/crates/punktfunk-core/src/session/perf.rs new file mode 100644 index 00000000..c8215bec --- /dev/null +++ b/crates/punktfunk-core/src/session/perf.rs @@ -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>, crate::fec::FecError> { + self.inner.encode(data, recovery_count) + } + fn encode_into( + &self, + data: &[&[u8]], + recovery_count: usize, + out: &mut Vec>, + ) -> 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>], + ) -> std::result::Result>, 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) + } +} diff --git a/crates/punktfunk-core/src/session/replay.rs b/crates/punktfunk-core/src/session/replay.rs new file mode 100644 index 00000000..a3679949 --- /dev/null +++ b/crates/punktfunk-core/src/session/replay.rs @@ -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); + } +} diff --git a/crates/punktfunk-core/src/session/seal.rs b/crates/punktfunk-core/src/session/seal.rs new file mode 100644 index 00000000..21703a8a --- /dev/null +++ b/crates/punktfunk-core/src/session/seal.rs @@ -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>, + 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, + pub(super) from_worker: std::sync::mpsc::Receiver, +} + +impl SealLane { + pub(super) fn spawn(crypto: std::sync::Arc) -> Option { + let (to_worker, jobs) = std::sync::mpsc::sync_channel::(1); + let (done_tx, from_worker) = std::sync::mpsc::sync_channel::(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], + 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(()) +}