From 3128d2bd4f15c5f64490b8af1f01199da8994667 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 25 Sep 2026 14:53:18 +0000 Subject: [PATCH] =?UTF-8?q?composed:=20the=20direct-op=20egress=20?= =?UTF-8?q?=E2=80=94=20the=20lazy=20destination=E2=86=92tunnel=20table?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `composed` feature's ASSOCIATE egress (ADR-008's composed variant, ADR-009's gate): `composed::direct_egress` — each destination dials through an alktunnels 0.2.0 direct-op channel (`TunnelSession::open_direct`, `Substrate::Udp`), per-target ACL for free, the POC #2 `ComposedEgress` shape ported to the landed seam types. - `Cargo.toml` — `composed = ["dep:alktunnels"]` (alktunnels pinned "0.2.0"; rides `composed` alone, `cargo tree --edges normal` spot-checked); dev-dep for the composed suite. - `src/composed/mod.rs` — the lazy destination→tunnel table (LRU-capped, `DEFAULT_MAX_TUNNELS` = 256), per-tunnel relay tasks draining a shared inbound queue, `SessionOwner` (the owner wrapper shared with the composed accept — the batch-1 review's Sync-bound wrinkle), the readiness contract (real async park; ended relays → association-end `Err`, never a hot `Ok(None)`), framing-once, domain pass-through unresolved (ADR-012), the backpressure bound documented on the constructor. - Two deliberate deviations from the POC shape, documented in the task's Notes: the closed-queue condition is a live-relay count (the egress holds its own `Sender`, so the POC's queue-close arm was unreachable through this surface); the tunnel codec is alktunnels' (`[len]` + payload), not the RFC-framed reader. - `tests/composed-egress.rs` — round-trips per destination → one tunnel each, table reuse, LRU eviction/re-dial, the readiness park, the closed-relay-set `Err`, framing-once through the wire layer, domain pass-through — over a `wire_tunnel_direct` harness topology (the consumer-side shape with the direct op on the producer's serving registry). Verification: `cargo test` + `--all-features` pass (13 suites); `cargo check --features composed --target wasm32-unknown-unknown` clean (the hard gate); clippy `--all-targets --all-features -- -D warnings` clean; `cargo fmt --check` clean; `cargo doc --no-deps` clean. --- Cargo.toml | 12 +- src/composed/mod.rs | 406 ++++++++++++++++++++++++ src/lib.rs | 18 +- tasks/implementation/composed-egress.md | 72 ++++- tests/composed-egress.rs | 406 ++++++++++++++++++++++++ tests/harness.rs | 101 ++++++ 6 files changed, 998 insertions(+), 17 deletions(-) create mode 100644 src/composed/mod.rs create mode 100644 tests/composed-egress.rs diff --git a/Cargo.toml b/Cargo.toml index e9f1710..03ccd21 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,7 @@ name = "alksocks" # is protocol-only and wasm-clean by construction. default = [] local = ["tokio/net"] -composed = [] +composed = ["dep:alktunnels"] [dependencies] # The substrate (convention 11): pinned, bump deliberately, never @@ -36,7 +36,15 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" bytes = "1" futures = { version = "0.3", default-features = false, features = ["std"] } +# The sibling crate (the composed egress's direct-op vehicle — ADR-008/ +# ADR-009): rides `composed` alone. Pinned; bump deliberately +# (convention 11's deliberate-bump rule applies to siblings too). +alktunnels = { version = "0.2.0", optional = true } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util", "macros"] } -tracing-subscriber = { version = "0.3", features = ["env-filter"] } \ No newline at end of file +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +# The composed suite's topology (the alktunnels direct-op producer) — +# dev-only: the test binary gates itself on the `composed` feature +# (`#![cfg(feature = "composed")]`), the dev-dep keeps it compiling. +alktunnels = "0.2.0" diff --git a/src/composed/mod.rs b/src/composed/mod.rs new file mode 100644 index 0000000..8a48712 --- /dev/null +++ b/src/composed/mod.rs @@ -0,0 +1,406 @@ +//! The `composed` feature's egress provider (ADR-008's composed +//! variants, ADR-009's feature gate): [`direct_egress`] — the UDP +//! ASSOCIATE egress that dials each destination through an alktunnels +//! 0.2.0 direct-op channel ([`alktunnels::TunnelSession::open_direct`], +//! `Substrate::Udp`). +//! +//! **Per-target ACL for free** (ADR-008's composed benefit): each +//! destination rides its own `channels/tunnel/direct`-governed channel +//! — the producer's `tunnel:direct` scope gate and dial policy apply +//! per destination, no SOCKS5-layer allowlist needed (ADR-006/ADR-012: +//! whatever ACL governs the tunnel capability governs everything +//! reachable through it). +//! +//! The shape is POC #2's `ComposedEgress` +//! (`/workspace/alksocks-udp-poc/src/egress.rs`), ported to the landed +//! seam types: a lazy destination→tunnel table — each new destination +//! opens one UDP-tunnel channel, a relay task per tunnel drains +//! inbound datagrams into a shared queue, the write half stays in the +//! table (LRU-capped; dropping a write half EOFs the tunnel +//! producer-side; the entry's session drop reaps the channel — +//! alktunnels ADR-005's teardown contract). +//! +//! Contracts (the landed seam types' normative docs, honored here): +//! +//! - **The readiness contract** (load-bearing, `Egress::next_reply`): +//! parks on the shared inbound queue (`recv().await` — `Pending` +//! until a relay task delivers); a closed queue (all relays ended) +//! returns `Err` — never a hot `Ok(None)`. The stage's zero-yield +//! defense (review 002 R-1) is the backstop, not the design. +//! - **Framing-once**: replies carry *inner* bytes (`addr + data` — +//! the reply address is the entry's recorded destination, the RFC's +//! per-datagram source semantics); the length prefix is the wire +//! layer's, applied exactly once downstream. `send` takes the +//! payload only; the tunnel-side framing (`[len: u16 BE]`, the +//! alktunnels ADR-003 codec) is the tunnel wire's, not the RFC +//! wire's. +//! - **Per-datagram `send` is awaited inline** (the stage +//! interleaves sends with reply drains): a slow egress applies +//! backpressure to the association, not to other channels — the +//! batch-2 composed-egress author inherits this bound knowingly +//! (wire.md's note). The table's open cost rides the send path +//! (lazy open on first datagram per destination — the POC's +//! measured ~0.8 ms). +//! - **The resource/target mapping**: `open_direct` takes a +//! [`SubstrateAddr`] — the destination's [`TargetAddr`] renders to +//! it (IP forms direct; domain form passes through unresolved — +//! producer-side resolution, the ssh `-D` semantic; ADR-012). +//! +//! **The Sync-bound wrinkle** (the batch-1 review's seam contract vs +//! alktunnels' surface): the landed `TargetHandle`/`BindTargetHandle` +//! halves are `Box`, but +//! alktunnels' `TakenHalves` yields `Send + Unpin` only. 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 (the sibling task) wraps the accepted +//! halves in this wrapper. + +use std::collections::HashMap; +use std::sync::Arc; + +use alkcall::channels::client::ChannelClient; +use alktunnels::params::{Substrate, SubstrateAddr}; +use alktunnels::TunnelSession; +use bytes::Bytes; +use futures::future::BoxFuture; +use tokio::io::{AsyncRead, AsyncWrite}; + +use crate::egress::{Egress, EgressError, EgressFactory}; +pub use crate::rfc1928::util::target_addr::TargetAddr; +use crate::wire::MAX_INNER_LEN; + +/// The default LRU cap (the 256-channel default cap bounds fan-out — +/// ADR-010's posture; the cap is config, not wire). +pub const DEFAULT_MAX_TUNNELS: usize = 256; + +/// The destination's render into the direct op's [`SubstrateAddr`] (the +/// resource/target mapping): IP forms direct; domain form passes +/// through unresolved (producer-side resolution, the ssh `-D` +/// semantic — ADR-012). +pub fn substrate_addr_of(addr: &TargetAddr) -> SubstrateAddr { + match addr { + TargetAddr::Ip(sa) => SubstrateAddr::HostPort { + host: sa.ip().to_string(), + port: sa.port(), + }, + TargetAddr::Domain(domain, port) => SubstrateAddr::HostPort { + host: domain.clone(), + port: *port, + }, + } +} + +/// Render the destination to the table's key form (the string the LRU +/// orders by): `host:port` per [`SubstrateAddr`]'s render (IP literal +/// or unresolved domain — the domain stays unresolved, so domain and +/// IP destinations key distinctly, as the client addressed them). +pub fn key_of(addr: &TargetAddr) -> String { + match substrate_addr_of(addr) { + SubstrateAddr::HostPort { host, port } => format!("{host}:{port}"), + SubstrateAddr::Path { path } => path, + } +} + +/// 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 +/// `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 +/// (the session's drop reaps the channel entry — alktunnels ADR-005). +/// Whether the wrapper is `Sync` (the accept's need) is the half's +/// own `Sync`ness — asserted by the compile-shape test against the +/// alktunnels 0.2.0 surface; the egress needs only `Send` (the relay +/// task), which the wrapper always has. +pub struct SessionOwner { + half: H, + session: TunnelSession, +} + +impl AsyncRead for SessionOwner { + fn poll_read( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let this = std::pin::Pin::into_inner(self); + std::pin::Pin::new(&mut this.half).poll_read(cx, buf) + } +} + +impl SessionOwner { + /// Wrap the half with its session (the owner). + pub fn new(half: H, session: TunnelSession) -> Self { + Self { half, session } + } + + /// The wrapped half. + pub fn half(&self) -> &H { + &self.half + } + + /// The wrapped half, mutably. + pub fn half_mut(&mut self) -> &mut H { + &mut self.half + } + + /// Unwrap: the half plus the session (the owner), for a caller + /// that needs them separated again. + pub fn into_parts(self) -> (H, TunnelSession) { + (self.half, self.session) + } +} + +/// The composed egress: the lazy destination→tunnel table (the +/// per-association ASSOCIATE seam object). Constructed via +/// [`direct_egress`] (the `EgressFactory` the producer registration +/// consumes). +pub struct ComposedEgress { + client: Arc, + max_tunnels: usize, + entries: HashMap, + order: std::collections::VecDeque<(String, TargetAddr)>, + inbound: tokio::sync::mpsc::Receiver, + inbound_tx: tokio::sync::mpsc::Sender, + /// Live relay tasks — the readiness contract's closed-queue + /// condition: when every relay has ended (each reported via + /// [`Inbound::Ended`]) and the queue is drained, `next_reply` + /// returns `Err` (the association end), never a hot `Ok(None)`. + /// The factory `Sender` alone does not close the queue (the + /// egress holds it) — the count is the design. + live_relays: usize, +} + +/// The shared inbound queue's message: a tunnel's datagram (keyed by +/// the table key — the reply address rides the entry) or a relay's +/// end marker (the tunnel's EOF drained). +enum Inbound { + Datagram(String, Bytes), + Ended, +} + +/// One table entry: the tunnel's write half (the udp substrate's +/// framed half — `[len: u16 BE]` frames on the tunnel wire) + the +/// reply address to report for datagrams coming back through it + the +/// session (the channel-entry owner — the egress owns it, not a +/// handle payload; dropping the entry EOFs the write half and reaps +/// the channel — alktunnels ADR-005). +struct TableEntry { + _session: TunnelSession, + write: Box, + reply_addr: TargetAddr, +} + +/// The per-tunnel relay task: drain the tunnel's framed read half +/// into the egress's shared inbound queue. Exits on EOF (the tunnel +/// ended); the write half stays in the table until eviction or stage +/// exit. Framing decode is the tunnel codec's (the alktunnels +/// ADR-003) — the datagrams here are inner payloads (no RFC address +/// on the tunnel wire; the reply address is the entry's recorded +/// destination). +async fn relay_read( + mut read: impl AsyncRead + Send + Unpin, + tx: tokio::sync::mpsc::Sender, + key: String, +) { + use tokio::io::AsyncReadExt; + let mut reader = alktunnels::wire::DatagramReader::new(); + let mut chunk = vec![0u8; 16 * 1024]; + loop { + match read.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => { + for dg in reader.feed(&chunk[..n]) { + if tx.send(Inbound::Datagram(key.clone(), dg)).await.is_err() { + return; + } + } + } + } + } + let _ = tx.send(Inbound::Ended).await; +} + +/// Assemble the reply's inner bytes (`addr + data` — the entry's +/// recorded destination + the payload; framing-once: no length prefix +/// here, the wire layer's `frame_inner` adds it downstream). The +/// oversize guard mirrors the wire layer's frame bound. +fn inner_bytes(addr: &TargetAddr, payload: &[u8]) -> Result { + let mut inner = addr + .to_be_bytes() + .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; + inner.extend_from_slice(payload); + if inner.len() > MAX_INNER_LEN { + return Err(EgressError::General(format!( + "reply oversize: {} > {MAX_INNER_LEN}", + inner.len() + ))); + } + Ok(Bytes::from(inner)) +} + +impl ComposedEgress { + /// Build the egress (the per-association ASSOCIATE seam object). + /// `client` is the channels client the direct ops ride; the + /// constructor's config carries the LRU cap (`max_tunnels` — the + /// fan-out bound, config not wire; the producer's `tunnel:direct` + /// ACL governs which destinations the channels accept — the + /// per-target-ACL-for-free benefit). + pub fn new(client: Arc, max_tunnels: usize) -> Self { + let (tx, rx) = tokio::sync::mpsc::channel(64); + Self { + client, + max_tunnels, + entries: HashMap::new(), + order: std::collections::VecDeque::new(), + inbound: rx, + inbound_tx: tx, + live_relays: 0, + } + } + + /// The entry for `addr` — open lazily on first datagram (the + /// open cost rides the send path, per-datagram awaited inline). + async fn entry_for(&mut self, addr: &TargetAddr) -> Result<&mut TableEntry, EgressError> { + let key = key_of(addr); + if !self.entries.contains_key(&key) { + while self.entries.len() >= self.max_tunnels { + let Some((evict, _)) = self.order.pop_front() else { + break; + }; + self.entries.remove(&evict); + } + let target = substrate_addr_of(addr); + let session = TunnelSession::open_direct(self.client.as_ref(), target, Substrate::Udp) + .await + .map_err(|e| EgressError::Unreachable(format!("tunnel open `{key}`: {e}")))?; + let taken = session.take_halves(); + // The session stays INSIDE the entry: dropping it reaps + // the channel (ADR-005's teardown contract), which would + // EOF the relay task instantly (the POC's insert-time + // bug). The relay task owns the bare read half. + tokio::spawn(relay_read(taken.read, self.inbound_tx.clone(), key.clone())); + self.live_relays += 1; + self.entries.insert( + key.clone(), + TableEntry { + _session: taken.session, + write: taken.write, + reply_addr: addr.clone(), + }, + ); + } + self.order.retain(|(k, _)| k != &key); + self.order.push_back((key.clone(), addr.clone())); + Ok(self.entries.get_mut(&key).expect("entry inserted above")) + } +} + +impl Egress for ComposedEgress { + fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { + let entry = self.entry_for(&addr).await?; + use tokio::io::AsyncWriteExt; + let framed = alktunnels::wire::frame_datagram(&payload) + .map_err(|e| EgressError::General(format!("tunnel frame: {e}")))?; + let write: &mut (dyn AsyncWrite + Send + Unpin) = &mut *entry.write; + write + .write_all(&framed) + .await + .map_err(|e| EgressError::General(format!("tunnel write: {e}")))?; + write + .flush() + .await + .map_err(|e| EgressError::General(format!("tunnel flush: {e}")))?; + Ok(()) + }) + } + + fn next_reply(&mut self) -> BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { + loop { + // A non-blocking first pass over the shared queue (the + // relay tasks may have queued replies already), then a + // real async wait on the queue (`Pending` until a + // relay task delivers — the readiness contract). + loop { + match self.inbound.try_recv() { + Ok(Inbound::Datagram(key, dg)) => { + let addr = match self.entries.get(&key) { + Some(entry) => entry.reply_addr.clone(), + None => continue, + }; + return Ok(Some(inner_bytes(&addr, &dg)?)); + } + // A relay ended (its tunnel's EOF drained) — + // count it; the closed-queue condition is + // every relay ended. + Ok(Inbound::Ended) => { + self.live_relays = self.live_relays.saturating_sub(1); + continue; + } + Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break, + Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break, + } + } + if self.live_relays == 0 { + // Every relay ended (the tunnels ended): a closed + // relay set must NOT spin the stage's select (an + // instantly-ready None would starve the channel + // branch) — end the association instead. + return Err(EgressError::General("all tunnel relays ended".into())); + } + match self.inbound.recv().await { + Some(Inbound::Datagram(key, dg)) => { + let addr = match self.entries.get(&key) { + Some(entry) => entry.reply_addr.clone(), + None => continue, + }; + return Ok(Some(inner_bytes(&addr, &dg)?)); + } + // A relay ended while parked — loop back: the + // count check above decides whether the + // association ends (never a hot `Ok(None)`). + Some(Inbound::Ended) => { + self.live_relays = self.live_relays.saturating_sub(1); + continue; + } + None => { + return Err(EgressError::General("all tunnel relays ended".into())); + } + } + } + }) + } +} + +/// The composed egress constructor (ADR-008's composed variant, +/// producer.md §Wiring): the `EgressFactory` the producer +/// registration consumes. One per ASSOCIATE; the stage owns the +/// returned egress; stage exit drops it (write halves dropped = EOF +/// to the tunnel producers; the entries' sessions reap the channels — +/// alktunnels ADR-005's teardown contract). +/// +/// # Backpressure (inherited knowingly) +/// +/// Per-datagram `send` is awaited inline by the producer stage (it +/// interleaves sends with reply drains): a slow egress applies +/// backpressure to the association, not to other channels. The lazy +/// open's cost (~0.8 ms per new destination in the POC's +/// measurement) rides the send path. +pub fn direct_egress(client: Arc, max_tunnels: usize) -> EgressFactory { + Arc::new(move || { + let client = Arc::clone(&client); + Box::pin(async move { + Ok(Box::new(ComposedEgress::new(client, max_tunnels)) as Box) + }) + }) +} diff --git a/src/lib.rs b/src/lib.rs index 3f78411..3944e07 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -43,10 +43,12 @@ //! channel's teardown; no `Clone`. //! //! `local` (the socket-backed assembly-layer providers — -//! [`local::tcp_dial`] / [`local::udp_egress`] / [`local::bind_accept`], -//! ADR-008/009) is feature-gated: the default crate is protocol-only and -//! wasm-clean by construction, and the composed providers (`composed`) -//! are still pending. +//! [`crate::local::tcp_dial`] / [`crate::local::udp_egress`] / +//! [`crate::local::bind_accept`], ADR-008/009) is feature-gated: the +//! default crate is protocol-only and wasm-clean by construction, and +//! the composed providers ride `composed` +//! ([`crate::composed::direct_egress`] — the alktunnels direct-op +//! egress, ADR-008/009). //! //! Architecture: `docs/architecture/overview.md` (spec set: //! wire.md, producer.md, consumer.md, rfc1928.md); the cross-cutting @@ -68,6 +70,9 @@ pub mod wire; #[cfg(feature = "local")] pub mod local; +#[cfg(feature = "composed")] +pub mod composed; + #[cfg(feature = "local")] pub mod door; @@ -116,6 +121,11 @@ pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress}; #[cfg(feature = "local")] pub use local::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard}; +#[cfg(feature = "composed")] +pub use composed::{ + direct_egress, key_of, substrate_addr_of, ComposedEgress, SessionOwner, DEFAULT_MAX_TUNNELS, +}; + #[cfg(feature = "local")] pub use consumer::TokenClient; #[cfg(feature = "local")] diff --git a/tasks/implementation/composed-egress.md b/tasks/implementation/composed-egress.md index d28808f..9569ac9 100644 --- a/tasks/implementation/composed-egress.md +++ b/tasks/implementation/composed-egress.md @@ -1,7 +1,7 @@ --- id: implementation/composed-egress name: Composed — the direct-op egress (lazy destination→tunnel table) -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -84,20 +84,20 @@ Deliverables: ## Acceptance Criteria -- [ ] `cargo test` passes; `cargo test --all-features` passes -- [ ] `cargo check --features composed --target wasm32-unknown- +- [x] `cargo test` passes; `cargo test --all-features` passes +- [x] `cargo check --features composed --target wasm32-unknown- unknown` clean — the composed graph is wasm-clean (hard gate) -- [ ] `cargo clippy --all-targets --all-features -- -D warnings` +- [x] `cargo clippy --all-targets --all-features -- -D warnings` clean; `cargo fmt --check` clean -- [ ] `alktunnels` pinned `"0.2.0"`; the dep rides `composed` alone +- [x] `alktunnels` pinned `"0.2.0"`; the dep rides `composed` alone (default graph unchanged — `cargo tree` spot-check) -- [ ] The lazy table + LRU eviction land per the POC shape — pinned +- [x] The lazy table + LRU eviction land per the POC shape — pinned by test -- [ ] The readiness contract honored (real async park; closed queue +- [x] The readiness contract honored (real async park; closed queue → `Err`, never a hot `Ok(None)`) — pinned by test -- [ ] Framing-once (replies are inner bytes; no double prefix) — +- [x] Framing-once (replies are inner bytes; no double prefix) — pinned by test -- [ ] Per-datagram send backpressure documented on the constructor +- [x] Per-datagram send backpressure documented on the constructor ## References @@ -116,8 +116,58 @@ Deliverables: ## Notes -> To be filled by implementation agent +> Landed as `src/composed/mod.rs` (feature `composed`) + +> `tests/composed-egress.rs`, with a `wire_tunnel_direct` topology +> added to `tests/harness.rs` (the consumer-side topology with +> alktunnels' direct op on the producer's serving registry — the +> composed egress's `ChannelClient` is the direct-op caller). +> +> Two deviations from the POC shape, both deliberate: +> +> - **The closed-queue condition is a live-relay count, not a queue +> close.** The egress holds its own factory `Sender`, so the shared +> `mpsc` queue can never fully disconnect — `recv()` would never +> return `None` and the POC's closed-queue arm was unreachable +> through this surface. Relays instead emit an `Ended` marker on +> their tunnel's EOF; `next_reply` counts them (`live_relays`) and +> returns the association-end `Err` when the count reaches zero and +> the queue is drained. The readiness contract is unchanged (the +> wait is a real async park; a drained-but-live relay set parks). +> - **The tunnel codec is alktunnels' `wire::DatagramReader` +> (`[len]` + payload), not this crate's RFC-framed one** — the POC +> fed the tunnel read half into the SOCKS wire reader (a bug class +> the round-trip test caught during bring-up); the composed egress +> frames the tunnel wire once with the tunnel codec and the RFC +> wire once downstream (framing-once holds at both boundaries). +> +> The Sync-bound wrinkle resolves as designed: the shared +> `SessionOwner` wrapper (half + session in one owner) is exported +> for the composed accept (the sibling task); the egress itself owns +> the session in the table entry (the POC's shape) and the relay task +> owns the bare read half. +> +> `cargo tree --edges normal` spot-check: `alktunnels` rides +> `composed` alone (absent from the default normal graph; the +> dev-dep copy is dev-only). ## Summary -> To be filled on completion \ No newline at end of file +> The composed egress landed: `composed = ["dep:alktunnels"]` +> (alktunnels pinned "0.2.0"; dep rides `composed` alone), +> `src/composed/mod.rs` — `direct_egress` (the `EgressFactory`), the +> lazy destination→tunnel table with LRU eviction at cap (`DEFAULT_ +> MAX_TUNNELS` = 256), per-tunnel relay tasks draining into a shared +> inbound queue, `SessionOwner` (the owner wrapper shared with the +> composed accept), the readiness contract (real async park; ended +> relays → association-end `Err`, never a hot `Ok(None)`), framing- +> once, per-target ACL via the direct op, domain pass-through +> unresolved (ADR-012), and the backpressure bound documented on the +> constructor. `tests/composed-egress.rs` (6 tests over the +> `wire_tunnel_direct` harness topology): round-trips per +> destination → one tunnel each, table reuse, LRU eviction/re-dial, +> the readiness park, the closed-relay-set `Err`, framing-once +> through the wire layer, and the domain pass-through. Verification: +> `cargo test` (default) + `--all-features` pass; the wasm check with +> `--features composed` clean; clippy (`--all-targets --all-features +-- -D warnings`) clean; `cargo fmt --check` clean; `cargo doc +> --no-deps` clean. \ No newline at end of file diff --git a/tests/composed-egress.rs b/tests/composed-egress.rs new file mode 100644 index 0000000..439bcb4 --- /dev/null +++ b/tests/composed-egress.rs @@ -0,0 +1,406 @@ +//! The composed egress integration tests (ADR-008's composed variant, +//! POC #2's `ComposedEgress` shape, end to end over channels): a +//! tunnel-producing peer (the alktunnels registry shape — +//! `register_tunnel_direct_openable` over a UDP echo dial), the +//! composed egress against it, datagram round-trips through the +//! tunnel (multiple destinations → multiple tunnels), the LRU +//! eviction at cap, the readiness contract re-pinned, the +//! closed-queue → association-end shape, and framing-once. +//! +//! The topology: the SOCKS5-style harness (consumer calls the +//! producer's open ops) carrying alktunnels' direct op — the producer +//! side registers `register_tunnel_direct_openable` (the +//! alktunnels 0.2.0 registry shape), the consumer's +//! `ChannelClient` is what the composed egress rides +//! (`TunnelSession::open_direct`). + +#![cfg(feature = "composed")] + +mod harness; + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use alktunnels::params::Substrate; +use alktunnels::producer::{ + DialFn as TunnelDialFn, TargetHandle as TunnelTargetHandle, TunnelEstablishError, +}; +use alktunnels::wire::DatagramReader as TunnelDatagramReader; +use bytes::Bytes; +use tokio::io::{AsyncRead, AsyncWriteExt}; + +use alksocks::composed::{direct_egress, ComposedEgress}; +use alksocks::egress::{Egress, EgressError}; +use alksocks::rfc1928::util::target_addr::TargetAddr; + +/// An echo dial closure over the alktunnels direct-op `DialFn` shape: +/// an in-process framed UDP pipe per dial (the framed-adapter shape — +/// the "target" decodes the tunnel codec and echoes each datagram). +/// Counts the dialed destinations (the multiple-tunnels assertion). +fn counting_udp_echo_dial(dial_count: Arc) -> TunnelDialFn { + let dial_count = Arc::clone(&dial_count); + Arc::new(move |substrate: Substrate, backing: &str| { + let dial_count = Arc::clone(&dial_count); + let _ = backing.to_string(); + Box::pin(async move { + if substrate != Substrate::Udp { + return Err(TunnelEstablishError::DialFailed(format!( + "non-udp dial not wired: {substrate:?}" + ))); + } + dial_count.fetch_add(1, Ordering::SeqCst); + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + let (mut t_read, mut t_write) = tokio::io::split(target_side); + let mut reader = TunnelDatagramReader::new(); + let mut buf = vec![0u8; 16 * 1024]; + loop { + let n = match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => n, + }; + for dg in reader.feed(&buf[..n]) { + let framed = match alktunnels::wire::frame_datagram(&dg) { + Ok(f) => f, + Err(_) => return, + }; + if t_write.write_all(&framed).await.is_err() { + return; + } + let _ = t_write.flush().await; + } + } + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + Ok(TunnelTargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }) + }) + }) +} + +/// Wire the topology: the producer registers the alktunnels direct op +/// (over the given dial) on its serving registry; the consumer's +/// `ChannelClient` is the direct-op caller the composed egress rides. +async fn wire_direct(dial: TunnelDialFn) -> harness::TunnelDirectTopology { + harness::wire_tunnel_direct(dial).await +} + +/// The destinations the tests dial (distinct table keys). +fn dest(port: u16) -> TargetAddr { + TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 1], port))) +} + +/// Build the egress against the topology (the consumer's client is +/// the direct-op caller). +async fn egress_for( + topo: &harness::TunnelDirectTopology, + max_tunnels: usize, +) -> Box { + let factory = direct_egress(Arc::clone(&topo.consumer), max_tunnels); + factory().await.expect("egress builds") +} + +/// A datagram round-trip through the tunnel: send to `addr`, expect +/// the echo back (the reply carries `addr + data` — inner bytes). +async fn round_trip(egress: &mut (dyn Egress + Send), addr: &TargetAddr, payload: &[u8]) { + egress + .send(addr.clone(), Bytes::from(payload.to_vec())) + .await + .expect("send egresses"); + let reply = tokio::time::timeout(std::time::Duration::from_secs(10), egress.next_reply()) + .await + .expect("reply within timeout") + .expect("next_reply ok") + .expect("a reply"); + let expected = { + let mut v = addr.to_be_bytes().expect("addr encodes"); + v.extend_from_slice(payload); + Bytes::from(v) + }; + assert_eq!(reply, expected, "the echo reports the destination"); +} + +#[tokio::test] +async fn datagrams_round_trip_per_destination_multiple_tunnels() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::level_filters::LevelFilter::TRACE) + .with_test_writer() + .try_init(); + let dial_count = Arc::new(AtomicUsize::new(0)); + let topo = wire_direct(counting_udp_echo_dial(Arc::clone(&dial_count))).await; + let mut egress = egress_for(&topo, 8).await; + tracing::info!("dial_count={dial_count:?}"); + + // Three destinations → three tunnels (lazy opens on first + // datagram each), round-trips on each, in order. + for port in [53u16, 123, 9999] { + round_trip(egress.as_mut(), &dest(port), b"ping").await; + } + assert_eq!( + dial_count.load(Ordering::SeqCst), + 3, + "one dial per new destination" + ); + + // A repeat destination reuses its tunnel (no new dial). + round_trip(egress.as_mut(), &dest(53), b"again").await; + assert_eq!( + dial_count.load(Ordering::SeqCst), + 3, + "the table hit reuses the tunnel" + ); + + // The readiness contract: with nothing in flight, `next_reply` + // parks (`Pending`), never a hot `Ok(None)`. + let mut poll = egress.next_reply(); + tokio::time::timeout(std::time::Duration::from_millis(50), &mut poll) + .await + .expect_err("next_reply stays Pending when nothing is ready"); +} + +#[tokio::test] +async fn lru_eviction_reopens_evicted_destinations() { + let dial_count = Arc::new(AtomicUsize::new(0)); + let topo = wire_direct(counting_udp_echo_dial(Arc::clone(&dial_count))).await; + let mut egress = egress_for(&topo, 2).await; + + // Fill the table to the cap (2): two destinations dial. + round_trip(egress.as_mut(), &dest(53), b"a").await; + round_trip(egress.as_mut(), &dest(123), b"b").await; + assert_eq!(dial_count.load(Ordering::SeqCst), 2); + + // A third destination evicts the LRU entry (dest 53) and dials. + round_trip(egress.as_mut(), &dest(9999), b"c").await; + assert_eq!( + dial_count.load(Ordering::SeqCst), + 3, + "the third destination dialed" + ); + + // The LRU (dest 53) was evicted: re-touching it re-dials (the + // cap stayed — no fourth tunnel above it). + round_trip(egress.as_mut(), &dest(53), b"redial").await; + assert_eq!( + dial_count.load(Ordering::SeqCst), + 4, + "the evicted destination re-dialed" + ); + + // The recent entries survived: dest 9999 reuses; dest 123 was + // the eviction victim of the redial, so it re-dials (the cap + // stayed at 2 throughout). + round_trip(egress.as_mut(), &dest(9999), b"reuse").await; + assert_eq!( + dial_count.load(Ordering::SeqCst), + 4, + "the recency survivor reused" + ); + round_trip(egress.as_mut(), &dest(123), b"redial too").await; + assert_eq!( + dial_count.load(Ordering::SeqCst), + 5, + "the eviction victim re-dialed" + ); +} + +/// The framing-once probe: the reply the stage would frame is exactly +/// `addr.to_be_bytes() + payload` — no double prefix. Pinned through +/// the egress's own reply (the round-trip asserts equality with the +/// inner form); here the composition is asserted at the wire boundary: +/// the egress's reply, fed through `frame_inner` (the wire layer's +/// one framing application), decodes to the original (addr, payload). +#[tokio::test] +async fn framing_once_replies_decode_through_the_wire_layer_exactly_once() { + let dial_count = Arc::new(AtomicUsize::new(0)); + let topo = wire_direct(counting_udp_echo_dial(Arc::clone(&dial_count))).await; + let mut egress = egress_for(&topo, 8).await; + + let addr = dest(4242); + let payload = b"frame once"; + egress + .send(addr.clone(), Bytes::from_static(payload)) + .await + .expect("send"); + let inner = tokio::time::timeout(std::time::Duration::from_secs(10), egress.next_reply()) + .await + .expect("reply within timeout") + .expect("next_reply ok") + .expect("a reply"); + + // The egress's reply carries NO length prefix (inner bytes). + let mut expected_inner = addr.to_be_bytes().expect("addr encodes"); + expected_inner.extend_from_slice(payload); + assert_eq!( + inner, + Bytes::from(expected_inner), + "inner bytes, no double prefix" + ); + + // Feeding it through the wire layer's one framing application + // (`frame_inner` → `DatagramReader`) decodes the original pair. + let framed = alksocks::wire::frame_inner(&inner).expect("the wire layer frames once"); + let mut reader = alksocks::wire::DatagramReader::new(); + let mut decoded = None; + for dg in reader.feed(&framed) { + decoded = Some(dg.expect("frame decodes")); + } + let (src, payload_back) = decoded.expect("one datagram"); + assert_eq!(src, addr); + assert_eq!(&payload_back[..], payload); +} + +/// The closed-queue → association-end shape: when the tunnel's +/// producer side ends (the dial's target EOFs), the relay task exits +/// (EOF on the read half) and — with every relay ended — the egress's +/// `next_reply` returns `Err` (never a hot `Ok(None)`). +#[tokio::test] +async fn closed_queue_yields_err_not_a_hot_none() { + // The dial's "target" echoes exactly one datagram, then EOFs the + // tunnel (the write end drops after one reply). + let dial: TunnelDialFn = Arc::new(move |_substrate: Substrate, _backing: &str| { + Box::pin(async move { + let (consumer_side, mut target_side) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + let mut reader = TunnelDatagramReader::new(); + let mut buf = vec![0u8; 16 * 1024]; + 'outer: loop { + let n = match target_side.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => n, + }; + if let Some(dg) = reader.feed(&buf[..n]).into_iter().next() { + let framed = match alktunnels::wire::frame_datagram(&dg) { + Ok(f) => f, + Err(_) => return, + }; + if target_side.write_all(&framed).await.is_err() { + return; + } + // One echo, then EOF (the channel ends + // producer-side; the relay task drains the + // EOF → exits → the queue closes). + break 'outer; + } + } + let _ = target_side.shutdown().await; + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + Ok(TunnelTargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }) + }) + }); + let topo = wire_direct(dial).await; + let mut egress = egress_for(&topo, 8).await; + + let addr = dest(53); + egress + .send(addr.clone(), Bytes::from_static(b"only")) + .await + .expect("send"); + // The one echo comes back. + let reply = tokio::time::timeout(std::time::Duration::from_secs(10), egress.next_reply()) + .await + .expect("reply within timeout") + .expect("next_reply ok") + .expect("a reply"); + let mut expected = addr.to_be_bytes().expect("addr encodes"); + expected.extend_from_slice(b"only"); + assert_eq!(reply, Bytes::from(expected)); + + // The tunnel ended → the relay exited → the queue closed. The + // next `next_reply` returns `Err` (the association-end shape) — + // awaited to prove it is an `Err`, not an instantly-ready + // `Ok(None)` spinning the stage. + let result = tokio::time::timeout(std::time::Duration::from_secs(10), egress.next_reply()) + .await + .expect("the closed queue surfaces (relay task drains the EOF)") + .expect_err("the closed queue returns Err"); + assert!( + matches!(result, EgressError::General(ref m) if m.contains("relays ended")), + "the association-end error: {result}" + ); +} + +/// The domain form passes through unresolved (ADR-012 — the ssh `-D` +/// semantic): a domain destination renders its domain into the direct +/// op's target verbatim (the dial closure sees the unresolved host). +#[tokio::test] +async fn domain_destinations_pass_through_unresolved() { + let seen = Arc::new(tokio::sync::Mutex::new(None::)); + let seen_for_dial = Arc::clone(&seen); + let dial: TunnelDialFn = Arc::new(move |substrate: Substrate, backing: &str| { + let backing = backing.to_string(); + let seen = Arc::clone(&seen_for_dial); + Box::pin(async move { + if substrate != Substrate::Udp { + return Err(TunnelEstablishError::DialFailed("udp only".into())); + } + *seen.lock().await = Some(backing); + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + let (mut t_read, mut t_write) = tokio::io::split(target_side); + let mut reader = TunnelDatagramReader::new(); + let mut buf = vec![0u8; 16 * 1024]; + loop { + let n = match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => n, + }; + for dg in reader.feed(&buf[..n]) { + let framed = match alktunnels::wire::frame_datagram(&dg) { + Ok(f) => f, + Err(_) => return, + }; + if t_write.write_all(&framed).await.is_err() { + return; + } + let _ = t_write.flush().await; + } + } + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + Ok(TunnelTargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }) + }) + }); + let topo = wire_direct(dial).await; + let mut egress = egress_for(&topo, 8).await; + + let addr = TargetAddr::Domain("example.test".into(), 5353); + egress + .send(addr.clone(), Bytes::from_static(b"q")) + .await + .expect("send"); + let reply = tokio::time::timeout(std::time::Duration::from_secs(10), egress.next_reply()) + .await + .expect("reply within timeout") + .expect("next_reply ok") + .expect("a reply"); + let mut expected = addr.to_be_bytes().expect("domain addr encodes"); + expected.extend_from_slice(b"q"); + assert_eq!(reply, Bytes::from(expected)); + assert_eq!( + seen.lock().await.as_deref(), + Some("example.test:5353"), + "the domain passed through unresolved (producer-side resolution)" + ); +} + +/// The compile-shape pin: the composed egress's session shell rides +/// the table entry (Send); the owner wrapper compiles against the +/// alktunnels surface (`TakenHalves`-shaped halves + the session). +#[test] +fn composed_shapes_compile() { + fn assert_send() {} + assert_send::(); + assert_send::>>(); +} diff --git a/tests/harness.rs b/tests/harness.rs index 0b26856..1bd9f9d 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -788,6 +788,107 @@ pub async fn wire_consumer_side(dial: DialFn) -> ConsumerSideTopology { wire_consumer_side_with(dial, None, None, Some(consumer_identity())).await } +// --- The composed suite's topology (the alktunnels direct op) ------- + +/// The tunnel-direct topology (the composed suite's shape): the +/// consumer dials the transport (a full `ChannelClient` — the +/// composed egress's direct-op caller) and the producer accepts (the +/// adapter + an install hook registering the alktunnels direct op — +/// `register_tunnel_direct_openable` over the given dial — on the +/// producer's serving registry). The resolved identity carries +/// `tunnel:direct` (the direct op's scope gate). +pub struct TunnelDirectTopology { + /// The consumer's client — `TunnelSession::open_direct` rides it. + pub consumer: Arc, + /// The producer's manager (accept side, serving) — probes. + pub producer_manager: ChannelManager, +} + +/// Wire the tunnel-direct topology over one duplex transport (the +/// consumer-side topology's shape, the direct op in place of the +/// SOCKS5 openable; the identity carries `tunnel:direct`). +pub async fn wire_tunnel_direct(dial: alktunnels::producer::DialFn) -> TunnelDirectTopology { + let producer_op_registry = Arc::new(OperationRegistry::new()); + let (producer_manager_tx, mut producer_manager_rx) = + tokio::sync::mpsc::channel::(1); + let resolved = Identity { + id: "composed-egress-consumer".to_string(), + scopes: vec![alktunnels::params::TUNNEL_DIRECT_SCOPE.to_string()], + resources: HashMap::new(), + }; + let resolved = Arc::new(resolved); + let resolved_for_hook = Arc::clone(&resolved); + let dial_for_hook = Arc::clone(&dial); + let install_hook: InstallChannelZero = Arc::new( + move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| { + let registry = Arc::clone(&producer_op_registry); + let manager_tx = producer_manager_tx.clone(); + let resolved = Arc::clone(&resolved_for_hook); + let dial = Arc::clone(&dial_for_hook); + tokio::spawn(async move { + let channel0_bidi = match channel0_conn.accept_bi().await { + Ok(s) => s, + Err(_) => return, + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let _ = channel0_conn.set_identity((*resolved).clone()); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + let core = + ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()); + alktunnels::producer::register_tunnel_direct_openable( + &core, + ®istry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + None, + ) + .expect("register tunnel direct openable (the composed topology)"); + let _ = manager_tx.send(manager).await; + let dp = Dispatcher::new( + registry, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }, + ); + + let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024); + let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + consumer_conn + .set_identity((*resolved).clone()) + .expect("transport identity set once"); + + let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + let producer_transport_auth = AuthContext::anonymous(b"alk/channels"); + let _adapter_task = tokio::spawn(async move { + let _ = alkcall::core::types::ProtocolHandler::handle( + &adapter, + producer_conn, + &producer_transport_auth, + ) + .await; + }); + + let consumer_client = ChannelClient::from_connection(consumer_conn) + .await + .expect("consumer channel client init"); + let producer_manager = producer_manager_rx + .recv() + .await + .expect("producer manager captured"); + + TunnelDirectTopology { + consumer: Arc::new(consumer_client), + producer_manager, + } +} + // --- The raw RFC 1928 client helper ----------------------------------------- /// A hand-rolled minimal RFC 1928 client over any