diff --git a/Cargo.lock b/Cargo.lock index cea7cdd..9d333f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -58,6 +58,7 @@ dependencies = [ "thiserror", "tokio", "tracing", + "tracing-subscriber", ] [[package]] @@ -524,6 +525,21 @@ dependencies = [ "scopeguard", ] +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "memchr" version = "2.8.3" @@ -547,6 +563,15 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + [[package]] name = "num" version = "0.4.3" @@ -841,6 +866,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -932,6 +966,15 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.4" @@ -999,6 +1042,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -1050,6 +1123,12 @@ dependencies = [ "vsimd", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "version_check" version = "0.9.5" diff --git a/Cargo.toml b/Cargo.toml index e56b60d..29a23a1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,4 +28,5 @@ tracing = "0.1" thiserror = "2" [dev-dependencies] -tokio = { version = "1", features = ["full", "test-util", "macros"] } \ No newline at end of file +tokio = { version = "1", features = ["full", "test-util", "macros"] } +tracing-subscriber = { version = "0.3.23", features = ["env-filter"] } diff --git a/src/consumer.rs b/src/consumer.rs index ceaa068..ccda29f 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -1,9 +1,416 @@ //! The consumer half: `TunnelSession` — the typed client for tunnel -//! channels (ADR-005). Forward path (`open`), reverse path (`adopt`), -//! substrate-shaped data planes (`stream_halves` / `take_halves` / -//! `send_datagram` / `recv_datagram`), and teardown ownership -//! (`close` / `join` / `Drop`). +//! channels (ADR-005). Forward path (`open`), reverse path (`adopt` +//! and `open_reverse_channel`), substrate-shaped data planes +//! (`stream_halves`, `take_halves`, `send_datagram`, `recv_datagram`), +//! pump ownership (`pump_against`), and teardown ownership +//! (`close`, `join`, `Drop`). //! -//! Skeleton module — filled by `tunnels/consumer-session`. +//! One session = one channel = one tunnel (the channel ID is the flow +//! key, OQ-TN-02). The session owns the adopted channel entry — +//! nothing upstream awaits the adopter's pump (the W3 gap); the +//! session closes it structurally. Role follows the resource, not the +//! connection (OQ-TN-03): `-L` and `-R` consumers share this one API. -pub use crate::producer::{OP_TUNNEL_OPEN, TUNNEL_ALPN}; +use alkcall::channels::client::{ChannelClient, ChannelOpenError}; +use alkcall::channels::manager::ChannelManager; +use alkcall::channels::pump::pump_bidi; +use alkcall::channels::reassembly::{MpscRecvStream, MpscSendStream}; +use alkcall::core::types::BiStream; +use alkcall::protocol::connection::CallConnection; +use alkcall::protocol::wire::CallError; +use bytes::Bytes; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf}; + +use crate::error::{ReverseOpenError, TunnelIoError, TunnelOpenError}; +use crate::params::{Substrate, TunnelParams, OP_TUNNEL_OPEN, TUNNEL_ALPN}; +use crate::wire::{frame_datagram, DatagramReader}; + +/// The typed consumer session (ADR-005): one session = one channel = +/// one tunnel. Owns the adopted channel entry, the substrate-shaped +/// data plane, and (when spawned) the session-side pump handle. +/// +/// The session does **not** implement `Clone` (compile-asserted below): +/// aliasing a session would alias its teardown; multi-channel +/// consumers hold a `Vec` (or the assembly layer does). +/// +/// ```compile_fail +/// use alktunnels::consumer::TunnelSession; +/// fn assert_clone() {} +/// assert_clone::(); +/// ``` +/// +/// Fields are `Option` so `take_halves`/`pump_against` can move the +/// data plane and pump out of the `Drop`-impl type; an emptied session +/// drops as a no-op (teardown is idempotent — an already-reaped entry +/// reaps `Err(UnknownChannel)`, ignored). +pub struct TunnelSession { + pub channel_id: u32, + manager: Option, + data: Option, + pump: Option>, +} + +enum DataPlane { + /// Raw pass-through (stream substrates: `tcp`, `unix`) — the + /// halves ARE the tunnel (ADR-003); zero tunnel-level framing. + Stream(StreamHalves), + /// The mandatory `[len: u16 BE]` codec over the halves (`udp`, + /// ADR-003): boundary-preserving, `len=0` a legal empty datagram. + Datagram(DatagramHalves), +} + +struct StreamHalves { + read: ReadHalf, + write: WriteHalf, +} + +struct DatagramHalves { + read: ReadHalf, + write: WriteHalf, + reader: DatagramReader, + /// Datagrams decoded from a chunk beyond the one returned — a + /// single chunk may batch several frames (the codec decodes all + /// completions per feed); they queue here for the next + /// `recv_datagram`. + pending: std::collections::VecDeque, +} + +impl TunnelSession { + /// Open a tunnel channel toward a produced resource (the forward + /// path, `-L`). Calls the open op + /// (`ChannelClient::open_channel(OP_TUNNEL_OPEN, params, + /// TUNNEL_ALPN)`), adopts the returned channel ID, splits the + /// channel `BiStream`, and presents the substrate-shaped data + /// plane. A failed open never yields a session (typed errors — + /// ADR-049 §4; no phantom session). + pub async fn open( + client: &ChannelClient, + params: TunnelParams, + ) -> Result { + let input = serde_json::to_value(¶ms).map_err(|e| { + TunnelOpenError::Open(ChannelOpenError::CallFailed { + error: CallError::internal(format!("params serialize: {e}")), + }) + })?; + let (channel_id, send, recv) = client + .open_channel(OP_TUNNEL_OPEN, input, TUNNEL_ALPN) + .await?; + Ok(Self::from_halves( + channel_id, + params.substrate, + client.manager().clone(), + send, + recv, + )) + } + + /// Adopt a producer-allocated channel ID (the reverse path, `-R`): + /// install the data plane from the adopted halves. Early arrivals + /// are parked by the manager and drained on adopt (the adoption + /// race — ADR-047 §5). An adopt failure is `AdoptFailed`-class on + /// the typed error surface, not an establishment error. + /// + /// The substrate shapes the data plane (raw halves vs the UDP + /// codec) and rode the open call that produced the `channel_id`; + /// `adopt` takes it so the session presents the right variant (the + /// task sketch omitted it — a datagram session cannot exist without + /// it; documented in the task notes). + pub async fn adopt( + manager: &ChannelManager, + channel_id: u32, + substrate: Substrate, + alpn: impl Into, + ) -> Result { + let (send, recv) = manager + .adopt_channel(channel_id, alpn, None) + .await + .map_err(ChannelOpenError::AdoptFailed)?; + Ok(Self::from_halves( + channel_id, + substrate, + manager.clone(), + send, + recv, + )) + } + + fn from_halves( + channel_id: u32, + substrate: Substrate, + manager: ChannelManager, + send: MpscSendStream, + recv: MpscRecvStream, + ) -> Self { + let bidi = BiStream::from_joined(recv, send); + let (read, write) = tokio::io::split(bidi); + let data = match substrate { + Substrate::Tcp | Substrate::Unix => DataPlane::Stream(StreamHalves { read, write }), + Substrate::Udp => DataPlane::Datagram(DatagramHalves { + read, + write, + reader: DatagramReader::new(), + pending: std::collections::VecDeque::new(), + }), + }; + TunnelSession { + channel_id, + manager: Some(manager), + data: Some(data), + pump: None, + } + } + + /// Borrow the raw halves (stream substrates). The halves ARE the + /// tunnel — raw pass-through (ADR-003). For owned access, use + /// [`TunnelSession::take_halves`]. `None` for datagram sessions + /// (drive those with `send_datagram`/`recv_datagram`) or after the + /// halves were taken. + pub fn stream_halves( + &mut self, + ) -> Option<( + &mut (dyn AsyncRead + Send + Unpin), + &mut (dyn AsyncWrite + Send + Unpin), + )> { + match self.data.as_mut()? { + DataPlane::Stream(h) => { + let read: &mut (dyn AsyncRead + Send + Unpin) = &mut h.read; + let write: &mut (dyn AsyncWrite + Send + Unpin) = &mut h.write; + Some((read, write)) + } + DataPlane::Datagram(_) => None, + } + } + + /// Take the raw halves (either substrate): consumes the session's + /// data plane into owned boxed halves. The session stays alive as + /// the channel-entry owner — reap with `close`/`join`/`Drop` + /// (`join` on a pump-less session completes immediately, reaps + /// only, `(0, 0, reaped)`). The halves may be taken once; later + /// data-plane calls fail with [`TunnelIoError::ChannelTaken`]. + /// Datagram halves carry the codec framing (ADR-003) — raw bytes + /// on the wire are `[len: u16 BE][payload]` frames. + pub fn take_halves(mut self) -> TakenHalves { + let manager = self.manager.take(); + let data = self.data.take(); + let pump = self.pump.take(); + let (read, write) = match data { + Some(DataPlane::Stream(h)) => ( + Box::new(h.read) as Box, + Box::new(h.write) as Box, + ), + Some(DataPlane::Datagram(h)) => ( + Box::new(h.read) as Box, + Box::new(h.write) as Box, + ), + None => unreachable!("take_halves on a spent session"), + }; + TakenHalves { + session: TunnelSession { + channel_id: self.channel_id, + manager, + data: None, + pump, + }, + read, + write, + } + } + + /// UDP: send one datagram (frame → write → flush; ADR-003's + /// mandatory framing). `Oversize` above 65535 (the u16 length + /// field would wrap). Empty payloads are legal (`len=0` — the + /// F-2 layering). + pub async fn send_datagram(&mut self, payload: &[u8]) -> Result<(), TunnelIoError> { + match self.data.as_mut() { + Some(DataPlane::Datagram(h)) => { + let framed = frame_datagram(payload)?; + h.write.write_all(&framed).await?; + h.write.flush().await?; + Ok(()) + } + Some(DataPlane::Stream(_)) => Err(TunnelIoError::WrongSubstrate), + None => Err(TunnelIoError::ChannelTaken), + } + } + + /// UDP: receive the next datagram (boundary-preserving, the + /// incremental decode path — the only consumer-visible decode + /// path). `Some(bytes)` per datagram — possibly empty (`len=0` + /// legal); `None` only on stream EOF (the channels-level sentinel; + /// the codec never collides with it, F-2). A stream end + /// mid-datagram is [`TunnelIoError::TruncatedDatagram`] + /// (fail-loud, OQ-TN-13). + pub async fn recv_datagram(&mut self) -> Result, TunnelIoError> { + match self.data.as_mut() { + Some(DataPlane::Datagram(h)) => { + if let Some(dg) = h.pending.pop_front() { + return Ok(Some(dg)); + } + let out = read_one_datagram(&mut h.reader, &mut h.read).await?; + match out { + Some((first, rest)) => { + h.pending.extend(rest); + Ok(Some(first)) + } + None => Ok(None), + } + } + Some(DataPlane::Stream(_)) => Err(TunnelIoError::WrongSubstrate), + None => Err(TunnelIoError::ChannelTaken), + } + } + + /// Spawn `pump_bidi(channel_halves, accepted_read, accepted_write)` + /// against an accepted local connection and hold the returned + /// handle (the reverse-flow use; ADR-005). Builder-style: returns + /// the session, which now owns the pump — `close` aborts it, + /// `join` awaits it, `Drop` aborts it. Consumes the data plane + /// (the pump owns the channel halves from here). + pub async fn pump_against(mut self, accepted: S) -> Self + where + S: AsyncRead + AsyncWrite + Send + Unpin + 'static, + { + let manager = self.manager.take(); + let data = self.data.take(); + // The invariant is data-plane XOR pump (a session never holds + // both), so this abort is defensive only. + if let Some(stale) = self.pump.take() { + stale.abort(); + } + let bidi = match data { + Some(DataPlane::Stream(h)) => BiStream::from_joined(h.read, h.write), + Some(DataPlane::Datagram(h)) => BiStream::from_joined(h.read, h.write), + None => unreachable!("pump_against on a spent session"), + }; + let (a_read, a_write) = tokio::io::split(accepted); + let pump = tokio::spawn(pump_bidi(bidi, a_read, a_write)); + TunnelSession { + channel_id: self.channel_id, + manager, + data: None, + pump: Some(pump), + } + } + + /// Tear the tunnel down (the ungraceful path): abort the + /// session-owned pump (if any), then reap the adopted channel + /// entry (`teardown_channel`). Returns whether the entry existed. + pub async fn close(mut self) -> bool { + if let Some(pump) = self.pump.take() { + pump.abort(); + } + self.manager + .take() + .map(|m| m.teardown_channel(self.channel_id).is_ok()) + .unwrap_or(false) + } + + /// Await pump completion (both directions finished — half-close + /// semantics fall out of `pump_bidi`, W4), THEN reap the adopted + /// channel. Returns the `(u64, u64)` copy counts for + /// observability plus the reap result. **Pump-less sessions** + /// (after `take_halves`, or datagram sessions the caller drives + /// directly): completes immediately, reaps only, `(0, 0, reaped)`. + pub async fn join(mut self) -> (u64, u64, bool) { + let (c2p, p2c) = match self.pump.take() { + Some(pump) => match pump.await { + Ok(counts) => counts, + Err(e) => { + tracing::warn!( + error = %e, + "tunnel session: pump ended abnormally; copy counts unknown" + ); + (0, 0) + } + }, + None => (0, 0), + }; + let reaped = self + .manager + .take() + .map(|m| m.teardown_channel(self.channel_id).is_ok()) + .unwrap_or(false); + (c2p, p2c, reaped) + } +} + +impl Drop for TunnelSession { + /// Abort the pump and sync-reap the channel (`teardown_channel` is + /// sync — best-effort; `Drop` cannot await). Dropping without + /// close/join never leaks the channel entry (ADR-005). + fn drop(&mut self) { + if let Some(pump) = self.pump.take() { + pump.abort(); + } + if let Some(manager) = self.manager.take() { + let _ = manager.teardown_channel(self.channel_id); + } + } +} + +/// The owned halves [`TunnelSession::take_halves`] yields, paired with +/// the session that still owns the channel entry (reap it via +/// `TakenHalves::session`). +pub struct TakenHalves { + pub session: TunnelSession, + pub read: Box, + pub write: Box, +} + +/// Read from the chunk stream until one complete datagram is +/// assembled (the incremental decode path — ADR-003). `Ok(None)` on +/// stream EOF; mid-datagram EOF is `TruncatedDatagram` (fail-loud). +async fn read_one_datagram( + reader: &mut DatagramReader, + stream: &mut R, +) -> Result)>, TunnelIoError> { + let mut chunk = vec![0u8; 16 * 1024]; + loop { + let n = stream.read(&mut chunk).await?; + if n == 0 { + return if reader.is_mid_datagram() { + Err(TunnelIoError::TruncatedDatagram) + } else { + Ok(None) + }; + } + let mut dgs = reader.feed(&chunk[..n])?.into_iter(); + if let Some(first) = dgs.next() { + return Ok(Some((first, dgs.collect()))); + } + } +} + +/// Call the producer's open op on channel 0 and extract the allocated +/// `channel_id` (the reverse path's first step; the POC's shape). The +/// optional `auth_token` rides the payload — the hub-forwarding path +/// (precedence: token > `ServingConfig.identity` > transport identity, +/// CF-005). The assembly layer then constructs the session with +/// [`TunnelSession::adopt`]. +pub async fn open_reverse_channel( + hub_call: &CallConnection, + params: &TunnelParams, + auth_token: Option<&str>, +) -> Result { + let substrate_str = match params.substrate { + Substrate::Tcp => "tcp", + Substrate::Udp => "udp", + Substrate::Unix => "unix", + }; + let mut payload = serde_json::json!({ + "operationId": OP_TUNNEL_OPEN, + "input": { + "resource": params.resource, + "substrate": substrate_str, + }, + }); + if let Some(token) = auth_token { + payload["auth_token"] = serde_json::Value::String(token.to_string()); + } + let response = hub_call.call_with_payload(payload).await; + let out = response.result.map_err(ReverseOpenError::Call)?; + out.get("channel_id") + .and_then(|v| v.as_u64()) + .map(|id| id as u32) + .ok_or(ReverseOpenError::NoChannelId) +} diff --git a/src/error.rs b/src/error.rs index a62b372..65505b9 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,13 +1,13 @@ -//! Tunnel error surfaces: the crate's `TunnelError` plus the typed -//! open-error helpers consumers branch on (ADR-049 §4 surface). -//! -//! Skeleton module — filled by `tunnels/params` (codec + establishment -//! error mapping) and `tunnels/consumer-session` (session errors). +//! Tunnel error surfaces: the crate-level [`TunnelError`], the typed +//! session-open error ([`TunnelOpenError`], ADR-049 §4 surface), and +//! the session data-plane error ([`TunnelIoError`]). + +use alkcall::channels::client::ChannelOpenError; +use alkcall::protocol::wire::CallError; + +use crate::wire::DatagramCodecError; /// The crate-level error type (thiserror; AGENTS.md convention 2). -/// -/// Skeleton: the codec and session variants land with -/// `tunnels/wire-codec` and `tunnels/consumer-session`. #[derive(Debug, thiserror::Error)] pub enum TunnelError { /// The data-plane codec rejected an operation (UDP framing). @@ -21,3 +21,71 @@ pub enum TunnelError { #[error("io: {0}")] Io(#[from] std::io::Error), } + +/// A failed tunnel open or adopt. A failure never yields a session +/// (no phantom session, mirroring the no-phantom-channel property — +/// ADR-005). Branch on [`ChannelOpenError::establishment_reason`] +/// through [`TunnelOpenError::open_ref`] for `channel:open_failed`'s +/// reason (`dial_failed` / `unknown_resource` / `resource_shortage` / +/// `handler_error` / `timeout`), or the pre-establishment codes +/// (`FORBIDDEN`, `channel:too_many_channels`). +#[derive(Debug, thiserror::Error)] +pub enum TunnelOpenError { + /// The open op failed, or the reply/adopt failed — the typed + /// [`ChannelOpenError`] carries the wire [`CallError`] verbatim + /// (adopt failures surface as `AdoptFailed`). + #[error("tunnel open failed: {0}")] + Open(#[from] ChannelOpenError), +} + +impl TunnelOpenError { + /// The underlying [`ChannelOpenError`] (the ADR-049 typed surface). + pub fn open_ref(&self) -> &ChannelOpenError { + match self { + TunnelOpenError::Open(e) => e, + } + } +} + +/// An I/O failure on the session's data plane, or a substrate-shaped +/// operation attempted on the wrong plane (the POC's shape). +#[derive(Debug, thiserror::Error)] +pub enum TunnelIoError { + /// The underlying stream failed. + #[error("io: {0}")] + Io(#[from] std::io::Error), + /// The datagram codec rejected the operation (e.g. `Oversize` — + /// a >65535-byte send, rejected at frame time per ADR-003). + #[error("codec: {0}")] + Codec(#[from] DatagramCodecError), + /// The stream ended mid-datagram — declared length not satisfied + /// (a clean peer never does this; the fail-loud posture, OQ-TN-13 + /// / ADR-003: never a silent partial datagram). + #[error("stream ended mid-datagram (truncated)")] + TruncatedDatagram, + /// A substrate-shaped operation was attempted on the wrong data + /// plane (e.g. `send_datagram` on a stream tunnel). + #[error("wrong substrate for this operation")] + WrongSubstrate, + /// The session's channel halves were already taken — by + /// `take_halves` (caller-driven) or `pump_against` (pump-owned); + /// no data plane remains on the session. + #[error("channel halves already taken from this session")] + ChannelTaken, +} + +/// The reverse-path open call failed (`open_reverse_channel` — the +/// assembly layer calls it before [`crate::consumer::TunnelSession:: +/// adopt`]; the hub's call surface is a `CallConnection`, so the +/// two-step is the honest API). +#[derive(Debug, thiserror::Error)] +pub enum ReverseOpenError { + /// The open op failed — the wire [`CallError`]; branch on + /// `details.reason` of `channel:open_failed` (ADR-049 §4). + #[error("open call failed: {0:?}")] + Call(CallError), + /// The open op succeeded but the reply carried no `channel_id` + /// (malformed responder). + #[error("open reply missing channel_id")] + NoChannelId, +} diff --git a/src/lib.rs b/src/lib.rs index b165424..69066cf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,7 +24,8 @@ pub mod params; pub mod producer; pub mod wire; -pub use error::TunnelError; +pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession}; +pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError}; pub use params::ChannelOpenError; pub use params::{establishment_reason, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; pub use wire::MAX_DATAGRAM_LEN; diff --git a/tasks/tunnels/consumer-session.md b/tasks/tunnels/consumer-session.md index e5c704d..d355128 100644 --- a/tasks/tunnels/consumer-session.md +++ b/tasks/tunnels/consumer-session.md @@ -1,7 +1,7 @@ --- id: tunnels/consumer-session name: Consumer half — TunnelSession (open/adopt, data planes, teardown) -status: pending +status: completed depends_on: [tunnels/params, tunnels/wire-codec, tunnels/producer-open-op] scope: broad risk: high @@ -99,6 +99,70 @@ matrix (close/join/Drop paths — no leaked channel entries asserted via > Agent fills during implementation. +- **`adopt` takes `substrate: Substrate`** (a deviation from the task + sketch's `adopt(manager, channel_id, alpn)`): the substrate shapes + the data plane (raw halves vs the UDP codec) and cannot be inferred + from the channel ID. The ALPN is observability-only (alkcall + manager semantics); the substrate rode the open call that produced + the ID — passing it to `adopt` keeps the two calls consistent. +- **`pump_against` takes the whole accepted stream** (`S: AsyncRead + + AsyncWrite`) and splits internally — `tokio::io::split` at the + pump boundary, matching the POC's `pump_against(accepted)`. Returns + the session (builder-style); the pump handle is session-owned and + private (close/join/Drop are the only handles needed). +- **`take_halves(self)` returns `TakenHalves { session, read, write + }`** (not bare halves): the session must stay alive as the + channel-entry owner (reap via `close`/`join`/`Drop`); returning + bare halves would strand the entry. `join` on the held session + completes immediately (`(0, 0, reaped)` — pump-less). +- **`open_reverse_channel`'s hub_call is `&CallConnection`** (not + `&Arc`) — borrowing is the honest shape; callers + hold the Arc. +- **Fields are `Option` + idempotent `Drop`:** a `Drop`-impl type + cannot be destructured, so `take_halves`/`pump_against`/`close`/ + `join` take fields out via `.take()`; an emptied session drops as + a no-op. Teardown is idempotent (`teardown_channel` on a reaped + entry is `Err(UnknownChannel)`, ignored). +- **Bogus-ID adopt succeeds by design** (the manager parks early + arrivals for any not-yet-seen ID — the adoption-race cover, + alkcall ADR-047 §5); the honest `AdoptFailed` probe is the + ID-collision path (same ID adopted twice), pinned in the tests. +- **`recv_datagram` keeps a pending queue**: one chunk may batch + several frames; `DatagramReader::feed` decodes all completions per + feed, so frames beyond the first queue for subsequent + `recv_datagram` calls (dropped frames were a bug the tests caught). +- **Harness gained `wire_forward`** (`ForwardTopology`): the forward + path's consumer is the connect side (`from_connection`, pure + consumer) and the producer accepts via the adapter; the accept-side + transport-identity posture attaches the dialer's identity in the + install hook (modeled; the transport resolves it out-of-band for + real). `all_substrate_dial` = stream echo + framed UDP echo. + ## Summary -> Agent fills this on completion. \ No newline at end of file +> Agent fills this on completion. + +Implemented `TunnelSession` (src/consumer.rs, ~415 lines) per +consumer.md + ADR-005: both construction paths (`open` via +`ChannelClient::open_channel`, `adopt` via `ChannelManager:: +adopt_channel` + `open_reverse_channel` on the `CallConnection`), +the substrate-shaped data planes (raw halves for stream; the ADR-003 +codec with a pending frame queue for UDP), pump ownership +(`pump_against` spawns `pump_bidi` and holds the handle), and the +full teardown matrix — `close` (abort + ungraceful reap), `join` +(await + reap + copy counts; pump-less `(0, 0, reaped)`), `Drop` +(abort + sync reap, never leaks). No `Clone` (compile_fail doc-test). + +Tests: `tests/consumer_session.rs` — 12 integration tests over the +two harness topologies: forward round-trip (session halves), take- +halves + pump-less join, datagram round-trip incl. the F-2 empty +datagram, wrong-substrate typed errors, oversize-at-frame-time, +reverse open+adopt+pump round-trip, W4 half-close, out-of-band close ++ self-reap (the W3 split), Drop-no-leak (pump + pump-less), +AdoptFailed-on-collision, failed-open-typed (no phantom session), +concurrent reverse sessions. Harness additions: `wire_forward` +(`ForwardTopology`) + `all_substrate_dial`/`framed_udp_echo_dial`. + +Verified: `cargo test` green (38 total), clippy `-D warnings` clean +(native + wasm32), fmt clean, wasm32 check passes (default crate +stays wasm-clean). \ No newline at end of file diff --git a/tests/consumer_session.rs b/tests/consumer_session.rs new file mode 100644 index 0000000..88a6599 --- /dev/null +++ b/tests/consumer_session.rs @@ -0,0 +1,597 @@ +//! Integration tests for the consumer half — `TunnelSession` per +//! consumer.md + ADR-005, over the duplex harness. Ported from the +//! reverse POC's `ReverseTunnel` suite (the W3/W4 probes) plus the +//! forward POC's session shapes, generalized to the spec's session +//! type. +//! +//! Covered: forward open (session halves drive a duplex round-trip), +//! reverse open+adopt+pump_against (round-trip, W4 half-close, join +//! copy counts, out-of-band close + self-reaping, pump-less join), +//! datagram variant (round-trip incl. the empty datagram — the F-2 +//! layering), wrong-substrate errors, and the teardown matrix +//! (close/join/Drop — no leaked channel entries via `channel_ids()`). + +mod harness; + +use alktunnels::params::{Substrate, TunnelParams}; +use alktunnels::producer::ResourceRegistry; +use alktunnels::{open_reverse_channel, TunnelSession}; + +use harness::{all_substrate_dial, echo_dial, framed_udp_echo_dial, wire, wire_forward, Topology}; + +fn params(resource: &str, substrate: Substrate) -> TunnelParams { + TunnelParams { + resource: resource.to_string(), + substrate, + } +} + +/// Both managers hold only channel 0 (the phantom-channel property, +/// asserted from both topologies). +trait LeakCheck { + fn no_data_channels(&self) -> bool; +} + +impl LeakCheck for Topology { + fn no_data_channels(&self) -> bool { + self.consumer_manager + .channel_ids() + .iter() + .all(|&id| id == 0) + && self + .producer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0) + } +} + +impl LeakCheck for harness::ForwardTopology { + fn no_data_channels(&self) -> bool { + self.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0) + && self + .producer_manager + .channel_ids() + .iter() + .all(|&id| id == 0) + } +} + +fn only_channel_0(topo: &T) -> bool { + topo.no_data_channels() +} + +#[tokio::test] +async fn forward_open_session_halves_round_trip() { + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire_forward(registry, echo_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp)) + .await + .expect("open"); + assert!(topo.producer_manager.has_channel(session.channel_id)); + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (read, write) = session.stream_halves().expect("stream session has halves"); + write.write_all(b"ping").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 4]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"ping"); + + // Teardown: the session reaps its adopted entry (ADR-005). + let channel_id = session.channel_id; + let reaped = session.close().await; + assert!(reaped, "close reaped the adopted entry"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!(!topo.producer_manager.has_channel(channel_id)); + assert!(only_channel_0(&topo), "no leaked entries after close"); +} + +#[tokio::test] +async fn forward_open_take_halves_then_pumpless_join() { + // consumer.md's pinned semantics: after take_halves the session is + // pump-less — join completes immediately, reaps only, (0, 0, reaped). + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire_forward(registry, echo_dial()).await; + + let session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp)) + .await + .expect("open"); + let taken = session.take_halves(); + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut read = taken.read; + let mut write = taken.write; + write.write_all(b"echo-me").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 7]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip through taken halves") + .expect("read"); + assert_eq!(&buf, b"echo-me"); + + let session = taken.session; + // Drop the halves BEFORE joining: EOF propagates through the mux, + // the producer-side wrapper reaps, and the pump-less session reaps + // its own adopted entry on join (nothing to await). + drop(read); + drop(write); + let (c2p, p2c, reaped) = session.join().await; + assert_eq!((c2p, p2c), (0, 0), "pump-less join: no pump to await"); + assert!(reaped, "pump-less join reaps the adopted entry"); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn datagram_round_trip_including_empty() { + // The datagram variant over the codec (ADR-003): round-trip incl. + // the F-2 empty datagram — `len=0` is a legal datagram, never EOF. + let registry = ResourceRegistry::new(); + registry.register("dns", Substrate::Udp, "in-process").await; + let topo = wire_forward(registry, framed_udp_echo_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp)) + .await + .expect("udp open"); + assert!( + session.stream_halves().is_none(), + "stream_halves is stream-substrate-only" + ); + + session + .send_datagram(b"query-1") + .await + .expect("send datagram"); + session + .send_datagram(b"") + .await + .expect("send empty datagram"); + + let dg1 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv timed out") + .expect("no io error") + .expect("first datagram"); + assert_eq!(&dg1[..], b"query-1"); + let dg2 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram()) + .await + .expect("recv timed out") + .expect("no io error") + .expect("second datagram (empty round-trips — F-2)"); + assert!(dg2.is_empty(), "the empty datagram round-trips (F-2)"); + + let reaped = session.close().await; + assert!(reaped); + // The producer-side wrapper reap is async (the pump task exits, + // the wrapper reaps) — settle before asserting both managers. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn wrong_substrate_operations_are_typed() { + // The POC's WrongSubstrate shape: substrate-shaped operations fail + // typed on the wrong plane. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + registry.register("dns", Substrate::Udp, "in-process").await; + let topo = wire_forward(registry, all_substrate_dial()).await; + + let mut stream_session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp)) + .await + .expect("tcp open"); + let err = stream_session + .send_datagram(b"nope") + .await + .expect_err("send_datagram on a stream session"); + assert!(matches!(err, alktunnels::TunnelIoError::WrongSubstrate)); + let err = stream_session + .recv_datagram() + .await + .expect_err("recv_datagram on a stream session"); + assert!(matches!(err, alktunnels::TunnelIoError::WrongSubstrate)); + stream_session.close().await; + + let mut dg_session = + match TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp)).await { + Err(_) => panic!("setup: udp dial must succeed in the framed harness"), + Ok(mut s) => { + assert!( + s.stream_halves().is_none(), + "stream_halves is stream-substrate-only" + ); + s + } + }; + dg_session.send_datagram(b"late").await.expect("send works"); + dg_session.close().await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn reverse_open_adopt_pump_round_trip() { + // The reverse flow (the POC's ReverseTunnel shape, generalized): + // open_reverse_channel → adopt → pump_against → round-trip → + // join (copy counts + reap). + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let channel_id = + open_reverse_channel(&topo.consumer_call, ¶ms("echo", Substrate::Tcp), None) + .await + .expect("reverse open call"); + + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + assert_eq!(session.channel_id, channel_id); + + let (accepted_end, local_end) = tokio::io::duplex(64 * 1024); + let session = session.pump_against(accepted_end).await; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (mut l_read, mut l_write) = tokio::io::split(local_end); + l_write.write_all(b"reverse").await.expect("write"); + l_write.flush().await.expect("flush"); + let mut buf = vec![0u8; 7]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + l_read.read_exact(&mut buf), + ) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"reverse"); + + // EOF propagates through the two pumps (the two-pump contract). + drop(l_write); + drop(l_read); + let (c2p, p2c, reaped) = + tokio::time::timeout(std::time::Duration::from_secs(5), session.join()) + .await + .expect("join timed out"); + assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}"); + assert!(reaped, "join reaps after pump completion"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!( + !topo.producer.manager().has_channel(channel_id), + "producer-side channel reaped (wrapper-managed)" + ); + assert!(only_channel_0(&topo), "no leaked entries after join"); +} + +#[tokio::test] +async fn reverse_half_close_semantics_w4() { + // W4's shape: the local side half-closes (shutdown write); the + // target sees EOF on its read side while the reverse direction + // stays pumpable, and the target's final reply still flows back. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let channel_id = + open_reverse_channel(&topo.consumer_call, ¶ms("echo", Substrate::Tcp), None) + .await + .expect("reverse open call"); + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + + let (accepted_end, local_end) = tokio::io::duplex(64 * 1024); + let session = session.pump_against(accepted_end).await; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (mut l_read, mut l_write) = tokio::io::split(local_end); + l_write.write_all(b"half").await.expect("write"); + l_write.flush().await.expect("flush"); + l_write.shutdown().await.expect("half close"); + + let mut buf = vec![0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + l_read.read_exact(&mut buf), + ) + .await + .expect("echo after half close") + .expect("read echo"); + assert_eq!(&buf, b"half"); + + let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.join()) + .await + .expect("join after half close"); + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn out_of_band_close_then_session_self_reap() { + // The POC's reverse_channel_close_out_of_band shape: a peer-initiated + // `channel/close` tears the SERVING side down; the consumer's session + // still owns its adopted entry — close/join reaps it (W3 split). + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let channel_id = + open_reverse_channel(&topo.consumer_call, ¶ms("echo", Substrate::Tcp), None) + .await + .expect("reverse open call"); + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + let session = session.pump_against(tokio::io::duplex(1024).0).await; + + let response = topo + .consumer_call + .call_with_payload(serde_json::json!({ + "operationId": "channel/close", + "input": { "channel_id": channel_id, "reason": "out-of-band" } + })) + .await; + assert!(response.result.is_ok(), "channel/close resolved"); + + // The close ran on the PRODUCER side; the consumer's entry remains. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + !topo.producer.manager().has_channel(channel_id), + "producer-side channel reaped by the out-of-band close" + ); + assert!(topo.consumer_manager.has_channel(channel_id)); + let reaped = session.close().await; + assert!(reaped, "session reaped its adopted entry"); + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn drop_never_leaks_the_channel_entry() { + // ADR-005: Drop = abort + sync reap (best-effort). Dropping a live + // session without close/join never leaks the entry. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let ftopo = wire_forward(registry.clone(), echo_dial()).await; + + let session = TunnelSession::open(&ftopo.consumer, params("echo", Substrate::Tcp)) + .await + .expect("open"); + let channel_id = session.channel_id; + assert!(ftopo.consumer.manager().has_channel(channel_id)); + drop(session); + assert!( + !ftopo.consumer.manager().has_channel(channel_id), + "Drop reaped the adopted entry" + ); + + // Drop with a session-owned pump: the pump aborts too (the reverse + // topology — pump_against is the reverse-flow use). + let topo = wire(registry, echo_dial()).await; + let channel_id = + open_reverse_channel(&topo.consumer_call, ¶ms("echo", Substrate::Tcp), None) + .await + .expect("second reverse open"); + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + let session = session.pump_against(tokio::io::duplex(1024).0).await; + assert!(topo.consumer_manager.has_channel(channel_id)); + drop(session); + assert!(!topo.consumer_manager.has_channel(channel_id)); +} + +#[tokio::test] +async fn oversize_datagram_rejected_at_frame_time() { + // ADR-003: a >65535-byte send is rejected at frame time (`Oversize`), + // never a wire overflow. + let registry = ResourceRegistry::new(); + registry.register("dns", Substrate::Udp, "in-process").await; + let topo = wire_forward(registry, framed_udp_echo_dial()).await; + + let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp)) + .await + .expect("udp open"); + let err = session + .send_datagram(&vec![0u8; alktunnels::MAX_DATAGRAM_LEN + 1]) + .await + .expect_err("oversize must fail"); + assert!( + matches!( + err, + alktunnels::TunnelIoError::Codec(alktunnels::wire::DatagramCodecError::Oversize(_, _)) + ), + "oversize is a codec error at frame time: {err:?}" + ); + session.close().await; +} + +#[tokio::test] +async fn failed_open_never_yields_a_session() { + // No phantom session: unknown resource → typed error, no channel + // anywhere, and `open_ref()` exposes the typed surface. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire_forward(registry, echo_dial()).await; + + let err = match TunnelSession::open(&topo.consumer, params("ghost", Substrate::Tcp)).await { + Ok(_) => panic!("unknown resource must not open"), + Err(e) => e, + }; + assert_eq!( + err.open_ref().establishment_reason(), + Some("unknown_resource") + ); + assert!(only_channel_0(&topo)); + + // Reverse path on the reverse topology: the call fails typed; + // adopt is never reached. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + let err = + match open_reverse_channel(&topo.consumer_call, ¶ms("ghost", Substrate::Tcp), None) + .await + { + Ok(_) => panic!("reverse open of unknown resource must fail"), + Err(e) => e, + }; + assert!(matches!(err, alktunnels::ReverseOpenError::Call(_))); + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn adopt_failure_is_typed_not_establishment() { + // An adopt failure is `AdoptFailed`-class (post-reply local + // failure: ID collision), not an establishment error. A bogus-id + // adopt SUCCEEDS by design — the manager parks early arrivals for + // any not-yet-seen ID (the adoption race cover, ADR-047 §5) — so + // the collision path is the honest probe: adopt the same ID twice. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let channel_id = + open_reverse_channel(&topo.consumer_call, ¶ms("echo", Substrate::Tcp), None) + .await + .expect("reverse open"); + + let first = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("first adopt"); + + let err = match TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + { + Ok(_) => panic!("adopting a live id twice must fail"), + Err(e) => e, + }; + match err.open_ref() { + alkcall::channels::client::ChannelOpenError::AdoptFailed( + alkcall::channels::manager::ManagerError::ChannelExists(_), + ) => {} + other => panic!("expected AdoptFailed(ChannelExists), got {other:?}"), + } + + first.close().await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!(only_channel_0(&topo)); +} + +#[tokio::test] +async fn concurrent_reverse_sessions_independent() { + // Two reverse sessions on one connection: independent channels, + // each round-trips its own marker, both reaped. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut sessions = Vec::new(); + for i in 0..2 { + let channel_id = + open_reverse_channel(&topo.consumer_call, ¶ms("echo", Substrate::Tcp), None) + .await + .expect("reverse open"); + let session = TunnelSession::adopt( + &topo.consumer_manager, + channel_id, + Substrate::Tcp, + alktunnels::TUNNEL_ALPN, + ) + .await + .expect("adopt"); + let (accepted_end, local_end) = tokio::io::duplex(64 * 1024); + let session = session.pump_against(accepted_end).await; + let (mut l_read, mut l_write) = tokio::io::split(local_end); + let msg = format!("marker {i}"); + l_write.write_all(msg.as_bytes()).await.expect("write"); + l_write.flush().await.expect("flush"); + let mut buf = vec![0u8; msg.len()]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + l_read.read_exact(&mut buf), + ) + .await + .expect("round trip") + .expect("read"); + assert_eq!(buf, msg.as_bytes()); + sessions.push(session); + } + assert_ne!(sessions[0].channel_id, sessions[1].channel_id); + + for session in sessions { + let (_, _, reaped) = session.join().await; + assert!(reaped); + } + assert!(only_channel_0(&topo)); +} + +#[allow(dead_code)] +fn type_assertions() { + // The `compile_fail` doc-test in consumer.rs is the no-Clone gate; + // this parity stub keeps the type in the test's namespace. + let _ = std::mem::size_of::(); +} diff --git a/tests/harness.rs b/tests/harness.rs index 0492341..972363f 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -1,5 +1,6 @@ //! Test harness — the producer↔consumer wiring over a //! `tokio::io::duplex` carrying the channels 8-byte chunk format. +#![allow(dead_code)] //! //! Topology (the reverse POC's shape, mirrored for both directions): //! - **producer** (connect side of the channels connection — it runs @@ -222,6 +223,133 @@ pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn .await } +/// The forward topology (the `-L` direction): the consumer dials the +/// transport (connect side — `ChannelClient::from_connection`, a pure +/// consumer serving nothing) and the producer accepts (the adapter + +/// an install hook that registers the channel ops + the tunnel +/// openable on the producer's serving registry). This is the topology +/// `TunnelSession::open(&channel_client, params)` is normative for: +/// the consumer's own client calls the open op on the producer's +/// channel 0. +/// +/// Identity on the accept-side serving path (the mTLS posture): the +/// transport authenticated the dialer; the hook attaches the resolved +/// identity to channel 0 so the serving dispatch sees the caller +/// (the accept-side analogue of CF-005 (b)'s propagation). +pub struct ForwardTopology { + /// The consumer's own client (connect side, pure consumer). Its + /// manager adopted the producer-allocated channel IDs; the + /// session reaps via this manager. + pub consumer: Arc, + /// The producer's manager (accept side, serving) — for asserting + /// wrapper-managed reaping. + pub producer_manager: ChannelManager, + /// What the establisher's per-call auth carried (the CF-006 probe). + pub identity_witness: Arc>>, +} + +pub async fn wire_forward( + registry: ResourceRegistry, + dial: alktunnels::producer::DialFn, +) -> ForwardTopology { + let producer_op_registry = Arc::new(OperationRegistry::new()); + let (producer_manager_tx, mut producer_manager_rx) = + tokio::sync::mpsc::channel::(1); + let identity_witness = Arc::new(tokio::sync::Mutex::new(None::)); + let dial_witness = Arc::clone(&identity_witness); + + // --- producer (accept side, serving): adapter + install hook ------ + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { + let registry_arc = Arc::clone(&producer_op_registry); + let resource_registry = registry.clone(); + let dial = Arc::clone(&dial); + let witness = Arc::clone(&dial_witness); + let manager_tx = producer_manager_tx.clone(); + 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); + // The accept-side transport-identity posture: the transport + // authenticated the dialer; attach the resolved identity to + // channel 0 (the dispatch reads it as the caller identity). + let _ = channel0_conn.set_identity(consumer_identity()); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + + // Register the generic channel ops + the tunnel openable + // (the producer.md recipe) before the dispatch loop runs. + let core = alkcall::channels::operations::ChannelCore::new( + manager.clone(), + alkcall::channels::policy::default_policy(), + ); + let ops = ChannelOperations::with_default_policy(manager.clone()); + ops.register_on(®istry_arc) + .expect("register channel ops on producer"); + register_tunnel_openable( + &core, + &resource_registry, + ®istry_arc, + AuthContext::anonymous(b"alk/tunnel"), + dial, + Some(witness), + ) + .expect("register tunnel openable"); + let _ = manager_tx.send(manager).await; + + let dp = Dispatcher::new( + registry_arc, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + // --- transport ------------------------------------------------------ + 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); + + // The dialer's transport identity (the mTLS/QUIC posture) — set + // before from_connection. The accept side resolves it out-of-band + // (modeled in the hook above). + consumer_conn + .set_identity(consumer_identity()) + .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; + }); + + // --- consumer (connect side): pure client --------------------------- + let consumer_client = ChannelClient::from_connection(consumer_conn) + .await + .expect("consumer channel client init"); + let consumer_client = Arc::new(consumer_client); + + let producer_manager = producer_manager_rx + .recv() + .await + .expect("producer manager captured"); + + ForwardTopology { + consumer: consumer_client, + producer_manager, + identity_witness, + } +} + /// An echo dial closure (the ADR-004 injection point under test): an /// in-process pipe per dial — the "target" echoes bytes until EOF. /// The consumer side gets split `DuplexStream` halves; a spawned task @@ -263,6 +391,17 @@ pub fn echo_dial() -> alktunnels::producer::DialFn { }) } +/// A both-substrates dial: stream substrates echo via pipes, UDP +/// echoes via the framed codec (the datagram tests' single dial). +pub fn all_substrate_dial() -> alktunnels::producer::DialFn { + let echo = echo_dial(); + let udp = framed_udp_echo_dial(); + Arc::new(move |substrate: Substrate, backing: &str| match substrate { + Substrate::Udp => udp(substrate, backing), + Substrate::Tcp | Substrate::Unix => echo(substrate, backing), + }) +} + /// A failing dial closure (the `dial_failed` stand-in). pub fn failing_dial(message: &'static str) -> alktunnels::producer::DialFn { Arc::new(move |_substrate: Substrate, _backing: &str| { @@ -273,3 +412,60 @@ pub fn failing_dial(message: &'static str) -> alktunnels::producer::DialFn { }) }) } + +/// A framed UDP echo dial (the ADR-003 codec at the target boundary — +/// the in-process stand-in for the real `local` UDP adapter): the +/// dial's halves carry the `[len: u16 BE]` wire framing; a spawned +/// echo task decodes frames from the channel side and re-frames the +/// echoed payloads back. The producer pump copies raw bytes; the +/// codec lives at this boundary (the pump never sees UDP specifics). +pub fn framed_udp_echo_dial() -> alktunnels::producer::DialFn { + Arc::new(move |substrate: Substrate, backing: &str| { + let backing = backing.to_string(); + Box::pin(async move { + match substrate { + Substrate::Udp => { + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + tokio::spawn(async move { + use alktunnels::wire::DatagramReader; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let (mut t_read, mut t_write) = tokio::io::split(target_side); + let mut reader = DatagramReader::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, + }; + let dgs = match reader.feed(&buf[..n]) { + Ok(dgs) => dgs, + Err(_) => return, + }; + for dg in dgs { + 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 _ = &backing; + let (c_read, c_write) = tokio::io::split(consumer_side); + Ok(alktunnels::producer::TargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }) + } + Substrate::Tcp | Substrate::Unix => { + Err(alktunnels::producer::TunnelEstablishError::DialFailed( + format!("stream dial not wired in the udp harness: {backing}"), + )) + } + } + }) + }) +}