From 65fb69db6164de25b0c9ed53086bc9d258642ff4 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Tue, 8 Sep 2026 09:44:23 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20local-socket-halves=20=E2=80=94=20the?= =?UTF-8?q?=20local=20feature=20(TCP/UDP/unix=20halves=20functions)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/local/mod.rs (feature = "local" -> tokio/net): dial_tcp (5s timeout), connect_udp (ephemeral bind + connect + the FRAMED UdpHalf adapter), dial_unix (OQ-TN-14 in v1), bind_tcp + TcpListenerHalves::accept_loop (the listen shape's assembly half), local_dial() (one DialFn covering all three substrates). - UdpHalf truncation fails loud (OQ-TN-13): poll_read recvs into a scratch buffer first and size-checks against the caller's buffer — an over-size datagram is io::ErrorKind::InvalidData, never a silent truncation (tokio's poll_recv-into-ReadBuf would truncate). - No socket type appears outside src/local/ (convention 16); the default crate stays wasm-clean; stdio bridging is NOT here (alktty owns process stdio). - Tests (tests/local_halves.rs, 7, feature-gated): TCP/UDP/unix dial round-trips through the real establisher path, the 1400-byte MTU datagram, the truncation fail-loud probe, dial-refusal -> dial_failed, listen producer over a real TCP listener end-to-end. Verified: cargo test green (default + --all-features, 44 + 7), clippy -D warnings (default + all-features + wasm32), fmt clean, cargo check --target wasm32-unknown-unknown passes. --- Cargo.toml | 1 + src/lib.rs | 3 + src/local/mod.rs | 290 +++++++++++++++++++++++ tasks/tunnels/local-socket-halves.md | 49 +++- tests/local_halves.rs | 332 +++++++++++++++++++++++++++ 5 files changed, 673 insertions(+), 2 deletions(-) create mode 100644 src/local/mod.rs create mode 100644 tests/local_halves.rs diff --git a/Cargo.toml b/Cargo.toml index 29a23a1..c181978 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ name = "alktunnels" [features] default = [] +local = ["tokio/net"] [dependencies] alkcall = "0.7.0" diff --git a/src/lib.rs b/src/lib.rs index 418d086..bd57abe 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,6 +24,9 @@ pub mod params; pub mod producer; pub mod wire; +#[cfg(feature = "local")] +pub mod local; + pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession}; pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError}; pub use params::ChannelOpenError; diff --git a/src/local/mod.rs b/src/local/mod.rs new file mode 100644 index 0000000..ee1bcd7 --- /dev/null +++ b/src/local/mod.rs @@ -0,0 +1,290 @@ +//! The `local` feature module: real-socket halves functions the +//! assembly layer injects as `DialFn`/`AcceptFn` closures (ADR-004 — +//! "produce boxed halves for a resource" is a function, not a trait). +//! Non-wasm by design: this module IS the platform boundary — the +//! default crate stays wasm-clean (convention 16: `local`-gated code +//! is never imported from the shared/producer/consumer modules). +//! +//! Substrate coverage (overview.md §Feature Gates): +//! - **TCP** — [`dial_tcp`] (5s connect timeout per the POCs) and the +//! [`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). +//! - **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); +//! remote command execution composes via alktty, not here. +//! +//! Error mapping: [`TunnelEstablishError`] variants (the POC's +//! `Into` path) — dial refusal → `dial_failed`, +//! bind/resource exhaustion → `resource_shortage`, handler-internal +//! → `handler_error`. +//! +//! `TargetHandle` re-used from producer.rs — the halves type lives +//! there; `local/` depends on it, never vice versa (convention 16). + +use std::sync::Arc; + +use crate::params::Substrate; +use crate::producer::{AcceptQueue, DialFn, TargetHandle, TunnelEstablishError}; +use tokio::io::{AsyncRead, AsyncWrite}; + +/// The dial/connect timeout (the POCs' 5s posture). +const DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +// --- TCP ---------------------------------------------------------------- + +/// Dial a TCP target (`host:port`): `TcpStream::connect` bounded by +/// [`DIAL_TIMEOUT`], `into_split` → boxed halves (raw pass-through — +/// the halves ARE the tunnel, ADR-003). +pub async fn dial_tcp(target: &str) -> Result { + let stream = tokio::time::timeout(DIAL_TIMEOUT, tokio::net::TcpStream::connect(target)) + .await + .map_err(|_| TunnelEstablishError::DialFailed(format!("dial `{target}` timed out")))? + .map_err(|e| TunnelEstablishError::DialFailed(format!("dial `{target}`: {e}")))?; + let (read, write) = stream.into_split(); + Ok(TargetHandle { + read: Box::new(read), + write: Box::new(write), + }) +} + +/// 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. +pub async fn bind_tcp(addr: &str) -> Result { + let listener = tokio::net::TcpListener::bind(addr) + .await + .map_err(|e| TunnelEstablishError::ResourceShortage(format!("bind `{addr}`: {e}")))?; + Ok(TcpListenerHalves { + listener: Arc::new(listener), + }) +} + +/// An assembly-owned TCP listener + its accept loop's feed point. The +/// accept loop ([`TcpListenerHalves::accept_loop`]) pushes accepted +/// handles into the assembly's [`AcceptQueue`]; the listen establisher +/// pops them (producer-listen's contract). +pub struct TcpListenerHalves { + listener: Arc, +} + +impl TcpListenerHalves { + /// The bound listener's local address. + pub fn local_addr(&self) -> Result { + self.listener + .local_addr() + .map_err(|e| TunnelEstablishError::HandlerError(format!("local_addr: {e}"))) + } + + /// The accept loop: accept connections until the listener fails, + /// pushing each accepted handle into the queue. Returns when the + /// listener is closed (accept errors end the loop — the assembly + /// layer aborts this task on teardown). + pub async fn accept_loop(&self, queue: AcceptQueue) -> Result<(), TunnelEstablishError> { + loop { + let (stream, _peer) = self + .listener + .accept() + .await + .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?; + let (read, write) = stream.into_split(); + let handle = TargetHandle { + read: Box::new(read), + write: Box::new(write), + }; + queue.push(handle).await.map_err(|dropped| { + TunnelEstablishError::HandlerError(format!( + "queue closed; accepted connection dropped: {dropped:?}" + )) + })?; + } + } +} + +// --- 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 +/// specifics. +pub async fn connect_udp(target: &str) -> Result { + let sock = tokio::net::UdpSocket::bind(("0.0.0.0", 0)) + .await + .map_err(|e| TunnelEstablishError::ResourceShortage(format!("udp bind failed: {e}")))?; + sock.connect(target) + .await + .map_err(|e| TunnelEstablishError::DialFailed(format!("udp connect `{target}`: {e}")))?; + let (read, write) = UdpHalf::split(sock); + 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. +/// +/// **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. +pub struct UdpHalf { + sock: Arc, + kind: UdpHalfKind, + /// The recv scratch buffer (the fail-loud truncation check reads + /// whole datagrams before copying into the caller's buffer). + recv_scratch: Vec, +} + +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +enum UdpHalfKind { + Read, + Write, +} + +impl UdpHalf { + pub fn split( + sock: tokio::net::UdpSocket, + ) -> ( + Box, + 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(), + }), + ) + } +} + +impl AsyncRead for UdpHalf { + fn poll_read( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> 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; + match this.sock.try_recv(scratch) { + Ok(n) => { + if n > buf.remaining() { + return std::task::Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "udp datagram {n} bytes exceeds caller buffer {} (truncation fails loud — OQ-TN-13)", + buf.remaining() + ), + ))); + } + buf.put_slice(&scratch[..n]); + std::task::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, + } + } + Err(e) => std::task::Poll::Ready(Err(e)), + } + } +} + +impl tokio::io::AsyncWrite for UdpHalf { + fn poll_write( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> 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, + } + } + Err(e) => std::task::Poll::Ready(Err(e)), + } + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + // UDP has no shutdown; drop is the close. + std::task::Poll::Ready(Ok(())) + } +} + +// --- Unix --------------------------------------------------------------- + +/// Dial a unix-socket target (OQ-TN-14 — ships in v1, the same halves +/// shape as TCP): `UnixStream::connect` bounded by [`DIAL_TIMEOUT`], +/// `into_split` → boxed halves (raw pass-through, ADR-003). +pub async fn dial_unix(path: &str) -> Result { + let stream = tokio::time::timeout(DIAL_TIMEOUT, tokio::net::UnixStream::connect(path)) + .await + .map_err(|_| TunnelEstablishError::DialFailed(format!("dial `{path}` timed out")))? + .map_err(|e| TunnelEstablishError::DialFailed(format!("dial `{path}`: {e}")))?; + let (read, write) = stream.into_split(); + Ok(TargetHandle { + read: Box::new(read), + write: Box::new(write), + }) +} + +/// The local dial closure (the ADR-004 injection point): one +/// [`DialFn`]-shaped function covering tcp/udp/unix — the assembly +/// layer injects this (closed over nothing substrate-specific beyond +/// this module). +pub fn local_dial() -> DialFn { + Arc::new(move |substrate, backing: &str| { + let backing = backing.to_string(); + Box::pin(async move { + match substrate { + Substrate::Tcp => dial_tcp(&backing).await, + Substrate::Udp => connect_udp(&backing).await, + Substrate::Unix => dial_unix(&backing).await, + } + }) + }) +} diff --git a/tasks/tunnels/local-socket-halves.md b/tasks/tunnels/local-socket-halves.md index 3f67747..2848b87 100644 --- a/tasks/tunnels/local-socket-halves.md +++ b/tasks/tunnels/local-socket-halves.md @@ -1,7 +1,7 @@ --- id: tunnels/local-socket-halves name: "local feature — real socket halves functions (TCP dial, UDP connect, unix)" -status: pending +status: completed depends_on: [tunnels/producer-open-op, tunnels/wire-codec] scope: moderate risk: medium @@ -104,6 +104,51 @@ POC's sizing test). > Agent fills during implementation. +- **`bind_tcp` + `TcpListenerHalves::accept_loop(queue)`** complete the + listen side (the task surface sketched `bind`/`accept_fn`): the + accept loop is a spawned task feeding an [`AcceptQueue`] — the + protocol-side queue contract from producer-listen. The assembly + layer aborts the task on teardown; accept errors end the loop. +- **`UdpHalf` truncation probe** (OQ-TN-13): `poll_read` recvs into a + 64KiB scratch buffer first, then size-checks against the caller's + buffer — a >buffer datagram is `io::ErrorKind::InvalidData`, never + a silent truncation. (`tokio`'s `poll_recv`-into-ReadBuf truncates + silently, which is exactly the invariant F-2/OQ-TN-13 forbids.) + WouldBlock polls `poll_recv_ready` for waker registration. +- **The task sketch's `TcpListenerHalves::accept_fn(&self, queue)` + became `accept_loop(&self, queue)`** (an owned async loop to spawn) + — the `accept_fn` borrow-shape fought the spawn model; the loop + returning on listener failure is the assembly-abort contract. +- **Test plumbing note** (tests): the listen-over-real-socket test + drives the client socket via std `try_clone` BEFORE `from_std` + (O_NONBLOCK is shared across dup — both ends go async); the session + uses borrowed halves (the producer's wrapper-managed pump already + pumps channel <-> accepted handle; `pump_against` there would loop + the same socket back into itself). + ## Summary -> Agent fills this on completion. \ No newline at end of file +> Agent fills this on completion. + +Implemented the `local` feature (src/local/mod.rs, feature = +`["tokio/net"]`): `dial_tcp` (5s timeout), `connect_udp` (ephemeral +bind + connect + the FRAMED `UdpHalf` adapter — truncation fails +loud), `dial_unix` (OQ-TN-14 in v1), `bind_tcp`/`TcpListenerHalves:: +accept_loop` (the listen shape's assembly half), and `local_dial()` +(the one `DialFn` covering all three substrates). No socket type +appears outside `src/local/` (convention 16: the module is never +imported from producer/consumer/shared; `TargetHandle` re-used from +producer.rs). Stdio bridging is NOT here (alktty's pipe mode owns +process stdio). + +Tests: `tests/local_halves.rs` (7, `#![cfg(feature = "local")]`) — +TCP dial round-trip through the real establisher path, UDP framed +round-trip, the 1400-byte MTU datagram, unix dial round-trip, the +OQ-TN-13 truncation fail-loud probe (60k datagram vs 16KiB read → +InvalidData), dial-refusal → `dial_failed`, and the listen producer +over a real TCP listener end-to-end. + +Verified: `cargo test` (default) and `cargo test --all-features` +green (44 + 7), clippy `-D warnings` clean (default + all-features + +wasm32), fmt clean, wasm32 check passes (default crate stays +wasm-clean; `local` is non-wasm by design). \ No newline at end of file diff --git a/tests/local_halves.rs b/tests/local_halves.rs new file mode 100644 index 0000000..f94f95c --- /dev/null +++ b/tests/local_halves.rs @@ -0,0 +1,332 @@ +//! Tests for the `local` feature: real-socket halves through the real +//! establisher path (`local_dial` injected as the `DialFn`), TCP/UDP/ +//! unix round-trips against real local echo servers, the OQ-TN-13 +//! truncation fail-loud probe, and the 1400-byte MTU datagram. +//! +//! The forward topology (`wire_forward`) carries the full path: the +//! consumer opens (`TunnelSession::open`), the producer's establisher +//! dials via `local_dial`, the pump handler pumps — real sockets at +//! both ends. + +#![cfg(feature = "local")] + +mod harness; + +use std::sync::Arc; + +use alktunnels::params::{Substrate, TunnelParams}; +use alktunnels::producer::ResourceRegistry; +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)) + .await + .expect("bind echo"); + let addr = listener.local_addr().expect("addr"); + tokio::spawn(async move { + loop { + let Ok((sock, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let (mut r, mut w) = tokio::io::split(sock); + let _ = tokio::io::copy(&mut r, &mut w).await; + }); + } + }); + addr +} + +/// A UDP echo server on an ephemeral port (one task, echo loop). +async fn udp_echo_server() -> std::net::SocketAddr { + let sock = tokio::net::UdpSocket::bind(("127.0.0.1", 0)) + .await + .expect("bind udp echo"); + let addr = sock.local_addr().expect("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; + } + } + }); + addr +} + +#[tokio::test] +async fn tcp_dial_round_trips_through_the_real_establisher_path() { + let addr = tcp_echo_server().await; + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, &addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp)) + .await + .expect("tcp open"); + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (read, write) = session.stream_halves().expect("halves"); + write.write_all(b"tcp-local").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 9]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"tcp-local"); + session.close().await; +} + +#[tokio::test] +async fn udp_datagram_round_trip_through_the_framed_adapter() { + // The FRAMED adapter end-to-end (ADR-003): session sends the codec + // frame; the pump copies raw bytes; the dialed UdpHalf strips the + // frame on recv and delivers ONE datagram to the target; the echo + // comes back through the same path. + let addr = udp_echo_server().await; + let registry = ResourceRegistry::new(); + registry + .register("dns", Substrate::Udp, &addr.to_string()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp)) + .await + .expect("udp open"); + session.send_datagram(b"query").await.expect("send"); + 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"query"); + session.close().await; +} + +#[tokio::test] +async fn udp_large_datagram_through_bounded_buffers() { + // The POC's sizing test: a 1400-byte datagram (max ethernet MTU + // payload) round-trips through the codec + the bounded-buffer path. + let addr = udp_echo_server().await; + let registry = ResourceRegistry::new(); + registry + .register("big", Substrate::Udp, &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"); + let payload: Vec = (0..1400u32).map(|i| (i % 7) as u8).collect(); + session.send_datagram(&payload).await.expect("send"); + let got = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv timed out") + .expect("io ok") + .expect("datagram"); + assert_eq!(got.len(), payload.len()); + assert_eq!(&got[..], &payload[..]); + session.close().await; +} + +#[tokio::test] +async fn unix_dial_round_trips() { + // OQ-TN-14: unix ships with local v1 (same halves shape as TCP). + let dir = std::env::temp_dir().join(format!("alktunnels-test-{}", std::process::id())); + let _ = std::fs::create_dir_all(&dir); + let path = dir.join("echo.sock"); + let _ = std::fs::remove_file(&path); + + let listener = tokio::net::UnixListener::bind(&path).expect("bind unix echo"); + tokio::spawn(async move { + loop { + let Ok((sock, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let (mut r, mut w) = tokio::io::split(sock); + let _ = tokio::io::copy(&mut r, &mut w).await; + }); + } + }); + + let registry = ResourceRegistry::new(); + registry + .register("uds", Substrate::Unix, &path.to_string_lossy()) + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("uds", Substrate::Unix)) + .await + .expect("unix open"); + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (read, write) = session.stream_halves().expect("halves"); + write.write_all(b"unix-local").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 10]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"unix-local"); + session.close().await; + let _ = std::fs::remove_file(&path); +} + +#[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. + 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"); + + // The producer-side adapter (what the establisher dials): sock_a + // wrapped as halves. Send a large datagram 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"); + + // The pump reads with the codec's chunk size (16 KiB) — smaller + // than 60000: the adapter must FAIL LOUD, not truncate. + use tokio::io::AsyncReadExt; + 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:?}"), + } +} + +#[tokio::test] +async fn dial_refusal_maps_to_dial_failed() { + // Error mapping: a refused dial (nothing listening on the port) is + // `dial_failed` on the wire (`channel:open_failed` reason). + let registry = ResourceRegistry::new(); + registry + .register("dead", Substrate::Tcp, "127.0.0.1:1") + .await; + let topo = wire_forward(registry, alktunnels::local::local_dial()).await; + + let err = match TunnelSession::open(&topo.consumer, params("dead", Substrate::Tcp)).await { + Ok(_) => panic!("refused dial must fail"), + Err(e) => e, + }; + assert_eq!(err.open_ref().establishment_reason(), Some("dial_failed")); +} + +#[tokio::test] +async fn listen_producer_over_a_real_tcp_listener() { + // The listen shape + local halves compose: a real TcpListenerHalves + // accept loop feeds an AcceptQueue; a listen-registered producer + // pops accepted sockets; the consumer pumps the tunnel against the + // accepted local socket. + let queue = alktunnels::producer::AcceptQueue::new(); + let listener = alktunnels::local::bind_tcp("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + + let accept_loop = tokio::spawn({ + let queue = queue.clone(); + async move { listener.accept_loop(queue).await } + }); + + // A local client connects (the accepted connection's source). The + // driving end comes from `try_clone` — the pump owns the accepted + // handle; the driver pushes/reads the echo through the same real + // socket. + // std-level try_clone BEFORE converting to tokio streams: one fd + // for the driver, one wrapped for the pump (two independent + // handles to the same socket). O_NONBLOCK is shared across the + // dup, so BOTH ends go nonblocking and the driver is driven + // async too. + let local_std = std::net::TcpStream::connect(addr).expect("std connect"); + let driver_std = local_std.try_clone().expect("try_clone driver"); + local_std.set_nonblocking(true).expect("nonblocking"); + let mut driver = tokio::net::TcpStream::from_std(driver_std).expect("tokio driver"); + + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let accept_fn: alktunnels::producer::AcceptFn = Arc::new(move || { + let queue = queue.clone(); + Box::pin(async move { + queue.pop().await.ok_or_else(|| { + alktunnels::producer::TunnelEstablishError::DialFailed( + "listener closed".to_string(), + ) + }) + }) + }); + let topo = harness::wire_listen(registry, accept_fn).await; + + let channel_id = alktunnels::open_reverse_channel( + &topo.consumer_call, + ¶ms("listener", Substrate::Tcp), + None, + ) + .await + .expect("listen open"); + + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + // The producer's pump (wrapper-managed) copies channel <-> the + // accepted handle; the accepted handle IS the client socket — so + // the session's borrowed halves drive it directly: the driver end + // writes into the client socket, the pump carries it over the + // channel, and the session halves read it (and back). + let mut session = session; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + driver.write_all(b"listen-local").await.expect("write"); + driver.flush().await.expect("flush"); + let (read, _write) = session.stream_halves().expect("halves"); + let mut buf = vec![0u8; 12]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"listen-local"); + + session.close().await; + accept_loop.abort(); +}