From 04c76e3a411ae9b81fd9b1ca7901ed1146f69954 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 26 Sep 2026 08:24:32 +0000 Subject: [PATCH] =?UTF-8?q?review:=20the=20batch-2=20gate=20closed=20?= =?UTF-8?q?=E2=80=94=20the=20assembly=20layer=20verified,=20R-1/R-2=20fixe?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review checkpoint closing batch 2 (the assembly layer). The gate opens: the never-binds posture, seam contracts, door fidelity, the bind-first graduation, feature-gate hygiene, and POC fidelity all verify clean; two findings were corrected in the review change. Findings corrected: - R-1 (major): the UDP front door's in-band decode created a fresh DatagramReader per call and kept only the first frame of a batch — POC #2's first-cut bug (reader state must persist across calls; batched datagrams were silently dropped, mid-frame EOF was silent). Fix: an owning InbandBridge (persistent reader + pending queue + fail-loud mid-frame EOF, the AssociateSession shape) threaded through the channel->socket pump; pinned by three bridge unit tests (which fail against the per-call reader) and an end-to-end doubled-echo door-udp test. - R-2 (convention 2): seven non-test expects -> typed errors — DoorBuildError (both door builders' .build() is now Result), ProducerDoorBuildError/ProducerDoorError (the bind decision is serve's typed error), the composed table invariant as an egress error, register_forwarded_serving -> Result (the alkcall diagnostic surfaced), TokenClientError (wrap on a spent client). New error types re-exported; test call sites updated. - R-3/R-4 (doc residues): producer.md's local-backend section described a channel pipe — the landed ProducerDoor serves serve_socks5_over on the accepted socket directly; associate_with_token's doc (the TCP door refuses ASSOCIATE, never calls it); SessionOwner's doc (the accept uses SyncHalves/ OwnedWrite). Verification: cargo test (default) green; cargo test --all-features green; cargo clippy --all-targets [--all-features] -- -D warnings clean; cargo fmt --check clean; both wasm checks clean (default + composed); local-on-wasm fails at mio as designed; cargo doc --no-deps clean. Task: tasks/implementation/review-assembly-layer.md (status: completed; the batch-2 gate opens release polish). --- docs/architecture/producer.md | 33 ++- src/composed/mod.rs | 59 ++-- src/consumer.rs | 33 ++- src/door.rs | 253 +++++++++++++++--- src/lib.rs | 6 +- src/local/mod.rs | 4 +- src/local/producer_door.rs | 39 ++- tasks/implementation/review-assembly-layer.md | 238 ++++++++++++++-- tests/assembly-integration.rs | 37 ++- tests/common/mod.rs | 51 ++++ tests/composed-accept.rs | 3 +- tests/door-tcp.rs | 52 +++- tests/door-udp.rs | 85 +++++- tests/harness.rs | 4 +- tests/local-backend-producer.rs | 3 +- 15 files changed, 767 insertions(+), 133 deletions(-) diff --git a/docs/architecture/producer.md b/docs/architecture/producer.md index 29f728e..9d21a27 100644 --- a/docs/architecture/producer.md +++ b/docs/architecture/producer.md @@ -130,24 +130,31 @@ Contracts (POC-pinned, load-bearing): The optional shape where the producer serves from a real listener: a feature-gated backend binds `socks5://127.0.0.1:port` (or any -address) and pipes each accepted TCP connection raw onto a channels -`BiStream` (`copy_bidirectional`, zero protocol translation — POC -#1's curl front door validated the pipe shape). The same door shape -in reverse: this is a *front door for the producer side* — vanilla -clients reach a locally-served SOCKS5 resource through it. The door -surface (NoAuth raw pipe vs RFC 1929 cred-bearing, ADR-005) is the -consumer door's code shape (see [consumer.md](consumer.md)). +address) and hands each accepted TCP connection to the landed +substrate-free entry `Socks5Producer::serve_socks5_over` directly — +the RFC conversation runs on the accepted socket (the NoAuth +in-stream shape: greeting, command, reply, data plane, all on the +accepted socket), and the seams dial/egress/accept locally. The same +door shape in reverse: this is a *front door for the producer side* +— vanilla clients reach a locally-served SOCKS5 resource through it +(`src/local/producer_door.rs`, the `ProducerDoor`/builder/guard +surface). A cred-bearing variant (RFC 1929) would wrap +`accept_password_auth` before `serve_socks5_over` — a documented +follow-up, not landed. The door's accept path owns the idle defense the protocol layer does -not provide (ADR-014): an accepted connection that sends no first -byte within a configured window is closed door-side (an accepted -socket is not a channel — the cap regime does not bound it), plus -per-credential/per-peer session caps and accept-queue bounds as the -deployment's policy. +not provide (ADR-014): an accepted connection that sends no first byte +within a configured window is closed door-side (an accepted socket is +not a channel — the cap regime does not bound it), plus the +max-concurrent-sessions cap (an accept beyond the bound is refused +door-side, no session spawned) — the two constructor knobs. The producer-side local *backend* for the I/O planes (the local-socket providers — TCP dial, UDP egress, BIND listener) is -ADR-008's variant set, also feature `local`. +ADR-008's variant set, also feature `local`. The BIND command gate +on this path is the builder's `bind_policy` (there is no open-op +identity on an accepted socket — the accepted socket is the whole +caller; the pump handler's identity-derived gate does not apply). ## Wiring (Assembly Layer) diff --git a/src/composed/mod.rs b/src/composed/mod.rs index aa27b7a..ebb65c9 100644 --- a/src/composed/mod.rs +++ b/src/composed/mod.rs @@ -56,19 +56,17 @@ //! `alkcall::BiStream` is `Send + Unpin` but NOT `Sync` (its inner is //! a `dyn AsyncReadWrite` trait object). The owner-wrapper posture //! (egress.rs's documented contract: "keep-alive state a substrate -//! needs rides inside the halves — a provider-side owner wrapper") is -//! designed here as [`SessionOwner`] and shared with the composed -//! accept (the sibling task): an alktunnels half wrapped with its -//! session inside, so the handle payload's `Sync` bound is met by -//! ownership and the channel entry reaps when the wrapper (and its -//! session) drops. For the egress the wrapper is not the handle -//! payload's burden — the egress owns the session in the table entry -//! (the POC's shape); the relay task owns the bare read half. The -//! accept variant meets the `+ Sync` bound with the [`SyncHalves`] -//! adapter (a `std::sync::Mutex` around the joined halves — a -//! `Mutex` is `Sync`, no unsafe) with the channel session -//! owned inside the payloads ([`ForwardedOwner`] — dropping the handle -//! drops the session AND the halves together). +//! needs rides inside the halves — a provider-side owner wrapper") +//! takes two shapes here: [`SessionOwner`] (the generic half-wraps- +//! session wrapper — the egress's design point, exported as the +//! wrapper-aware deployment's surface) and the accept's own +//! [`SyncHalves`]/[`OwnedWrite`] adapter pair (a `std::sync::Mutex` +//! around the half — `Mutex` is `Sync`, no unsafe — with +//! the channel session owned inside the payload, [`ForwardedOwner`]: +//! dropping the handle drops the session AND the halves together). +//! For the egress the wrapper is not the handle payload's burden — +//! the egress owns the session in the table entry (the POC's shape); +//! the relay task owns the bare read half. use std::collections::HashMap; use std::net::SocketAddr; @@ -126,9 +124,11 @@ pub fn key_of(addr: &TargetAddr) -> String { } /// The owner wrapper (the batch-1 review's seam contract vs -/// alktunnels' surface — the shared design point with the composed -/// accept): an alktunnels half wrapped with its [`TunnelSession`] -/// inside. The landed handle payloads carry `+ Sync` (the +/// alktunnels' surface — the owner-wrapper posture's generic shape; +/// the composed accept meets the `+ Sync` bound with its own +/// [`SyncHalves`]/[`OwnedWrite`] adapter pair instead): an alktunnels +/// half wrapped with its [`TunnelSession`] inside. The landed handle +/// payloads carry `+ Sync` (the /// `ChannelPlan` payload bound), alktunnels' `TakenHalves` yields /// `Send + Unpin` only — the wrapper owns the session beside the half, /// so dropping the wrapper drops the half and the session together @@ -327,7 +327,15 @@ impl ComposedEgress { } self.order.retain(|(k, _)| k != &key); self.order.push_back((key.clone(), addr.clone())); - Ok(self.entries.get_mut(&key).expect("entry inserted above")) + // The entry was inserted above (or the key was already + // present): an absent entry here is a structural invariant + // break, not a runtime condition — surfacing it as an + // egress error instead of a panic (convention 2). + self.entries.get_mut(&key).ok_or_else(|| { + EgressError::General(format!( + "tunnel table invariant broken: `{key}` absent after insert" + )) + }) } } @@ -641,26 +649,25 @@ impl AsyncWrite for OwnedWrite { /// `tunnel:open`). The returned registry handle feeds the factories; /// keep it alongside the client they ride. /// -/// # Panics +/// # Errors /// -/// If the underlying registration fails (a duplicate -/// `channels/tunnel/forwarded` op id on the registry — an assembly -/// bug, not a runtime condition). `register_openable`'s failure type -/// is a `String` diagnostic, not a typed error. +/// A duplicate `channels/tunnel/forwarded` op id on the registry (an +/// assembly bug, not a runtime condition) — the underlying +/// registration's `String` diagnostic, surfaced (convention 2: no +/// panic path; the caller decides). pub fn register_forwarded_serving( core: &ChannelCore, on_registry: &Arc, auth: AuthContext, -) -> CorrelationRegistry { +) -> Result { let registry: CorrelationRegistry = Arc::new(tokio::sync::Mutex::new(HashMap::new())); alktunnels::producer::register_tunnel_forwarded_openable( core, on_registry, auth, make_forwarded_handler(Arc::clone(®istry)), - ) - .expect("register forwarded openable (the composed accept's serving half)"); - registry + )?; + Ok(registry) } /// The composed accept factory (ADR-007 variant 2, producer.md diff --git a/src/consumer.rs b/src/consumer.rs index 01ce393..9cea84a 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -440,10 +440,13 @@ impl AssociateSession { /// The cred-bearing front door's ASSOCIATE re-origination (the /// token-carrying twin of [`Self::associate`]): the open rides - /// the payload `auth_token` (CF-005 precedence (1)). The TCP door - /// uses this only for its RFC-faithful UDP refusal probe (the - /// door's datagram plane is the UDP door's task); the session is - /// otherwise identical. + /// the payload `auth_token` (CF-005 precedence (1)). Not called + /// by the landed TCP door — its ASSOCIATE arm refuses + /// RFC-faithfully before any session is opened (the datagram + /// plane is the UDP door's task, ADR-003); the constructor is + /// the session family's completeness (every command carries the + /// token twin) and the wrapper-aware deployment's surface. The + /// session is otherwise identical. pub async fn associate_with_token( client: &TokenClient, token: Vec, @@ -1081,15 +1084,20 @@ impl TokenClient { /// Wrap a `ChannelClient`: take its call connection (the open-op /// calling surface) and clone the manager. The door's /// token-bearing sessions then ride this client. - pub async fn wrap(client: &ChannelClient) -> Self { + /// + /// [`TokenClientError::CallConnectionTaken`] when the client's + /// call connection was already taken (a client is a single + /// calling surface — the door-only contract; convention 2: no + /// panic path on a double wrap). + pub async fn wrap(client: &ChannelClient) -> Result { let call = client .take_call_connection() .await - .expect("the channel client's call connection exists until taken"); - Self { + .ok_or(TokenClientError::CallConnectionTaken)?; + Ok(Self { call, manager: client.manager().clone(), - } + }) } /// The manager (adopt/teardown — the token sessions' prologue uses @@ -1124,6 +1132,15 @@ impl TokenClient { } } +/// Why a `TokenClient` refused to construct (the typed construction +/// error — the client's calling surface was already taken; +/// convention 2: no panic path). +#[derive(Debug, thiserror::Error)] +pub enum TokenClientError { + #[error("the channel client's call connection was already taken (a client is one calling surface — the door-only contract)")] + CallConnectionTaken, +} + impl std::fmt::Debug for TokenClient { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("TokenClient").finish_non_exhaustive() diff --git a/src/door.rs b/src/door.rs index 531451d..83fa5fb 100644 --- a/src/door.rs +++ b/src/door.rs @@ -248,6 +248,9 @@ impl FrontDoorBuilder { /// The local bind address — the door's explicit bind decision /// (the never-binds posture: the door is the assembly layer's /// spelled-out choice; loopback is the documented posture). + /// Required: `.build()` is a `Result` — omitting it is the + /// builder's typed construction error (convention 2: no panic + /// path). pub fn bind(mut self, addr: impl Into) -> Self { self.bind_addr = Some(addr.into()); self @@ -273,16 +276,26 @@ impl FrontDoorBuilder { self } - pub fn build(self) -> FrontDoor { - FrontDoor { - bind_addr: self.bind_addr.expect("the door binds an address"), + pub fn build(self) -> Result { + let bind_addr = self.bind_addr.ok_or(DoorBuildError::MissingBindAddr)?; + Ok(FrontDoor { + bind_addr, credentials: self.credentials, idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)), max_sessions: self.max_sessions.unwrap_or(256), - } + }) } } +/// Why a door builder refused to construct (the builder's typed +/// construction error — a missing required field; convention 2: no +/// panic path on an omitted `.bind`). +#[derive(Debug, thiserror::Error)] +pub enum DoorBuildError { + #[error("the door requires a bind address (.bind) — the bind decision is never implied")] + MissingBindAddr, +} + impl FrontDoor { /// Bind the configured address and run the accept loop /// (consumer.md §Wiring: `door.serve(channel_client)`). Per- @@ -313,10 +326,13 @@ impl FrontDoor { // NoAuth door keeps the client's own calling surface // (`open_channel` — the raw-pipe shape never carries a // token). - let token_client = if credentials.is_some() { - Some(Arc::new(TokenClient::wrap(&client).await)) - } else { - None + let token_client = match credentials { + Some(_) => Some(Arc::new( + TokenClient::wrap(&client) + .await + .map_err(|err| std::io::Error::other(err.to_string()))?, + )), + None => None, }; loop { @@ -750,11 +766,12 @@ impl UdpFrontDoorBuilder { self } - pub fn build(self) -> UdpFrontDoor { - UdpFrontDoor { - bind_addr: self.bind_addr.expect("the UDP door binds an address"), + pub fn build(self) -> Result { + let bind_addr = self.bind_addr.ok_or(DoorBuildError::MissingBindAddr)?; + Ok(UdpFrontDoor { + bind_addr, idle: self.idle.unwrap_or(Duration::from_secs(30)), - } + }) } } @@ -850,13 +867,17 @@ impl UdpFrontDoor { // encode the RFC reply header (the codec boundary), send the // RFC-header'd datagram back to the client's source address. // The association's end (the channel EOF) is this pump's clean - // exit. + // exit. The decode state (the incremental reader + the + // batched-datagram pending queue) persists across pump + // iterations — the bridge owns it for the association's + // lifetime. let c2s = { let socket = Arc::clone(&socket); let client_addr = Arc::clone(&client_addr); async move { + let mut bridge = InbandBridge::new(); loop { - let (addr, payload) = match recv_inband(&mut channel_read).await { + let (addr, payload) = match recv_inband(&mut bridge, &mut channel_read).await { Ok(Some(pair)) => pair, Ok(None) => return UdpDoorExit::AssociationEnded, Err(err) => return UdpDoorExit::Session(err.to_string()), @@ -904,24 +925,81 @@ async fn send_inband( Ok(()) } -/// The in-band frame read (the read half's shape — the incremental -/// decoder; its state persists across calls, the same `DatagramReader` -/// the session runs). -async fn recv_inband( - read: &mut tokio::io::ReadHalf<&mut AssociationStream>, -) -> Result, SocksSessionError> { - let mut chunk = vec![0u8; 16 * 1024]; - let mut reader = DatagramReader::new(); - loop { - let n = tokio::io::AsyncReadExt::read(read, &mut chunk).await?; - if n == 0 { - return Ok(None); - } - let dgs = reader.feed(&chunk[..n]); - if let Some(res) = dgs.into_iter().next() { - return res.map(Some).map_err(wire_error); +/// The in-band frame read (the read half's shape): a fresh-looking +/// async read whose decoder state **persists across calls** — the +/// bridge's [`InbandBridge`] owns the incremental `DatagramReader` +/// and the batched-datagram pending queue across calls (a datagram +/// split across channel chunks reassembles across reads; frames +/// decoded beyond the one returned queue for later reads — the POC +/// #2 reader-state finding, the same shape +/// `AssociateSession::recv_datagram` owns). +struct InbandBridge { + reader: DatagramReader, + pending: std::collections::VecDeque<(TargetAddr, Bytes)>, +} + +impl InbandBridge { + fn new() -> Self { + Self { + reader: DatagramReader::new(), + pending: std::collections::VecDeque::new(), } } + + /// The next in-band datagram: the pending queue first (a chunk + /// may have batched several frames — drain in order), then an + /// incremental decode off the read half. `Ok(None)` on channel + /// EOF (the association's end); fail-loud mid-frame (never a + /// silent partial datagram). + async fn recv( + &mut self, + read: &mut R, + ) -> Result, SocksSessionError> + where + R: tokio::io::AsyncRead + Unpin, + { + if let Some(dg) = self.pending.pop_front() { + return Ok(Some(dg)); + } + let mut chunk = vec![0u8; 16 * 1024]; + loop { + let n = tokio::io::AsyncReadExt::read(read, &mut chunk).await?; + if n == 0 { + if self.reader.is_mid_frame() { + return Err(SocksSessionError::Rfc(SocksError::Io(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "association ended mid-datagram (fail-loud)", + )))); + } + return Ok(None); + } + let mut dgs = self.reader.feed(&chunk[..n]).into_iter(); + if let Some(first) = dgs.next() { + for res in dgs { + match res { + Ok(pair) => self.pending.push_back(pair), + Err(err) => return Err(wire_error(err)), + } + } + match first { + Ok(pair) => return Ok(Some(pair)), + Err(err) => return Err(wire_error(err)), + } + } + } + } +} + +/// The channel→socket pump's decode call (the bridge the pump loop +/// drives — the decode state lives in the bridge, not per call). +async fn recv_inband( + bridge: &mut InbandBridge, + read: &mut R, +) -> Result, SocksSessionError> +where + R: tokio::io::AsyncRead + Unpin, +{ + bridge.recv(read).await } /// Map a wire-decode error into the RFC vocabulary (the consumer @@ -940,3 +1018,118 @@ fn wire_error(err: WireError) -> SocksSessionError { /// The association's framed stream type (the split halves' target — /// the forked client type over the channel's `BiStream`). type AssociationStream = crate::rfc1928::client::Socks5Stream; + +#[cfg(test)] +mod tests { + use super::*; + use crate::wire::frame as wire_frame; + use tokio::io::AsyncWriteExt; + use tokio::io::{AsyncRead, DuplexStream}; + + /// Drive one pump-shaped decode call: feed `bytes` from the peer + /// side, then decode (the recv parks until a frame completes or + /// the stream ends — the pump loop's shape). + async fn feed_then_recv( + bridge: &mut InbandBridge, + read: &mut R, + writer: &mut DuplexStream, + bytes: &[u8], + ) -> Result, SocksSessionError> + where + R: AsyncRead + Unpin, + { + if !bytes.is_empty() { + writer.write_all(bytes).await.expect("feed write"); + writer.flush().await.expect("feed flush"); + } + bridge.recv(read).await + } + + #[tokio::test] + async fn batched_frames_queue_not_drop() { + // Two frames batched into one feed: the first returns, the + // second queues and drains on the next call (the per-call + // reader dropped it — review 003 R-1; the pending queue is + // the state that persists across the call boundary). + let (mut writer, reader) = tokio::io::duplex(1024); + let (mut channel_read, _channel_write) = tokio::io::split(reader); + let addr = TargetAddr::Ip("10.0.0.2:53".parse().expect("addr")); + let mut wire = wire_frame(&addr, b"first").unwrap().to_vec(); + wire.extend_from_slice(&wire_frame(&addr, b"second").unwrap()); + let mut bridge = InbandBridge::new(); + + let first = feed_then_recv(&mut bridge, &mut channel_read, &mut writer, &wire).await; + match first { + Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"first"), + other => panic!("the first datagram returned: {other:?}"), + } + let second = bridge.recv(&mut channel_read).await; + match second { + Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"second"), + other => panic!("the batched datagram queued, not lost: {other:?}"), + } + } + + #[tokio::test] + async fn eof_mid_frame_fails_loud_not_silent() { + // A frame in flight when the stream ends: the decode errors + // loud (never a silent partial datagram) — the same shape + // AssociateSession::recv_datagram pins. + let (mut writer, reader) = tokio::io::duplex(1024); + let (mut channel_read, _channel_write) = tokio::io::split(reader); + let addr = TargetAddr::Ip("10.0.0.3:53".parse().expect("addr")); + let framed = wire_frame(&addr, b"truncated").expect("frame"); + let mut bridge = InbandBridge::new(); + + let feeder = tokio::spawn(async move { + writer + .write_all(&framed[..3]) + .await + .expect("the partial frame"); + writer.flush().await.expect("flush"); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let _ = writer.shutdown().await; + }); + + let ended = bridge.recv(&mut channel_read).await; + assert!( + ended.is_err(), + "EOF mid-frame is the fail-loud truncated-datagram condition" + ); + feeder.await.expect("feeder"); + } + + #[tokio::test] + async fn partial_frame_state_persists_across_calls() { + // One chunk carries frame 1 complete + the first bytes of + // frame 2: call 1 returns frame 1 (frame 2's bytes stay in + // the bridge's reader), call 2 — after the stream ends — + // fails loud on the surviving mid-frame state (a per-call + // reader would report a clean Ok(None) EOF here). + let (mut writer, reader) = tokio::io::duplex(1024); + let (mut channel_read, _channel_write) = tokio::io::split(reader); + let addr = TargetAddr::Ip("10.0.0.4:53".parse().expect("addr")); + let mut wire = wire_frame(&addr, b"whole").unwrap().to_vec(); + wire.extend_from_slice(&wire_frame(&addr, b"partial").unwrap()[..3]); + let mut bridge = InbandBridge::new(); + + let feeder = tokio::spawn(async move { + writer.write_all(&wire).await.expect("the mixed chunk"); + writer.flush().await.expect("flush"); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let _ = writer.shutdown().await; + }); + + let first = bridge.recv(&mut channel_read).await; + match first { + Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"whole"), + other => panic!("frame 1 returned: {other:?}"), + } + let second = bridge.recv(&mut channel_read).await; + assert!( + second.is_err(), + "frame 2's partial bytes survived the call boundary (fail-loud at EOF)" + ); + feeder.await.expect("feeder"); + } +} diff --git a/src/lib.rs b/src/lib.rs index 7940969..6200cac 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -137,6 +137,8 @@ pub use composed::{ }; #[cfg(feature = "local")] -pub use consumer::TokenClient; +pub use consumer::{TokenClient, TokenClientError}; #[cfg(feature = "local")] -pub use door::{CredentialTable, DoorCredential, FrontDoor, UdpFrontDoor}; +pub use door::{CredentialTable, DoorBuildError, DoorCredential, FrontDoor, UdpFrontDoor}; +#[cfg(feature = "local")] +pub use local::{ProducerDoorBuildError, ProducerDoorError}; diff --git a/src/local/mod.rs b/src/local/mod.rs index 469f32d..1ebf4d4 100644 --- a/src/local/mod.rs +++ b/src/local/mod.rs @@ -283,4 +283,6 @@ mod tests; mod producer_door; -pub use producer_door::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard}; +pub use producer_door::{ + ProducerDoor, ProducerDoorBuildError, ProducerDoorBuilder, ProducerDoorError, ProducerDoorGuard, +}; diff --git a/src/local/producer_door.rs b/src/local/producer_door.rs index 462181a..cc2cd27 100644 --- a/src/local/producer_door.rs +++ b/src/local/producer_door.rs @@ -141,16 +141,39 @@ impl ProducerDoorBuilder { self } - pub fn build(self) -> ProducerDoor { - ProducerDoor { + pub fn build(self) -> Result { + let producer = self + .producer + .ok_or(ProducerDoorBuildError::MissingProducer)?; + Ok(ProducerDoor { bind_addr: self.bind_addr, - producer: self.producer.expect("the backend serves a producer"), + producer, idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)), max_sessions: self.max_sessions.unwrap_or(256), - } + }) } } +/// Why the producer-door builder refused to construct (the builder's +/// typed construction error — a missing required field; convention 2: +/// no panic path). +#[derive(Debug, thiserror::Error)] +pub enum ProducerDoorBuildError { + #[error("the backend requires a producer (.producer) — there is nothing to serve")] + MissingProducer, +} + +/// Why the producer door's `serve` refused to start (the typed +/// construction error — `serve` carries the bind decision, so an +/// omitted `.bind` is its error, not a panic; convention 2). +#[derive(Debug, thiserror::Error)] +pub enum ProducerDoorError { + #[error("the backend requires a bind address (.bind) before serve — the bind decision is never implied")] + MissingBindAddr, + #[error("tcp bind: {0}")] + Bind(#[source] std::io::Error), +} + /// The serving guard: [`ProducerDoor::serve`] returns one; `close()` /// ends the accept loop gracefully (the listener is dropped, the /// loop's end is awaited — the spawned sessions already running own @@ -196,15 +219,17 @@ impl ProducerDoor { /// (the alktunnels per-accept isolation posture). The listener's /// lifetime is the caller's: the returned [`ProducerDoorGuard`] /// ends it (`close()` graceful; drop best-effort abort). - pub async fn serve(self) -> Result { + pub async fn serve(self) -> Result { let ProducerDoor { bind_addr, producer, idle_first_byte, max_sessions, } = self; - let bind_addr = bind_addr.expect("the backend binds an address (builder .bind)"); - let listener = TcpListener::bind(bind_addr).await?; + let bind_addr = bind_addr.ok_or(ProducerDoorError::MissingBindAddr)?; + let listener = TcpListener::bind(bind_addr) + .await + .map_err(ProducerDoorError::Bind)?; Ok(serve_on_parts( listener, idle_first_byte, diff --git a/tasks/implementation/review-assembly-layer.md b/tasks/implementation/review-assembly-layer.md index 65d7714..81a93a9 100644 --- a/tasks/implementation/review-assembly-layer.md +++ b/tasks/implementation/review-assembly-layer.md @@ -1,7 +1,7 @@ --- id: review-assembly-layer name: Review — the assembly layer (the batch-2 gate) -status: pending +status: completed depends_on: [implementation/assembly-integration-suite] scope: narrow risk: low @@ -62,23 +62,233 @@ Review against the architecture, in this order: 8. **Docs sync**: the specs match what landed (the integration task's reconciliation verified; any residue is a finding). -## Notes +## Summary -> To be filled by the review agent — see the Summary and the Findings. +Review performed 2026-09-26 against the eight-criteria order; the gate +opens (the crate is feature-complete against the Phase 1 spec set — +release polish remains). One major correctness finding was corrected +in the review change itself (R-1: the UDP door's in-band bridge lost +the reader-state contract), plus a convention sweep residue (R-2: +seven non-test `expect`s → typed errors, the batch-1 R-2 class) and +two doc residues (R-3, R-4). No seam-contract, never-binds, +graduation-shape, feature-gate, or POC-fidelity drift was found +beyond these. + +**1. The never-binds posture — clean.** The protocol layer +(`producer.rs`/`consumer.rs`/`wire.rs`/`params.rs`/`egress.rs`/ +`error.rs`/`rfc1928/`) names no kernel-socket type (`tokio::net` +appears only in doc prose; zero code references — swept). The default +dependency graph carries no `tokio/net`/`mio`/`socket2` +(`cargo tree -e normal -f "{p} [{f}]"`: tokio rides the wasm-clean +subset `bytes,io-util,macros,rt,sync,time`; `net` appears only under +`--features local`, which also brings `mio`+`socket2`; `composed` +adds alktunnels 0.2.0 only and its tokio stays on the clean subset). +Every kernel bind is an explicit constructor decision in `local`-gated +code (`local/mod.rs` — the dial's connect, `LocalEgress::bind`'s +loopback `127.0.0.1:0` (a wildcard bind is the caller's variant, doc +pinned), `local_bind_accept`'s explicit `bind_host` param, the doors' +`TcpListener::bind`/`UdpSocket::bind` behind builder `.bind()`) — +the doors' binds are the doors' own (assembly-layer, ADR-009). + +**2. Seam-contract compliance (ADR-004) — clean.** *Readiness*: the +local egress parks on `sock.readable().await` then `try_recv_from` +(`WouldBlock` → the spurious `Ok(None)`, never hot); the composed +egress parks on the shared queue's `recv().await`, and its +closed-queue condition is a **count-based design** (`live_relays` + +`ever_opened`) — a real improvement over the POC (the POC held its +own `inbound_tx` clone, so its queue-`None` arm was unreachable; the +landed egress actually detects every-relay-ended and returns `Err`, +pinned by `closed_queue_yields_err_not_a_hot_none`). *Ownership*: +rides the handle structurally everywhere — the local providers box +their socket halves into the payloads; the composed accept wraps the +`Send + Unpin`-only alktunnels halves in `SyncHalves`/`OwnedWrite` +(`Mutex` is `Sync`, no unsafe) with the `TunnelListener` +riding INSIDE the write payload (`ForwardedOwner` — the +disjoint-capture trap is doc-pinned and the whole-struct rebinding is +in place); the egress keeps the session in the table entry (the POC's +insert-time-bug lesson, doc-pinned). No `_` bindings of owning values +(swept). *The accept future owns its resources end-to-end* in both +variants (the listener lives inside the future; the composed future +owns the listener + the oneshot receiver — the client-gone select! +drop tears both down). *Framing-once*: the local egress returns +`addr + data` inner bytes; the composed egress returns inner bytes +with its own oversize guard mirroring the wire bound; the tunnel-side +framing is the tunnel codec's (`frame_datagram`), never the RFC +frame's. + +**3. Door fidelity (ADR-005/ADR-003) — clean, one finding corrected +(R-1).** The NoAuth door is a literal raw pipe +(`copy_bidirectional`, zero translation — pinned by +`door-tcp.rs`'s raw-byte probes). The cred-bearing door terminates +RFC 1929 locally through the forked `accept_password_auth` surface, +maps through the static table (the check IS the lookup — one table, +one truth), and re-originates in-band via the token sessions; the +never-pipes-raw-post-auth property is pinned by the re-origination +probe (the vanilla client's command bytes are consumed door-side; +the in-band conversation is fresh). The token is the mapped +identity's id riding the open-op `auth_token` payload (CF-005 +precedence (1)) — the password never crosses the channel (pinned); +the identity never reaches the vanilla client. The RFC UDP-header +codec (`parse_udp_request`/`new_udp_header`) appears only at the +door's boundary (the in-band wire stays `{addr, data}` frames — +pinned end to end through a real echo target); FRAG ≠ 0 is refused by +dropping the datagram, the association stays healthy (pinned). The +TCP door's ASSOCIATE arm refuses RFC-faithfully before any session +opens (no phantom data plane). R-1 is the UDP door's in-band decode +state — see the Findings. + +**4. The graduation's shape (ADR-007 variant 2) — clean.** The +composed accept advertises the real OS-chosen `bound` (the open-op +reply's `bound` field via the reply projection; a missing +`bound`/`listen` FAILS the factory — `Shortage` with an explicit +no-stand-in message, never an expected-addr stand-in) and the real +forwarded-op `peer` (a `Path` peer maps to the RFC sentinel — the +documented addressless case, not a stand-in). The POC's superseded +workaround (reply#1-first + carried `bind_addr` + sentinel peer) is +not resurrected — `tests/composed-accept.rs` pins the real addresses +end to end (`reply#1 advertises the real OS-chosen bound`, `reply#2 +carries the forwarded op's real peer`), and the establishment bound +covers only the open-op round trip (the listener open is the +factory's; the accept wait is the teardown-governed quiet-wait phase). + +**5. Convention compliance — one residue corrected (R-2).** After +R-2, no `unwrap`/`expect`/`panic!`/`unreachable!` in non-test library +code (the seven batch-2 additions were the builder-required-field +`expect`s and two internal-invariant `expect`s — same class as +batch-1's R-2; all now typed). `thiserror` surfaces throughout (the +new builder errors included); naming/module structure hold — the +feature-gated backend modules (`local/`, `composed/`, `door`) are +never imported from the protocol modules (swept: zero +`crate::local`/`crate::door`/`crate::composed` references in the +protocol layer); producer/consumer vocabulary holds (RFC-facing +"server"/"client" only inside `src/rfc1928/`). + +**6. Feature-gate hygiene (ADR-009) — green, all four targets.** +`cargo test` (default) green; `cargo test --all-features` green; both +wasm checks clean for default AND `--features composed` (the +composed wasm gate verified live); `local` on wasm fails to build +with exactly the expected shape — `mio` cannot compile on +`wasm32-unknown-unknown` (the alktty posture's build-error +guarantee, verified). `local` brings only `tokio/net` (+ `mio`/ +`socket2` transitively); `composed` brings only alktunnels; the +default graph is unchanged from batch 1 (spot-checked per feature). + +**7. POC-fidelity spot-check — clean, with the R-1 exception and one +documented improvement.** +- *The composed egress's LRU + closed-queue posture*: the lazy + destination→tunnel table, the LRU eviction-before-insert loop, the + touch-on-access ordering, the session-inside-the-entry placement, + and the relay task's shape are all line-faithful to POC #2's + `ComposedEgress` (`/workspace/alksocks-udp-poc/src/egress.rs`). + The one deliberate divergence is the closed-queue detection — the + POC's `recv().await`-`None` arm was structurally unreachable (it + held its own `Sender`); the crate replaces it with the + `live_relays`/`ever_opened` count so the association actually ends + when every tunnel ends (the POC's stated *intent*, realized; + drift-in-the-right-direction, doc-pinned in the struct docs). +- *The cred-bearing door's re-origination*: matches POC #1's + terminate-locally + re-speak shape (the POC's raw-pipe door is the + NoAuth variant; the cred-bearing extension is the ADR-005 design, + validated by the integration suite's vanilla-client round trips). +- *The bind-first correlation*: the token→oneshot registry shape + with the forwarded handler owning the data-plane pump (R-02 held — + the handler's inline `pump_bidi` await tracks the plane) matches + the alktunnels 0.2.0 forwarded-op surface; the three-party + half-close deadlock note (the bridge completes on EITHER + direction's EOF) is the pinned non-obvious-correctness comment + done right. + +**8. Docs sync — two residues fixed (R-3, R-4).** The integration +task's reconciliation was verified: overview.md (module map, feature +gates, dependencies), consumer.md (the front doors, the TokenClient +door-only contract, the BIND take-guard), producer.md §Wiring, and +README.md Current State all match the tree. Residues fixed: +producer.md's local-backend section described the landed shape as a +channel pipe (`copy_bidirectional` onto a `BiStream`) — the landed +`ProducerDoor` serves the RFC conversation on the accepted socket +directly via `serve_socks5_over` (no channel in the path; rewritten, +including the `bind_policy` gate note the docs omitted); +`associate_with_token`'s doc claimed "the TCP door uses this" — the +landed TCP door refuses ASSOCIATE before any session (reworded); +`SessionOwner`'s docs claimed it is "shared with the composed +accept" — the accept uses its own `SyncHalves`/`OwnedWrite` pair +(reworded). No other spec drift. ## Acceptance Criteria -- [ ] The never-binds posture verified (protocol layer + default +- [x] The never-binds posture verified (protocol layer + default graph) -- [ ] Seam-contract compliance verified for every provider -- [ ] Door fidelity verified (both variants + the UDP boundary) -- [ ] The graduation's shape verified (no stand-ins) -- [ ] Convention compliance swept -- [ ] Feature-gate hygiene verified (all four verification targets) -- [ ] POC-fidelity spot-check of the trickiest assembly behaviors -- [ ] Docs sync verified; findings recorded; the gate to release +- [x] Seam-contract compliance verified for every provider +- [x] Door fidelity verified (both variants + the UDP boundary) +- [x] The graduation's shape verified (no stand-ins) +- [x] Convention compliance swept +- [x] Feature-gate hygiene verified (all four verification targets) +- [x] POC-fidelity spot-check of the trickiest assembly behaviors +- [x] Docs sync verified; findings recorded; the gate to release polish opens +### Findings + +- **R-1 (major → fixed in the review change)**: the UDP front door's + in-band decode (`recv_inband`, `src/door.rs`) created a fresh + `DatagramReader` per call and returned only the first frame of + each feed — a datagram split across channel chunks would have + restarted the reassembly from scratch per read, and datagrams + batched beyond the first in one chunk were **silently dropped**. + This is exactly POC #2's first-cut bug (ADR-003: "reader state + persists across calls"; `AssociateSession::recv_datagram` owns the + pinned shape) — the door bridge did not carry it, and its own doc + comment claimed persistence. Unexercised by the suite (every + landed test drives one frame per chunk; the failure needs two + frames coalescing into one read — a burst shape real echo targets + produce under load). Fix: an owning `InbandBridge` (persistent + `DatagramReader` + batched-datagram pending queue, the + `AssociateSession` shape) threaded through the channel→socket + pump, plus the fail-loud mid-frame EOF (never a silent partial + datagram). Pinned by three unit tests on the bridge (batched + frames queue-not-drop; partial-frame state persists across calls + — the test fails against a per-call reader; EOF mid-frame errors + loud) and an end-to-end door-udp test with a doubled-echo egress + (`batched_in_band_replies_both_reach_the_vanilla_client`). +- **R-2 (convention, fixed in the review change)**: seven non-test + `expect`s landed with batch 2 (convention 2's rule is the rule; + batch-1's R-2 precedent applies): the `FrontDoor`/`UdpFrontDoor` + builders' missing-`.bind` panics → `DoorBuildError` + (`MissingBindAddr`, `.build()` is now `Result`); the + `ProducerDoor` builder's missing-`.producer` panic and `serve`'s + missing-`.bind` panic → `ProducerDoorBuildError`/`ProducerDoorError` + (the bind decision is `serve`'s typed error, the io error is + `#[source]`-wrapped); `ComposedEgress::entry_for`'s + inserted-above invariant → an `EgressError::General` structural + diagnostic; `register_forwarded_serving`'s registration panic → + `Result` (the underlying alkcall + diagnostic surfaced, `# Panics` → `# Errors`); + `TokenClient::wrap`'s call-connection-taken panic → + `TokenClientError::CallConnectionTaken` (a client is one calling + surface — the door-only contract). Test call sites updated + (`.expect("the door builds")` etc. — test-side, allowed); the new + error types are re-exported (`lib.rs`). +- **R-3 (doc residue, fixed)**: producer.md's producer-side + local-backend section described the landed shape as a raw channel + pipe (`copy_bidirectional` onto a `BiStream`, "the consumer door's + code shape") — the landed `ProducerDoor` hands each accepted + socket to `serve_socks5_over` (the RFC conversation runs on the + socket; no channel pipe). Reconciled, including the + `bind_policy`-is-this-path's-BIND-gate note and the cred-bearing + follow-up posture (not landed). +- **R-4 (doc residue, fixed)**: `associate_with_token`'s doc claimed + the TCP door calls it (it refuses ASSOCIATE before any session); + `SessionOwner`'s module + type docs claimed the composed accept + shares it (the accept uses `SyncHalves`/`OwnedWrite`). Both + reworded to the landed reality. +- **R-5 (accepted, no fix demanded)**: the end-to-end + `batched_in_band_replies...` test does not deterministically + exercise the batching (the two echo frames may ride separate + channel chunks — timing-dependent); the deterministic pins are the + bridge unit tests (which demonstrably fail against the per-call + reader). The end-to-end test still guards the regression class + through the real stack; the unit tests carry the determinism. + ## References - docs/architecture/overview.md, wire.md, producer.md, consumer.md @@ -88,8 +298,4 @@ Review against the architecture, in this order: fidelity sources) - AGENTS.md (conventions 1–17; the verification commands) - The batch-1 review precedent: - `tasks/implementation/review-protocol-core.md` - -## Summary - -> To be filled on completion \ No newline at end of file + `tasks/implementation/review-protocol-core.md` \ No newline at end of file diff --git a/tests/assembly-integration.rs b/tests/assembly-integration.rs index 1e573b7..b2f5d4e 100644 --- a/tests/assembly-integration.rs +++ b/tests/assembly-integration.rs @@ -389,7 +389,8 @@ async fn wire_composed_suite() -> ComposedSuite { ), &topo.opener_registry, alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), - ); + ) + .expect("register forwarded openable"); // The SOCKS5 producer: the composed accept seam (BIND) + the // composed direct-op egress (ASSOCIATE) + a real TCP dial @@ -504,7 +505,8 @@ async fn full_dash_d_story_one_registration_three_commands_through_the_doors() { let door = FrontDoor::builder() .credentials(cred_table()) .bind(door_addr) - .build(); + .build() + .expect("the door builds"); let _tcp_serve = tokio::spawn({ let client = Arc::clone(&tcp_topo.consumer); async move { @@ -515,7 +517,10 @@ async fn full_dash_d_story_one_registration_three_commands_through_the_doors() { }); let udp_relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); let udp_relay_addr = udp_relay.local_addr().expect("relay addr"); - let udp_door = UdpFrontDoor::builder().bind(udp_relay_addr).build(); + let udp_door = UdpFrontDoor::builder() + .bind(udp_relay_addr) + .build() + .expect("the door builds"); let _udp_serve = tokio::spawn({ let client = Arc::clone(&udp_topo.consumer); async move { @@ -725,7 +730,10 @@ async fn door_client_disconnect_tears_the_association_down_no_leaks() { let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); let relay_addr = relay.local_addr().expect("relay addr"); - let door = UdpFrontDoor::builder().bind(relay_addr).build(); + let door = UdpFrontDoor::builder() + .bind(relay_addr) + .build() + .expect("the door builds"); let serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(relay, client).await.expect("the door serves") } @@ -780,7 +788,10 @@ async fn door_client_disconnect_tears_the_association_down_no_leaks() { // poisoning). let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2"); let relay2_addr = relay2.local_addr().expect("relay addr 2"); - let door2 = UdpFrontDoor::builder().bind(relay2_addr).build(); + let door2 = UdpFrontDoor::builder() + .bind(relay2_addr) + .build() + .expect("the door builds"); let _serve2 = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -963,7 +974,8 @@ async fn mixed_assembly_channels_native_and_door_clients_concurrent() { let door = FrontDoor::builder() .credentials(cred_table()) .bind(door_addr) - .build(); + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&door_topo.consumer); async move { @@ -974,7 +986,10 @@ async fn mixed_assembly_channels_native_and_door_clients_concurrent() { }); let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); let relay_addr = relay.local_addr().expect("relay addr"); - let udp_door = UdpFrontDoor::builder().bind(relay_addr).build(); + let udp_door = UdpFrontDoor::builder() + .bind(relay_addr) + .build() + .expect("the door builds"); let _udp_serve = tokio::spawn({ let client = Arc::clone(&udp_door_topo.consumer); async move { @@ -1277,7 +1292,8 @@ async fn producer_door_one_registration_three_commands() { let door_addr = listener.local_addr().expect("door addr"); let door = alksocks::local::ProducerDoor::builder() .producer(producer) - .build(); + .build() + .expect("the door builds"); let _guard = door.serve_on(listener).await; // CONNECT (live). @@ -1446,7 +1462,10 @@ async fn dash_d_domain_targets_resolve_at_the_producer_seam() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let door_addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().bind(door_addr).build(); + let door = FrontDoor::builder() + .bind(door_addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { diff --git a/tests/common/mod.rs b/tests/common/mod.rs index b119875..f0dadf0 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -148,6 +148,57 @@ pub fn echo_egress_factory() -> EgressFactory { }) } +/// A doubled-echo egress factory: every datagram echoes back **twice +/// back-to-back** (two frames queued as one reply burst — the +/// producer's frame writer puts both length-prefixed frames on the +/// channel; the consumer's read may batch them into one chunk). The +/// reader-state probe for the front doors' bridge (review 003 R-1): +/// a bridge that re-creates its decoder per read loses the second +/// datagram; the landed bridge queues it. +pub fn doubled_echo_egress_factory() -> EgressFactory { + alksocks::egress::egress_factory(|| { + let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel::(); + Ok(Box::new(DoubledEchoEgress { + inbound, + outbound: Mutex::new(outbound), + }) as Box) + }) +} + +/// The doubled echo behind [`doubled_echo_egress_factory`] (the +/// [`QueueEgress`] shape with the echo count two). +pub struct DoubledEchoEgress { + pub inbound: tokio::sync::mpsc::UnboundedSender, + pub outbound: Mutex>, +} + +impl Egress for DoubledEchoEgress { + fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { + let mut inner = addr + .to_be_bytes() + .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; + inner.extend_from_slice(&payload); + let inner = Bytes::from(inner); + self.inbound + .send(inner.clone()) + .map_err(|_| EgressError::General("egress closed".into()))?; + self.inbound + .send(inner) + .map_err(|_| EgressError::General("egress closed".into())) + }) + } + + fn next_reply(&mut self) -> BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { + match self.outbound.lock().await.recv().await { + Some(inner) => Ok(Some(inner)), + None => Err(EgressError::General("egress closed".into())), + } + }) + } +} + /// A failing egress factory (the ASSOCIATE-refusal probe): refuses /// with the given error, which the producer maps to reply#1's code. /// The variant is reconstructed per call (the seam errors are not diff --git a/tests/composed-accept.rs b/tests/composed-accept.rs index 5c50a19..23d161b 100644 --- a/tests/composed-accept.rs +++ b/tests/composed-accept.rs @@ -67,7 +67,8 @@ async fn wire_suite_with_identity(identity: alkcall::core::auth::Identity) -> Su ), &topo.opener_registry, alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), - ); + ) + .expect("register forwarded openable"); // The SOCKS5 openable registers post-hoc (W2's shape) on the // producer's registry — with the REAL composed factory, which // closes over the opener's client + the correlation registry (both diff --git a/tests/door-tcp.rs b/tests/door-tcp.rs index 527e241..ac4a39d 100644 --- a/tests/door-tcp.rs +++ b/tests/door-tcp.rs @@ -222,7 +222,10 @@ async fn noauth_door_raw_pipe_vanilla_connect_round_trip() { // the door's own bind). let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().bind(addr).build(); + let door = FrontDoor::builder() + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -288,7 +291,11 @@ async fn cred_door_good_creds_connect_round_trip_with_mapped_identity() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let door = FrontDoor::builder() + .credentials(table) + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -368,7 +375,11 @@ async fn cred_door_bad_creds_rejected_no_channel_opened() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let door = FrontDoor::builder() + .credentials(table) + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -445,7 +456,11 @@ async fn cred_door_never_pipes_raw_post_auth() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let door = FrontDoor::builder() + .credentials(table) + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -518,7 +533,11 @@ async fn bind_door_vanilla_conversation_reply_fidelity() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let door = FrontDoor::builder() + .credentials(table) + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -601,7 +620,8 @@ async fn idle_first_byte_timeout_closes_a_silent_connection() { let door = FrontDoor::builder() .bind(addr) .idle_first_byte(Duration::from_millis(300)) - .build(); + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -658,7 +678,8 @@ async fn session_cap_refuses_beyond_the_bound() { .bind(addr) .idle_first_byte(Duration::from_secs(30)) .max_sessions(1) - .build(); + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -723,7 +744,10 @@ async fn per_connection_failure_isolation() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().bind(addr).build(); + let door = FrontDoor::builder() + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -776,7 +800,10 @@ async fn native_sessions_work_beside_a_noauth_door() { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("door addr"); - let door = FrontDoor::builder().bind(addr).build(); + let door = FrontDoor::builder() + .bind(addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -822,8 +849,11 @@ async fn bind_session_with_token_round_trip() { ) .await; - let token_client = - Arc::new(alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref()).await); + let token_client = Arc::new( + alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref()) + .await + .expect("the door's token surface wraps"), + ); let mut session = BindSession::bind_with_token( token_client.as_ref(), ("bind.token", 0u16).to_target_addr().unwrap(), diff --git a/tests/door-udp.rs b/tests/door-udp.rs index ebef714..f29471b 100644 --- a/tests/door-udp.rs +++ b/tests/door-udp.rs @@ -100,7 +100,10 @@ async fn udp_door_vanilla_client_round_trip_and_empty_datagram() { let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); let relay_addr = relay.local_addr().expect("relay addr"); let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); - let door = UdpFrontDoor::builder().bind(relay_addr).build(); + let door = UdpFrontDoor::builder() + .bind(relay_addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(relay, client).await.expect("the door serves") } @@ -165,6 +168,65 @@ async fn udp_door_vanilla_client_round_trip_and_empty_datagram() { ); } +/// The reader-state pin through the door's bridge (review 003 R-1, +/// the POC #2 finding carried to the doors): two replies arriving +/// back-to-back (the doubled echo — both frames ride the producer's +/// frame writer; the door's channel read may batch them into one +/// chunk) both reach the vanilla client. A bridge that re-created +/// its decoder per read lost the second datagram. +#[tokio::test] +async fn batched_in_band_replies_both_reach_the_vanilla_client() { + let echo_addr = udp_echo_server().await; + let topo = harness::wire_consumer_side_with( + echo_dial(), + Some(harness::doubled_echo_egress_factory()), + None, + Some(harness::consumer_identity()), + ) + .await; + + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let relay_addr = relay.local_addr().expect("relay addr"); + let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + let door = UdpFrontDoor::builder() + .bind(relay_addr) + .build() + .expect("the door builds"); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { door.serve_on(relay, client).await.expect("the door serves") } + }); + + // One datagram in → two echo datagrams out (both from the relay + // socket, both carrying the target as their source address — the + // bridge's pending queue drained in order, none lost). + client + .send_to(&rfc_datagram(echo_addr, b"once"), relay_addr) + .await + .expect("send"); + let mut rx = vec![0u8; 65535]; + for _ in 0..2 { + let (n, from) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("reply within bound") + .expect("recv"); + assert_eq!(from, relay_addr, "the reply comes from the relay socket"); + let (addr, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(addr, echo_addr); + assert_eq!(payload, b"once"); + } + + // No third reply: exactly the two echoes arrived (the burst was + // queued, not duplicated). + let mut probe = vec![0u8; 65535]; + assert!( + timeout(Duration::from_millis(300), client.recv_from(&mut probe)) + .await + .is_err(), + "exactly two replies arrived" + ); +} + /// FRAG ≠ 0 is refused at the door (the v1 posture): the fragment is /// dropped — nothing is egressed — and the association stays healthy /// (a following normal datagram round-trips). @@ -182,7 +244,10 @@ async fn frag_nonzero_refused_and_association_stays_healthy() { let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); let relay_addr = relay.local_addr().expect("relay addr"); let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); - let door = UdpFrontDoor::builder().bind(relay_addr).build(); + let door = UdpFrontDoor::builder() + .bind(relay_addr) + .build() + .expect("the door builds"); let _serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(relay, client).await.expect("the door serves") } @@ -253,7 +318,8 @@ async fn idle_bound_closes_an_idle_association_and_a_fresh_one_succeeds() { let door = UdpFrontDoor::builder() .bind(relay.local_addr().unwrap()) .idle(Duration::from_millis(300)) - .build(); + .build() + .expect("the door builds"); let serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(relay, client).await.expect("the door serves") } @@ -285,7 +351,10 @@ async fn idle_bound_closes_an_idle_association_and_a_fresh_one_succeeds() { // is healthy — the POC's strongest observable). let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2"); let relay2_addr = relay2.local_addr().unwrap(); - let door2 = UdpFrontDoor::builder().bind(relay2_addr).build(); + let door2 = UdpFrontDoor::builder() + .bind(relay2_addr) + .build() + .expect("the door builds"); let _serve2 = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { @@ -332,7 +401,8 @@ async fn fresh_association_succeeds_after_teardown() { let door = UdpFrontDoor::builder() .bind(relay_addr) .idle(Duration::from_millis(200)) - .build(); + .build() + .expect("the door builds"); let serve = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { door.serve_on(relay, client).await.expect("the door serves") } @@ -358,7 +428,10 @@ async fn fresh_association_succeeds_after_teardown() { // A fresh association on the same topology succeeds end to end. let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2"); let relay2_addr = relay2.local_addr().expect("relay2 addr"); - let door2 = UdpFrontDoor::builder().bind(relay2_addr).build(); + let door2 = UdpFrontDoor::builder() + .bind(relay2_addr) + .build() + .expect("the door builds"); let serve2 = tokio::spawn({ let client = Arc::clone(&topo.consumer); async move { diff --git a/tests/harness.rs b/tests/harness.rs index a28cf34..760d328 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -77,8 +77,8 @@ pub mod common; #[allow(unused_imports)] pub use common::{ - datagram_frame as udp_frame, decode_datagram as udp_unframe, echo_dial, - echo_dial_witnessed as witnessing_echo_dial, echo_egress_factory as echo_egress, + datagram_frame as udp_frame, decode_datagram as udp_unframe, doubled_echo_egress_factory, + echo_dial, echo_dial_witnessed as witnessing_echo_dial, echo_egress_factory as echo_egress, parked_accept_factory, raw_greet, raw_read_reply, raw_request, read_reply_v4, refusing_dial, refusing_egress, write_request, QueueEgress, }; diff --git a/tests/local-backend-producer.rs b/tests/local-backend-producer.rs index a6d558c..4f2a7a7 100644 --- a/tests/local-backend-producer.rs +++ b/tests/local-backend-producer.rs @@ -80,7 +80,8 @@ async fn serve( .producer(producer) .idle_first_byte(idle) .max_sessions(cap) - .build(); + .build() + .expect("the door builds"); let guard = door.serve_on(listener).await; (addr, guard) }