From 60c625899a5d8c4304f8b48f756fa14aff65b568 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Tue, 8 Sep 2026 07:16:18 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20wire-codec=20=E2=80=94=20UDP=20datagram?= =?UTF-8?q?=20framing=20(frame=5Fdatagram/DatagramReader)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - [len: u16 BE] per-datagram framing per ADR-003; len=0 is a legal empty datagram (F-2 invariant), never EOF - DatagramCodecError: Oversize (frame-time, never a wire overflow) + InvalidLength (truncated stream — the OQ-TN-13 fail-loud codec-side anchor) - DatagramReader: incremental decoder — datagrams split across chunks, batched in one chunk, partial headers; feed -> Vec (zero-copy freeze handoff) - is_mid_datagram teardown diagnostic - 8 test families pinning the ADR-003/BAST invariants incl. 7-byte chunk splits and truncated-stream-never-yields-partial Verified: cargo test (14 passed), clippy --all-targets -D warnings (native + wasm32), fmt --check, wasm32 check — all clean --- src/wire.rs | 230 +++++++++++++++++++++++++++++++++++- tasks/tunnels/wire-codec.md | 44 +++++-- 2 files changed, 263 insertions(+), 11 deletions(-) diff --git a/src/wire.rs b/src/wire.rs index ab9e504..16880c4 100644 --- a/src/wire.rs +++ b/src/wire.rs @@ -3,9 +3,233 @@ //! framing for `udp`. Normative WHAT in `wire.md`; the binary framing //! carries its BAST contract at `docs/architecture/bast.md`. //! -//! Skeleton module — `frame_datagram`, `DatagramReader`, and -//! `DatagramCodecError` land with `tunnels/wire-codec`. +//! The channels layer already length-prefixes every chunk (8-byte +//! header), so stream substrates need zero tunnel-level framing — the +//! halves are the tunnel, and a zero-length read is genuinely EOF +//! (stream semantics). UDP needs boundary re-framing only: datagram +//! boundaries do not survive the chunk stream, and `u16` suffices +//! (UDP's max payload 65507 < 65535). +//! +//! Sentinel layering (the F-2 invariant, load-bearing): `len == 0` is +//! a legal *empty datagram* — a real datagram, never an EOF signal. +//! EOF is the channels-level sentinel on the `BiStream` and lives at +//! a different layer; the codec never emits zero-length reads, so the +//! layers never collide. This is why raw pass-through is structurally +//! broken for UDP: an empty datagram and a zero-byte read are the same +//! wire shape at the pump level. +//! +//! No chunk-type byte, no 5-byte header: a tunnel has exactly one data +//! stream per direction (the channel's read/write halves), so there is +//! no sub-demux key — the structural difference from alktty's 5-byte +//! `[stream_type][len]` header. + +use bytes::{Buf, BufMut, Bytes, BytesMut}; +use thiserror::Error; /// The u16 length-field bound: datagrams larger than this are /// rejected at frame time (ADR-003 — never a wire overflow). -pub const MAX_DATAGRAM_LEN: usize = 65535; +pub const MAX_DATAGRAM_LEN: usize = u16::MAX as usize; + +const DATAGRAM_LEN_FIELD: usize = 2; + +/// Codec errors: oversize at frame time, and the truncated-stream +/// error a mid-datagram stream end maps to (the fail-loud posture, +/// OQ-TN-13 — never a silent partial datagram). +#[derive(Debug, Error)] +pub enum DatagramCodecError { + /// A datagram larger than [`MAX_DATAGRAM_LEN`] was framed — the + /// u16 length field would wrap. Rejected at frame time. + #[error("datagram too large for u16 length prefix: {0} bytes (max {1})")] + Oversize(usize, usize), + /// The stream ended mid-datagram (declared length not satisfied). + /// A clean peer never ends the stream mid-datagram; this is a + /// truncated-stream error, not a partial datagram. + #[error("unexpected end of datagram buffer: declared {declared}, have {have}")] + InvalidLength { declared: usize, have: usize }, +} + +/// Frame one datagram into the chunk stream: `[len: u16 BE][payload]`. +/// `len = 0` is a legal empty datagram; payloads larger than +/// [`MAX_DATAGRAM_LEN`] are rejected at frame time. +pub fn frame_datagram(payload: &[u8]) -> Result { + if payload.len() > MAX_DATAGRAM_LEN { + return Err(DatagramCodecError::Oversize( + payload.len(), + MAX_DATAGRAM_LEN, + )); + } + let mut buf = BytesMut::with_capacity(DATAGRAM_LEN_FIELD + payload.len()); + buf.put_u16(payload.len() as u16); + buf.put_slice(payload); + Ok(buf.freeze()) +} + +/// Incremental datagram decoder over the channel's read half. Feed it +/// whatever `read()` yields; it reassembles `[len: u16 BE][payload]` +/// frames across chunk boundaries — a datagram may be split across +/// two chunks, two datagrams may share one chunk, and partial headers +/// wait for more bytes. This is the only consumer-visible decode path +/// (ADR-003; chunk splitting/batching is transparent). +#[derive(Debug)] +pub struct DatagramReader { + buf: BytesMut, + current: Option<(usize, BytesMut)>, +} + +impl Default for DatagramReader { + fn default() -> Self { + Self::new() + } +} + +impl DatagramReader { + pub fn new() -> Self { + Self { + buf: BytesMut::new(), + current: None, + } + } + + /// Feed raw bytes from the read half. Returns every datagram that + /// completed with this feed, in order — zero or more per chunk. + /// Buffers partial frames across chunk boundaries. + pub fn feed(&mut self, chunk: &[u8]) -> Result, DatagramCodecError> { + self.buf.extend_from_slice(chunk); + let mut out = Vec::new(); + loop { + if self.current.is_none() { + if self.buf.len() < DATAGRAM_LEN_FIELD { + break; + } + let declared = u16::from_be_bytes([self.buf[0], self.buf[1]]) as usize; + self.buf.advance(DATAGRAM_LEN_FIELD); + self.current = Some((declared, BytesMut::new())); + } + let Some((declared, acc)) = self.current.as_mut() else { + unreachable!("current set above when not none") + }; + if *declared == 0 { + self.current = None; + out.push(Bytes::new()); + continue; + } + let take = (*declared - acc.len()).min(self.buf.len()); + if take > 0 { + acc.extend_from_slice(&self.buf[..take]); + self.buf.advance(take); + } + if acc.len() == *declared { + let payload = std::mem::take(acc).freeze(); + self.current = None; + out.push(payload); + } else { + break; + } + } + Ok(out) + } + + /// `true` while a partial datagram is in flight (for teardown + /// diagnostics — a clean peer never ends the stream mid-datagram; + /// an end while this is set is [`DatagramCodecError::InvalidLength`]'s + /// condition). + pub fn is_mid_datagram(&self) -> bool { + self.current.is_some() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn feed_one(reader: &mut DatagramReader, chunk: &[u8]) -> Option { + reader.feed(chunk).unwrap().into_iter().next() + } + + #[test] + fn single_datagram_round_trips() { + let framed = frame_datagram(b"hello").unwrap(); + assert_eq!(&framed[..2], &[0, 5]); + let mut r = DatagramReader::new(); + let dgs = r.feed(&framed).unwrap(); + assert_eq!(dgs.len(), 1); + assert_eq!(&dgs[0][..], b"hello"); + } + + #[test] + fn empty_datagram_survives_and_is_not_eof() { + let framed = frame_datagram(b"").unwrap(); + assert_eq!(framed.len(), 2); + let mut r = DatagramReader::new(); + let dgs = r.feed(&framed).unwrap(); + assert_eq!(dgs.len(), 1); + assert!(dgs[0].is_empty()); + } + + #[test] + fn datagram_split_across_awkward_chunks_reassembles() { + let framed = frame_datagram(b"0123456789").unwrap(); + let mut r = DatagramReader::new(); + for piece in framed.chunks(7) { + let dgs = r.feed(piece).unwrap(); + if !dgs.is_empty() { + assert_eq!(&dgs[0][..], b"0123456789"); + return; + } + } + panic!("datagram never reassembled across 7-byte chunks"); + } + + #[test] + fn two_datagrams_batched_in_one_chunk() { + let mut buf = BytesMut::new(); + buf.extend_from_slice(&frame_datagram(b"abc").unwrap()); + buf.extend_from_slice(&frame_datagram(b"de").unwrap()); + let mut r = DatagramReader::new(); + let dgs = r.feed(&buf.freeze()).unwrap(); + assert_eq!(dgs.len(), 2); + assert_eq!(&dgs[0][..], b"abc"); + assert_eq!(&dgs[1][..], b"de"); + } + + #[test] + fn partial_header_at_chunk_boundary_waits() { + let framed = frame_datagram(b"xyz").unwrap(); + let mut r = DatagramReader::new(); + assert!(feed_one(&mut r, &framed[..1]).is_none()); + let dg = feed_one(&mut r, &framed[1..]).expect("datagram after header completes"); + assert_eq!(&dg[..], b"xyz"); + } + + #[test] + fn mid_datagram_state_is_observable() { + let framed = frame_datagram(b"0123").unwrap(); + let mut r = DatagramReader::new(); + r.feed(&framed[..3]).unwrap(); + assert!(r.is_mid_datagram()); + r.feed(&framed[3..]).unwrap(); + assert!(!r.is_mid_datagram()); + } + + #[test] + fn oversize_rejected_at_frame_time() { + let err = frame_datagram(&vec![0u8; MAX_DATAGRAM_LEN + 1]).unwrap_err(); + assert!(matches!(err, DatagramCodecError::Oversize(_, _))); + } + + #[test] + fn truncated_stream_never_yields_partial_datagram() { + let framed = frame_datagram(b"payload").unwrap(); + let mut r = DatagramReader::new(); + r.feed(&framed[..3]).unwrap(); + assert!(r.is_mid_datagram()); + assert!(feed_one(&mut r, &[]).is_none()); + assert!(r.is_mid_datagram()); + assert_eq!( + r.feed(&framed[3..framed.len() - 1]).unwrap().len(), + 0, + "one byte short must stay buffered, not yield a partial datagram" + ); + assert!(r.is_mid_datagram()); + } +} diff --git a/tasks/tunnels/wire-codec.md b/tasks/tunnels/wire-codec.md index 74ad62e..76a7b9c 100644 --- a/tasks/tunnels/wire-codec.md +++ b/tasks/tunnels/wire-codec.md @@ -1,7 +1,7 @@ --- id: tunnels/wire-codec name: Data-plane codec (frame_datagram / DatagramReader) + sentinel layering tests -status: pending +status: completed depends_on: [tunnels/crate-init, tunnels/params] scope: narrow risk: medium @@ -65,13 +65,13 @@ the POC proved no collision). ## Acceptance Criteria -- [ ] All 8 test families above pass (the POC's 7 + the truncation pin) -- [ ] `frame_datagram` returns `Bytes` (zero-copy handoff to the mux) -- [ ] `DatagramReader` state is incremental across arbitrary chunk +- [x] All 8 test families above pass (the POC's 7 + the truncation pin) +- [x] `frame_datagram` returns `Bytes` (zero-copy handoff to the mux) +- [x] `DatagramReader` state is incremental across arbitrary chunk boundaries -- [ ] `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check` +- [x] `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check` clean -- [ ] wasm32 check passes (the codec is pure byte work — it must be +- [x] wasm32 check passes (the codec is pure byte work — it must be wasm-clean) ## References @@ -84,8 +84,36 @@ the POC proved no collision). ## Notes -> Agent fills during implementation. +- Direct port of the POC's wire.rs (17 tests rode it) with the crate + conventions: thiserror doc comments, module doc per ADR-003 + + wire.md + bast.md layering (sentinel layering called out — the F-2 + invariant is load-bearing, so the module doc states why raw + pass-through is structurally broken for UDP). +- Error naming per the task's API: `Oversize(usize, usize)` (was + POC's `TooLarge`) + `InvalidLength {declared, have}` (the + truncated-stream shape — the OQ-TN-13 fail-loud posture's codec-side + anchor). +- `feed` returns `Vec` per the task API (the POC's + `Vec` wrapper dropped — `Bytes` is the payload; the mux + handoff is zero-copy via `BytesMut::freeze`). +- `MAX_DATAGRAM_LEN`/`DATAGRAM_LEN_FIELD` constants; `DATAGRAM_LEN_FIELD` + is private (implementation detail), `MAX_DATAGRAM_LEN` public (the + frame-time bound consumers need). +- 8 test families: single round-trip, empty-datagram-not-EOF, + 7-byte-chunk split reassembly, two-datagram batch, partial header, + mid-datagram observability, oversize at frame time, truncated + stream (feeds an empty chunk + a one-byte-short tail — never yields + a partial datagram; the codec-side half of OQ-TN-13; the + adapter-level receive shape lands with local-socket-halves). +- POC's `Datagram` struct dropped: `Bytes` suffices (the task's + `feed -> Vec` signature); `is_empty()` moves to callers + (trivial on `Bytes`). ## Summary -> Agent fills this on completion. \ No newline at end of file +`src/wire.rs` complete: `frame_datagram` (Oversize at frame time, +`len=0` legal), `DatagramReader` (incremental decoder, `is_mid_datagram` +teardown diagnostic), `DatagramCodecError {Oversize, InvalidLength}` — +the ADR-003/BAST contract in executable form. 8 codec tests + 6 params +tests = 14 passing. Verified: cargo test, clippy --all-targets -D +warnings (native + wasm32), fmt --check, wasm32 check — all clean. \ No newline at end of file