alkdev/alktunnels
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: UdpHalf is the FRAMED adapter (U-1) — U-2 bind posture, U-3 lazy scratch

Browse Source 
Review 001 remediation (tasks/tunnels/fix-udp-framed-adapter):

- U-1 [major]: UdpHalf now composes the ADR-003 codec at the
  substrate boundary — poll_write de-frames channel bytes (incremental
  DatagramReader) and sends each payload as one raw datagram;
  poll_read frames raw target datagrams and hands frame bytes to the
  pump. Targets see payload-only bytes; empty datagrams round-trip
  (the F-2 fix end-to-end); >8 KiB datagrams arrive as ONE datagram.
  Mechanics the verification gates forced beyond the sketch:
  poll_write accepts all offered bytes (tokio copy re-offers a full
  window unsatisfied — partial-frame holds deadlock >8 KiB frames),
  backpressure is an accepted-byte high-water mark, socket drain
  pumps on write/flush/shutdown; >caller-buffer frames stage across
  reads (the truncation probe's assertion surface updated per the
  re-validation correction).
- U-2 [minor]: connect_udp resolves the target and binds the
  unspecified address of the matching family (v6 targets dial from
  [::]:0); bind_tcp gains the wildcard-exposure doc warning.
- U-3 [minor]: the 65535 recv scratch allocates lazily on first recv
  (full-size read kept — try_recv truncates silently; the MTU-shrink
  idea confirmed unsafe in re-validation).
- N-1: the module/connect_udp/UdpHalf docs now describe the composed
  codec truthfully.

Gates (permanent tests in tests/local_halves.rs): recording-target
suite (payload-only at the target, empty-datagram round-trip, 9000-
byte single datagram, unframed-target resolution), the 65507-byte
max-datagram staging probe, and the v6 dial gate.

Docs: ADR-003 §Amendment (the boundary mechanics), wire.md truncation
bullet, consumer.md adopt sketch (substrate param). Review 001
resolution notes on U-1/U-2/U-3/N-1; task marked completed.

Verified: 63 default / 76 local tests ×3 repeat runs; clippy
-D warnings (all-targets + wasm32), fmt --check, doc --no-deps
(warning-free), wasm check, publish dry-run.
This commit is contained in:
glm-5.3-flash committed 2026-09-08 21:22:52 +00:00
1 parent aa7b5b7169
commit 8cf4cc8bb5
7 files changed
+765 -106
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/local/mod.rs
+293 -67
View File
@@ -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<TargetHandle, TunnelEstablishError
/// Bind a TCP listener (the listen shape's assembly-side half).
/// The binding decision belongs to the caller (OQ-TN-04) — this is a
/// `local`-feature convenience, not protocol behavior.
///
/// **Bind posture (review 001 U-2):** the caller-supplied address is
/// deliberate — a wildcard bind (`0.0.0.0` or `::`; note `::1` is
/// loopback, not a wildcard) is the caller's explicit choice and
/// exposes the listener per the host's firewall. There is no implicit
/// wildcard default here.
pub async fn bind_tcp(addr: &str) -> Result<TcpListenerHalves, TunnelEstablishError> {
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<TargetHandle, TunnelEstablishError> {
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<TargetHandle, TunnelEstablishEr
Ok(TargetHandle { read, write })
}
/// A UDP socket presented as `AsyncRead`/`AsyncWrite` halves: reads
/// yield whole datagrams (one `poll_recv` per read); writes send one
/// datagram per `poll_write` buffer. The FRAMED adapter — the codec
/// wraps at this boundary (ADR-003); the pump never sees UDP
/// specifics.
/// The FRAMED UDP adapter (ADR-003's substrate-boundary codec): the
/// pump-facing face speaks the tunnel's `[len: u16 BE][payload]`
/// channel framing; the socket face speaks whole raw datagrams.
/// Orientation (review 001 §Re-validation — the fix sketch's caution):
/// `UdpHalf`'s caller is the pump, and the CHANNEL side is the framed
/// side —
///
/// **Truncation fails loud** (OQ-TN-13's resolved posture, ADR-003):
/// `poll_recv` into a too-small caller buffer returns
/// `io::ErrorKind::InvalidData` (the datagram is dropped with the
/// error — silent truncation would corrupt the framing invariants).
/// The codec's u16 bound keeps well-formed peers from ever hitting
/// this path; only a mis-framed wire does.
/// - `poll_write` is fed CHANNEL bytes (already-framed frames, from
/// the session's codec) → it **de-frames** (an incremental
/// [`DatagramReader`]) and sends each decoded payload to the target
/// as one raw datagram. A frame may straddle successive poll_write
/// buffers (the pump's 8 KiB copy chunks split large frames); the
/// decoder carries the partial state until the frame completes.
/// `len=0` decodes to a zero-payload datagram — a real empty
/// datagram, NOT EOF (the F-2 fix end-to-end).
/// - `poll_read` yields bytes destined for the CHANNEL (the session's
/// codec decodes there) → it **frames** ([`frame_datagram`]) each
/// raw target datagram and hands the frame bytes to the pump. A
/// frame larger than the caller's buffer is staged and handed out
/// across successive reads — a partial frame is never misread as a
/// shorter one.
///
/// The target sees payload-only bytes; an empty datagram crosses the
/// pump as its 2-byte frame (unambiguous with EOF).
///
/// **Truncation never happens silently** (OQ-TN-13's fail-loud
/// posture, ADR-003): the socket-facing recv uses a full 65535-byte
/// scratch (`try_recv` truncates at the buffer size — a smaller
/// scratch would corrupt the framing before any size check runs), and
/// a datagram beyond [`MAX_DATAGRAM_LEN`] (impossible for a
/// well-formed peer — the u16 frame bound) errors loudly instead of
/// truncating.
///
/// The scratch is allocated lazily on the first recv (review 001 U-3:
/// a 64 KiB eager allocation per tunnel is needless for the write
/// half and for tunnels that never recv).
pub struct UdpHalf {
sock: Arc<tokio::net::UdpSocket>,
kind: UdpHalfKind,
/// The recv scratch buffer (the fail-loud truncation check reads
/// whole datagrams before copying into the caller's buffer).
recv_scratch: Vec<u8>,
/// 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<Vec<u8>>,
}
/// 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<bytes::Bytes>,
/// The datagram mid-send (the socket returned WouldBlock under
/// it — retried first on every poll).
sending: Option<bytes::Bytes>,
/// 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<dyn AsyncWrite + Send + Sync + Unpin>,
) {
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<std::io::Result<()>> {
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<std::io::Result<usize>> {
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::io::Result<()>> {
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<std::io::Result<()>> {
// 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<std::io::Result<()>> {
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