From 50182d72983495d69f6781dfe39e613656b4c946 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 18 Sep 2026 05:15:29 +0000 Subject: [PATCH] feat(review 008 Unit 3c): the gate-2 e2e harness; 0.8.0 release bookkeeping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/channels/gate2_tests.rs (ADR-051 gates 6/7, review 008 gate 2): the full consumer -> hub (HubLegTemplate) -> spoke relay e2e — the open resolves with the hub-allocated channel_id, `bound` survives the relay, data flows both directions with a fake 8-byte chunk header riding verbatim (the hub never parses the data plane, ADR-034/035), spoke-side close cascades to clean reclaim on both legs; hub-side disconnect (the consumer's duplex end dropped via a killable transport proxy) tears down both legs with the ledger decremented; the mid-establishment window (ADR-051 §6) pinned with its two reclaim signals — the consumer leg reclaims at its own transport EOF, the spoke channel (allocated before the establisher replied) is the honest residual, reclaimed when the spoke-leg transport ends; the channels/tty/sub standard-shape companion pins no derivation regression. - src/channels/relay.rs: the relay's adopted producer-leg channel entry now reclaims when the relayed pump completes (teardown after pump_bidi) — the consumer-leg wrapper's teardown cannot see the producer leg's manager; the RelayPlan carries the spoke id for the reclaim. - Release bookkeeping: 0.7.1 -> 0.8.0, the CHANGELOG entry covering Units 1-3 (Establishment reply projection, open_channel_with_reply, flavor-form discovery derivation, ChannelRelay, HubLegTemplate, gate-2 harness); review 008 Status -> Resolved with the U-1/U-2 commit refs and the 955->945 errata note; ADR-051 Status -> all units landed + the §6 mid-establishment residual expanded to the two-reclaim-signal shape the gate pins; the stale "0.7.2" version mentions in ADR-047/049 corrected to 0.8.0 (the units land unreleased). Verification: cargo test 669 passed / 0 failed; clippy --all-targets -- -D warnings clean; fmt --check clean; cargo doc --no-deps clean; cargo check + clippy on wasm32-unknown-unknown clean; cargo publish --dry-run --allow-dirty passed. --- CHANGELOG.md | 87 ++ Cargo.lock | 2 +- Cargo.toml | 2 +- .../047-openable-alpns-are-operations.md | 2 +- .../049-channel-open-establishment-phase.md | 2 +- .../051-channel-relay-and-hub-leg.md | 28 +- .../008-tunnels-graduation-upstream-review.md | 29 +- src/channels/gate2_tests.rs | 742 ++++++++++++++++++ src/channels/hub_leg.rs | 4 + src/channels/relay.rs | 26 +- 10 files changed, 903 insertions(+), 21 deletions(-) create mode 100644 src/channels/gate2_tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index aaf8b22..8a129c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,93 @@ All notable changes to this crate are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this crate adheres to [Semantic Versioning](https://semver.org/). +## [0.8.0] - 2026-09-18 + +Review 008's remediation lands in full — the graduation upstream asks +(U-1 flavor-form open-op ids in discovery; U-2 the establisher → reply +projection) plus the in-tree channel relay they demanded +(`ChannelRelay`, the hub-leg install template, and the review's gate-2 +e2e harness). The channels data plane is untouched (ADR-034/035's +one-way doors stay closed); the new surfaces are additive on the +call-plane JSON and the registry seams. + +### Added + +- **`Establishment` reply projection (review 008 U-2, ADR-049 + amendment 3)** — `Establishment::with_reply_field` / + `with_reply_fields` / `reply_fields()` (an optional map riding the + typed-opaque plan carrier). The open-op wrapper merges the fields + into the success output after reserving `channel_id` — an + establisher supplying the reserved key fails loudly + (`channel:open_failed`, reason `handler_error`), never shadows. + Absent fields leave the reply byte-identical to the pre-amendment + shape. +- **`ChannelClient::open_channel_with_reply` (review 008 U-2)** — the + full open-op success output (the extra reply fields, e.g. a + bind-first listener's `bound`) alongside the streams; + `open_channel` delegates with the fields discarded, its signature + unchanged. +- **Flavor-form open-op ids in discovery (review 008 U-1, ADR-047 + amendment 3)** — `spec_to_json_pub` emits an explicit + `channel_open_alpn` string alongside the boolean marker when the op + name is not a standard `channels//(sub|pub)` shape (standard + shapes stay byte-stable); `rebuild_spec_for` prefers the explicit + string and generalizes the derivation to strip the LAST path + segment when the marker is present. `channels/tunnel/direct`-shaped + ops now reconstruct WITH the `channel_open` marker through + discovery — hubs relaying through `from_call` never see them as + plain forwarding stubs. +- **`ChannelRelay` (review 008 Unit 3a, ADR-042 as amended by + ADR-051)** — the in-tree relay component: + `ProducerLeg` (`Arc` + producer-leg + `ChannelManager` — not a `ChannelClient`, whose `take_call_ + connection` detach is wrong for a hub with multiple connection + claimants) and `ChannelRelay::register_relay_openable` (registers a + from_call-imported marked spec on a consumer-leg registry via the + full open-op wrapper: ACL, per-identity cap/ledger, establishment + bound, teardown). The translate hop calls the producer leg with the + forwarded payload (`forwarded_for` from the consumer's per-call + identity, ADR-026 §3) and adopts the spoke-allocated id; the + byte-forward hop is `pump_bidi` awaited inline — the hub never + parses chunk framing, ids are per-connection (no rewrite exists). + The spoke reply's `channel_id` is stripped (the consumer reply + carries the hub-allocated id); the reply's other fields ride + through (`bound` flows end-to-end, per-hop truthful). Reason + mapping per ADR-051 §3 — the spoke's code and message are preserved + (`timeout` maps to `dial_failed`, the one non-1:1 row). The + adopted producer-leg channel entry reclaims when the relayed pump + completes. Rejection postures are loud: unmarked and Pub-typed + marked specs fail registration (`RelayError`), never silent stubs. +- **`HubLegImports` + `HubLegTemplate` (review 008 Unit 3b, ADR-051 + §5)** — the hub-leg install template as an in-tree export: the + `Clone` discover/stash (from_call bundles split by the marker, with + the per-consumer `filtered`/`only` subset filter) and the + `install_channel_zero` hook composing fork + generic channel ops + + bootstrap discovery closed over the fork (review 004 F-06) + plain + bundles as-is + relay openables + serving-identity resolution + (CF-005 precedence) + the single-stream dispatch loop. A + Pub-typed (or unregistrable) marked spec ends the leg's install + task — channel 0 never dispatches. Discovered bootstrap-discovery + op names are skipped on re-registration (the template's own + install supersedes the imported copies). + +### Changed + +- **New module `src/channels/relay.rs` and `src/channels/hub_leg.rs` + re-exported from `channels`**; the relay's teardown fix is part of + this landing (no pre-relay behavior to preserve). + +### Verified + +- 669 tests pass (`cargo test`), `clippy --all-targets -- -D warnings` + and `fmt --check` clean. The gate-2 e2e harness + (`src/channels/gate2_tests.rs`) pins the review's gate 2: the full + consumer → hub → producer relay through the template (hub-allocated + id, `bound` surviving, data both directions with a fake 8-byte + chunk header riding verbatim), hub-side disconnect tearing down + both legs, the mid-establishment window's two reclaim signals + (ADR-051 §6), and the `channels/tty/sub` standard-shape companion. + ## [0.7.1] - 2026-09-09 The MSRV floor becomes honest: `rust-version` raises from 1.85 to diff --git a/Cargo.lock b/Cargo.lock index e61a940..34755b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -27,7 +27,7 @@ dependencies = [ [[package]] name = "alkcall" -version = "0.7.1" +version = "0.8.0" dependencies = [ "async-trait", "bytes", diff --git a/Cargo.toml b/Cargo.toml index bb4dea0..5dabd9b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "alkcall" -version = "0.7.1" +version = "0.8.0" edition = "2021" rust-version = "1.88" license = "MIT OR Apache-2.0" diff --git a/docs/architecture/decisions/047-openable-alpns-are-operations.md b/docs/architecture/decisions/047-openable-alpns-are-operations.md index 99eea3f..92e1006 100644 --- a/docs/architecture/decisions/047-openable-alpns-are-operations.md +++ b/docs/architecture/decisions/047-openable-alpns-are-operations.md @@ -86,7 +86,7 @@ pinned by test). The `op_name_is_standard_channel_open_shape` helper and the strip-last derivation are two-way-door implementation details within the one-way wire shape. -Implemented surface (alkcall 0.7.2): the generalized +Implemented surface (alkcall 0.8.0): the generalized `derive_alpn_from_op_name` (strip-last), the explicit-field emission (`spec_to_json_pub` + the advertised schema property), and the preference order in `rebuild_spec_for`. Verification gates landed as diff --git a/docs/architecture/decisions/049-channel-open-establishment-phase.md b/docs/architecture/decisions/049-channel-open-establishment-phase.md index 3ac6299..c801feb 100644 --- a/docs/architecture/decisions/049-channel-open-establishment-phase.md +++ b/docs/architecture/decisions/049-channel-open-establishment-phase.md @@ -468,7 +468,7 @@ unknown fields by the envelope's own parse posture); the field set an op may contribute is the producing crate's schema concern, not a protocol vocabulary. No data-plane change. -Implemented surface (alkcall 0.7.2): `Establishment::with_reply_field` +Implemented surface (alkcall 0.8.0): `Establishment::with_reply_field` / `with_reply_fields` / `reply_fields`, the wrapper's reservation check (`RESERVED_REPLY_KEY` = `channel_id`) + `merge_reply_fields`, `ChannelClient::open_channel_with_reply`. diff --git a/docs/architecture/decisions/051-channel-relay-and-hub-leg.md b/docs/architecture/decisions/051-channel-relay-and-hub-leg.md index 7fece2c..0fd7316 100644 --- a/docs/architecture/decisions/051-channel-relay-and-hub-leg.md +++ b/docs/architecture/decisions/051-channel-relay-and-hub-leg.md @@ -5,14 +5,15 @@ Accepted (2026-09-17) — amends ADR-042 (the relay contract's implementation home moves into alkcall; the translate-not-forward contract is unchanged). Sequenced after review 008 remediation Units 1 -and 2 (both landed, alkcall 0.7.2): the relay's establisher projects +and 2 (both landed, alkcall 0.8.0): the relay's establisher projects the spoke reply's extra fields via `Establishment` reply fields (ADR-049 amendment 3), and the spoke's flavor-form open ops arrive reconstructed WITH the `channel_open` marker through discovery -(ADR-047 amendment 3). Unit 3a (the `ChannelRelay` component, §1–§4, -§6) and Unit 3b (the hub-leg install template, §5) are implemented — -`src/channels/relay.rs` and `src/channels/hub_leg.rs`; Unit 3c (the -review's gate-2 e2e harness) remains. +(ADR-047 amendment 3). All three implementation units are landed: +Unit 3a (the `ChannelRelay` component, §1–§4, §6) in +`src/channels/relay.rs`, Unit 3b (the hub-leg install template, §5) in +`src/channels/hub_leg.rs`, and Unit 3c (the review's gate-2 e2e +harness, gates 6/7) in `src/channels/gate2_tests.rs` — alkcall 0.8.0. ## Context @@ -201,11 +202,18 @@ relays. The end consumer never authenticates to the spoke directly. bounds each hop, not the chain. No fix — noted for capacity planning. - **Mid-establishment consumer disconnect** — the one teardown window - with a delay: the spoke channel can exist briefly after the consumer - leg dies. Not a leak: once the wrapper spawns the pump, hub-leg - `clear_all` (transport EOF) EOFs the consumer recv, the pump exits, - the spoke leg EOFs, and both wrappers decrement their ledgers. The - e2e gate asserts this window explicitly. + with a delay, with two reclaim signals pinned by the Unit 3c gate + (`gate2_mid_establishment_disconnect_reclaims_without_leak`): (a) + the consumer leg's just-allocated channel reclaims at the leg's own + transport EOF (`clear_all` + ledger decrement) — the hub has not + adopted, so the producer leg stays clean; (b) the spoke's channel — + the spoke allocates before its establisher replies — is the honest + residual: it cannot be reclaimed by the hub (no adoption exists), + so it lingers until a spoke-side teardown signal (the spoke↔hub leg + ending, or the spoke's own establishment bound). The gate bounds + the window deterministically by severing the spoke leg and asserts + the spoke-side reclaim + ledger decrement. A channel never outlives + its leg lifetimes; never a leak past the connection's lifetime. - **Multi-hop relays** — a chain of relays composes (each hop is a hub leg pair), with the compounding-bound note above. No special machinery. diff --git a/docs/reviews/008-tunnels-graduation-upstream-review.md b/docs/reviews/008-tunnels-graduation-upstream-review.md index f21ce97..77fbe86 100644 --- a/docs/reviews/008-tunnels-graduation-upstream-review.md +++ b/docs/reviews/008-tunnels-graduation-upstream-review.md @@ -2,6 +2,24 @@ ## Status +Resolved — all three remediation units landed in alkcall 0.8.0 +(2026-09-18): Unit 1 `82ddddf` (U-2, ADR-049 amendment 3), Unit 2 +`620180d` (U-1, ADR-047 amendment 3), Unit 3a `90c1820` (the in-tree +`ChannelRelay`, ADR-051), Unit 3b `4e52bd4` (the hub-leg install +template, ADR-051 §5), Unit 3c (the gate-2 e2e harness in +`src/channels/gate2_tests.rs` + the relay's producer-leg teardown + +this release's bookkeeping). Two notes on the as-filed text: the U-2 +line citation is 945, not 955 (see Errata), and U-1's gate 2 landed as +an in-tree relay component plus a permanent e2e gate rather than a +test-only harness (the escalation the remediation plan pinned above). +Original findings below, retained as filed (verified against tree +`3b36b40`, 0.7.1). + +> **Errata (post-landing):** U-2 cites the reply construction at +> `src/channels/operations.rs:955`; the cf-006/007 commit had already +> shifted it to 945 by filing time. The citation in the finding below +> is left as filed; the verified line is 945. + Open — filed 2026-09-16 from the alktunnels graduation spec work (ADRs 007/008 at `/workspace/@alkdev/alktunnels/docs/architecture/decisions/`; @@ -29,7 +47,11 @@ changes the channels data-plane wire format (ADR-034/071 untouched). ## U-1: Flavor-form open-op ids in discovery derivation -**Finding.** ADR-047 §1 pins the open-op naming to exactly two shapes +**Finding.** *(Resolved — ADR-047 amendment 3, Unit 2 `620180d`; the +explicit `channel_open_alpn` field plus the last-segment derivation, +exactly the pinned (b)+(c) shape. Gate 2 landed as the in-tree relay + +`src/channels/gate2_tests.rs`, Unit 3.)* ADR-047 §1 pins the open-op +naming to exactly two shapes per ALPN — `channels//sub` and `channels//pub` — and Gap F's marker derivation (`rebuild_spec_for`, `src/client/from_call.rs:209` + `derive_alpn_from_op_name`, @@ -107,7 +129,10 @@ relay), which is the worst failure mode for a relay. ## U-2: Establisher → reply projection (`Establishment` contributing reply fields) -**Finding.** The open-op success reply is constructed by the wrapper +**Finding.** *(Resolved — ADR-049 amendment 3, Unit 1 `82ddddf`: the +`reply_fields` carrier, the `channel_id` reservation failing loudly, +`open_channel_with_reply` on the client.)* The open-op success reply +is constructed by the wrapper and hardcoded to the channel id: `ResponseEnvelope::ok(request_id, json!({ "channel_id": channel_id }))` (`src/channels/operations.rs:955` in `run_open_wrapper`). ADR-049 §1 diff --git a/src/channels/gate2_tests.rs b/src/channels/gate2_tests.rs new file mode 100644 index 0000000..140e34a --- /dev/null +++ b/src/channels/gate2_tests.rs @@ -0,0 +1,742 @@ +//! Review 008 gate 2 — the full e2e harness (ADR-051 gates 6/7; Unit 3c). +//! +//! Topology: consumer —(alk/channels duplex, the consumer's end held +//! by a kill handle)— hub (HubLegTemplate on the consumer leg; +//! ChannelRelay + producer-leg surface toward the spoke) +//! —(alk/channels duplex, the hub's spoke-side end held by a kill +//! handle)— spoke (the direct op with a `bound`-contributing +//! establisher + an echo data plane, the tty op, a plain op). +//! +//! Killing a leg = dropping that side's duplex end. Dropping one end +//! of a `tokio::io::duplex` pair wakes the peer's pending reads with +//! EOF and errors the peer's writes — the same shape a real transport +//! close walks (verified by the kill-probe experiment). +//! +//! Gates: +//! - The open resolves on the consumer with the hub-allocated +//! `channel_id`; the establisher-contributed `bound` survives the +//! relay to the consumer reply; data flows both directions with the +//! hub never parsing the data plane (a payload that embeds a fake +//! 8-byte chunk header arrives verbatim); spoke-side close cascades +//! to clean reclaim on both legs. +//! - Hub-side disconnect (the consumer's duplex end dropped +//! mid-stream): the consumer leg's demux `clear_all` reclaims the +//! leg's channels and decrements the ledger; the relay pump sees EOF +//! and exits, EOFing the spoke leg; the spoke handler exits and the +//! spoke's wrapper reclaims. Both legs end clean — no leak. +//! - The mid-establishment window (ADR-051 §6): the consumer leg dies +//! while the spoke establisher is parked — the spoke channel +//! (allocated before the establisher parked) is the honest residual; +//! it reclaims when the spoke↔hub leg ends. The consumer leg's +//! ledger reclaims with its own teardown, no cap leak. +//! - A companion `channels/tty/sub` relay run pins no regression on +//! standard shapes. + +use super::{HubLegImports, HubLegTemplate}; +use crate::channels::adapter::ChannelsAdapter; +use crate::channels::operations::{ChannelCore, Establishment, OpenEstablisher, OpenHandler}; +use crate::channels::policy::{ChannelLifecyclePolicy, NoCap, PerIdentityChannelPolicy}; +use crate::channels::relay::{ChannelRelay, ProducerLeg}; +use crate::client::{from_call, FromCallConfig}; +use crate::core::auth::{AuthContext, Identity, IdentityProvider}; +use crate::core::types::Connection; +use crate::protocol::connection::{split_single_stream, CallConnection}; +use crate::protocol::dispatch::Dispatcher; +use crate::protocol::wire::ResponseEnvelope; +use crate::registry::discovery::install_bootstrap_discovery; +use crate::registry::registration::{ + make_handler, HandlerKind, HandlerRegistration, OperationProvenance, OperationRegistry, +}; +use crate::registry::spec::{ + AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility, +}; +use serde_json::json; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::pin::Pin; +use std::sync::Arc; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; + +const TEST_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321); +const OP: &str = "channels/tunnel/direct"; +const TTY_OP: &str = "channels/tty/sub"; +const TTY_ALPN: &str = "alk/tty"; +const ALPN: &str = "alk/tunnel"; +const PLAIN_OP: &str = "fs/read"; + +fn identity(id: &str, scopes: &[&str]) -> Identity { + Identity { + id: id.to_string(), + scopes: scopes.iter().map(|s| (*s).to_string()).collect(), + resources: Default::default(), + } +} + +fn open_spec(name: &str, alpn: &str) -> OperationSpec { + OperationSpec::new( + name, + OperationType::Sub, + Visibility::External, + json!({ "type": "object" }), + json!({ "type": "object" }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new(alpn.to_string())) +} + +fn plain_spec(name: &str) -> OperationSpec { + OperationSpec::new( + name, + OperationType::Query, + Visibility::External, + json!({ "type": "object" }), + json!({ "type": "object" }), + vec![], + AccessControl::default(), + None, + ) +} + +struct NoopIdProvider; +impl IdentityProvider for NoopIdProvider { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option { + None + } +} + +/// A killable transport end: the shared slot holds the inner duplex +/// end the proxy reads/writes through; `sever()` drops it. The +/// kill-probe experiment verified both directions: the peer's parked +/// reads wake with EOF and the proxy's own pending polls complete +/// (the demux loop's read errors, the dispatch loop breaks). +#[derive(Clone, Default)] +struct Kill(Arc>>); + +impl Kill { + fn new(inner: tokio::io::DuplexStream) -> Self { + Self(Arc::new(std::sync::Mutex::new(Some(inner)))) + } + fn sever(&self) { + let _ = self.0.lock().unwrap_or_else(|e| e.into_inner()).take(); + } +} + +struct KillableStream(Kill); + +impl AsyncRead for KillableStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> std::task::Poll> { + let mut guard = self.0 .0.lock().unwrap_or_else(|e| e.into_inner()); + match guard.as_mut() { + Some(inner) => Pin::new(inner).poll_read(cx, buf), + None => std::task::Poll::Ready(Ok(())), + } + } +} + +impl AsyncWrite for KillableStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let mut guard = self.0 .0.lock().unwrap_or_else(|e| e.into_inner()); + match guard.as_mut() { + Some(inner) => Pin::new(inner).poll_write(cx, buf), + None => std::task::Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::ConnectionAborted, + "killed", + ))), + } + } + fn poll_flush( + self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let mut guard = self.0 .0.lock().unwrap_or_else(|e| e.into_inner()); + match guard.as_mut() { + Some(inner) => Pin::new(inner).poll_flush(cx), + None => std::task::Poll::Ready(Ok(())), + } + } + fn poll_shutdown( + self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let mut guard = self.0 .0.lock().unwrap_or_else(|e| e.into_inner()); + match guard.as_mut() { + Some(inner) => Pin::new(inner).poll_shutdown(cx), + None => std::task::Poll::Ready(Ok(())), + } + } +} + +// --- the spoke (producer leg) --------------------------------------------- + +/// The spoke's echo data-plane handler. +fn echo_handler() -> OpenHandler { + Arc::new(move |_input, _plan, channel_conn, _auth| { + tokio::spawn(async move { + let Ok(mut bidi) = channel_conn.accept_bi().await else { + return; + }; + let mut buf = Vec::new(); + if tokio::io::AsyncReadExt::read_to_end(&mut bidi, &mut buf) + .await + .is_err() + { + return; + } + let _ = bidi.write_all(&buf).await; + let _ = bidi.shutdown().await; + }) + }) +} + +/// The spoke's install hook: the direct op (a `bound`-contributing +/// establisher gated on `bound_latch` — released at spoke startup +/// unless a test holds it — plus an echo data plane) and the tty op +/// (no establisher, echo data plane), plus a plain op and bootstrap +/// discovery, then the channel-0 dispatch. +fn make_spoke_install( + bound_latch: Arc>, +) -> crate::channels::adapter::InstallChannelZero { + Arc::new(move |manager, channel0_conn, auth| { + let bound_latch = bound_latch.clone(); + tokio::spawn(async move { + let Ok(channel0_bidi) = channel0_conn.accept_bi().await else { + return; + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let core = ChannelCore::new(manager.clone(), crate::channels::policy::default_policy()); + let registry = Arc::new(OperationRegistry::new()); + install_bootstrap_discovery(®istry).expect("spoke bootstrap discovery"); + + let bound_establisher: OpenEstablisher = { + let rx = bound_latch.subscribe(); + Arc::new(move |_input, _auth| { + let mut rx = rx.clone(); + Box::pin(async move { + while !*rx.borrow_and_update() { + if rx.changed().await.is_err() { + break; + } + } + Ok(Establishment::default().with_reply_field( + "bound", + json!({ "host": "203.0.113.9", "port": 42113 }), + )) + }) + }) + }; + core.register_openable_with_establisher( + open_spec(OP, ALPN), + Some(bound_establisher), + echo_handler(), + ®istry, + auth.clone(), + None, + ) + .expect("spoke registers the direct open op"); + + core.register_openable_with_establisher( + open_spec(TTY_OP, TTY_ALPN), + None, + echo_handler(), + ®istry, + auth.clone(), + None, + ) + .expect("spoke registers the tty open op"); + + registry + .register(HandlerRegistration::new( + plain_spec(PLAIN_OP), + HandlerKind::Once(make_handler(|input, ctx| async move { + ResponseEnvelope::ok(ctx.request_id, input) + })), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + )) + .expect("spoke registers the plain op"); + + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + Dispatcher::new( + registry, + Arc::new(NoopIdProvider) as Arc, + ) + .run_loop_single_stream(call_connection, reader, writer) + .await; + }) + }) +} + +// --- the harness ----------------------------------------------------------- + +struct Harness { + consumer: crate::channels::client::ChannelClient, + /// The consumer leg's manager on the hub (captured by wrapping the + /// template's install hook) — the observability point for the + /// hub-side ledger/channel-map assertions. + hub_consumer_manager: crate::channels::manager::ChannelManager, + /// The hub's producer-leg manager (the relay's adopt side). + hub_producer_manager: crate::channels::manager::ChannelManager, + /// The spoke's manager (captured by wrapping the spoke install + /// hook) — the observability point for the spoke-side ledger. + spoke_manager: crate::channels::manager::ChannelManager, + /// Releases the spoke's `bound` establisher latch. + release_bound: Arc>, + /// Kills the consumer↔hub transport (drops the hub-side end). + kill_consumer_leg: Kill, + /// Kills the hub↔spoke transport (drops the spoke-side end). + kill_spoke_leg: Kill, + consumer_policy: Arc, +} + +async fn start() -> Harness { + let (release_tx, _release_rx) = tokio::sync::watch::channel(false); + let h = start_with_latch(Arc::new(release_tx)).await; + h.release_bound.send(true).expect("release the bound latch"); + h +} + +async fn start_with_latch(release_bound: Arc>) -> Harness { + // ---- the spoke ---- + let (hub_spoke_client_end, spoke_server_end) = tokio::io::duplex(64 * 1024); + // The spoke's end is plain; the hub's end rides the killable + // proxy whose inner the Kill slot holds. Severing drops the HUB's + // end — the spoke's parked demux/dispatch (the PEER side of the + // pair) wakes with EOF. The hub's client to the spoke is not + // needed after the mid-establishment test severs the leg. + let kill_spoke_leg = Kill::new(hub_spoke_client_end); + let hub_conn_to_spoke = Connection::from_bidi( + KillableStream(kill_spoke_leg.clone()), + b"alk/channels".to_vec(), + Some(TEST_ADDR), + ); + let spoke_side = + Connection::from_bidi(spoke_server_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let captured_spoke: Arc>> = + Arc::new(tokio::sync::Mutex::new(None)); + let captured_spoke_for_hook = Arc::clone(&captured_spoke); + let spoke_install = { + let inner = make_spoke_install(Arc::clone(&release_bound)); + Arc::new( + move |manager: crate::channels::manager::ChannelManager, conn, auth| { + let _ = captured_spoke_for_hook + .try_lock() + .map(|mut g| *g = Some(manager.clone())); + inner(manager, conn, auth) + }, + ) + }; + let spoke_adapter = ChannelsAdapter::new(spoke_install, Arc::new(NoCap)); + let _spoke_task = tokio::spawn(async move { + let auth = AuthContext::anonymous(b"alk/channels"); + let _ = + crate::core::types::ProtocolHandler::handle(&spoke_adapter, spoke_side, &auth).await; + }); + + // ---- the hub: dial the spoke, discover, stash ---- + let hub_client_to_spoke = + crate::channels::client::ChannelClient::from_connection(hub_conn_to_spoke) + .await + .expect("hub's channel client to the spoke"); + let producer_call = hub_client_to_spoke + .take_call_connection() + .await + .expect("hub takes its channel-0 CallConnection"); + let bundles = from_call(&producer_call, FromCallConfig::new()) + .await + .expect("from_call discovers the spoke's ops"); + assert!( + bundles + .iter() + .any(|b| b.spec.name == OP && b.spec.channel_open.is_some()), + "the flavor-form direct op is discovered WITH the marker" + ); + let producer_leg = ProducerLeg { + call: producer_call, + manager: hub_client_to_spoke.manager().clone(), + }; + let hub_producer_manager = hub_client_to_spoke.manager().clone(); + drop(hub_client_to_spoke); + + let imports = HubLegImports::from_bundles(bundles); + let template = HubLegTemplate::new(ChannelRelay::new(producer_leg), imports); + let template_install = template.install_hook(); + + // ---- the hub's consumer leg: the template wrapped to capture the + // leg's manager ---- + let captured: Arc>> = + Arc::new(tokio::sync::Mutex::new(None)); + let consumer_policy = Arc::new(PerIdentityChannelPolicy::new(8)); + let policy_dyn: Arc = + Arc::clone(&consumer_policy) as Arc; + let captured_for_hook = Arc::clone(&captured); + let consumer_install: crate::channels::adapter::InstallChannelZero = + Arc::new(move |manager, conn, auth| { + let inner = Arc::clone(&template_install); + let captured = Arc::clone(&captured_for_hook); + let handle = inner(manager.clone(), conn, auth); + let _ = captured.try_lock().map(|mut g| *g = Some(manager)); + handle + }); + + let (consumer_end, hub_consumer_end) = tokio::io::duplex(64 * 1024); + // The hub side is plain; the consumer's end rides the killable + // proxy whose inner the Kill slot holds. Severing drops the + // consumer's end — the hub's parked demux/dispatch (the PEER side + // of the pair) wakes with EOF (the kill-probe semantics: dropping + // one end wakes the peer's pending reads; the dropped end's own + // waker is lost, but the consumer client is not needed after the + // disconnect tests sever it). + let kill_consumer_leg = Kill::new(consumer_end); + let consumer_conn = Connection::from_bidi( + KillableStream(kill_consumer_leg.clone()), + b"alk/channels".to_vec(), + Some(TEST_ADDR), + ); + let hub_consumer_side = + Connection::from_bidi(hub_consumer_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let consumer_adapter = ChannelsAdapter::new(consumer_install, policy_dyn); + let _hub_task = tokio::spawn(async move { + let auth = AuthContext::anonymous(b"alk/channels"); + let _ = crate::core::types::ProtocolHandler::handle( + &consumer_adapter, + hub_consumer_side, + &auth, + ) + .await; + }); + + let consumer = crate::channels::client::ChannelClient::from_connection(consumer_conn) + .await + .expect("consumer's channel client"); + + let hub_consumer_manager = captured + .lock() + .await + .clone() + .expect("the template install captured the consumer-leg manager"); + + let spoke_manager = captured_spoke + .lock() + .await + .clone() + .expect("the spoke install captured the spoke manager"); + + Harness { + consumer, + hub_consumer_manager, + hub_producer_manager, + spoke_manager, + release_bound, + kill_consumer_leg, + kill_spoke_leg, + consumer_policy, + } +} + +/// Poll until `pred` holds (the teardown cascade is asynchronous — +/// multiple spawned tasks must run to completion). +async fn wait_for(pred: impl Fn() -> bool, what: &str) { + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + while !pred() { + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for {what}" + ); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } +} + +/// Data-channel count on a leg's manager. Channel 0 is pre-installed +/// at connection setup (so `open_count()` counts it too); after a +/// transport EOF `clear_all` drains the map entirely — including +/// channel 0 — so the count saturates at zero. +fn data_channels(manager: &crate::channels::manager::ChannelManager) -> usize { + manager.open_count().saturating_sub(1) +} + +// --- gate 6: the full harness ---------------------------------------------- + +/// The review's gate 2 main run: consumer → hub (the template) → +/// producer, opening `channels/tunnel/direct` end to end. The open +/// resolves with the hub-allocated `channel_id`, the establisher's +/// `bound` survives the relay to the consumer reply, data flows both +/// directions (the hub never parses it — a payload embedding a fake +/// 8-byte chunk header arrives verbatim), and spoke-side close +/// cascades to clean reclaim on both legs (both ledgers decremented). +#[tokio::test] +async fn gate2_direct_op_relays_end_to_end_through_the_template() { + let h = start().await; + + // The open resolves with the hub-allocated id and `bound`. + let (consumer_id, reply, mut send, mut recv) = tokio::time::timeout( + std::time::Duration::from_secs(10), + h.consumer + .open_channel_with_reply(OP, json!({ "target": "example.internal:443" }), ALPN), + ) + .await + .expect("relay open timed out") + .expect("open through the template"); + assert!(consumer_id > 0, "the consumer leg allocated the id"); + assert_eq!(reply["channel_id"], json!(consumer_id)); + assert_eq!( + reply["bound"], + json!({ "host": "203.0.113.9", "port": 42113 }), + "the spoke establisher's `bound` survives the relay through the template" + ); + + // The hub adopted the spoke's channel on its producer leg. + assert_eq!( + data_channels(&h.hub_producer_manager), + 1, + "the hub's producer leg holds the adopted spoke channel" + ); + + // Data both directions; the hub never parses the data plane. The + // outbound payload embeds a valid 8-byte chunk header for channel + // 999 with a plausible length — if any hop parsed the data plane, + // the spoke's echo would differ (the framing would be corrupted or + // the payload dropped by a bogus route). + let mut payload = vec![0u8; 8]; + payload[..4].copy_from_slice(&999u32.to_be_bytes()); + payload[4..8].copy_from_slice(&64u32.to_be_bytes()); + payload.extend_from_slice(b"gate2-data-plane"); + send.write_all(&payload).await.expect("write"); + send.shutdown().await.expect("shutdown"); + drop(send); + let mut echoed = Vec::new(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + recv.read_to_end(&mut echoed), + ) + .await + .expect("echo timed out") + .expect("read echo"); + assert_eq!( + echoed, payload, + "the fake chunk header rode verbatim — no hop parsed the data plane" + ); + + // Spoke-side close: drop the consumer's streams (the consumer's + // handler exits → pump EOFs both ways → spoke handler exits → + // spoke's wrapper decrements its ledger; the consumer leg's + // wrapper teardown reclaims the consumer-leg entry). + drop(recv); + wait_for( + || data_channels(&h.hub_producer_manager) == 0, + "the producer-leg adopted channel to reclaim after the spoke-side close cascade", + ) + .await; + wait_for( + || data_channels(&h.hub_consumer_manager) == 0, + "the consumer-leg channel to reclaim after the teardown cascade", + ) + .await; + assert_eq!( + h.consumer_policy.count_for(&identity("anonymous", &[])), + 0, + "the consumer leg's ledger decremented" + ); +} + +/// Hub-side disconnect mid-stream (ADR-051 §6's teardown-window +/// sibling): the consumer's duplex end drops while a relayed channel +/// is live. The consumer leg's demux `clear_all` reclaims the leg's +/// channels and decrements the ledger; the relay pump sees EOF and +/// exits, EOFing the spoke leg; the spoke handler exits and the +/// spoke's wrapper reclaims. Both legs end clean — no leak. +#[tokio::test] +async fn gate2_hub_side_disconnect_tears_down_both_legs() { + let h = start().await; + + let (_id, _reply, mut send, recv) = tokio::time::timeout( + std::time::Duration::from_secs(10), + h.consumer + .open_channel_with_reply(OP, json!({ "target": "example.internal:443" }), ALPN), + ) + .await + .expect("relay open timed out") + .expect("open through the template"); + assert_eq!(data_channels(&h.hub_producer_manager), 1); + assert_eq!(data_channels(&h.hub_consumer_manager), 1); + + // Some data first so the channel is live mid-stream. + send.write_all(b"pre-disconnect").await.expect("write"); + drop(send); + drop(recv); + + // Kill the consumer↔hub transport (the hub-side end drops; the + // kill-probe semantics: the hub's parked demux reads EOF). + h.kill_consumer_leg.sever(); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + wait_for( + || data_channels(&h.hub_consumer_manager) == 0, + "the consumer leg's demux clear_all to reclaim the channel", + ) + .await; + wait_for( + || data_channels(&h.hub_producer_manager) == 0, + "the relay pump to EOF the spoke leg and the adopted entry to reclaim", + ) + .await; + assert_eq!( + h.consumer_policy.count_for(&identity("anonymous", &[])), + 0, + "the consumer leg's ledger decremented on transport EOF" + ); +} + +/// The mid-establishment disconnect window (ADR-051 §6). The spoke +/// allocates its channel before its establisher replies, so the window +/// has two reclaim signals; the test pins both: +/// +/// 1. The consumer leg dies while the spoke establisher is parked — +/// the consumer leg's demux `clear_all` reclaims the leg's +/// just-allocated channel and decrements the ledger; the hub has +/// not adopted (the spoke has not replied), so the hub's +/// producer-leg map stays clean. +/// 2. The spoke's channel is the honest residual (ADR-051 §6): the +/// spoke-side establisher is parked and the spoke↔hub leg is still +/// alive, so the spoke channel lingers until a teardown signal. +/// Killing the spoke↔hub leg bounds the window deterministically; +/// the spoke's demux `clear_all` reclaims its channel and +/// decrements its ledger. Never a leak past the leg lifetimes. +#[tokio::test] +async fn gate2_mid_establishment_disconnect_reclaims_without_leak() { + let (release_tx, _release_rx) = tokio::sync::watch::channel(false); + let release_tx = Arc::new(release_tx); + let h = start_with_latch(Arc::clone(&release_tx)).await; + + // Open through the relay — the establisher is parked, so the open + // hangs in the hub's translate hop (establishment bound pending). + // Poll once so the request is on the wire, then let it run. + let open_fut = + h.consumer + .open_channel_with_reply(OP, json!({ "target": "parked.internal:1" }), ALPN); + tokio::pin!(open_fut); + let _ = futures::poll!(open_fut.as_mut()); + // The consumer leg's wrapper has allocated its channel + // (establishment in flight); the spoke has allocated its own + // (the establisher parked after the spoke's open_channel). + wait_for( + || data_channels(&h.hub_consumer_manager) == 1, + "the consumer leg's wrapper to allocate the channel", + ) + .await; + wait_for( + || data_channels(&h.spoke_manager) == 1, + "the spoke to allocate its channel before parking its establisher", + ) + .await; + assert_eq!( + data_channels(&h.hub_producer_manager), + 0, + "the hub has not adopted — the spoke has not replied yet" + ); + + // Kill the consumer leg mid-establishment. + h.kill_consumer_leg.sever(); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + wait_for( + || data_channels(&h.hub_consumer_manager) == 0, + "the consumer leg's demux clear_all to reclaim the in-flight channel", + ) + .await; + assert_eq!( + h.consumer_policy.count_for(&identity("anonymous", &[])), + 0, + "the consumer leg's ledger decremented on the transport EOF" + ); + + // The spoke's channel is the residual — reclaim it by ending the + // spoke↔hub leg. + h.kill_spoke_leg.sever(); + wait_for( + || data_channels(&h.spoke_manager) == 0, + "the spoke channel to reclaim when the spoke↔hub leg ends", + ) + .await; + + // The open never succeeds after both legs died. + let outcome = tokio::time::timeout(std::time::Duration::from_secs(10), open_fut).await; + match outcome { + Err(_) => {} + Ok(Ok((id, _, _, _))) => { + panic!("the open must not succeed after both legs died, got id {id}") + } + Ok(Err(_)) => {} + } + assert_eq!( + h.consumer_policy.count_for(&identity("anonymous", &[])), + 0, + "no consumer-leg cap leak" + ); +} + +/// Gate 7's companion: a standard-shape `channels/tty/sub` relay run +/// through the same template harness — no regression on the shapes the +/// existing derivation produced before the flavor amendment. +#[tokio::test] +async fn gate2_tty_standard_shape_relays_with_no_regression() { + let h = start().await; + + let (consumer_id, reply, mut send, mut recv) = tokio::time::timeout( + std::time::Duration::from_secs(10), + h.consumer + .open_channel_with_reply(TTY_OP, json!({}), TTY_ALPN), + ) + .await + .expect("tty open timed out") + .expect("tty open through the template"); + assert!(consumer_id > 0); + assert_eq!( + reply.as_object().expect("object").len(), + 1, + "no establisher reply fields: the reply is exactly channel_id" + ); + + send.write_all(b"tty-gate2").await.expect("write"); + send.shutdown().await.expect("shutdown"); + drop(send); + let mut echoed = Vec::new(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + recv.read_to_end(&mut echoed), + ) + .await + .expect("echo timed out") + .expect("read echo"); + assert_eq!(echoed, b"tty-gate2"); + + wait_for( + || data_channels(&h.hub_producer_manager) == 0, + "the adopted tty channel to reclaim after the echo completed", + ) + .await; + wait_for( + || data_channels(&h.hub_consumer_manager) == 0, + "the consumer-leg tty channel to reclaim", + ) + .await; + assert_eq!( + h.consumer_policy.count_for(&identity("anonymous", &[])), + 0, + "the tty relay's ledger decremented" + ); +} diff --git a/src/channels/hub_leg.rs b/src/channels/hub_leg.rs index c539891..ad8bd63 100644 --- a/src/channels/hub_leg.rs +++ b/src/channels/hub_leg.rs @@ -290,3 +290,7 @@ impl HubLegTemplate { #[cfg(test)] #[path = "hub_leg_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "gate2_tests.rs"] +mod gate2_tests; diff --git a/src/channels/relay.rs b/src/channels/relay.rs index 54920c9..5929e54 100644 --- a/src/channels/relay.rs +++ b/src/channels/relay.rs @@ -62,10 +62,12 @@ pub struct ProducerLeg { } /// The plan payload the relay's establisher hands its pump handler: -/// the spoke-leg streams. The establisher and the handler agree on -/// this concrete type (the `ChannelPlan` downcast contract, ADR-049 -/// amendment 2); alkcall never inspects it. +/// the spoke-leg streams plus the adopted spoke `channel_id`. The +/// establisher and the handler agree on this concrete type (the +/// `ChannelPlan` downcast contract, ADR-049 amendment 2); alkcall +/// never inspects it. struct RelayPlan { + spoke_id: u32, spoke_send: MpscSendStream, spoke_recv: MpscRecvStream, } @@ -139,6 +141,7 @@ impl ChannelRelay { let producer_call = Arc::clone(&self.producer_leg.call); let producer_manager = self.producer_leg.manager.clone(); let op_name = spec.name.clone(); + let handler_manager = producer_manager.clone(); let establisher: OpenEstablisher = Arc::new(move |input, per_call_auth| { let producer_call = Arc::clone(&producer_call); let producer_manager = producer_manager.clone(); @@ -156,6 +159,7 @@ impl ChannelRelay { }); let open_handler: OpenHandler = Arc::new(move |_input, plan, channel_conn, _auth| { + let producer_manager = handler_manager.clone(); tokio::spawn(async move { let Some(plan) = plan else { return; @@ -179,8 +183,19 @@ impl ChannelRelay { // The byte-forward hop (ADR-042 layer 2, ADR-050): // two pumps joined inline (R-02). On completion the // pump's drops EOF the spoke leg; the wrapper's - // handler-exit teardown covers the consumer leg. - let _ = pump_bidi(bidi, relay_plan.spoke_recv, relay_plan.spoke_send).await; + // handler-exit teardown covers the consumer leg. The + // adopted spoke channel entry is reclaimed here so the + // producer-leg manager holds no stale routing state + // after the relayed channel is done (the consumer-leg + // wrapper's teardown cannot see it — it is on the + // producer leg's manager). + let RelayPlan { + spoke_id, + spoke_send, + spoke_recv, + } = relay_plan; + let _ = pump_bidi(bidi, spoke_recv, spoke_send).await; + let _ = producer_manager.teardown_channel(spoke_id); }) }); @@ -252,6 +267,7 @@ async fn open_on_producer_leg( } let establishment = Establishment::new(Arc::new(RelayPlan { + spoke_id, spoke_send, spoke_recv, }) as super::operations::ChannelPlan);