diff --git a/docs/architecture/consumer.md b/docs/architecture/consumer.md index 45dce91..05406d4 100644 --- a/docs/architecture/consumer.md +++ b/docs/architecture/consumer.md @@ -46,8 +46,8 @@ are identical: ```rust // Per local accept (assembly-owned listener): let channel_id = open_reverse_channel(&hub_call, ¶ms, auth_token).await?; -let session = TunnelSession::adopt(hub_manager, channel_id, TUNNEL_ALPN).await?; -let pump_handle = session.pump_against(accepted_halves).await; // or drive the halves directly +let session = TunnelSession::adopt(hub_manager, channel_id, substrate, TUNNEL_ALPN).await?; +let pumped = session.pump_against(accepted_halves).await; // or drive the halves directly ``` - `open_reverse_channel` — call the worker's open op on channel 0 @@ -57,11 +57,13 @@ let pump_handle = session.pump_against(accepted_halves).await; // or drive the h before dialing); the payload `auth_token` is the optional hub-forwarding path (precedence: token > `ServingConfig.identity` > transport). -- `TunnelSession::adopt(manager, channel_id, alpn)` — adopt the - worker-allocated ID (connection-owner rule, ADR-047 §5 — the - SERVING side allocates; the initiator adopts; early-arrival parking - covers the adoption race), install the session's data plane from - the adopted halves. `adopt` takes the same typed-error surface as +- `TunnelSession::adopt(manager, channel_id, substrate, alpn)` — + adopt the worker-allocated ID (connection-owner rule, ADR-047 §5 — + the SERVING side allocates; the initiator adopts; early-arrival + parking covers the adoption race), install the session's data plane + from the adopted halves. The substrate parameter is required — a + datagram session cannot exist without it (the data plane variant is + substrate-shaped). `adopt` takes the same typed-error surface as `open` (an adopt failure is `AdoptFailed`-class, not an establishment error). - `pump_against(accepted_halves)` — spawn `pump_bidi` against the diff --git a/docs/architecture/decisions/003-codec-and-udp-framing.md b/docs/architecture/decisions/003-codec-and-udp-framing.md index a8297b3..fffb2e6 100644 --- a/docs/architecture/decisions/003-codec-and-udp-framing.md +++ b/docs/architecture/decisions/003-codec-and-udp-framing.md @@ -3,7 +3,12 @@ ## Status Accepted (2026-09-07; supersedes the Phase 0 "codec direction" hunch -by making the UDP framing mandatory — reverse POC finding F-2) +by making the UDP framing mandatory — reverse POC finding F-2). +Amended (2026-09-08, review 001 U-1 remediation): the substrate- +boundary placement is pinned mechanically — the `local` feature's +`UdpHalf` composes the codec at the boundary (channel side framed, +target side raw), with the pump-window and recv-scratch disciplines +the composition requires. See §Amendment (2026-09-08). ## Context @@ -77,6 +82,55 @@ The data plane inside a `alk/tunnel` channel is: ## Consequences +### Amendment (2026-09-08 — the boundary mechanics, from review 001 U-1's remediation) + +The placement sentence ("the establisher wraps the dialed UDP socket +in the framed adapter before returning the plan") leaves the adapter's +*mechanics* open; review 001 found `UdpHalf` shipping as a raw +boundary-preserving adapter (the codec on one end only — real targets +saw the tunnel's length prefixes; empty datagrams collapsed to EOF at +the pump's zero-read check; >8 KiB datagrams split at the copy +buffer). The remediation pins the mechanics: + +- **Orientation is channel-relative.** The adapter's caller is the + pump; the CHANNEL side is the framed side. `poll_write` (fed + channel bytes) **de-frames** (an incremental `DatagramReader`) and + sends each decoded payload as one raw datagram; `poll_read` **frames** + (`frame_datagram`) each raw target datagram and hands frame bytes to + the pump. "Frame on write, de-frame on read" describes the TARGET's + face and is the inverted (wrong) reading — the review's original + fix sketch had it inverted and was corrected in re-validation. +- **The write half accepts all offered bytes and reports them + consumed.** The pump's copy engine re-offers a *full* window + unsatisfied until it drains (tokio `copy`'s `read_done` path never + tops the window up), so a partial-frame tail could sit forever in a + full window if the adapter accepted only complete frames — the + >8 KiB frame would deadlock the pump. Completed datagrams queue for + the socket; a trailing partial frame waits in the `DatagramReader` + (its state is bounded by construction — the u16 cap, review 001 + N-2). Backpressure is an accepted-byte high-water mark on + `poll_write`, and the socket drain pumps opportunistically on + write/flush/shutdown (tokio's copy polls flush after every window + drain). +- **The recv scratch is full-size (65535) and lazy.** `try_recv` + truncates SILENTLY at the buffer size — a smaller scratch corrupts + the framing before any size check could reject it (the review's + "size to the MTU" suggestion is unsafe as written). The 64 KiB + eager allocation is instead lazy (first recv), the U-3 mitigation. +- **A >caller-buffer frame is staged, not an error.** The frame's + length prefix reaches the pump on the first read; the remainder + hands out across reads. The pump's far-side de-framer reassembles + the true length — the raw adapter's read-level `InvalidData` is + replaced by staging (the fail-loud posture now protects the *frame + stream*, and a wire-level over-bound datagram remains a loud + `Oversize`-class error at the frame path). + +The v1 verification gates (review 001 §Verification gates) pin these +end-to-end over real sockets in `tests/local_halves.rs` (the +recording-target suite). + +## Consequences (original) + - **TCP/unix tunnels have 0 B tunnel overhead** — the SSH `direct-tcpip` experience (the payload is the bytes). - **UDP correctness requires the codec** — raw pass-through over UDP diff --git a/docs/architecture/wire.md b/docs/architecture/wire.md index 72dfb68..0b9597e 100644 --- a/docs/architecture/wire.md +++ b/docs/architecture/wire.md @@ -111,9 +111,14 @@ the codec: payload, not the theoretical 1472 max) ride the bounded-buffer path; the 64-parked-chunks bound (alkcall early-arrival cap) is upstream sizing, not tunnel policy. -- **Truncation:** a receive buffer smaller than the datagram fails - loudly at the adapter level (OQ-TN-13, resolved fail-loud per - ADR-003); silent truncation would corrupt the framing invariants. +- **Truncation:** silent truncation never happens (OQ-TN-13, resolved + fail-loud per ADR-003): the `local` adapter recvs on a full + 65535-byte scratch (UDP `try_recv` truncates silently at the buffer + size — a smaller scratch would corrupt the framing) and a + >caller-buffer frame is staged across reads, the length prefix + intact, so the far-side de-framer reassembles the true length. A + wire-level over-bound datagram (>65535 — impossible for a + well-formed peer) is a loud error at the frame path. ### Sentinels and lifecycle (both codecs) diff --git a/docs/reviews/001-implementation-review.md b/docs/reviews/001-implementation-review.md index 4462272..f407972 100644 --- a/docs/reviews/001-implementation-review.md +++ b/docs/reviews/001-implementation-review.md @@ -12,6 +12,7 @@ consolidated here per the review-impl deliverable). **U-1 is load-bearing for the UDP wire promise and is NOT yet fixed** — see the remediation section (it owns the remaining remediation task). +**FIXED 2026-09-08** — see §Remediation resolution below. **C-1 (`AcceptQueue` lost wakeup) is fixed with this review's commit**, including its regression test. The trivial minors (N-4, N-5, N-7, N-8, N-9, N-10) are fixed with the same commit — the review's "trivial @@ -263,6 +264,30 @@ the codec path. **Remediation task:** `tasks/tunnels/fix-udp-framed-adapter.md` (created with this review). +**RESOLUTION (2026-09-08, remediation commit):** `UdpHalf` is the +framed adapter — de-frames on `poll_write` (channel bytes → raw +datagrams, via an incremental `DatagramReader`), frames on `poll_read` +(raw datagrams → channel frames, via `frame_datagram`). All five +verification gates landed as permanent tests in `tests/local_halves.rs` +(the recording-target suite: payload-only at the target, empty +round-trip, 9000-byte single datagram, unframed-target resolution, +staging-not-truncation at 65507) plus a v6-dial gate for U-2. Two +mechanics the gates forced beyond the sketch (pinned in ADR-003's +amendment): + +1. The write half ACCEPTS ALL offered bytes (reports them consumed) + and queues completed datagrams for the socket — tokio `copy` + re-offers a FULL window unsatisfied until drained, so accepting + only complete frames deadlocks a >8 KiB frame (its tail sits in a + full window forever). Backpressure is an accepted-byte high-water + mark; the socket drain pumps on write/flush/shutdown (copy polls + flush after every window). +2. A >caller-buffer frame is STAGED across reads (length prefix + first) rather than the raw adapter's read-level `InvalidData` — + gate 4's predicted assertion-surface change; the 65507-byte + max-datagram round-trip through 16 KiB reads is the new fail-loud + probe (silent truncation would still corrupt the framing). + ## U-2 [minor] — `connect_udp` binds `0.0.0.0` unconditionally — no v6 or dual-stack posture `src/local/mod.rs:116`: `UdpSocket::bind(("0.0.0.0", 0))`. A UDP @@ -306,6 +331,14 @@ The rule the remediation implements: Recorded here so the posture survives the implementation; the executable rider lives on `tunnels/fix-udp-framed-adapter.md` §Riders. +**RESOLUTION (2026-09-08, remediation commit):** `connect_udp` +resolves the target first and binds the UNSPECIFIED address of the +matching family (`0.0.0.0:0` for v4 targets, `[::]:0` for v6 — no +implicit wildcard mismatch, no loopback default); no bind-address +parameter added. `bind_tcp` gained the wildcard-exposure doc warning. +Pinned by `connect_udp_binds_the_matching_family_v6_targets_dial` (a +real `::1` echo round-trip through the full path). + ## U-3 [minor] — `UdpHalf` recv scratch allocates 65535 bytes per read half `src/local/mod.rs:164`: `recv_scratch: vec![0u8; 65535]` — a 64 KiB @@ -318,6 +351,13 @@ the codec's MTU discipline (1400) with an oversized-datagram the fail-loud check (a >MTU datagram currently *is* received then error-matched against the caller buffer). Fold into the U-1 task. +**RESOLUTION (2026-09-08, remediation commit):** the scratch is +allocated LAZILY on the first recv (the re-validation's mitigation — +the MTU-shrink idea was unsafe as written; the full-size read is +correctness-load-bearing on the de-frame path). The write half +carries no scratch. Memory posture: one 64 KiB allocation per +*receiving* UDP tunnel, on first recv. + ## C-1 [major] — `AcceptQueue::pop` can miss the final `push` (lost wakeup in the check→register window) — FIXED in-review **Status: FIXED during the review** (commit with this doc). @@ -374,6 +414,11 @@ Filed separately from U-1 because even if U-1's fix is deferred, the docs currently describe behavior that does not exist — a conformance review must not leave that standing. +**RESOLUTION (2026-09-08, remediation commit):** U-1's fix landed — +the docs are now true. The module header, `connect_udp`, and +`UdpHalf` docs describe the composed codec (channel side framed, +target side raw) with the ADR-003 amendment's mechanics. + ## N-2 — NON-FINDING (retracted during the review): `DatagramReader` buffering is bounded by construction Filed as a suspect ("a lying `len` header buffers without bound"), @@ -588,6 +633,13 @@ U-1 remediation docs. # Verification gates for the U-1 remediation +(All five landed 2026-09-08 as permanent tests in +`tests/local_halves.rs` — see §U-1's resolution note. Gates 1–3 are +the `gate_*` recording-target suite; gate 4's truncation-surface +update is `truncation_fails_loud_oq_tn_13` (rewritten for the staged +frame) + `oversized_frame_stages_across_reads`; gate 5 is the standing +battery.) + 1. A real-socket probe test (the review's probe shape, made permanent in `tests/local_halves.rs`): a recording UDP echo server asserts the datagrams it receives are **payload bytes only** (no `[len]` diff --git a/src/local/mod.rs b/src/local/mod.rs index ee1bcd7..a2f40de 100644 --- a/src/local/mod.rs +++ b/src/local/mod.rs @@ -10,8 +10,9 @@ //! [`TcpListenerHalves`] accept half for listen producers. //! - **UDP** — [`connect_udp`]: a connected `UdpSocket` wrapped in the //! FRAMED adapter ([`UdpHalf`], ADR-003 — the codec composes at this -//! boundary; the pump never sees UDP specifics). Truncation fails -//! loud (OQ-TN-13's resolved posture). +//! boundary: the channel side speaks the `[len: u16 BE]` frames, the +//! target side sees payload-only datagrams; the pump never sees UDP +//! specifics). Truncation fails loud (OQ-TN-13's resolved posture). //! - **Unix** — [`dial_unix`], in v1 per OQ-TN-14 (same halves shape //! as TCP). **Stdio is NOT here** — a spawned process's stdio is //! alktty's pipe mode (exit codes + signals are tty semantics); @@ -29,6 +30,7 @@ use std::sync::Arc; use crate::params::Substrate; use crate::producer::{AcceptQueue, DialFn, TargetHandle, TunnelEstablishError}; +use crate::wire::{frame_datagram, DatagramCodecError, DatagramReader, MAX_DATAGRAM_LEN}; use tokio::io::{AsyncRead, AsyncWrite}; /// The dial/connect timeout (the POCs' 5s posture). @@ -54,6 +56,12 @@ pub async fn dial_tcp(target: &str) -> Result Result { let listener = tokio::net::TcpListener::bind(addr) .await @@ -106,14 +114,36 @@ impl TcpListenerHalves { // --- UDP ---------------------------------------------------------------- -/// Connect a UDP "tunnel" to a target: bind an ephemeral socket, -/// `connect(target)` (the associated-socket posture — one peer), and -/// wrap it in the FRAMED adapter (ADR-003): `UdpHalf` enforces one -/// datagram per read/write at the substrate boundary; the `[len: u16 -/// BE]` codec composes at this boundary — the pump never sees UDP +/// Connect a UDP "tunnel" to a target: resolve the target, bind an +/// ephemeral socket on the MATCHING address family (never an implicit +/// wildcard mismatch — a v6-only target dials from a v6-bound socket; +/// review 001 U-2's posture), `connect(target)` (the +/// associated-socket posture — one peer), and wrap it in the FRAMED +/// adapter (ADR-003): `UdpHalf` composes the `[len: u16 BE]` codec at +/// this boundary — the channel side speaks framed bytes, the target +/// side sees payload-only datagrams; the pump never sees UDP /// specifics. pub async fn connect_udp(target: &str) -> Result { - let sock = tokio::net::UdpSocket::bind(("0.0.0.0", 0)) + use std::net::ToSocketAddrs; + // Same posture as tokio's own `UdpSocket::connect(target: &str)` — + // it resolves inline too; numeric targets (the registry's shape) + // make this a non-blocking parse. + let resolved = target + .to_socket_addrs() + .map_err(|e| TunnelEstablishError::DialFailed(format!("udp resolve `{target}`: {e}")))? + .next() + .ok_or_else(|| { + TunnelEstablishError::DialFailed(format!("udp resolve `{target}`: no addresses")) + })?; + let bind_addr = match resolved { + std::net::SocketAddr::V4(_) => { + std::net::SocketAddr::new(std::net::Ipv4Addr::UNSPECIFIED.into(), 0) + } + std::net::SocketAddr::V6(_) => { + std::net::SocketAddr::new(std::net::Ipv6Addr::UNSPECIFIED.into(), 0) + } + }; + let sock = tokio::net::UdpSocket::bind(bind_addr) .await .map_err(|e| TunnelEstablishError::ResourceShortage(format!("udp bind failed: {e}")))?; sock.connect(target) @@ -123,24 +153,55 @@ pub async fn connect_udp(target: &str) -> Result, kind: UdpHalfKind, - /// The recv scratch buffer (the fail-loud truncation check reads - /// whole datagrams before copying into the caller's buffer). - recv_scratch: Vec, + /// Read-half state: the recv scratch (lazily grown to the full + /// 65535 on first use — `try_recv` truncates silently at the + /// buffer size, so the full-size read is correctness-load-bearing) + /// plus the frame-staging state for handing a >caller-buffer + /// frame out across reads. + stage: UdpReadStage, + /// Write-half state: the incremental de-framer fed by successive + /// poll_write buffers, the decoded datagrams queued for the + /// socket, and the accepted-byte backpressure accounting. + deframe: UdpWriteDeFrame, } #[derive(Clone, Copy, PartialEq, Eq, Debug)] @@ -149,6 +210,40 @@ enum UdpHalfKind { Write, } +/// Read-half staging: framed bytes awaiting the pump, chunked to the +/// caller's buffer. +struct UdpReadStage { + /// The frame bytes not yet handed out (a full frame when `None` + /// staged-partial state exists; empty between datagrams). + pending: bytes::Bytes, + /// Lazily allocated recv scratch (full 65535 on first recv). + recv_scratch: Option>, +} + +/// Write-half staging: the incremental de-framer, the decoded +/// datagrams queued for the socket, and the datagram mid-send (blocked +/// on socket send readiness). +struct UdpWriteDeFrame { + reader: DatagramReader, + /// Decoded datagrams awaiting send (FIFO). poll_write accepts all + /// offered bytes into the de-framer — the pump's copy window + /// cannot be held closed for later top-up (a full window is + /// re-offered unsatisfied until drained); backpressure is the + /// accepted-byte high-water mark instead. + queue: std::collections::VecDeque, + /// The datagram mid-send (the socket returned WouldBlock under + /// it — retried first on every poll). + sending: Option, + /// Accepted-byte accounting for the high-water mark (the bytes + /// accepted but not yet sent on the socket). + accepted: usize, +} + +/// The accepted-byte bound before poll_write applies backpressure +/// (Pending). The pump's copy window is 8 KiB; this bounds the queue +/// at ~16 full-size frames' worth of accepted-but-unsent bytes. +const WRITE_QUEUE_HIGH_WATER: usize = 128 * 1024; + impl UdpHalf { pub fn split( sock: tokio::net::UdpSocket, @@ -157,18 +252,35 @@ impl UdpHalf { Box, ) { let sock = Arc::new(sock); - ( - Box::new(UdpHalf { - sock: Arc::clone(&sock), - kind: UdpHalfKind::Read, - recv_scratch: vec![0u8; 65535], - }), - Box::new(UdpHalf { - sock, - kind: UdpHalfKind::Write, - recv_scratch: Vec::new(), - }), - ) + let read_half = UdpHalf { + sock: Arc::clone(&sock), + kind: UdpHalfKind::Read, + stage: UdpReadStage { + pending: bytes::Bytes::new(), + recv_scratch: None, + }, + deframe: UdpWriteDeFrame { + reader: DatagramReader::new(), + queue: std::collections::VecDeque::new(), + sending: None, + accepted: 0, + }, + }; + let write_half = UdpHalf { + sock, + kind: UdpHalfKind::Write, + stage: UdpReadStage { + pending: bytes::Bytes::new(), + recv_scratch: None, + }, + deframe: UdpWriteDeFrame { + reader: DatagramReader::new(), + queue: std::collections::VecDeque::new(), + sending: None, + accepted: 0, + }, + }; + (Box::new(read_half), Box::new(write_half)) } } @@ -178,40 +290,71 @@ impl AsyncRead for UdpHalf { cx: &mut std::task::Context<'_>, buf: &mut tokio::io::ReadBuf<'_>, ) -> std::task::Poll> { + use std::task::Poll; let this = self.get_mut(); debug_assert_eq!(this.kind, UdpHalfKind::Read); - // One recv yields exactly one datagram into the caller's - // buffer. The fail-loud posture (OQ-TN-13, ADR-003): a - // datagram LARGER than the caller's buffer is a framing - // violation — reported as an error, never silently truncated - // (`poll_recv`-into-ReadBuf would truncate; the scratch-buffer - // recv + size check keeps the invariant observable). The - // codec's u16 bound keeps well-formed peers off this path. - let scratch = &mut this.recv_scratch; + + // Hand out staged frame bytes first (a frame larger than the + // caller's buffer was split across reads; the frame's length + // prefix already crossed, so the pump's de-framer on the far + // side reassembles — no shorter-datagram misread is possible). + if !this.stage.pending.is_empty() { + let n = this.stage.pending.len().min(buf.remaining()); + buf.put_slice(&this.stage.pending.split_to(n)); + return Poll::Ready(Ok(())); + } + + // Recv one raw datagram (full-size scratch — try_recv + // truncates silently at the buffer size, so the scratch must + // hold the largest possible datagram before any size check). + if this.stage.recv_scratch.is_none() { + this.stage.recv_scratch = Some(vec![0u8; MAX_DATAGRAM_LEN]); + } + let scratch = this + .stage + .recv_scratch + .as_deref_mut() + .expect("scratch allocated above"); match this.sock.try_recv(scratch) { Ok(n) => { - if n > buf.remaining() { - return std::task::Poll::Ready(Err(std::io::Error::new( + let framed = frame_datagram(&scratch[..n]).map_err(|e| match e { + // A >u16 datagram cannot exist on a well-formed + // wire (the IP theoretical max is 65507); treat it + // as the fail-loud framing violation it is. + DatagramCodecError::Oversize(len, max) => std::io::Error::new( std::io::ErrorKind::InvalidData, format!( - "udp datagram {n} bytes exceeds caller buffer {} (truncation fails loud — OQ-TN-13)", - buf.remaining() + "udp datagram {len} bytes exceeds the codec frame bound {max} \ + (truncation fails loud — OQ-TN-13)" ), - ))); + ), + DatagramCodecError::InvalidLength { .. } => { + unreachable!("frame_datagram never yields InvalidLength") + } + })?; + if framed.len() <= buf.remaining() { + buf.put_slice(&framed); + } else { + // The frame exceeds the caller's buffer: hand out + // the first slice, stage the rest (the length + // prefix rode along — the far-side de-framer + // reassembles across reads). + let n = buf.remaining(); + buf.put_slice(&framed[..n]); + this.stage.pending = framed.slice(n..); } - buf.put_slice(&scratch[..n]); - std::task::Poll::Ready(Ok(())) + Poll::Ready(Ok(())) } Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { // Not ready: poll readiness to register the waker, // then Pending (the next wake retries the recv). match this.sock.poll_recv_ready(cx) { - std::task::Poll::Ready(Ok(())) => std::task::Poll::Pending, - std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)), - std::task::Poll::Pending => std::task::Poll::Pending, + Poll::Ready(Ok(())) => Poll::Pending, + Poll::Ready(Err(e)) => Poll::Ready(Err(e)), + Poll::Pending => Poll::Pending, } } - Err(e) => std::task::Poll::Ready(Err(e)), + Err(e) => Poll::Ready(Err(e)), } } } @@ -222,39 +365,122 @@ impl tokio::io::AsyncWrite for UdpHalf { cx: &mut std::task::Context<'_>, buf: &[u8], ) -> std::task::Poll> { + use std::task::Poll; let this = self.get_mut(); debug_assert_eq!(this.kind, UdpHalfKind::Write); - match this.sock.try_send(buf) { - Ok(n) => std::task::Poll::Ready(Ok(n)), - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { - // Poll send-readiness to register the waker, then - // Pending (the next wake retries the send). - match this.sock.poll_send_ready(cx) { - std::task::Poll::Ready(Ok(())) => std::task::Poll::Pending, - std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)), - std::task::Poll::Pending => std::task::Poll::Pending, + let offered = buf.len(); + + // Backpressure: if too many accepted bytes are still queued + // for the socket, try to drain first; if the queue is still + // over the line, return Pending (the pump's copy re-offers + // its window unsatisfied — safe, nothing is lost). + if this.deframe.accepted > WRITE_QUEUE_HIGH_WATER { + match this.pump_queue_to_socket(cx) { + Poll::Ready(Ok(_)) => {} + Poll::Ready(Err(e)) => return Poll::Ready(Err(e)), + Poll::Pending => { + if this.deframe.accepted > WRITE_QUEUE_HIGH_WATER { + return Poll::Pending; + } } } - Err(e) => std::task::Poll::Ready(Err(e)), } + + // Accept ALL offered bytes into the incremental de-framer and + // report them all consumed. The pump's copy window cannot be + // held closed for a later top-up (a full window is re-offered + // unsatisfied until drained — a partial-frame tail could sit + // forever in a full window), and reporting fewer than all + // bytes risks the copy's `WriteZero` on an empty re-offer. + // The DatagramReader's partial-frame state is bounded by + // construction (the u16 cap — review 001 N-2), so accepting + // everything is bounded memory; socket-side backpressure is + // the accepted-byte high-water mark above. + let dgs = match this.deframe.reader.feed(buf) { + Ok(dgs) => dgs, + Err(e) => { + return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e))) + } + }; + for dg in dgs { + this.deframe.accepted += dg.len() + 2; + this.deframe.queue.push_back(dg); + } + + // Opportunistic send: drain whatever the socket accepts now. + // The waker is registered on WouldBlock (the next + // poll_send_ready wake retries); the queue never blocks + // poll_write's accept path unless the high-water mark is hit. + let _ = this.pump_queue_to_socket(cx); + + Poll::Ready(Ok(offered)) } fn poll_flush( self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, + cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { - std::task::Poll::Ready(Ok(())) + // The send pump: tokio's copy polls flush after draining its + // window, so the queued datagrams drain here whenever the + // pump cycles — and a flush-resolved Ready(Ok(())) tells the + // copy the bytes are on the wire. UDP sends are all-or-nothing + // per datagram; a WouldBlock keeps the datagram queued (the + // send waker is registered) and the queue persists until it + // drains. + self.get_mut().pump_queue_to_socket(cx) } fn poll_shutdown( self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, + cx: &mut std::task::Context<'_>, ) -> std::task::Poll> { - // UDP has no shutdown; drop is the close. + // UDP has no shutdown; drop is the close. Drain the queue + // best-effort on the way out (readiness is already there if + // the datagrams were sendable). + let _ = self.get_mut().pump_queue_to_socket(cx); std::task::Poll::Ready(Ok(())) } } +impl UdpHalf { + /// Send as many queued datagrams as the socket accepts. Returns + /// `Ready(Ok(()))` when the queue is empty; WouldBlock registers + /// the send waker (via `poll_send_ready`) and keeps the remainder + /// queued for the next flush/wake. + fn pump_queue_to_socket( + &mut self, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + use std::task::Poll; + loop { + let dg = match self + .deframe + .sending + .take() + .or_else(|| self.deframe.queue.pop_front()) + { + Some(dg) => dg, + None => return Poll::Ready(Ok(())), // queue drained + }; + match self.sock.try_send(&dg) { + Ok(_) => { + self.deframe.accepted -= dg.len() + 2; + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + self.deframe.sending = Some(dg); + // Register the waker for the retry. + return match self.sock.poll_send_ready(cx) { + Poll::Ready(Ok(())) => Poll::Pending, + Poll::Ready(Err(e)) => Poll::Ready(Err(e)), + Poll::Pending => Poll::Pending, + }; + } + Err(e) => return Poll::Ready(Err(e)), + } + } + } +} + // --- Unix --------------------------------------------------------------- /// Dial a unix-socket target (OQ-TN-14 — ships in v1, the same halves diff --git a/tasks/tunnels/fix-udp-framed-adapter.md b/tasks/tunnels/fix-udp-framed-adapter.md index 7b0a8fd..c562a9e 100644 --- a/tasks/tunnels/fix-udp-framed-adapter.md +++ b/tasks/tunnels/fix-udp-framed-adapter.md @@ -1,13 +1,14 @@ --- id: tunnels/fix-udp-framed-adapter name: "U-1 remediation — make UdpHalf the FRAMED adapter (ADR-003 at the local boundary)" -status: pending +status: completed depends_on: [tunnels/review-impl] scope: moderate risk: medium impact: component level: implementation tags: [local, udp, codec, adr-003, review-remediation] +completed: 2026-09-08 --- ## Description @@ -126,11 +127,45 @@ mitigation must keep the full-size read underneath; see U-3 below. ## Acceptance Criteria -- [ ] `UdpHalf` frames on write and de-frames on read (ADR-003's +- [x] `UdpHalf` frames on write and de-frames on read (ADR-003's placement; the pump stays raw) -- [ ] The 5 verification gates pass; suites green 3× in both configs -- [ ] U-2/U-3 riders resolved or explicitly deferred with rationale -- [ ] Review 001's U-1/U-2/U-3/N-1 sections get resolution notes +- [x] The 5 verification gates pass; suites green 3× in both configs +- [x] U-2/U-3 riders resolved or explicitly deferred with rationale +- [x] Review 001's U-1/U-2/U-3/N-1 sections get resolution notes + +## Resolution (2026-09-08) + +All acceptance criteria met. Key implementation notes beyond the +sketch (pinned in ADR-003's §Amendment): + +- **Orientation** (the caution held): de-frame on write, frame on + read — the channel side is the framed side. +- **The write half accepts ALL offered bytes** and queues completed + datagrams for the socket: tokio's `copy` re-offers a FULL window + unsatisfied until drained (its `read_done` path never tops up), so + the "hold the partial tail, re-feed" shape deadlocked gate 2 (the + 9000-byte test) and `Ready(Ok(0))` on a partial-only feed tripped + the copy's `WriteZero` abrupt-close. The `DatagramReader`'s + partial-frame state is bounded by construction (u16 cap — review + 001 N-2); backpressure is an accepted-byte high-water mark + (128 KiB), and the socket drain pumps on write/flush/shutdown + (copy polls flush after every window drain). +- **A >caller-buffer frame stages across reads** (length prefix + first, remainder on the next read) — gate 4's predicted + assertion-surface change. The fail-loud probe is now the 65507-byte + max datagram reassembling through 16 KiB reads + (`truncation_fails_loud_oq_tn_13`, rewritten) plus the direct + staging test (`oversized_frame_stages_across_reads`). +- **U-2**: `connect_udp` resolves the target and binds the + unspecified address of the MATCHING family; `bind_tcp` gained the + wildcard-exposure doc warning. Pinned by a real `::1` round-trip + (`connect_udp_binds_the_matching_family_v6_targets_dial`). +- **U-3**: the 65535 scratch allocates lazily on first recv (the + full-size read kept — the MTU-shrink idea confirmed unsafe). + +Verification: 63 default / 76 `local` tests green ×3 repeat runs; +clippy `-D warnings` (all-targets + wasm32), fmt, doc, publish +dry-run, wasm check all clean. ## References diff --git a/tests/local_halves.rs b/tests/local_halves.rs index f94f95c..1b6859b 100644 --- a/tests/local_halves.rs +++ b/tests/local_halves.rs @@ -20,13 +20,6 @@ use alktunnels::TunnelSession; use harness::wire_forward; -fn params(resource: &str, substrate: Substrate) -> TunnelParams { - TunnelParams { - resource: resource.to_string(), - substrate, - } -} - /// A TCP echo server on an ephemeral port. async fn tcp_echo_server() -> std::net::SocketAddr { let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) @@ -190,10 +183,17 @@ async fn unix_dial_round_trips() { #[tokio::test] async fn truncation_fails_loud_oq_tn_13() { - // A receive buffer smaller than the datagram surfaces as an error - // (never a silent truncation). Probed at the adapter level: feed a - // >buffer datagram directly into a UdpHalf pair and read with a - // small buffer. + // The fail-loud posture (OQ-TN-13) at the FRAMED adapter: the + // socket-facing recv uses the full 65535-byte scratch (try_recv + // truncates silently at the buffer size), so a well-formed + // >caller-buffer datagram is received whole and STAGED — never + // silently truncated into a shorter frame. The staging surface is + // pinned by `oversized_frame_stages_across_reads` (below); this + // test pins the fail-loud path directly at the unit level: a + // scratch smaller than the datagram WOULD have corrupted the + // framing (the U-3 re-validation's warning), so the scratch must + // be full-size. A datagram at the IP/UDP max (65507) is within + // the u16 frame bound (65535) — it must arrive intact. let sock_a = tokio::net::UdpSocket::bind(("127.0.0.1", 0)) .await .expect("bind a"); @@ -210,24 +210,309 @@ async fn truncation_fails_loud_oq_tn_13() { .expect("connect b"); // The producer-side adapter (what the establisher dials): sock_a - // wrapped as halves. Send a large datagram from the "wire" side. + // wrapped as halves. A 65507-byte datagram (the UDP theoretical + // max — a well-formed wire's largest possible frame) from the + // wire side. let (read, _write) = alktunnels::local::UdpHalf::split(sock_a); let mut read = read; - let payload = vec![0u8; 60000]; // well-formed for UDP but big - sock_b.send(&payload).await.expect("send big"); + let payload: Vec = (0..65507u32).map(|i| (i % 251) as u8).collect(); + sock_b.send(&payload).await.expect("send max-size"); - // The pump reads with the codec's chunk size (16 KiB) — smaller - // than 60000: the adapter must FAIL LOUD, not truncate. + // The pump reads with the codec's chunk size (16 KiB) — far + // smaller than the datagram: the adapter must STAGE, never + // truncate. Drain through small reads and reassemble the frame. use tokio::io::AsyncReadExt; + let mut got = Vec::new(); let mut chunk = vec![0u8; 16 * 1024]; - let result = - tokio::time::timeout(std::time::Duration::from_secs(5), read.read(&mut chunk)).await; - match result { - Ok(Err(e)) => { - assert_eq!(e.kind(), std::io::ErrorKind::InvalidData); - } - other => panic!("expected InvalidData error, got {other:?}"), + while got.len() < payload.len() + 2 { + let n = tokio::time::timeout(std::time::Duration::from_secs(5), read.read(&mut chunk)) + .await + .expect("read timed out (datagram truncated?)") + .expect("read ok"); + assert!(n > 0, "zero-length read before the frame completed"); + got.extend_from_slice(&chunk[..n]); } + assert_eq!(&got[..2], &[0xff, 0xe3], "declared length = 65507"); + assert_eq!( + &got[2..], + &payload[..], + "full payload intact — no truncation" + ); +} + +/// The >caller-buffer (but within-frame-bound) datagram: the framed +/// adapter STAGES the frame and hands it out across reads — the pump's +/// far-side de-framer reassembles the true length (no silent +/// truncation, no shorter-datagram misread). The reassembly is pinned +/// by the end-to-end gates below; here the staging surface is pinned +/// directly: successive reads yield the full frame. +#[tokio::test] +async fn oversized_frame_stages_across_reads() { + let sock_a = tokio::net::UdpSocket::bind(("127.0.0.1", 0)) + .await + .expect("bind a"); + let sock_b = tokio::net::UdpSocket::bind(("127.0.0.1", 0)) + .await + .expect("bind b"); + sock_a + .connect(sock_b.local_addr().expect("b addr")) + .await + .expect("connect a"); + sock_b + .connect(sock_a.local_addr().expect("a addr")) + .await + .expect("connect b"); + + let (read, _write) = alktunnels::local::UdpHalf::split(sock_a); + let mut read = read; + let payload: Vec = (0..20000u32).map(|i| (i % 251) as u8).collect(); + sock_b.send(&payload).await.expect("send 20k"); + + // Read through a small caller buffer (16 KiB — smaller than the + // ~20002-byte frame): the first read fills it completely, the + // second hands out the staged remainder. + use tokio::io::AsyncReadExt; + let mut got = Vec::new(); + let mut chunk = vec![0u8; 16 * 1024]; + let n1 = tokio::time::timeout(std::time::Duration::from_secs(5), read.read(&mut chunk)) + .await + .expect("first read") + .expect("read ok"); + got.extend_from_slice(&chunk[..n1]); + let n2 = tokio::time::timeout(std::time::Duration::from_secs(5), read.read(&mut chunk)) + .await + .expect("second read") + .expect("read ok"); + got.extend_from_slice(&chunk[..n2]); + + assert_eq!(got.len(), payload.len() + 2, "frame length read out"); + assert_eq!(&got[..2], &[len_hi(payload.len()), len_lo(payload.len())]); + assert_eq!(&got[2..], &payload[..], "payload bytes intact"); +} + +fn len_hi(n: usize) -> u8 { + (n >> 8) as u8 +} +fn len_lo(n: usize) -> u8 { + (n & 0xff) as u8 +} + +/// A RECORDING UDP "target" (review 001's verification-gate probe +/// shape, made permanent): it records every datagram it receives +/// (byte-for-byte, to the shared log) and replies with canned, +/// configurable datagrams — including EMPTY and RAW-UNFRAMED ones the +/// echo server can't produce. +struct RecordingTarget { + addr: std::net::SocketAddr, + received: Arc>>>, +} + +async fn recording_udp_target(replies: Vec>) -> RecordingTarget { + let sock = tokio::net::UdpSocket::bind(("127.0.0.1", 0)) + .await + .expect("bind recording target"); + let addr = sock.local_addr().expect("addr"); + let received = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let log = Arc::clone(&received); + tokio::spawn(async move { + let mut buf = vec![0u8; 65535]; + // Reply cyclically (raw, as a real target would) to each + // datagram that arrives, then keep recording. + let mut next = 0usize; + loop { + let Ok((n, peer)) = sock.recv_from(&mut buf).await else { + return; + }; + log.lock().await.push(buf[..n].to_vec()); + if !replies.is_empty() { + let reply = &replies[next % replies.len()]; + next += 1; + if sock.send_to(reply, peer).await.is_err() { + return; + } + } + } + }); + RecordingTarget { addr, received } +} + +fn params(resource: &str, substrate: Substrate) -> TunnelParams { + TunnelParams { + resource: resource.to_string(), + substrate, + } +} + +/// GATE 1a (review 001): the target receives PAYLOAD BYTES ONLY — no +/// `[len: u16 BE]` codec prefix (the U-1 defect's primary face). +#[tokio::test] +async fn gate_target_receives_payload_bytes_only() { + let target = recording_udp_target(vec![]).await; + let registry = ResourceRegistry::new(); + registry + .register("rec", Substrate::Udp, &target.addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("rec", Substrate::Udp)) + .await + .expect("udp open"); + session.send_datagram(b"query").await.expect("send"); + // Give the datagram time to cross the pump. + for _ in 0..50 { + if !target.received.lock().await.is_empty() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + let received = target.received.lock().await; + assert_eq!(received.len(), 1, "exactly one datagram at the target"); + assert_eq!( + &received[0][..], + b"query", + "the target sees the payload, not the codec frame (no 00 05 prefix)" + ); + session.close().await; +} + +/// GATE 1b: an EMPTY datagram round-trips over the REAL socket path — +/// `recv_datagram` yields `Some(b"")` and the channel stays alive +/// (the F-2 fix end-to-end; the raw adapter collapsed it to EOF). +#[tokio::test] +async fn gate_empty_datagram_round_trips_over_the_real_socket_path() { + // The target replies cyclically: empty datagram first (the + // round-trip under test), then `still-alive` (the liveness probe). + let target = recording_udp_target(vec![vec![], b"still-alive".to_vec()]).await; + let registry = ResourceRegistry::new(); + registry + .register("empty", Substrate::Udp, &target.addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("empty", Substrate::Udp)) + .await + .expect("udp open"); + session.send_datagram(b"").await.expect("send empty"); + let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv timed out (empty datagram collapsed to EOF?)") + .expect("io ok") + .expect("empty datagram is Some, NOT None (EOF)"); + assert!(dg.is_empty(), "the empty datagram round-trips"); + + // The channel is still alive: a second exchange works (the + // canned reply cycles — the target sends b"still-alive" back). + session.send_datagram(b"second").await.expect("send 2"); + let dg2 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv 2 timed out") + .expect("io ok 2") + .expect("datagram 2"); + assert_eq!(&dg2[..], b"still-alive"); + session.close().await; +} + +/// GATE 2: a >8 KiB datagram round-trips as ONE datagram at the +/// target (the 8 KiB copy-buffer split is gone — the frame reassembles +/// at the adapter before crossing the pump). +#[tokio::test] +async fn gate_large_datagram_arrives_as_one_datagram() { + let target = recording_udp_target(vec![]).await; + let registry = ResourceRegistry::new(); + registry + .register("big", Substrate::Udp, &target.addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("big", Substrate::Udp)) + .await + .expect("udp open"); + // 9000 bytes — the review's probe size that split as 8192 + 810. + let payload: Vec = (0..9000u32).map(|i| (i % 251) as u8).collect(); + session.send_datagram(&payload).await.expect("send 9000"); + for _ in 0..100 { + if !target.received.lock().await.is_empty() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + let received = target.received.lock().await; + assert_eq!(received.len(), 1, "ONE datagram at the target, not two"); + assert_eq!( + received[0].len(), + 9000, + "full payload in one datagram (no 8192 split)" + ); + assert_eq!(&received[0][..], &payload[..], "payload intact"); + session.close().await; +} + +/// GATE 3: an unframed target-initiated datagram resolves at the +/// session (the session codec decodes the framed bytes the adapter +/// produced — real-target traffic is not parked mid-frame). +#[tokio::test] +async fn gate_unframed_target_datagram_resolves_at_the_session() { + // A raw "HELLO" — a real server's greeting the raw adapter parked + // mid-frame (the probe's 2s timeout). + let target = recording_udp_target(vec![b"HELLO".to_vec()]).await; + let registry = ResourceRegistry::new(); + registry + .register("greet", Substrate::Udp, &target.addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("greet", Substrate::Udp)) + .await + .expect("udp open"); + session.send_datagram(b"ping").await.expect("send"); + let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv timed out — unframed target traffic was parked/dropped") + .expect("io ok") + .expect("datagram"); + assert_eq!(&dg[..], b"HELLO"); + session.close().await; +} + +/// U-2 (review 001): `connect_udp`'s bind is family-correct — a v6 +/// target dials from a v6-bound socket (the hardcoded `0.0.0.0` bind +/// broke every v6-only target at connect). A real ::1 target +/// round-trips through the full path. +#[tokio::test] +async fn connect_udp_binds_the_matching_family_v6_targets_dial() { + let sock = tokio::net::UdpSocket::bind(("::1", 0)) + .await + .expect("bind v6 echo"); + let addr = sock.local_addr().expect("v6 addr"); + tokio::spawn(async move { + let mut buf = vec![0u8; 65535]; + loop { + let Ok((n, peer)) = sock.recv_from(&mut buf).await else { + return; + }; + if sock.send_to(&buf[..n], peer).await.is_err() { + return; + } + } + }); + + let registry = ResourceRegistry::new(); + registry + .register("v6", Substrate::Udp, &addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("v6", Substrate::Udp)) + .await + .expect("v6 udp open (family-correct bind)"); + session.send_datagram(b"v6-query").await.expect("send v6"); + let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv timed out") + .expect("io ok") + .expect("datagram"); + assert_eq!(&dg[..], b"v6-query"); + session.close().await; } #[tokio::test]