feat: wire-codec — UDP datagram framing (frame_datagram/DatagramReader)
- [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<Bytes> (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
This commit is contained in:
+227
-3
@@ -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<Bytes, DatagramCodecError> {
|
||||
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<Vec<Bytes>, 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<Bytes> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Bytes>` per the task API (the POC's
|
||||
`Vec<Datagram>` 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<Bytes>` signature); `is_empty()` moves to callers
|
||||
(trivial on `Bytes`).
|
||||
|
||||
## Summary
|
||||
|
||||
> Agent fills this on completion.
|
||||
`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.
|
||||
Reference in New Issue
Block a user