From 7d23a89791dea44ba96afabee38e8e9989481549 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 19 Sep 2026 09:07:28 +0000 Subject: [PATCH] feat: the bind-first establisher + accept dispatcher (shape 3, ADR-008 as amended) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BindFn + BoundListener: the platform-boundary bind seam (the protocol never binds, OQ-TN-04); the opaque listener rides the plan - bindfirst_establisher: registry namespace gate -> token mint (non-zero-seeded AtomicU32, per-open fresh, per-producer unique, token 0 never minted — the wrap shape unit-pinned) -> bind -> `bound` + `listen` reply fields -> the BindFirstPlan plan; the typed errors pass through (bind failure dial_failed, exhaustion resource_shortage, registry miss unknown_resource) - BindFirstAcceptLoop: the type-erased accept closure (receives the opaque listener; yields (peer, halves) per accept; Err is loop-critical) - ForwardedOpener: the per-accept forwarded-open caller — the call plane (call_with_payload on channel 0 + manager.adopt_channel; the reverse POC's open_reverse_channel shape) - make_bindfirst_accept_handler: the accept dispatcher — accept_bi first (R-02 telemetry), the idle listen-stream EOF sentinel cancels the loop + tears the listener down (select!), per-accept pump_bidi tasks tracked by the handler's JoinHandle (awaited on loop exit — a forwarded pump never outlives its listener), per-accept failure isolation (log + drop; v1 no retry), loop-critical accept errors end the loop - register_tunnel_bindfirst_openable: same spec + tunnel:open scope gate (the per-registration establisher posture); the injected seams ride BindFirstInjections {bind, accept_loop, open_forwarded} (8-arg sketch -> the bundle; clippy too_many_arguments) + the optional CF-006 identity witness (witness_establisher wrapper) - harness: RegistrationMode::BindFirst + wire_bindfirst; the consumer hook gained the accept-side identity propagation + a shared serving registry (Topology.consumer_registry exposed); the forwarded-op serving stub (make_forwarded_stub_handler — correlate by token, accept the BiStream, pump_bidi against the correlated halves; the opener-side registration lands with tunnels/forwarded-receipt) - tests: 6 new integration tests (the full flow e2e — the acceptance gate incl. the byte round-trip through the whole path, the dial-open neither-field projection, EOF teardown, per-accept isolation, typed establishment errors, token discipline) + 6 unit tests (the mint discipline, the establisher flow at the Establishment surface, the witness wrapper) Verified: cargo test green (118 default; 138 --all-features), clippy --all-targets -D warnings clean (native + wasm32), fmt clean, wasm32-unknown-unknown check passes, cargo doc --no-deps clean. --- CHANGELOG.md | 29 ++ src/lib.rs | 6 +- src/producer.rs | 630 +++++++++++++++++++++++++++- tasks/tunnels/bindfirst-producer.md | 140 ++++++- tests/harness.rs | 250 ++++++++++- tests/producer_bindfirst.rs | 630 ++++++++++++++++++++++++++++ 6 files changed, 1663 insertions(+), 22 deletions(-) create mode 100644 tests/producer_bindfirst.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 2395597..047d00c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,35 @@ dependency bump below is the only behavior-affecting change so far ### Added +- **The bind-first establisher + accept dispatcher (shape 3, ADR-008 + as amended)** — the `-R`/ingress establisher whose establishment + phase BINDS (the address-first-vs-accept-first deadlock dissolves; + the accept wait lives in the teardown-governed pump phase): + `BindFn` (the substrate bind function — the platform boundary, the + protocol never binds) + `BoundListener` (the opaque listener + the + real OS-chosen bound address); `bindfirst_establisher` (registry + namespace gate → token mint → bind → `bound` + `listen` reply + fields via `Establishment::with_reply_field` → the `BindFirstPlan` + plan payload; the token mint is the pinned discipline — non-zero- + seeded `AtomicU32`, per-open fresh, per-producer unique, token 0 + never minted); `BindFirstAcceptLoop` (the type-erased accept + closure — receives the opaque listener, yields `(peer, halves)` per + accept); `ForwardedOpener` (the per-accept forwarded-open caller — + the call plane: `call_with_payload` on channel 0 + + `manager.adopt_channel`, the reverse POC's shape); + `make_bindfirst_accept_handler` (the dispatcher: `accept_bi` first + per R-02, the idle listen-stream EOF cancels the loop and tears the + listener down, per-accept `pump_bidi` tasks tracked by the handler's + `JoinHandle` — a forwarded pump never outlives its listener; + forwarded-open failures are per-accept isolated — log + drop, the + loop continues, v1 no retry); `register_tunnel_bindfirst_openable` + (same spec + `tunnel:open` scope gate as the dial/listen variants — + the per-registration establisher posture; the injected seams ride + `BindFirstInjections {bind, accept_loop, open_forwarded}` + the + optional CF-006 identity witness). The opener-side forwarded-op + serving shape (`register_openable`, no establisher) lands with + `tunnels/forwarded-receipt`. + - **The forwarded op + the bind-first params surface (ADR-008 as amended)** — the wire-stable types pin first (the ordering `tunnels/params` used for the base op): `OP_TUNNEL_FORWARDED` / diff --git a/src/lib.rs b/src/lib.rs index 08eade4..a018472 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,9 +36,13 @@ pub use params::{ }; pub use params::{tunnel_open_spec, ChannelOpenError, Substrate, TunnelParams, OP_TUNNEL_OPEN}; pub use params::{TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; +pub use producer::{ + bindfirst_establisher, make_bindfirst_accept_handler, register_tunnel_bindfirst_openable, +}; pub use producer::{ direct_establisher, direct_establisher_with_witness, register_tunnel_direct_openable, - register_tunnel_listen_openable, AcceptFn, AcceptQueue, BindFirstPlan, DialFn, + register_tunnel_listen_openable, AcceptFn, AcceptQueue, BindFirstAcceptLoop, + BindFirstInjections, BindFirstPlan, BindFn, BoundListener, DialFn, ForwardedOpener, ResourceRegistry, TargetHandle, TunnelEstablishError, }; pub use wire::MAX_DATAGRAM_LEN; diff --git a/src/producer.rs b/src/producer.rs index aaade0e..9920f5a 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -42,13 +42,15 @@ use alkcall::core::auth::AuthContext; use alkcall::core::types::Connection; use alkcall::registry::registration::OperationRegistry; use serde_json::Value; -use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite}; -use crate::params::{Substrate, TunnelDirectParams, TunnelParams}; +use crate::params::{ + Substrate, SubstrateAddr, TunnelDirectParams, TunnelForwardedParams, TunnelParams, +}; pub use crate::params::{ - OP_TUNNEL_DIRECT, OP_TUNNEL_FORWARDED, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, - TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE, + OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_DIRECT_SCOPE, TUNNEL_FORWARDED_SCOPE, + TUNNEL_OPEN_SCOPE, }; /// The dialed target handle — the plan payload (typed-opaque @@ -539,6 +541,409 @@ pub fn register_tunnel_listen_openable( ) } +// --- Bind-first (shape 3, ADR-008) --------------------------------------- + +/// The substrate bind function the assembly layer injects (shape 3 — +/// ADR-008 §1): given a substrate + a registry backing (the same +/// backing-string form the dial shape's `DialFn` receives), **bind the +/// listener** and produce it. The protocol never binds (OQ-TN-04) — +/// this closure is the platform boundary, the same inversion point as +/// `DialFn`. The returned [`BoundListener`] pairs the opaque +/// assembly-typed listener (the accept loop's business) with the +/// OS-chosen bound address (the `bound` reply field's source). +pub type BindFn = Arc< + dyn Fn( + Substrate, + &str, + ) + -> futures::future::BoxFuture<'static, Result> + + Send + + Sync, +>; + +/// What the assembly's bind produces (ADR-008 §1): the opaque listener +/// (the accept loop's business) + the OS-chosen bound address (the +/// `bound` reply field's source — the REAL resolved address, port `0` +/// resolved; informational surface, never addressing). +pub struct BoundListener { + /// The assembly-typed listener (e.g. the `TcpListener` wrapper the + /// assembly chose). Opaque to the protocol — rides the plan's + /// `Any`, downcast only inside the assembly's accept closure. + pub listener: Arc, + /// The real OS-chosen bound address (the `bound` reply field's + /// source). + pub bound: SubstrateAddr, +} + +/// The token mint (the discipline pinned on [`BindFirstPlan`]): a +/// non-zero-seeded `AtomicU32` counter held in the establisher's +/// closure state — fresh per open, unique per producer, wraps +/// acceptable (a wrap collision requires 2^32 concurrent opens on one +/// producer). The counter seeds at 1: `fetch_add` returns the +/// PREVIOUS value, so the first mint is 1 and the minted stream is +/// 1, 2, 3, … — token 0 (an illegal wire value; the forwarded params' +/// `listen` pin rejects it) is never minted, and a u32 wrap that lands +/// on 0 re-increments (the loop skips it; the 2^32 collision window +/// makes wrap uniqueness a non-issue at the discipline's own terms). +fn mint_token(counter: &std::sync::atomic::AtomicU32) -> u32 { + use std::sync::atomic::Ordering; + loop { + let prev = counter.fetch_add(1, Ordering::Relaxed); + if prev != 0 { + return prev; + } + } +} + +/// The bind-first establisher (shape 3 — producer.md §The Establisher, +/// ADR-008 §1): the same open op, same params, same typed-error table; +/// the establisher **binds** (via the injected [`BindFn`]) instead of +/// dialing or popping an accept, mints the correlation token +/// (per-open-fresh, per-producer-unique — the discipline pinned on +/// [`BindFirstPlan`]), contributes +/// the `bound` + `listen` reply fields (the establisher → reply +/// projection, ADR-049 amendment 3 — the wrapper merges them into the +/// success output after its reserved `channel_id` key), and returns +/// the [`BindFirstPlan`] as the plan payload. +/// +/// The registry check is a namespace gate only (the same posture as +/// [`listen_establisher`]): the bind-first producer's "backing" is the +/// [`BindFn`], not a dial target; a resource the producer does not +/// produce is still `unknown_resource`. The resolved backing string +/// rides the same signature the dial shape's `DialFn` takes — the bind +/// closure sees the same `(substrate, backing)` form. +/// +/// Error mapping (producer.md's shape-3 table): unbindable/held port → +/// `dial_failed`-class (the bind IS the dial here); bind resource +/// exhaustion → `resource_shortage`; registry miss → +/// `unknown_resource`; internal → `handler_error`. The `BindFn` +/// returns the typed errors; the establisher passes them through. +/// +/// The token lives in BOTH the reply (`listen` — the opener's +/// correlation key) and the plan (the accept dispatcher stamps it into +/// every forwarded open). Two opens mint distinct tokens (the mint is +/// per-open); each dispatcher stamps ITS OWN plan's token (the plan +/// flows per-open — R-01). +pub fn bindfirst_establisher(registry: ResourceRegistry, bind: BindFn) -> OpenEstablisher { + use std::sync::atomic::AtomicU32; + let counter = Arc::new(AtomicU32::new(1)); + Arc::new(move |input: Value, _auth| { + let registry = registry.clone(); + let bind = Arc::clone(&bind); + let counter = Arc::clone(&counter); + Box::pin(async move { + let params: TunnelParams = parse_params(&input)?; + let backing = registry + .lookup(params.substrate, ¶ms.resource) + .await + .ok_or_else(|| { + TunnelEstablishError::UnknownResource(format!( + "resource `{}` ({:?}) not produced by this producer", + params.resource, params.substrate + )) + })?; + let token = mint_token(&counter); + let BoundListener { listener, bound } = bind(params.substrate, &backing).await?; + let plan = Arc::new(BindFirstPlan { listener, token }) as ChannelPlan; + let bound = serde_json::to_value(&bound) + .map_err(|e| TunnelEstablishError::HandlerError(format!("bound serialize: {e}")))?; + Ok(Establishment::new(plan) + .with_reply_field("bound", bound) + .with_reply_field("listen", serde_json::json!(token))) + }) + }) +} + +/// The accepted-connection source the assembly layer injects for the +/// BIND-FIRST producer (shape 3 — ADR-008 §1): pops one accepted +/// connection from the plan's opaque listener and yields +/// `(SubstrateAddr /* peer */, TargetHandle /* accepted halves */)`. +/// The closure receives the listener as the opaque `Any` — the +/// assembly downcasts it to ITS concrete listener type inside the +/// closure (the same type-erased injection pattern as +/// `DialFn`/`AcceptFn`; substrate code lives behind the closure, +/// ADR-004). Substrate code lives behind the closure (ADR-004); the +/// protocol crate never sees a socket type. +/// +/// A returned `Err` is a LOOP-CRITICAL condition: the accept loop's +/// substrate source is broken (the listener closed, accept errored +/// unrecoverably) — the dispatcher ends the loop, tears the listener +/// down with the plan, and lets the handler task finish (the channel +/// reaps). Per-ACCEPT failures (forwarded-open failures) do NOT come +/// from this closure — see `make_bindfirst_accept_handler`'s error +/// posture. +pub type BindFirstAcceptLoop = Arc< + dyn Fn( + Arc, + ) -> futures::future::BoxFuture< + 'static, + Result<(SubstrateAddr, TargetHandle), TunnelEstablishError>, + > + Send + + Sync, +>; + +/// The per-accept forwarded-open caller (ADR-008 §3): opens +/// `OP_TUNNEL_FORWARDED` with `TunnelForwardedParams {listen, +/// substrate, peer}` on the channel-0 call plane toward the listen +/// opener, adopts the opener-allocated ID on the producer-side +/// manager, and returns the adopted channel's halves (the pump side). +/// +/// **This closure is the call plane** (the design pin — producer.md +/// §The Pump Handler item 5): the per-channel `Connection` the +/// dispatcher receives is stream-less (`open_bi` is `StreamClosed` — +/// ADR-035) and carries no call surface, so the registration closes +/// over the serving session's `Arc` (the assembly has +/// it — it constructed the serving session). The shape is the reverse +/// POC's `open_reverse_channel`: `call_with_payload` on channel 0 → +/// extract `channel_id` from the reply → `manager.adopt_channel`. The +/// accepted halves pump against the adopted channel's halves. +pub type ForwardedOpener = Arc< + dyn Fn( + TunnelForwardedParams, + ) -> futures::future::BoxFuture< + 'static, + Result< + ( + alkcall::channels::reassembly::MpscSendStream, + alkcall::channels::reassembly::MpscRecvStream, + ), + TunnelEstablishError, + >, + > + Send + + Sync, +>; + +/// The bind-first pump handler — the ACCEPT DISPATCHER (ADR-008 §1). +/// +/// Handler contract (producer.md §The Pump Handler item 5 + R-02): +/// +/// 1. `conn.accept_bi()` FIRST (the yield-once contract — the +/// wrapper's accepted-flag telemetry requires it), downcast the +/// plan to [`BindFirstPlan`]. +/// 2. The listen channel's own `BiStream` stays idle (no pumps on +/// it); **its EOF cancels the loop and tears the listener down** — +/// a 1-byte `read` on the idle stream observes the opener dropping +/// the listener session (the zero-byte EOF sentinel; ADR-050's +/// teardown contract). `select!` races the EOF watch against the +/// accept loop. +/// 3. Per accept: open the forwarded channel ([`ForwardedOpener`] — +/// the call plane, `{listen: token, substrate, peer}` params), +/// then `pump_bidi` the accepted halves against the adopted +/// channel inside a spawned task tracked in the handler's handle +/// set. A forwarded-open failure is per-accept isolated: log and +/// drop that accepted connection — the loop continues (the +/// listener's overall health is not hostage to one failed accept; +/// v1: no retry/backoff). A loop-critical accept error ([`BindFirstAcceptLoop`]'s +/// contract) ends the loop. +/// 4. **Lifetime (R-02):** the handler's returned `JoinHandle` tracks +/// the whole accept loop + every per-accept pump it spawned — the +/// handler task awaits the pump handles on loop exit (a forwarded +/// pump outliving its listener is a leak). Handler exit → wrapper +/// teardown (the listen channel reaps) → the far listener tears +/// down (the plan's listener drops with the loop). +/// +/// The substrate discriminator for the forwarded opens rides the +/// open-op INPUT params (`TunnelParams.substrate` — the listen +/// tunnel's substrate; ADR-008 §3 "the listen tunnel's substrate +/// shapes the forwarded channel's data-plane framing"). +pub fn make_bindfirst_accept_handler( + accept_loop: BindFirstAcceptLoop, + open_forwarded: ForwardedOpener, +) -> OpenHandler { + Arc::new( + move |input: Value, plan: Option, conn: Connection, _auth: AuthContext| { + let accept_loop = Arc::clone(&accept_loop); + let open_forwarded = Arc::clone(&open_forwarded); + tokio::spawn(async move { + let Some(bidi) = conn.accept_bi().await.ok() else { + tracing::error!("tunnel bind-first: accept_bi failed"); + return; + }; + let Some(plan) = plan.and_then(|p| p.downcast::().ok()) else { + tracing::error!("tunnel bind-first: no BindFirstPlan in plan"); + return; + }; + let substrate = serde_json::from_value::(input.clone()) + .map(|p| p.substrate) + .unwrap_or(Substrate::Tcp); + let (mut idle_read, _idle_write) = tokio::io::split(bidi); + + // The per-accept pump task set — tracked by the + // handler's JoinHandle (R-02: every pump the handler + // spawned is awaited on loop exit). + let mut pumps: Vec> = Vec::new(); + + let loop_result = loop { + // The idle-stream EOF watch vs the accept loop + // (select!): the opener dropping the listener + // session EOFs the listen channel's stream — the + // zero-byte sentinel — and cancels the loop. + let eof_watch = async { + // The 1-byte read on the idle stream: `Ok(0)` + // (or an error) = the opener dropped the + // session; `Ok(1..)` never happens on a well- + // behaved opener (the stream is idle — a peer + // that writes is misbehaving; treat any + // non-empty read as an EOF trigger too — the + // channel is not a data plane). + let mut byte = [0u8; 1]; + match idle_read.read(&mut byte).await { + Ok(0) | Err(_) => true, + Ok(_) => true, + } + }; + let accept = accept_loop(Arc::clone(&plan.listener)); + tokio::select! { + _ = eof_watch => break "eof", + accepted = accept => match accepted { + Ok((peer, handle)) => { + let forwarded = TunnelForwardedParams { + listen: plan.token, + substrate, + peer: peer.clone(), + }; + match open_forwarded(forwarded).await { + Ok((send, recv)) => { + let channel_bidi = + alkcall::core::types::BiStream::from_joined(recv, send); + pumps.push(tokio::spawn(pump_bidi( + channel_bidi, + handle.read, + handle.write, + ))); + } + Err(e) => { + // Per-accept isolation: the + // forwarded open failed (scope- + // less opener → FORBIDDEN; old + // opener → NOT_FOUND; correlation + // miss → handler_error-class) — + // drop this accepted connection, + // keep serving. v1: no retry/ + // backoff (documented posture). + tracing::warn!( + listen = plan.token, + error = %e, + "tunnel bind-first: forwarded open failed; dropping the accepted connection" + ); + drop(handle); + } + } + } + Err(e) => { + // Loop-critical (the listener's + // substrate source broke): end the + // loop — the listener tears down with + // the plan. + tracing::warn!( + listen = plan.token, + error = %e, + "tunnel bind-first: accept loop ended; tearing the listener down" + ); + break "accept-error"; + } + } + } + }; + tracing::debug!( + listen = plan.token, + reason = loop_result, + "tunnel bind-first: accept loop exited" + ); + // The listener drops with the plan (the last Arc) — the + // far listener tears down. + drop(plan); + // R-02: the handler's JoinHandle tracks every per-accept + // pump it spawned — await them on loop exit (a forwarded + // pump outliving its listener is a leak). + for pump in pumps.drain(..) { + let _ = pump.await; + } + }) + }, + ) +} + +/// The bind-first registration's injected seams (the assembly layer +/// owns all four; bundling keeps the registration's signature to the +/// core pair + the bundle — the seams are one coherent injection +/// point, the same posture as `DialFn` riding the dial registration): +/// - `bind` — the substrate bind function ([`BindFn`]; the protocol +/// never binds, OQ-TN-04). +/// - `accept_loop` — the per-listener accept loop +/// ([`BindFirstAcceptLoop`]; receives the opaque listener, yields +/// `(peer, halves)` per accept). +/// - `open_forwarded` — the forwarded-open caller ([`ForwardedOpener`]; +/// the call plane: the serving session's `Arc` + +/// the producer-side `ChannelManager` for the adopt — the assembly +/// has both, it constructed the serving session; the reverse POC's +/// `open_reverse_channel` shape). +pub struct BindFirstInjections { + /// The substrate bind function (the platform boundary). + pub bind: BindFn, + /// The per-listener accept loop (pops the opaque listener). + pub accept_loop: BindFirstAcceptLoop, + /// The forwarded-open caller (the call plane + the adopt). + pub open_forwarded: ForwardedOpener, +} + +/// Register the bind-first openable (shape 3, ADR-008 §1): the SAME +/// spec + scope gate (`tunnel:open`) as the dial/listen variants — the +/// bind-first shape is a per-registration establisher choice over the +/// same op (ADR-008 §5's per-registration posture) — with +/// [`bindfirst_establisher`] as the establisher and +/// [`make_bindfirst_accept_handler`] as the pump. One establisher per +/// op id per registry (same note as +/// `register_tunnel_listen_openable`). +/// +/// `identity_witness` is the optional CF-006 probe seam (the +/// establisher records the per-call opener identity it saw). +pub fn register_tunnel_bindfirst_openable( + core: &ChannelCore, + registry: &ResourceRegistry, + on_registry: &Arc, + auth: AuthContext, + injections: BindFirstInjections, + identity_witness: Option, +) -> Result<(), String> { + let BindFirstInjections { + bind, + accept_loop, + open_forwarded, + } = injections; + let establisher = bindfirst_establisher(registry.clone(), bind); + let establisher = match identity_witness { + Some(witness) => witness_establisher(establisher, witness), + None => establisher, + }; + core.register_openable_with_establisher( + crate::params::tunnel_open_spec(), + Some(establisher), + make_bindfirst_accept_handler(accept_loop, open_forwarded), + on_registry, + auth, + None, + ) +} + +/// [`tunnel_establisher_with_witness`]-shaped CF-006 probe for the +/// bind-first establisher: records the per-call opener identity into +/// the witness, then delegates to the inner establisher (the same +/// seam the dial/direct variants inline). +fn witness_establisher( + inner: OpenEstablisher, + identity_witness: IdentityWitness, +) -> OpenEstablisher { + Arc::new(move |input: Value, auth| { + let inner = Arc::clone(&inner); + let witness = Arc::clone(&identity_witness); + Box::pin(async move { + *witness.lock().await = auth.identity.as_ref().map(|i| i.id.clone()); + inner(input, auth).await + }) + }) +} + #[cfg(test)] mod tests { use super::*; @@ -572,4 +977,221 @@ mod tests { }); assert!(other.downcast_ref::().is_none()); } + + // --- bind-first establisher (shape 3, ADR-008) ----------------------- + + /// The mint discipline: the counter seeds at 1, the first mint is + /// 1, the stream is per-open fresh + per-producer unique, and + /// token 0 is never minted (the loop skips it — a wrap that lands + /// on 0 re-increments). + #[test] + fn mint_token_seeds_nonzero_and_increments() { + let counter = std::sync::atomic::AtomicU32::new(1); + let first = mint_token(&counter); + let second = mint_token(&counter); + assert_eq!(first, 1, "a 1-seeded counter's first mint is 1"); + assert_eq!(second, 2); + // Two mints never collide (per-open fresh). + assert_ne!(first, second); + } + + /// Token 0 is never minted: a counter that has wrapped around to 0 + /// re-increments (the loop skips the illegal wire value). + #[test] + fn mint_token_never_yields_zero() { + // Force the wrap shape: seed the counter at u32::MAX — the + // next fetch_add returns u32::MAX (legal), then wraps to 0, + // which the loop must skip (the mint after the wrap is 1). + let counter = std::sync::atomic::AtomicU32::new(u32::MAX); + let before_wrap = mint_token(&counter); + assert_eq!(before_wrap, u32::MAX); + let after_wrap = mint_token(&counter); + assert_eq!(after_wrap, 1, "the wrap lands on 0; the mint skips it"); + } + + /// The establisher flow (registry gate → mint → bind → reply + /// fields → plan), pinned at the `Establishment` surface: the plan + /// is the `BindFirstPlan`, the reply fields are `bound` (the + /// `BindFn`'s address) + `listen` (the minted token), and the + /// token in the plan matches the token in the reply. + #[tokio::test] + async fn bindfirst_establisher_flow_end_to_end() { + struct TestListener(u32); + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let bind: BindFn = Arc::new(|_substrate, _backing| { + Box::pin(async { + Ok(BoundListener { + listener: Arc::new(TestListener(7)) as Arc, + bound: SubstrateAddr::HostPort { + host: "127.0.0.1".to_string(), + port: 42113, + }, + }) + }) + }); + let establisher = bindfirst_establisher(registry, bind); + let input = serde_json::json!({"resource": "listener", "substrate": "tcp"}); + let establishment = establisher(input, AuthContext::anonymous(b"alk/tunnel")) + .await + .expect("establishment"); + // The reply fields: `bound` (substrate-shaped, the BindFn's + // address) + `listen` (the token). + let fields = establishment.reply_fields().expect("reply fields"); + assert_eq!( + fields.get("bound"), + Some(&serde_json::json!({"host": "127.0.0.1", "port": 42113})) + ); + let listen = fields + .get("listen") + .and_then(|v| v.as_u64()) + .expect("listen"); + assert!(listen > 0, "the minted token is positive"); + // The plan: the BindFirstPlan carries the SAME token. + let plan = establishment.plan.expect("plan"); + let plan = plan.downcast::().expect("BindFirstPlan"); + assert_eq!(plan.token as u64, listen, "plan token == reply token"); + let listener = plan + .listener + .downcast_ref::() + .expect("listener"); + assert_eq!(listener.0, 7); + } + + /// The token discipline across TWO opens on the SAME producer: the + /// establisher mints distinct tokens per open (the per-open-fresh + /// half; the per-producer-unique half is the same counter). + #[tokio::test] + async fn bindfirst_establisher_mints_distinct_tokens_per_open() { + struct TestListener; + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let bind: BindFn = Arc::new(|_substrate, _backing| { + Box::pin(async { + Ok(BoundListener { + listener: Arc::new(TestListener) as Arc, + bound: SubstrateAddr::Path { + path: "/tmp/x".to_string(), + }, + }) + }) + }); + let establisher = bindfirst_establisher(registry, bind); + let input = serde_json::json!({"resource": "listener", "substrate": "tcp"}); + let e1 = establisher(input.clone(), AuthContext::anonymous(b"alk/tunnel")) + .await + .expect("open 1"); + let e2 = establisher(input, AuthContext::anonymous(b"alk/tunnel")) + .await + .expect("open 2"); + let t1 = e1 + .reply_fields() + .and_then(|m| m.get("listen")) + .and_then(|v| v.as_u64()) + .expect("listen 1"); + let t2 = e2 + .reply_fields() + .and_then(|m| m.get("listen")) + .and_then(|v| v.as_u64()) + .expect("listen 2"); + assert_ne!(t1, t2, "two opens mint distinct tokens"); + } + + /// The registry miss → `unknown_resource` (the namespace gate — + /// the same posture as the listen establisher). + #[tokio::test] + async fn bindfirst_establisher_registry_miss_is_unknown_resource() { + let registry = ResourceRegistry::new(); + let bind: BindFn = Arc::new(|_substrate, _backing| { + Box::pin(async { + panic!("the bind closure must not run past the gate"); + }) + }); + let establisher = bindfirst_establisher(registry, bind); + let input = serde_json::json!({"resource": "ghost", "substrate": "tcp"}); + let err = establisher(input, AuthContext::anonymous(b"alk/tunnel")) + .await + .expect_err("registry miss"); + assert!(matches!(err, EstablishmentError::UnknownResource { .. })); + } + + /// The `BindFn`'s typed errors pass through: a held port (the + /// bind IS the dial) surfaces `dial_failed`; bind resource + /// exhaustion surfaces `resource_shortage`. + #[tokio::test] + async fn bindfirst_establisher_bind_errors_pass_through() { + for expected in ["dial_failed", "resource_shortage"] { + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let err = expected; + let bind: BindFn = Arc::new(move |_substrate, _backing| { + let e = match err { + "dial_failed" => TunnelEstablishError::DialFailed("held".to_string()), + _ => TunnelEstablishError::ResourceShortage("exhausted".to_string()), + }; + Box::pin(async move { Err(e) }) + }); + let establisher = bindfirst_establisher(registry, bind); + let input = serde_json::json!({"resource": "listener", "substrate": "tcp"}); + let got = establisher(input, AuthContext::anonymous(b"alk/tunnel")) + .await + .expect_err("bind failure"); + let reason = match &got { + EstablishmentError::DialFailed { .. } => "dial_failed", + EstablishmentError::ResourceShortage { .. } => "resource_shortage", + other => panic!("unexpected error shape {other:?}"), + }; + assert_eq!(reason, expected); + } + } + + /// The witness seam (`register_tunnel_bindfirst_openable`'s + /// identity_witness): the wrapping establisher records the per-call + /// opener identity, then delegates (the CF-006 probe contract). + #[tokio::test] + async fn bindfirst_witness_wrapper_records_then_delegates() { + struct TestListener; + let registry = ResourceRegistry::new(); + registry + .register("listener", Substrate::Tcp, "assembly-owned") + .await; + let bind: BindFn = Arc::new(|_substrate, _backing| { + Box::pin(async { + Ok(BoundListener { + listener: Arc::new(TestListener) as Arc, + bound: SubstrateAddr::Path { + path: "/p".to_string(), + }, + }) + }) + }); + let witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None)); + let inner = bindfirst_establisher(registry, bind); + let wrapped = witness_establisher(inner, Arc::clone(&witness)); + let input = serde_json::json!({"resource": "listener", "substrate": "tcp"}); + let auth = AuthContext { + identity: Some(alkcall::core::auth::Identity { + id: "opener-9".to_string(), + scopes: vec![], + resources: Default::default(), + }), + ..AuthContext::anonymous(b"alk/tunnel") + }; + let establishment = wrapped(input, auth).await.expect("establishment"); + assert!( + establishment.plan.is_some(), + "delegation preserved the plan" + ); + assert_eq!( + *witness.lock().await, + Some("opener-9".to_string()), + "the per-call opener identity was recorded" + ); + } } diff --git a/tasks/tunnels/bindfirst-producer.md b/tasks/tunnels/bindfirst-producer.md index 28647cb..15d5619 100644 --- a/tasks/tunnels/bindfirst-producer.md +++ b/tasks/tunnels/bindfirst-producer.md @@ -1,7 +1,7 @@ --- id: tunnels/bindfirst-producer name: Bind-first establisher + accept dispatcher + registration (ADR-008 shape 3) -status: pending +status: completed depends_on: [tunnels/bindfirst-params] scope: broad risk: high @@ -171,22 +171,22 @@ pub fn register_tunnel_bindfirst_openable( ## Acceptance Criteria -- [ ] `bindfirst_establisher`: registry gate → mint token → bind → +- [x] `bindfirst_establisher`: registry gate → mint token → bind → `bound`+`listen` reply fields → `BindFirstPlan` plan-flow -- [ ] The accept dispatcher: `accept_bi` first (R-02 telemetry), +- [x] The accept dispatcher: `accept_bi` first (R-02 telemetry), idle-stream EOF cancels the loop + tears the listener down, per-accept `pump_bidi` tasks tracked by the handler JoinHandle -- [ ] The forwarded-open caller rides the call plane +- [x] The forwarded-open caller rides the call plane (`call_with_payload` + adopt — the reverse POC's shape); the registration wiring documents where the call plane comes from -- [ ] `register_tunnel_bindfirst_openable` wires spec + establisher + +- [x] `register_tunnel_bindfirst_openable` wires spec + establisher + dispatcher + injections -- [ ] Integration tests: the full flow e2e (the acceptance gate), +- [x] Integration tests: the full flow e2e (the acceptance gate), reply-field projection, EOF teardown, per-accept isolation, typed establishment errors, token discipline -- [ ] Clippy/fmt clean; wasm32 check passes (closures injected; no +- [x] Clippy/fmt clean; wasm32 check passes (closures injected; no substrate code in the default crate) -- [ ] `cargo test` green +- [x] `cargo test` green ## References @@ -205,8 +205,128 @@ pub fn register_tunnel_bindfirst_openable( ## Notes -> Agent fills during implementation. +- **Closure signatures decided (the design pin):** `BindFn` / + `BindFirstAcceptLoop` / `ForwardedOpener` follow the type-erased + closure precedent: + - `BindFn = Arc BoxFuture<'static, + Result> + Send + Sync>` — + same `(substrate, backing)` form as `DialFn` (the registry + backing string rides the same signature; the bind closure stays + substrate-agnostic from the protocol's view). + - `BindFirstAcceptLoop = Arc) -> + BoxFuture<'static, Result<(SubstrateAddr, TargetHandle), + TunnelEstablishError>> + Send + Sync>` — the closure receives the + plan's OPAQUE listener (`Arc`, no + downcast in the protocol: the assembly downcasts to ITS concrete + type inside the closure; the same injection pattern as + `DialFn`/`AcceptFn`). A returned `Err` is LOOP-CRITICAL (the + substrate accept source broke) — the dispatcher ends the loop; + per-accept forwarded-open failures never come from this closure. + - `ForwardedOpener = Arc + BoxFuture<'static, Result<(MpscSendStream, MpscRecvStream), + TunnelEstablishError>> + Send + Sync>` — closes over the serving + session's `Arc` + the producer-side + `ChannelManager` (the reverse POC's `open_reverse_channel` + shape: `call_with_payload` on channel 0 → extract `channel_id` → + `manager.adopt_channel`). **Where the call plane comes from:** the + assembly layer holds it — it constructed the serving session + (the harness's `wire_bindfirst` shows the wiring: the + `ForwardedOpener` closure closes over the `Arc` + and calls `call_open_op` + `adopt_channel`; in production the + assembly hands its `CallConnection` the same way). +- **Registration signature deviation (documented):** the task sketch + had 8 parameters (clippy's `too_many_arguments`); the injected + seams bundle into `BindFirstInjections {bind, accept_loop, + open_forwarded}` — one coherent injection point passed beside the + core pair + the optional witness. Same posture as `DialFn` riding + the dial registration. The `identity_witness` seam is wired via a + small `witness_establisher` wrapper (records the per-call opener + identity, then delegates — the same CF-006 seam the dial/direct + variants inline). +- **The token mint** (`mint_token`, private): `AtomicU32` counter + seeded at 1 inside the establisher's closure state; `fetch_add` + returns the PREVIOUS value, so the minted stream is 1, 2, 3, … — + token 0 (an illegal wire value) is never minted, and a u32 wrap + that lands on 0 re-increments (the loop skips it). Per-open fresh, + per-producer unique; unit-pinned including the wrap shape. +- **The idle-stream EOF watch:** a 1-byte `read` on the split idle + half inside a `select!` against the accept loop — `Ok(0)` (the + zero-byte EOF sentinel), a read error, or (defensively) any + non-empty read on an idle stream all cancel the loop. The listen + channel's stream is never pumped (no `pump_bidi` on it). +- **Per-accept failure posture (v1, documented):** a forwarded-open + failure (scope-less opener → FORBIDDEN; old opener → NOT_FOUND; + correlation miss → handler_error-class) logs a warning and drops + that accepted connection — the loop continues; NO retry/backoff + (v1 posture per the task). +- **Lifetime (R-02):** the handler collects every per-accept + `pump_bidi` `JoinHandle` in a `Vec` and awaits them all on loop + exit (before the handler task returns) — a forwarded pump never + outlives its listener. Handler exit → wrapper teardown → the listen + channel reaps → the plan's listener drops with the loop. +- **Harness changes (test infrastructure):** `wire_bindfirst` + (RegistrationMode::BindFirst); the consumer's install hook gained a + shared serving registry + the accept-side identity propagation + (the transport identity attaches to channel 0 — the same posture + `wire_forward` models; needed because the forwarded opens arrive + ON the consumer's serving path); `Topology.consumer_registry` + exposed; the forwarded-op serving stub + (`register_forwarded_stub_on_consumer` / + `make_forwarded_stub_handler` — correlate by `listen` token → + accept the `BiStream` → `pump_bidi` against the correlated halves; + the opener-side registration shape lands for real with + `tunnels/forwarded-receipt`). +- **Test-data-plane shape:** the e2e round-trip drives the local + BIND client end directly (the correlation map's halves are one end + of a raw duplex; the stub's pump bridges forwarded channel ↔ those + halves; the accepted side's echo task closes the loop). The + forwarded channel on the opener side is wrapper-managed (the + serving wrapper allocated it — `adopt_channel` on it a second time + fails `ChannelExists`), so the test drives the client end, not a + re-adopt. +- **EOF-teardown observability:** the test pins the listen channel + reaping on the producer side after the opener drops its session + (`has_channel` false) and the accept loop's death via the queue + (a follow-on push is never consumed). ## Summary -> Agent fills this on completion. \ No newline at end of file +Implemented the bind-first producer (shape 3, ADR-008 as amended) per +the task: `src/producer.rs` — `BindFn` + `BoundListener` (the +platform-boundary bind seam); the token mint (non-zero-seeded +`AtomicU32`, per-open fresh, per-producer unique, token 0 never +minted — the wrap shape unit-pinned); `bindfirst_establisher` +(registry namespace gate → mint → bind → `bound` + `listen` reply +fields → the `BindFirstPlan` plan; the typed errors pass through — +bind failure `dial_failed`, exhaustion `resource_shortage`, registry +miss `unknown_resource`); `BindFirstAcceptLoop` (the type-erased +accept closure receiving the opaque listener); `ForwardedOpener` (the +call-plane forwarded-open caller: call + adopt); +`make_bindfirst_accept_handler` (the accept dispatcher: `accept_bi` +first, the idle-stream EOF sentinel cancels the loop via `select!`, +per-accept `pump_bidi` in spawned tasks tracked by the handler's +`JoinHandle`, per-accept failure isolation, loop-critical accept +errors end the loop); `register_tunnel_bindfirst_openable` (same +spec + scope gate, `BindFirstInjections` bundle + witness seam) — +re-exported through `lib.rs`. `tests/harness.rs` — +`RegistrationMode::BindFirst` + `wire_bindfirst` (the consumer hook +gained the accept-side identity propagation + a shared serving +registry, `Topology.consumer_registry` exposed) + the forwarded-op +serving stub (`make_forwarded_stub_handler`). `tests/ +producer_bindfirst.rs` — 6 integration tests: the full flow e2e (the +acceptance gate: reply projection → token correlation → forwarded +open → byte round-trip through the whole path), the dial-open +neither-field projection, EOF teardown (listen reaped, loop gone), +per-accept isolation (scope-less opener; the loop serves the next +accept), typed establishment errors, and the token discipline (two +opens distinct tokens; each dispatcher stamps its own plan's token). +Plus 6 unit tests in producer.rs (the mint discipline incl. the wrap +shape, the establisher flow at the `Establishment` surface — plan +token == reply token — distinct tokens per open, registry miss, +bind-error passthrough, the witness wrapper). + +Verified: `cargo test` green (118 default; 138 `--all-features`), +clippy `--all-targets -D warnings` clean (native + wasm32), fmt +clean, `wasm32-unknown-unknown` check passes (closures injected; the +default crate stays wasm-clean), `cargo doc --no-deps` clean. +CHANGELOG updated (Unreleased → Added). \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs index a26b7ee..41e738b 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -22,6 +22,7 @@ //! dialing). CF-006 (the per-call opener identity on the establisher) //! is probed via the `identity_witness` the establisher records. +use serde_json::Value; use std::sync::Arc; use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; @@ -34,7 +35,9 @@ use alkcall::protocol::connection::{split_single_stream, CallConnection}; use alkcall::protocol::dispatch::Dispatcher; use alkcall::registry::registration::OperationRegistry; -use alktunnels::params::{Substrate, TUNNEL_DIRECT_SCOPE, TUNNEL_OPEN_SCOPE}; +use alktunnels::params::{ + Substrate, TUNNEL_DIRECT_SCOPE, TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE, +}; use alktunnels::producer::{ register_tunnel_direct_openable, register_tunnel_openable, ResourceRegistry, }; @@ -107,6 +110,9 @@ pub struct Topology { pub consumer_call: Arc, /// The producer's serving registry (probes). pub producer_registry: Arc, + /// The consumer's serving registry (the opener-side op stubs — + /// e.g. the forwarded-op serving stub — pre-register here). + pub consumer_registry: Arc, /// What the establisher's per-call auth carried as the opener /// identity id (the CF-006 probe — `None` until an open runs). pub identity_witness: Arc>>, @@ -136,6 +142,16 @@ pub enum RegistrationMode { /// rides the topology's `identity_witness` (the direct op's /// witness and the base op's never mix in one test). Direct(alktunnels::producer::DialFn), + /// The bind-first establisher (shape 3, ADR-008): the bind-first + /// establisher + the accept dispatcher over the injected bind + + /// accept loop + forwarded-open caller. The forwarded-op caller + /// rides the producer-side call plane (`producer_client`'s call + /// connection — captured in the topology) and adopts on the + /// producer-side manager. + BindFirst( + alktunnels::producer::BindFn, + alktunnels::producer::BindFirstAcceptLoop, + ), } pub async fn wire_with( @@ -151,28 +167,47 @@ pub async fn wire_with( let (consumer_manager_tx, mut consumer_manager_rx) = tokio::sync::mpsc::channel::(1); + // The consumer's serving registry: empty by default; a test that + // installs opener-side ops (the forwarded-op serving stub — the + // ADR-008 §3 ingress shape) pre-registers them here so the + // dispatch loop serves them from the first inbound call. + let consumer_serving_registry = Arc::new(OperationRegistry::new()); + let registry_for_hook = Arc::clone(&consumer_serving_registry); + let identity_for_hook = transport_identity.clone(); + // --- consumer (accept side): ChannelsAdapter + capturing hook ------ let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { let call_tx = consumer_call_tx.clone(); let manager_tx = consumer_manager_tx.clone(); + let serving_registry = Arc::clone(®istry_for_hook); + let peer_identity = identity_for_hook.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 same + // propagation `wire_forward` models): the transport + // authenticated the dialer; attach the resolved identity + // to channel 0 so the serving dispatch sees the caller + // (CF-005 (b)'s accept-side analogue). + if let Some(identity) = peer_identity { + let _ = channel0_conn.set_identity(identity); + } let call_connection = Arc::new(CallConnection::new_single_stream( channel0_conn, Arc::clone(&writer), )); let _ = call_tx.send(Arc::clone(&call_connection)).await; let _ = manager_tx.send(manager).await; - // The consumer serves no ops — an empty registry. The - // dispatcher keeps the call loop alive (its read loop - // resolves the producer's responses to the consumer's - // pending entries). + // The consumer serves the ops the test pre-registered on + // its serving registry (empty in the default topology — + // the dispatcher keeps the call loop alive either way; its + // read loop resolves the producer's responses to the + // consumer's pending entries). let dp = Dispatcher::new( - Arc::new(OperationRegistry::new()), + serving_registry, Arc::new(alkcall::core::auth::NoopIdentityProvider), ); dp.serve_single_stream(call_connection, reader, writer) @@ -231,6 +266,7 @@ pub async fn wire_with( producer_client.manager().clone(), alkcall::channels::policy::default_policy(), ); + let producer_client = Arc::new(producer_client); match mode { RegistrationMode::Dial(dial) => { register_tunnel_openable( @@ -263,6 +299,75 @@ pub async fn wire_with( ) .expect("register tunnel listen openable"); } + RegistrationMode::BindFirst(bind, accept_loop) => { + // The bind-first shape (ADR-008). The forwarded-op caller + // closes over the producer-side call plane + manager (the + // design pin: the per-channel `Connection` the handler + // receives is stream-less and carries no call surface, so + // the registration takes the call-plane handle — here the + // serving session's `CallConnection`, which the client + // holds; in production the assembly layer has the same + // handle because it constructed the serving session). The + // harness reaches it via `producer_client.call_open_op` + + // `adopt_channel` — the reverse POC's + // `open_reverse_channel` shape, inline (the reply's extra + // fields are not needed for the forwarded op; the channel + // ID extraction + adopt is the whole dance). + let manager = producer_client.manager().clone(); + let client_for_opener = Arc::clone(&producer_client); + let open_forwarded: alktunnels::producer::ForwardedOpener = Arc::new( + move |forwarded: alktunnels::params::TunnelForwardedParams| { + let client = Arc::clone(&client_for_opener); + let manager = manager.clone(); + Box::pin(async move { + let input = serde_json::to_value(&forwarded).map_err(|e| { + alktunnels::producer::TunnelEstablishError::HandlerError(format!( + "forwarded params serialize: {e}" + )) + })?; + let response = client + .call_open_op(alktunnels::params::OP_TUNNEL_FORWARDED, input) + .await; + let out = response.result.map_err(|e| { + alktunnels::producer::TunnelEstablishError::HandlerError(format!( + "forwarded open: {}", + e.message + )) + })?; + let channel_id = + out.get("channel_id") + .and_then(|v| v.as_u64()) + .ok_or_else(|| { + alktunnels::producer::TunnelEstablishError::HandlerError( + "forwarded open reply missing channel_id".to_string(), + ) + })? as u32; + let (send, recv) = manager + .adopt_channel(channel_id, alktunnels::params::TUNNEL_ALPN, None) + .await + .map_err(|e| { + alktunnels::producer::TunnelEstablishError::HandlerError(format!( + "forwarded adopt: {e}" + )) + })?; + Ok((send, recv)) + }) + }, + ); + alktunnels::producer::register_tunnel_bindfirst_openable( + &core, + ®istry, + &producer_op_registry, + AuthContext::anonymous(b"alk/tunnel"), + alktunnels::producer::BindFirstInjections { + bind, + accept_loop, + open_forwarded, + }, + Some(Arc::clone(&identity_witness)), + ) + .expect("register tunnel bindfirst openable"); + } RegistrationMode::DialNoWitness(dial) => { // The public no-witness wrapper (tunnel_establisher) — // registration + establisher body identical to the @@ -320,7 +425,6 @@ pub async fn wire_with( .expect("register hanging establisher"); } } - let producer_client = Arc::new(producer_client); // --- consumer pieces (captured from the hook) ------------------------- let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn"); @@ -331,6 +435,7 @@ pub async fn wire_with( consumer_manager, consumer_call, producer_registry: producer_op_registry, + consumer_registry: consumer_serving_registry, identity_witness, } } @@ -413,6 +518,137 @@ pub async fn wire_listen( .await } +/// The bind-first topology (shape 3, ADR-008): the producer's +/// establisher binds via the injected [`BindFn`]; the accept +/// dispatcher runs the injected accept loop and opens forwarded +/// channels back toward the opener (whose serving stub the test +/// installs on the consumer side — see +/// [`serve_forwarded_on_consumer`]). Same identity posture as +/// [`wire`], plus the forwarded scope (the opener-side ingress grant +/// — the consumer identity must hold `tunnel:forwarded` to receive +/// producer-initiated forwarded opens). +pub async fn wire_bindfirst( + registry: ResourceRegistry, + bind: alktunnels::producer::BindFn, + accept_loop: alktunnels::producer::BindFirstAcceptLoop, +) -> Topology { + wire_with( + registry, + RegistrationMode::BindFirst(bind, accept_loop), + Some(forwarded_opener_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +/// The forwarded-op opener identity: `tunnel:forwarded` AND +/// `tunnel:open` (the opener receives producer-initiated forwarded +/// opens — the ingress grant — and opens the listen tunnel itself — +/// the egress grant; ADR-008 §3 / ADR-006 Amendment 1). +pub fn forwarded_opener_identity() -> Identity { + Identity { + id: "opener".to_string(), + scopes: vec![ + TUNNEL_OPEN_SCOPE.to_string(), + TUNNEL_FORWARDED_SCOPE.to_string(), + ], + resources: std::collections::HashMap::new(), + } +} + +/// [`forwarded_opener_identity`] minus the forwarded grant (the +/// per-accept isolation probe's identity — the forwarded open fails +/// FORBIDDEN). +pub fn forwarded_scopeless_identity() -> Identity { + Identity { + id: "opener-scopeless".to_string(), + scopes: vec![TUNNEL_OPEN_SCOPE.to_string()], + resources: std::collections::HashMap::new(), + } +} + +/// Serve the forwarded op (`channels/tunnel/forwarded`, ADR-008 §3) on +/// the CONSUMER side (the listen opener's serving stub — the test +/// stand-in for the opener-side registration that lands with +/// `tunnels/forwarded-receipt`): a `register_openable` (no +/// establisher — the data plane starts immediately) whose handler +/// correlates the open by the `listen` token against the test's +/// correlation map, accepts the `BiStream`, and pumps against the +/// "local BIND client" halves the test registered for that token. +/// +/// `correlations` maps `listen` token → the pump side to pump the +/// forwarded channel against (the assembly-side bookkeeping posture — +/// the protocol provides the op + params, the test provides the map). +/// The registration is post-hoc on the consumer's captured registry — +/// but the CONSUMER serves no ops in the default harness (its +/// dispatch registry is empty and the dispatcher runs on it), so this +/// helper must be called BEFORE the topology is wired; hence it is a +/// registry-level helper the caller wires via a custom `wire_with` +/// variant when needed. +pub fn register_forwarded_stub_on_consumer( + consumer_registry: &Arc, + manager: ChannelManager, + correlations: CorrelationMap, +) -> Result<(), String> { + let core = alkcall::channels::operations::ChannelCore::new( + manager, + alkcall::channels::policy::default_policy(), + ); + core.register_openable( + alktunnels::params::tunnel_forwarded_spec(), + make_forwarded_stub_handler(correlations), + consumer_registry, + AuthContext::anonymous(b"alk/tunnel"), + ) +} + +/// The correlation map: `listen` token → the TargetHandle the +/// forwarded channel pumps against (the "local BIND client" halves). +pub type CorrelationMap = + Arc>>; + +/// The forwarded-op handler (the opener-side serving stub's handler): +/// correlate by `listen` → accept the `BiStream` → pump against the +/// correlated halves (`pump_bidi`, convention 8). A correlation miss +/// is a loud `handler_error`-class open failure (the typed posture — +/// the accept dies at birth; the loop keeps serving). +pub fn make_forwarded_stub_handler( + correlations: CorrelationMap, +) -> alkcall::channels::operations::OpenHandler { + Arc::new( + move |input: Value, + _plan: Option, + conn: alkcall::core::types::Connection, + _auth: AuthContext| { + let correlations = Arc::clone(&correlations); + tokio::spawn(async move { + let forwarded: alktunnels::params::TunnelForwardedParams = + match serde_json::from_value(input) { + Ok(p) => p, + Err(e) => { + tracing::error!(error = %e, "forwarded stub: params parse failed"); + return; + } + }; + let target = correlations.lock().await.remove(&forwarded.listen); + let Some(target) = target else { + tracing::error!( + listen = forwarded.listen, + "forwarded stub: correlation miss" + ); + return; + }; + let Some(bidi) = conn.accept_bi().await.ok() else { + tracing::error!("forwarded stub: accept_bi failed"); + return; + }; + alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).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 + diff --git a/tests/producer_bindfirst.rs b/tests/producer_bindfirst.rs new file mode 100644 index 0000000..b9cc2df --- /dev/null +++ b/tests/producer_bindfirst.rs @@ -0,0 +1,630 @@ +//! Integration tests for the bind-first establisher + accept +//! dispatcher (shape 3 — ADR-008 §1): the producer whose establisher +//! BINDS (via the injected `BindFn`), mints the correlation token, +//! contributes `bound` + `listen` to the open reply, and whose pump +//! handler is the accept dispatcher — per accept it opens +//! `channels/tunnel/forwarded` back toward the listen opener and pumps +//! the accepted halves against the adopted channel. +//! +//! Covered (the task's test plan): the full bind-first flow e2e (the +//! acceptance gate), the reply-field projection (both fields seen; a +//! dial open sees neither), EOF teardown (the idle-stream sentinel +//! cancels the loop; the far listener is gone), per-accept failure +//! isolation (a rejected forwarded open drops that accept; the loop +//! serves the next), the typed establishment errors (bind failure → +//! `dial_failed`; registry miss → `unknown_resource`), and the token +//! discipline (two opens mint distinct tokens; each dispatcher stamps +//! ITS OWN plan's token). + +mod harness; + +use tracing_subscriber::EnvFilter; + +fn init_tracing() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); +} + +use std::sync::Arc; + +use alkcall::protocol::wire::CallError; +use alktunnels::params::{parse_listen_reply, Substrate, SubstrateAddr, TunnelParams}; +use alktunnels::producer::{ + BindFirstAcceptLoop, BindFn, BoundListener, ResourceRegistry, TargetHandle, + TunnelEstablishError, +}; + +use harness::{ + make_forwarded_stub_handler, wire, wire_bindfirst, wire_with, CorrelationMap, RegistrationMode, + Topology, +}; + +fn listen_params(resource: &str) -> TunnelParams { + TunnelParams { + resource: resource.to_string(), + substrate: Substrate::Tcp, + } +} + +/// Call the producer's open op on channel 0 with the FULL reply +/// (the consumer's bind-first surface — the reply projection probe). +async fn open_listen_with_reply( + topo: &Topology, + params: &TunnelParams, +) -> Result<(u32, serde_json::Value), CallError> { + let payload = serde_json::json!({ + "operationId": "channels/tunnel/sub", + "input": serde_json::json!({ + "resource": params.resource, + "substrate": "tcp", + }), + }); + let response = topo.consumer_call.call_with_payload(payload).await; + let out = response.result?; + let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32; + Ok((channel_id, out)) +} + +/// The in-process "accepted connection" the test accept loop yields: +/// a duplex pair, the far end an echo task (the same shape the dial +/// harness echoes with). The peer address is the loop's choice. +fn accepted_halves(peer_addr: (u32, u16)) -> (SubstrateAddr, TargetHandle) { + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + let (mut t_read, mut t_write) = tokio::io::split(target_side); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buf = vec![0u8; 8192]; + loop { + match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if t_write.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + let (c_read, c_write) = tokio::io::split(consumer_side); + ( + SubstrateAddr::HostPort { + host: peer_addr.0.to_string(), + port: peer_addr.1, + }, + TargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }, + ) +} + +/// The test accept loop's feed point (the test plays the app server): +/// push an accepted pair; the dispatcher's loop picks it up. +async fn push_accept( + queue: &Arc>>, + notify: &Arc, + peer: (u32, u16), +) { + let (peer_addr, handle) = accepted_halves(peer); + queue.lock().await.push_back((peer_addr, handle)); + notify.notify_waiters(); +} + +/// The assembly's bind: produces a "listener" (the queue's identity) +/// and the OS-chosen bound address (the test's fake — a real +/// `bind_tcp` variant lives in the `local` feature). +fn test_bind() -> BindFn { + Arc::new(|_substrate, _backing| { + Box::pin(async { + Ok(BoundListener { + listener: Arc::new(4242u32) as Arc, + bound: SubstrateAddr::HostPort { + host: "127.0.0.1".to_string(), + port: 42113, + }, + }) + }) + }) +} + +/// A failing bind (the `dial_failed`-class probe — the bind IS the +/// dial here). +fn failing_bind(message: &'static str) -> BindFn { + Arc::new(move |_substrate, _backing| { + Box::pin(async move { Err(TunnelEstablishError::DialFailed(message.to_string())) }) + }) +} + +fn only_channel_0(topo: &Topology) -> bool { + topo.consumer_manager + .channel_ids() + .iter() + .all(|&id| id == 0) + && topo + .producer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0) +} + +/// The bind-first topology keeps the LISTEN channel open (the +/// dispatcher holds it) — the "no stray channels" probe excludes it +/// (and channel 0, the call plane) on both sides. +fn no_stray_channels(topo: &Topology, listen_id: u32) -> bool { + topo.consumer_manager + .channel_ids() + .iter() + .all(|&id| id == 0 || id == listen_id) + && topo + .producer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0 || id == listen_id) +} + +/// The forwarded-op serving stub on the opener (consumer) side + +/// its correlation map (the assembly-side bookkeeping the real +/// opener-side registration owns). +async fn install_forwarded_stub(topo: &Topology) -> CorrelationMap { + let correlations: CorrelationMap = Arc::new(tokio::sync::Mutex::new(Default::default())); + let core = alkcall::channels::operations::ChannelCore::new( + topo.consumer_manager.clone(), + alkcall::channels::policy::default_policy(), + ); + core.register_openable( + alktunnels::params::tunnel_forwarded_spec(), + make_forwarded_stub_handler(Arc::clone(&correlations)), + &topo.consumer_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), + ) + .expect("register forwarded stub on the opener"); + correlations +} + +#[tokio::test] +async fn bindfirst_full_flow_end_to_end() { + init_tracing(); + // THE acceptance gate (ADR-008): opener opens `channels/tunnel/sub` + // with the bind-first producer registered → the reply carries + // `bound` + `listen` → the opener serves `channels/tunnel/forwarded` + // with a correlation map keyed on the token → the test accept loop + // accepts → the producer's dispatcher opens forwarded with + // `{listen: token, substrate, peer}` → the opener correlates, + // pumps against the local BIND client halves → byte round-trip + // through the forwarded channel. + let registry = ResourceRegistry::new(); + registry + .register("bind", Substrate::Tcp, "assembly-owned") + .await; + let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let topo = wire_bindfirst( + registry, + test_bind(), + harness_queue_loop(queue.clone(), notify.clone()), + ) + .await; + + // The opener opens the listen tunnel (with-reply surface — the + // bind-first reply projection). + let (channel_id, reply) = open_listen_with_reply(&topo, &listen_params("bind")) + .await + .expect("listen open"); + + // Reply-field projection: `bound` + `listen` ride the success + // output alongside the wrapper's `channel_id`. + let projected = parse_listen_reply(&reply).expect("projected reply"); + assert_eq!( + projected.bound, + Some(SubstrateAddr::HostPort { + host: "127.0.0.1".to_string(), + port: 42113, + }), + "the `bound` field is the OS-chosen (test-chosen) address" + ); + let token = projected.listen.expect("the `listen` token"); + assert!(token > 0); + + // The opener correlates on the token (the assembly-side map). + let correlations = install_forwarded_stub(&topo).await; + + // The LOCAL BIND client (the app-server-side session on the + // opener): a raw duplex pair — the map's halves are the stub's + // pump target, the far end is the test's client end. The stub's + // `pump_bidi` bridges the forwarded channel ↔ these halves, so + // bytes the test writes on the far end traverse the whole path + // (local BIND client ↔ opener stub pump ↔ forwarded channel ↔ + // producer-side pump ↔ accepted handle's echo task) and echo back. + let (client_read, client_write, l_read, l_write) = bind_client_pair(); + correlations.lock().await.insert( + token, + TargetHandle { + read: Box::new(l_read), + write: Box::new(l_write), + }, + ); + + // The opener adopts the LISTEN channel (the idle side — its EOF + // is the teardown signal). + topo.consumer_manager + .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None) + .await + .expect("adopt listen channel"); + + // The test accept loop accepts (the app server dialed). + push_accept(&queue, ¬ify, (198, 55231)).await; + + // The dispatcher opens forwarded; the opener's stub handler + // correlates by token and pumps against the local halves. The + // forwarded channel ID arrives on the opener's manager — find it + // (the only non-0, non-listen channel on the opener side after + // the adopt + forwarded open settle). + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let forwarded_ids: Vec = topo + .consumer_manager + .channel_ids() + .into_iter() + .filter(|&id| id != 0 && id != channel_id) + .collect(); + assert_eq!(forwarded_ids.len(), 1, "one forwarded channel arrived"); + + // The round-trip: bytes the test writes on the local BIND + // client's end cross the stub's pump → the forwarded channel → + // the producer's per-accept pump → the accepted handle's echo → + // and back (the full bind-first data path, ADR-008's gate). + let (mut client_read, mut client_write) = (client_read, client_write); + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + client_write.write_all(b"bind-ping").await.expect("write"); + client_write.flush().await.expect("flush"); + let mut buf = vec![0u8; 9]; + tokio::time::timeout( + std::time::Duration::from_secs(5), + client_read.read_exact(&mut buf), + ) + .await + .expect("round trip") + .expect("read"); + assert_eq!(&buf, b"bind-ping"); + + // Teardown follows the accepted side: the far echo closes when the + // accepted handle's pump sees the test client's EOF. (No channel + // assertions here — the listen channel legitimately stays open.) +} + +/// The listen-channel EOF teardown: drop the listener session (close +/// the listen channel's stream) → the accept loop cancels, the far +/// listener is gone (a follow-on accept attempt fails), and the +/// forwarded channels already adopted own their teardown +/// independently. +#[tokio::test] +async fn listen_channel_eof_cancels_loop_and_tears_listener_down() { + let registry = ResourceRegistry::new(); + registry + .register("bind", Substrate::Tcp, "assembly-owned") + .await; + let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let topo = wire_bindfirst( + registry, + test_bind(), + harness_queue_loop(queue.clone(), notify.clone()), + ) + .await; + + let (channel_id, reply) = open_listen_with_reply(&topo, &listen_params("bind")) + .await + .expect("listen open"); + let token = parse_listen_reply(&reply) + .expect("projected") + .listen + .expect("token"); + + // The opener adopts the listen channel and then DROPS its stream + // (the listener session is dropped — the zero-byte EOF sentinel). + let (send, _recv) = topo + .consumer_manager + .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None) + .await + .expect("adopt listen channel"); + drop(send); + + // The opener-side teardown reaps the listen channel. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + !topo.producer.manager().has_channel(channel_id), + "the listen channel reaped after the session dropped" + ); + + // A follow-on accept attempt: the loop is gone — the accept queue + // accepts nothing new (the listener was torn down with the plan). + // Observable via the queue: nothing pumps it anymore. + let _ = token; + push_accept(&queue, ¬ify, (198, 55232)).await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert_eq!( + queue.lock().await.len(), + 1, + "the accept loop is gone: the pushed accept was never consumed" + ); +} + +/// Per-accept failure isolation: the opener's forwarded registry +/// rejects one forwarded open (scope-less identity) → that accept's +/// connection drops, the loop serves the next accept. +#[tokio::test] +async fn forwarded_open_failure_isolates_the_accept() { + init_tracing(); + // The topology's consumer identity holds `tunnel:open` but NOT + // `tunnel:forwarded` — every forwarded open fails FORBIDDEN at the + // opener's ACL. The loop must keep serving (a second accept + // produces a second forwarded-open attempt). + let registry = ResourceRegistry::new(); + registry + .register("bind", Substrate::Tcp, "assembly-owned") + .await; + let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let topo = wire_with( + registry, + RegistrationMode::BindFirst( + test_bind(), + harness_queue_loop(queue.clone(), notify.clone()), + ), + Some(harness::forwarded_scopeless_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + let (listen_id, _reply) = open_listen_with_reply(&topo, &listen_params("bind")) + .await + .expect("listen open"); + + // First accept: the forwarded open fails (scope-less opener) — + // the accepted connection drops, the loop continues. + push_accept(&queue, ¬ify, (198, 55233)).await; + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + assert!( + no_stray_channels(&topo, listen_id), + "a failed forwarded open leaves no channel beyond the listen channel" + ); + + // Second accept: the loop is still alive (per-accept isolation) — + // this one fails the same way, but the LOOP's health is what the + // test pins: the accept was CONSUMED (the queue drained). + push_accept(&queue, ¬ify, (198, 55234)).await; + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let remaining = queue.lock().await.len(); + assert_eq!( + remaining, 0, + "the second accept was consumed too (the loop served it)" + ); + assert!(no_stray_channels(&topo, listen_id)); +} + +/// Establishment error paths: bind failure → `dial_failed`-class on +/// the wire (`channel:open_failed`); registry miss → +/// `unknown_resource`. +#[tokio::test] +async fn establishment_errors_are_typed_on_the_wire() { + // Bind failure → dial_failed (the bind IS the dial here). + let registry = ResourceRegistry::new(); + registry + .register("bind", Substrate::Tcp, "assembly-owned") + .await; + let topo = wire_bindfirst( + registry, + failing_bind("address already in use"), + noop_loop(), + ) + .await; + match open_listen_with_reply(&topo, &listen_params("bind")).await { + Ok(_) => panic!("bind failure must fail the open"), + Err(e) => { + assert_eq!(e.code, "channel:open_failed"); + let reason = e + .details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) + .expect("reason in details"); + assert_eq!(reason, "dial_failed"); + } + } + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo), "a failed open leaves no channel"); + + // Registry miss → unknown_resource. + let registry = ResourceRegistry::new(); + let topo = wire_bindfirst(registry, test_bind(), noop_loop()).await; + match open_listen_with_reply(&topo, &listen_params("ghost")).await { + Ok(_) => panic!("registry miss must fail the open"), + Err(e) => { + assert_eq!(e.code, "channel:open_failed"); + let reason = e + .details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) + .expect("reason in details"); + assert_eq!(reason, "unknown_resource"); + } + } + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!(only_channel_0(&topo)); +} + +/// Token discipline: two opens on the same producer mint distinct +/// tokens; each dispatcher stamps ITS OWN plan's token (the second +/// open's forwarded opens carry the second token — pinned via the +/// correlation stub's map: the FIRST token's entry is left +/// unconsumed). +#[tokio::test] +async fn token_discipline_two_opens_distinct_tokens() { + let registry = ResourceRegistry::new(); + registry + .register("bind", Substrate::Tcp, "assembly-owned") + .await; + let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let topo = wire_bindfirst( + registry, + test_bind(), + harness_queue_loop(queue.clone(), notify.clone()), + ) + .await; + let correlations = install_forwarded_stub(&topo).await; + + let (id1, reply1) = open_listen_with_reply(&topo, &listen_params("bind")) + .await + .expect("open 1"); + let (id2, reply2) = open_listen_with_reply(&topo, &listen_params("bind")) + .await + .expect("open 2"); + assert_ne!(id1, id2); + let t1 = parse_listen_reply(&reply1) + .expect("reply 1") + .listen + .expect("token 1"); + let t2 = parse_listen_reply(&reply2) + .expect("reply 2") + .listen + .expect("token 2"); + assert_ne!(t1, t2, "two opens mint distinct tokens"); + + // Adopt both listen channels (both sessions live). + topo.consumer_manager + .adopt_channel(id1, alktunnels::TUNNEL_ALPN, None) + .await + .expect("adopt listen 1"); + topo.consumer_manager + .adopt_channel(id2, alktunnels::TUNNEL_ALPN, None) + .await + .expect("adopt listen 2"); + + // Both dispatchers are live. Correlate BOTH tokens (each opener + // session keys its own), but only register the local halves for + // t1 — a forwarded open stamped t2 would correlation-miss. Push + // one accept AFTER open 2: whichever dispatcher consumed it, the + // STAMPED token is the plan's own. To pin "each dispatcher stamps + // ITS OWN plan's token", correlate both but track which token the + // forwarded open carried. + correlations.lock().await.insert(t1, { + let (l_read, l_write, _, _) = bind_client_pair(); + TargetHandle { + read: Box::new(l_read), + write: Box::new(l_write), + } + }); + correlations.lock().await.insert(t2, { + let (l_read, l_write, _, _) = bind_client_pair(); + TargetHandle { + read: Box::new(l_read), + write: Box::new(l_write), + } + }); + + push_accept(&queue, ¬ify, (198, 55235)).await; + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + // The accept was consumed by exactly one dispatcher, and the + // forwarded open succeeded (correlated by its stamped token). + assert_eq!(queue.lock().await.len(), 0, "one accept consumed"); + assert_eq!( + correlations.lock().await.len(), + 1, + "one token's halves were consumed" + ); +} + +/// A plain dial open sees NEITHER reply field (the projection +/// property — the reserved-key guard upstream guarantees `channel_id` +/// stays wrapper-owned; a dial establisher contributes nothing). +#[tokio::test] +async fn dial_open_reply_has_neither_field() { + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, harness::echo_dial()).await; + let payload = serde_json::json!({ + "operationId": "channels/tunnel/sub", + "input": {"resource": "echo", "substrate": "tcp"}, + }); + let response = topo.consumer_call.call_with_payload(payload).await; + let out = response.result.expect("dial open"); + let projected = parse_listen_reply(&out).expect("projected"); + assert_eq!( + projected, + alktunnels::params::TunnelListenReply { + bound: None, + listen: None + }, + "a dial establisher's open carries neither field" + ); +} + +/// The accept loop's queue-shaped stand-in (the harness_queue_loop in +/// the tests above): a closure popping from the test's VecDeque + +/// Notify pair. (Named separately from the generic helper so the +/// tests read: queue in, loop out.) +fn harness_queue_loop( + queue: Arc>>, + notify: Arc, +) -> BindFirstAcceptLoop { + Arc::new(move |_listener| { + let queue = Arc::clone(&queue); + let notify = Arc::clone(¬ify); + Box::pin(async move { + // The queue-driven accept: check once, wait for a push, + // check again (the Notify lost-wakeup pattern — the + // `Notified` future is created BEFORE the state check on + // the waiting path). One accepted pair per closure call + // (the dispatcher invokes the closure per accept). + let popped = { + let notified = std::pin::pin!(notify.notified()); + let mut q = queue.lock().await; + if let Some(pair) = q.pop_front() { + Some(pair) + } else { + drop(q); + notified.await; + let mut q = queue.lock().await; + q.pop_front() + } + }; + match popped { + Some(pair) => Ok(pair), + None => Err(TunnelEstablishError::DialFailed( + "listener closed".to_string(), + )), + } + }) + }) +} + +/// A never-accepting loop (the establishment-error probes — the loop +/// never runs: the open fails at the establisher). +fn noop_loop() -> BindFirstAcceptLoop { + Arc::new(|_listener| { + Box::pin(async { Err(TunnelEstablishError::DialFailed("noop loop".to_string())) }) + }) +} + +/// The "local BIND client" pair (the opener-side pump target): a raw +/// duplex pair — the map's halves are the stub's pump target, the far +/// end is the test's client end (driven directly; the accepted side's +/// echo task closes the loop). +fn bind_client_pair() -> ( + tokio::io::ReadHalf, + tokio::io::WriteHalf, + tokio::io::ReadHalf, + tokio::io::WriteHalf, +) { + let (a, b) = tokio::io::duplex(64 * 1024); + let (a_read, a_write) = tokio::io::split(a); + let (b_read, b_write) = tokio::io::split(b); + (b_read, b_write, a_read, a_write) +}