feat(review 008 Unit 3c): the gate-2 e2e harness; 0.8.0 release bookkeeping
- 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.
This commit is contained in:
@@ -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/<seg>/(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<CallConnection>` + 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
|
||||
|
||||
Generated
+1
-1
@@ -27,7 +27,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "alkcall"
|
||||
version = "0.7.1"
|
||||
version = "0.8.0"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
|
||||
+1
-1
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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`.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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/<alpn>/sub` and `channels/<alpn>/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
|
||||
|
||||
@@ -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<Identity> {
|
||||
None
|
||||
}
|
||||
fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option<Identity> {
|
||||
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<std::sync::Mutex<Option<tokio::io::DuplexStream>>>);
|
||||
|
||||
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<std::io::Result<()>> {
|
||||
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<std::io::Result<usize>> {
|
||||
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<std::io::Result<()>> {
|
||||
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<std::io::Result<()>> {
|
||||
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<tokio::sync::watch::Sender<bool>>,
|
||||
) -> 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<dyn IdentityProvider>,
|
||||
)
|
||||
.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<tokio::sync::watch::Sender<bool>>,
|
||||
/// 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<PerIdentityChannelPolicy>,
|
||||
}
|
||||
|
||||
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<tokio::sync::watch::Sender<bool>>) -> 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<tokio::sync::Mutex<Option<crate::channels::manager::ChannelManager>>> =
|
||||
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<tokio::sync::Mutex<Option<crate::channels::manager::ChannelManager>>> =
|
||||
Arc::new(tokio::sync::Mutex::new(None));
|
||||
let consumer_policy = Arc::new(PerIdentityChannelPolicy::new(8));
|
||||
let policy_dyn: Arc<dyn ChannelLifecyclePolicy> =
|
||||
Arc::clone(&consumer_policy) as Arc<dyn ChannelLifecyclePolicy>;
|
||||
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"
|
||||
);
|
||||
}
|
||||
@@ -290,3 +290,7 @@ impl HubLegTemplate {
|
||||
#[cfg(test)]
|
||||
#[path = "hub_leg_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "gate2_tests.rs"]
|
||||
mod gate2_tests;
|
||||
|
||||
+21
-5
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user