diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 8edda7b..5104b9f 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -1,10 +1,10 @@ --- status: draft -last_updated: 2026-09-22 (the targeted fork — ADR-013; the module is -`src/rfc1928/`; the vendor posture retired. The vendor extraction -amendment ADR-011 §Amendment 2026-09-21; Phase 1 initial draft -2026-09-20 — the OQ-SK ledger promoted into ADRs 001-012 and -open-questions.md) +last_updated: 2026-09-24 (the protocol core landed + the cross-cutting +integration suite; the rfc1928 doc's targeted fork 2026-09-22; the +vendor extraction amendment ADR-011 §Amendment 2026-09-21; Phase 1 +initial draft 2026-09-20 — the OQ-SK ledger promoted into ADRs +001-012 and open-questions.md) --- # alksocks — Architecture @@ -20,16 +20,27 @@ architecture spec docs and the ADRs. ## Current State -Phase 1 (Architecture) — initial draft. Phase 0 -(`docs/research/phase-0.md`) is converged (2026-09-14) and -POC-complete (all numbered POCs passed or were dropped with the -scope decision); the OQ-SK ledger's Phase 1 inputs are collected and -promoted into ADRs 001–012. All docs are `draft` — they transition +Phase 1 (Architecture) — the **protocol core is implemented**. Phase +0 (`docs/research/phase-0.md`) is converged (2026-09-14) and +POC-complete (all numbered POCs passed or were dropped with the scope +decision); the OQ-SK ledger's Phase 1 inputs are collected and +promoted into ADRs 001–013. All docs are `draft` — they transition to `reviewed` when their open questions resolve and the architecture -review passes. The RFC 1928 core is implemented (the targeted fork, +review passes. + +The landed core (the protocol-only batch; no `local`/`composed` code +exists yet): the RFC 1928 core in `src/rfc1928/` (the targeted fork, ADR-013 — the vendor group's six tasks landed and passed the -extraction review); the remaining layers follow the decomposition -phase. +extraction review), the ASSOCIATE datagram frame codec (`src/wire.rs`, +ADR-003), the open-op params + typed error surface (`src/params.rs` / +`src/error.rs`, ADR-001), the three seam types (`src/egress.rs`, +ADR-004), the producer half (`src/producer.rs` — the session-gate +establisher, the command-dispatch pump, the three commands, the scope +gates), and the consumer half (`src/consumer.rs` — the +connect/associate/bind session types with teardown ownership). The +batch-1 cross-cutting integration suite (`tests/core-integration.rs`) +pins the core's whole behavior end to end; the `local` and +`composed` provider groups follow (batch 2). ## Documents diff --git a/docs/architecture/consumer.md b/docs/architecture/consumer.md index fc9f01b..9f8c6fe 100644 --- a/docs/architecture/consumer.md +++ b/docs/architecture/consumer.md @@ -1,6 +1,8 @@ --- status: draft -last_updated: 2026-09-20 (Phase 1 initial draft) +last_updated: 2026-09-24 (the landed session signatures reconciled — +`&ChannelClient` constructors, `impl ToTargetAddr` datagrams, the +`Option`-shaped `recv_datagram`; Phase 1 initial draft 2026-09-20) --- # alksocks — The Consumer Half @@ -120,12 +122,12 @@ address) is assembly-layer configuration. ```rust let session = Socks5ChannelSession::connect( - &mut channel_client, target)?; + &channel_client, target).await?; let (mut read, mut write) = session.split(); // post-reply payload pipe -let assoc = AssociateSession::associate(&mut channel_client)?; -assoc.send_datagram(dst, &bytes).await?; -let (src, bytes) = assoc.recv_datagram().await?; +let assoc = AssociateSession::associate(&channel_client).await?; +assoc.send_datagram(dst, &bytes).await?; // frames once, ADR-003 +let (src, bytes) = assoc.recv_datagram().await?.expect("a datagram"); let door = FrontDoor::builder() .credentials(cred_table) // RFC 1929 mapping, ADR-005 @@ -134,6 +136,14 @@ let door = FrontDoor::builder() door.serve(channel_client).await?; ``` +The landed session signatures: the constructors take `&ChannelClient` +(shared — many sessions ride one client, the concurrency posture); +`send_datagram` takes `impl ToTargetAddr` (the CONNECT session's +target-construction posture, ADR-012); `recv_datagram()` returns +`Result, SocksSessionError>` — `None` is +the channel EOF (the association's clean end), never an empty +datagram (F-2: an empty payload is a legal datagram). + ## Design Decisions | ADR | Decision | Summary | diff --git a/docs/architecture/decisions/006-access-control.md b/docs/architecture/decisions/006-access-control.md index 7a98d29..463e0f8 100644 --- a/docs/architecture/decisions/006-access-control.md +++ b/docs/architecture/decisions/006-access-control.md @@ -3,7 +3,8 @@ ## Status Accepted (2026-09-20, Phase 1; scope gates POC-validated in shape — -POC #1 (`socks5:open`), POC #5 (`socks5:bind` at command dispatch)) +POC #1 (`socks5:open`), POC #5 (`socks5:bind` at command dispatch); +consequence asymmetry pinned by the integration suite, 2026-09-24) ## Context @@ -60,7 +61,14 @@ distinction at command dispatch from the per-call opener identity. - Two grants, clean separation: `socks5:open` (egress) and `socks5:bind` (ingress listeners) — never implied by each other (the alktunnels ADR-006 Amendment 1 posture: a capability on its - own scope). + own scope). The implication asymmetry is pinned by the integration + suite (`tests/core-integration.rs`, the identity/scope matrix): + `socks5:open` alone serves CONNECT + ASSOCIATE and is refused BIND + (0x02) at the command gate; `socks5:bind` alone grants *nothing* + on the op — the open gate's `required_scopes = ["socks5:open"]` + fails closed for every command (the establisher never runs, no + phantom channel); both scopes, granted explicitly, serve + everything. A BIND-capable deployment grants both. - BIND's security boundary is enforced even though BIND rides the same op (ADR-001): the command gate is the mechanism, and it is POC-validated (POC #5). diff --git a/src/lib.rs b/src/lib.rs index 90f9f26..082d3a1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,14 +9,49 @@ //! binds a port — kernel binds exist only as explicit, feature-gated //! assembly shapes (`local`), never in the protocol layer. //! -//! The RFC wire vocabulary and the explicit typestate live in -//! [`rfc1928`]: a targeted fork of fast-socks5 v1.0.0's `T`-generic -//! subset (MIT, third-party upstream) carrying two deliberate changes -//! (BIND support, `new_reply` publication) — see -//! `docs/architecture/rfc1928.md`, ADR-013, and ADR-011. +//! ## The landed surface (the protocol core) +//! +//! - **The wire vocabulary and the explicit typestate** live in +//! [`rfc1928`]: a targeted fork of fast-socks5 v1.0.0's `T`-generic +//! subset (MIT, third-party upstream) carrying two deliberate +//! changes (BIND support, `new_reply` publication) — see +//! `docs/architecture/rfc1928.md`, ADR-013, and ADR-011. +//! - **The wire codec** ([`wire`]): the ASSOCIATE datagram frame +//! (`[len: u16 BE][addr + data]`, ADR-003) — framing is applied +//! exactly once, by this layer; the egress speaks inner bytes. +//! - **The wire-stable open-op surface** ([`params`]): the +//! `channels/socks5/sub` op id, the `alk/socks5` ALPN, the +//! trivial `{}` params (ADR-001), and the two scope constants +//! (`socks5:open`, `socks5:bind` — two grants, never implied, +//! ADR-006). The typed open-error vocabulary lives in [`error`]. +//! - **The three seams** ([`egress`]): `DialFn` / `EgressFactory` / +//! `BindAcceptFactory` (ADR-004) — functions, not traits, at the +//! three I/O planes; ownership rides the handle structurally; the +//! egress readiness contract (`next_reply` `Pending` when nothing +//! is ready) is load-bearing API. +//! - **The producer half** ([`producer`]): the session-gate +//! establisher, the command-dispatch pump (CONNECT pass-through / +//! ASSOCIATE datagram stage / BIND phase model), the scope gates +//! (the open gate `socks5:open` at the registry; the command gate +//! `socks5:bind` at BIND dispatch), and +//! [`register_socks5_openable`] — substrate-agnostic over the +//! seams; never a bind in the protocol layer. +//! - **The consumer half** ([`consumer`]): the typed channels-native +//! sessions — [`Socks5ChannelSession`] (CONNECT), +//! [`AssociateSession`] (the framed datagram stage), +//! [`BindSession`] (the reply#1/reply#2 reader) — each owning its +//! channel's teardown; no `Clone`. +//! +//! `local` and `composed` (the feature-gated assembly-layer +//! providers: the front doors, the socket-backed/composed seams) are +//! not yet present — the default crate is protocol-only and +//! wasm-clean by construction (ADR-009). //! //! Architecture: `docs/architecture/overview.md` (spec set: -//! wire.md, producer.md, consumer.md, rfc1928.md). +//! wire.md, producer.md, consumer.md, rfc1928.md); the cross-cutting +//! behavior is pinned end to end by `tests/core-integration.rs` +//! (one registration, all three commands; concurrency; the teardown +//! matrix; the identity/scope matrix; the readiness contract). #![forbid(unsafe_code)] diff --git a/tasks/implementation/core-integration-suite.md b/tasks/implementation/core-integration-suite.md index 0ac1dc7..bac699f 100644 --- a/tasks/implementation/core-integration-suite.md +++ b/tasks/implementation/core-integration-suite.md @@ -1,7 +1,7 @@ --- id: implementation/core-integration-suite name: Core — cross-cutting integration suite + docs sync -status: pending +status: completed depends_on: [implementation/core-consumer-associate, implementation/core-consumer-bind] scope: moderate risk: medium @@ -56,15 +56,97 @@ Deliverables: `--all-targets -- -D warnings`, `cargo fmt --check`, both wasm checks, and `cargo doc --no-deps` (docs changed). -## Acceptance Criteria +## Notes -- [ ] The three-command regression + concurrency + teardown-matrix + - identity-matrix tests all pass -- [ ] Docs reconciled (README Current State; any drift fixed) -- [ ] The full verification sweep is green (test, all-features, - clippy, fmt, both wasm checks, doc) -- [ ] No `local`/`composed` code exists yet (the batch boundary - holds — the default crate is protocol-only) +- **The identity-matrix spec drift, reconciled (the one real drift + the implementation surfaced):** this task's Description row 2 + expected a `socks5:bind`-only identity to serve BIND ("BIND yes, + CONNECT/ASSOCIATE FORBIDDEN at the open gate"). The landed ADR-006 + + implementation pin the open op's `required_scopes = + ["socks5:open"]` **unconditionally** (the single op, all three + commands; ADR-001) — a bind-only identity fails closed at the open + gate for every command (FORBIDDEN, the establisher never runs, no + phantom channel), and a BIND-capable deployment grants both scopes + explicitly. The suite pins the landed posture + (`identity_matrix_bind_only_fails_closed_at_the_open_gate`), + producer- and consumer-side; ADR-006 §Consequences now states the + asymmetry explicitly (never implied, in either direction). The + task's phrasing was a Phase-0-era guess at the gate shape, not a + spec row the code contradicted — the spec (ADR-006 + producer.md) + and the implementation already agreed. +- **Test-shape notes:** the suite rides the established + `tests/harness.rs` topologies (the raw producer-side topology — + the producer's serving side allocates channel ids — and the + consumer-side topology — the consumer allocates; the mux/demux is + pinned in both allocation directions). The mixed-command + concurrency tests box the per-command round-trip futures into one + `Vec` (`Pin>` — the three async fns + have distinct opaque types, so a shared boxed type is the + `join_all` vehicle). The BIND client-gone row uses a parked accept + factory so the quiet-wait exit is the only possible end (the + deterministic client-gone probe). +- **Readiness contract under load:** the integration-level pin + reuses the producer-associate unit test's hot-`Ok(None)` egress + shape across concurrent associations (with a CONNECT interleaved) + — the stage's re-park tolerance must hold concurrently, not just + sequentially. + +## Summary + +Landed the batch-1 closing task: `tests/core-integration.rs` — nine +cross-cutting integration tests pinning the whole protocol core end +to end over the channels harness (the alkcall 0.8.0 establishment +surface, both topology directions): + +1. **One registration, all three commands** (the POC #2/#5 + regression guard, complete): CONNECT + ASSOCIATE + BIND opened + back to back on the one `channels/socks5/sub` op, all three live + simultaneously, each exercised mid-life of the others; a teardown + ends only its own session (mux/demux isolation). +2. **Concurrency**: nine concurrent mixed-command sessions over one + channels connection (the raw topology), six through the typed + consumer sessions (the consumer-side topology — the other + allocation direction), then a fresh round-trip of each command + after the burst; no channel entry leaks on either side. +3. **The teardown matrix** (wire.md §Sentinels and lifecycle): a + client half-close ends a CONNECT data plane cleanly (the + producer's pump exits); channel EOF ends the association; + the BIND client-gone teardown releases the parked accept; fresh + sessions of every command succeed immediately after each + teardown (no connection poisoning). +4. **The identity/scope matrix** (ADR-006 end to end): open-only → + CONNECT + ASSOCIATE serve, BIND refused 0x02 at the command gate + (CONNECT still serves after); bind-only → fail-closed at the + open gate for all commands (the reconciled posture, above); + both-scopes → everything, granted explicitly; the CF-006 + identity witness asserted per row. +5. **The readiness contract** re-pinned under concurrent load: a + hot-`Ok(None)` egress across concurrent associations (with a + CONNECT interleaved) does not starve the datagram stage's channel + branch. + +Docs sync (the architecture-sync check): `docs/architecture/README.md` +Current State rewritten — "the protocol core is implemented," the +landed per-module inventory, `tests/core-integration.rs` named as the +batch-1 pin, batch 2 (`local`/`composed`) as the follow; the +"remaining layers follow the decomposition phase" residue gone. +`consumer.md` §Wiring reconciled to the landed signatures +(`&ChannelClient` constructors, `impl ToTargetAddr` datagrams, the +`Option`-shaped `recv_datagram` — `None` is the channel-EOF +association end, never an empty datagram). ADR-006 §Consequences now +states the grant asymmetry explicitly. `src/lib.rs` crate docs +rewritten to describe the landed surface (rfc1928 fork, wire codec, +params, seams, producer, consumer) pointing at the specs and the +integration suite. + +Verification: `cargo test` 302 green (118 lib + 38 connect + 39 +associate + 40 bind + 40 integration + 30 differential + 37 +rfc1928 + 3 doctests), `cargo test --all-features` green (the empty +features `local`/`composed` compile), `cargo clippy --all-targets -- +-D warnings` clean, `cargo fmt --check` clean, both wasm checks +(`check` + `clippy -D warnings` on `wasm32-unknown-unknown`) clean, +`cargo doc --no-deps` clean. The batch boundary holds: no +`local`/`composed` code exists; the default crate is protocol-only. ## References @@ -79,8 +161,8 @@ Deliverables: ## Notes -> To be filled by implementation agent +> Filled by the implementation agent — see the Notes above. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion — see the Summary above. \ No newline at end of file diff --git a/tests/core-integration.rs b/tests/core-integration.rs new file mode 100644 index 0000000..f068fde --- /dev/null +++ b/tests/core-integration.rs @@ -0,0 +1,1013 @@ +//! The cross-cutting integration suite — the whole protocol core's +//! behavior pinned in one place (the batch-1 closing task; the +//! architecture-sync check's test half). +//! +//! The per-command tasks own their verticals (tests/consumer-*.rs and +//! the `src/` unit suites); this suite owns the cross-cutting +//! invariants: +//! +//! - **One registration, all three commands** (the POC #2/#5 +//! regression guard, complete): CONNECT + ASSOCIATE + BIND all +//! served on the same `channels/socks5/sub` op, interleaved live +//! sessions on one channels connection. +//! - **Concurrency**: mixed concurrent sessions over one channels +//! connection — the mux/demux under load (the POCs tested +//! sequentially), pinned in both allocation directions (the raw +//! topology where the producer's serving side allocates the channel +//! ids, and the consumer-side topology where the consumer does). +//! - **The teardown matrix** (wire.md §Sentinels and lifecycle, the +//! normative source): a client half-close ends a CONNECT data plane +//! cleanly; ASSOCIATE EOF ends the association; a BIND client-gone +//! teardown releases the accept; a fresh session succeeds +//! immediately after each teardown (no connection poisoning). +//! - **The identity/scope matrix** (ADR-006, end to end): +//! `socks5:open`-only (CONNECT + ASSOCIATE serve, BIND refused 0x02 +//! at the command gate), `socks5:bind`-only (fail-closed at the +//! open gate — all three commands; ADR-006's open gate requires +//! `socks5:open` unconditionally and the two grants are never +//! implied), both scopes (everything, granted explicitly). +//! - **The readiness contract** re-pinned under concurrent load (the +//! producer-associate task owns the unit-level test): a hot +//! `Ok(None)` egress across concurrent associations must not starve +//! the datagram stage's channel branch. + +mod harness; + +use std::future::Future; +use std::net::SocketAddr; +use std::sync::Arc; + +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +use alksocks::consumer::{AssociateSession, BindSession, Socks5ChannelSession}; +use alksocks::egress::{ + BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, Egress, EgressError, + EgressFactory, +}; +use alksocks::error::SocksSessionError; +use alksocks::params::Socks5Params; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::rfc1928::Socks5Command; + +use harness::{ + bind_identity, both_scopes_identity, consumer_identity, echo_accept, echo_dial, echo_egress, + wire_consumer_side_with, wire_with, ConsumerSideTopology, RawRfcClient, RegisterMode, Topology, + ECHO_BIND_PORT, +}; + +const TIMEOUT_SECS: u64 = 10; + +type RawPump = tokio::task::JoinHandle<(u64, u64)>; +type RawClient = RawRfcClient; +type RoundTrip<'a> = std::pin::Pin> + Send + 'a>>; + +async fn within(fut: impl Future) -> T { + tokio::time::timeout(std::time::Duration::from_secs(TIMEOUT_SECS), fut) + .await + .expect("operation timed out") +} + +// --- Topology builders ------------------------------------------------------ + +/// The full raw topology: all three seams wired, the both-scopes +/// identity (everything serves). +async fn full_raw_topology() -> Topology { + wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: Some(echo_accept()), + }, + Some(both_scopes_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +// --- Raw-session helpers (the harness's raw-RFC pattern) --------------------- + +/// Open a socks5 channel + adopt it + pump the raw RFC client through +/// it (the smoke test's opening shape). +async fn open_adopt_pump(topo: &Topology) -> (RawClient, RawPump) { + let response = topo + .consumer_call + .call(alksocks::OP_SOCKS5_OPEN, Socks5Params::as_json()) + .await; + let out = response.result.expect("the socks5 channel opens"); + let channel_id = out + .get("channel_id") + .and_then(|v| v.as_u64()) + .expect("the reply carries channel_id") as u32; + let (send, recv) = topo + .consumer_manager + .adopt_channel(channel_id, alksocks::SOCKS5_ALPN, None) + .await + .expect("adopt the producer-allocated id"); + let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024); + let pump = tokio::spawn(alkcall::channels::pump::pump_bidi( + stream_self, + Box::new(recv) as Box, + Box::new(send) as Box, + )); + (RawRfcClient::new(stream_peer), pump) +} + +/// The ASSOCIATE conversation over a fresh channel: greet → request → +/// the sentinel `127.0.0.1:0` reply (ADR-003). +async fn raw_associate(topo: &Topology) -> (RawClient, RawPump) { + let (mut raw, pump) = open_adopt_pump(topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::UDPAssociate, + &TargetAddr::Domain("relay.test".into(), 1080), + ) + .await + .expect("request written"); + let (code, addr) = raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED, "the associate reply"); + assert_eq!( + (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]), + (127, 0, 0, 1, 0, 0), + "the reply carries the sentinel BND.ADDR" + ); + (raw, pump) +} + +/// The BIND conversation over a fresh channel through the echo +/// accept: greet → request → reply#1 (the advertised port) → reply#2 +/// (the accepted peer, the harness's fixed stand-in). +async fn raw_bind(topo: &Topology, host: &str) -> (RawClient, RawPump) { + let (mut raw, pump) = open_adopt_pump(topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::TCPBind, + &TargetAddr::Domain(host.into(), ECHO_BIND_PORT), + ) + .await + .expect("request written"); + let (code1, addr1) = raw.read_reply().await.expect("reply#1"); + assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED, "reply#1 success"); + assert_eq!( + u16::from_be_bytes([addr1[addr1.len() - 2], addr1[addr1.len() - 1]]), + ECHO_BIND_PORT, + "reply#1 carries the factory's advertised port" + ); + let (code2, addr2) = raw.read_reply().await.expect("reply#2"); + assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED, "reply#2 success"); + let peer_port = u16::from_be_bytes([addr2[addr2.len() - 2], addr2[addr2.len() - 1]]); + assert_eq!(peer_port, 44444, "reply#2 carries the accepted peer"); + (raw, pump) +} + +async fn connect_round_trip(topo: &Topology, host: String, tag: String) -> Result<(), String> { + let (mut raw, _pump) = open_adopt_pump(topo).await; + let code = raw + .connect(&TargetAddr::Domain(host, 443)) + .await + .map_err(|e| e.to_string())?; + if code != consts::SOCKS5_REPLY_SUCCEEDED { + return Err(format!("reply code {code}")); + } + raw.stream + .write_all(tag.as_bytes()) + .await + .map_err(|e| e.to_string())?; + let mut buf = vec![0u8; tag.len()]; + raw.stream + .read_exact(&mut buf) + .await + .map_err(|e| e.to_string())?; + if buf != tag.as_bytes() { + return Err("echo mismatch".into()); + } + Ok(()) +} + +/// A raw ASSOCIATE round-trip: sentinel reply + one datagram echoed +/// in order. +async fn associate_round_trip( + topo: &Topology, + octet: u8, + tag: &'static [u8], +) -> Result<(), String> { + let (mut raw, _pump) = raw_associate(topo).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, octet], 53))); + raw.stream + .write_all(&udp_frame(&target, tag)) + .await + .map_err(|e| e.to_string())?; + let mut buf = vec![0u8; 2 + 7 + tag.len()]; + raw.stream + .read_exact(&mut buf) + .await + .map_err(|e| e.to_string())?; + let (src, payload) = udp_unframe(&buf); + if src != target { + return Err("echo source mismatch".into()); + } + if &payload[..] != tag { + return Err("echo payload mismatch".into()); + } + Ok(()) +} + +/// A raw BIND round-trip: reply#1 + reply#2 + one pass-through echo +/// exchange on the data plane. +async fn bind_round_trip( + topo: &Topology, + host: impl ToString, + tag: &'static [u8], +) -> Result<(), String> { + let (mut raw, _pump) = raw_bind(topo, &host.to_string()).await; + raw.stream.write_all(tag).await.map_err(|e| e.to_string())?; + let mut buf = vec![0u8; tag.len()]; + raw.stream + .read_exact(&mut buf) + .await + .map_err(|e| e.to_string())?; + if buf != tag { + return Err("echo mismatch".into()); + } + Ok(()) +} + +/// A raw CONNECT round-trip boxed for mixed concurrent joining. +fn tcp_fut<'a>(topo: &'a Topology, host: String, tag: String) -> RoundTrip<'a> { + Box::pin(connect_round_trip(topo, host, tag)) +} + +/// A raw ASSOCIATE round-trip boxed for mixed concurrent joining. +fn udp_fut<'a>(topo: &'a Topology, octet: u8, tag: &'static [u8]) -> RoundTrip<'a> { + Box::pin(associate_round_trip(topo, octet, tag)) +} + +/// A raw BIND round-trip boxed for mixed concurrent joining. +fn bind_fut<'a>(topo: &'a Topology, host: String, tag: &'static [u8]) -> RoundTrip<'a> { + Box::pin(bind_round_trip(topo, host, tag)) +} + +// --- The ASSOCIATE wire-frame helpers (the harness's shape) ------------------ + +fn udp_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { + alksocks::wire::frame(addr, payload) + .expect("frame") + .to_vec() +} + +fn udp_unframe(wire: &[u8]) -> (TargetAddr, bytes::Bytes) { + let mut reader = alksocks::wire::DatagramReader::new(); + let mut got = None; + for dg in reader.feed(wire) { + got = Some(dg); + } + got.expect("one datagram").expect("frame decodes") +} + +// --- Seam stand-ins ---------------------------------------------------------- + +/// A parked accept factory: the accept future never resolves (the +/// deterministic quiet-wait probe — the client-gone exit is the only +/// end this session can take). +fn parked_accept() -> BindAcceptFactory { + Arc::new(move |_target: TargetAddr| { + Box::pin(async move { + let bind_addr = SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)); + Ok(BindAccept::new( + bind_addr, + std::future::pending::>(), + )) + }) + }) +} + +/// The hot egress (the readiness contract's violator, the +/// producer-associate unit test's shape): `next_reply` drains one +/// queued reply if any, then returns an instantly-ready `Ok(None)` +/// instead of parking. The stage must tolerate this under load — a +/// biased select starved into livelock would hang every session here. +struct HotNoneEgress { + echo: tokio::sync::mpsc::UnboundedSender, + replies: tokio::sync::Mutex>, +} + +impl Egress for HotNoneEgress { + fn send( + &mut self, + addr: TargetAddr, + payload: bytes::Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { + let mut inner = addr + .to_be_bytes() + .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; + inner.extend_from_slice(&payload); + self.echo + .send(bytes::Bytes::from(inner)) + .map_err(|_| EgressError::General("egress closed".into())) + }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { + match self.replies.lock().await.try_recv() { + Ok(inner) => Ok(Some(inner)), + Err(_) => Ok(None), + } + }) + } +} + +fn hot_none_egress() -> EgressFactory { + Arc::new(|| { + Box::pin(async { + let (echo, replies) = tokio::sync::mpsc::unbounded_channel(); + Ok(Box::new(HotNoneEgress { + echo, + replies: tokio::sync::Mutex::new(replies), + }) as Box) + }) + }) +} + +// --- The consumer-session probes ---------------------------------------------- + +/// The no-leaked-channels probe (the per-command suites' shape): both +/// managers hold only channel 0; the producer's entry reaps by its +/// open-op wrapper — the probe polls past the reap race. +async fn no_leaked_channels(topo: &ConsumerSideTopology) { + let deadline = std::time::Duration::from_secs(5); + let consumer_ids = topo.consumer.manager().channel_ids(); + assert!( + consumer_ids.iter().all(|&id| id == 0), + "consumer leaked channel entries: {consumer_ids:?}" + ); + tokio::time::timeout(deadline, async { + loop { + let producer_ids = topo.producer_manager.channel_ids(); + if producer_ids.iter().all(|&id| id == 0) { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("the producer's handler-exit teardown reaps the entry"); +} + +/// A `FORBIDDEN`-carrying open failure (the fail-closed probe's +/// expectation shape). +fn expect_forbidden(err: &SocksSessionError) -> bool { + match err { + SocksSessionError::Open(open) => match open.open_ref() { + alkcall::channels::client::ChannelOpenError::CallFailed { error } => { + error.code == "FORBIDDEN" + } + _ => false, + }, + _ => false, + } +} + +/// A consumer CONNECT round-trip, result-shaped for concurrent +/// joining. +async fn consumer_connect_round_trip( + topo: &ConsumerSideTopology, + host: String, + tag: String, +) -> Result<(), String> { + let mut session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + (host.as_str(), 443u16).to_target_addr().unwrap(), + ) + .await + .map_err(|e| format!("connect: {e}"))?; + session + .split() + .expect("pipe") + .1 + .write_all(tag.as_bytes()) + .await + .map_err(|e| e.to_string())?; + let (mut read, _) = session.split().expect("pipe"); + let mut buf = vec![0u8; tag.len()]; + read.read_exact(&mut buf).await.map_err(|e| e.to_string())?; + if buf != tag.as_bytes() { + return Err("echo mismatch".into()); + } + session.close().await; + Ok(()) +} + +/// A consumer ASSOCIATE round-trip: one datagram echoed in order on +/// the framed stage. +async fn consumer_associate_round_trip( + topo: &ConsumerSideTopology, + octet: u8, + tag: &'static [u8], +) -> Result<(), String> { + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .map_err(|e| format!("associate: {e}"))?; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 1, octet], 53))); + assoc + .send_datagram(&target, tag) + .await + .map_err(|e| format!("send: {e}"))?; + let (src, payload) = within(assoc.recv_datagram()) + .await + .map_err(|e| format!("recv: {e}"))? + .expect("a datagram (not eof)"); + if src != target { + return Err("echo source mismatch".into()); + } + if &payload[..] != tag { + return Err("echo payload mismatch".into()); + } + assoc.close().await; + Ok(()) +} + +/// A consumer CONNECT round-trip boxed for mixed concurrent joining. +fn consumer_tcp_fut<'a>( + topo: &'a ConsumerSideTopology, + host: String, + tag: String, +) -> RoundTrip<'a> { + Box::pin(consumer_connect_round_trip(topo, host, tag)) +} + +/// A consumer ASSOCIATE round-trip boxed for mixed concurrent joining. +fn consumer_udp_fut<'a>( + topo: &'a ConsumerSideTopology, + octet: u8, + tag: &'static [u8], +) -> RoundTrip<'a> { + Box::pin(consumer_associate_round_trip(topo, octet, tag)) +} + +/// A consumer BIND round-trip boxed for mixed concurrent joining. +fn consumer_bind_fut<'a>( + topo: &'a ConsumerSideTopology, + host: String, + tag: &'static [u8], +) -> RoundTrip<'a> { + Box::pin(consumer_bind_round_trip(topo, host, tag)) +} + +/// A consumer BIND round-trip: reply#1 + reply#2 + one echo exchange +/// on the pass-through plane. +async fn consumer_bind_round_trip( + topo: &ConsumerSideTopology, + host: String, + tag: &'static [u8], +) -> Result<(), String> { + let mut session = BindSession::bind( + topo.consumer.as_ref(), + (host.as_str(), ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .map_err(|e| format!("bind: {e}"))?; + let mut plane = session + .await_inbound() + .await + .map_err(|e| format!("reply#2: {e}"))? + .expect("the data plane"); + plane.write_all(tag).await.map_err(|e| e.to_string())?; + plane.flush().await.map_err(|e| e.to_string())?; + let mut buf = vec![0u8; tag.len()]; + within(plane.read_exact(&mut buf)) + .await + .map_err(|e| e.to_string())?; + if buf != tag { + return Err("echo mismatch".into()); + } + drop(plane); + session.close().await; + Ok(()) +} + +// --- The cross-cutting suite --------------------------------------------------- + +/// One registration, all three commands, interleaved live sessions on +/// one channels connection (the POC #2/#5 regression guard, +/// complete): all three sessions are opened back to back and are live +/// simultaneously; each is exercised mid-life of the others; a +/// teardown ends only its own session (the others keep serving — the +/// mux/demux isolation). +#[tokio::test] +async fn one_registration_all_three_commands_interleaved() { + let topo = full_raw_topology().await; + + // Three interleaved opens on the one op; all three sessions live + // at once (CONNECT, ASSOCIATE, BIND — fresh channel per session, + // convention 10). + let (mut connect_raw, connect_pump) = open_adopt_pump(&topo).await; + let (mut assoc_raw, assoc_pump) = raw_associate(&topo).await; + let (mut bind_raw, bind_pump) = raw_bind(&topo, "bind.test").await; + + // The CONNECT session's RFC handshake (the other two helpers ran + // theirs on their own channels). + connect_raw.greet().await.expect("greet"); + connect_raw + .request( + Socks5Command::TCPConnect, + &TargetAddr::Domain("interleaved.test".into(), 443), + ) + .await + .expect("request written"); + let (code, _addr) = connect_raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + + // Exercise each mid-life of the others. + connect_raw.stream.write_all(b"tcp1").await.expect("write"); + let mut buf = [0u8; 4]; + connect_raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"tcp1"); + + let udp_target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53))); + assoc_raw + .stream + .write_all(&udp_frame(&udp_target, b"udp1")) + .await + .expect("datagram"); + let mut frame_buf = vec![0u8; 2 + 7 + 4]; + assoc_raw + .stream + .read_exact(&mut frame_buf) + .await + .expect("frame"); + let (_, payload) = udp_unframe(&frame_buf); + assert_eq!(&payload[..], b"udp1"); + + bind_raw.stream.write_all(b"bin1").await.expect("write"); + bind_raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"bin1"); + + // A teardown ends only its own session. + drop(connect_raw); + within(connect_pump).await.expect("the CONNECT pump ends"); + + // ASSOCIATE + BIND still serve (no connection poisoning). + assoc_raw + .stream + .write_all(&udp_frame(&udp_target, b"still")) + .await + .expect("datagram"); + let mut frame_buf = vec![0u8; 2 + 7 + 5]; + assoc_raw + .stream + .read_exact(&mut frame_buf) + .await + .expect("frame"); + let (_, payload) = udp_unframe(&frame_buf); + assert_eq!(&payload[..], b"still"); + + bind_raw.stream.write_all(b"alive").await.expect("write"); + let mut buf = [0u8; 5]; + bind_raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"alive"); + + drop(assoc_raw); + drop(bind_raw); + within(assoc_pump).await.expect("associate pump"); + within(bind_pump).await.expect("bind pump"); +} + +/// Concurrency over one channels connection (the mux/demux under +/// load): nine concurrent sessions of mixed command types — three +/// CONNECTs, three ASSOCIATEs, three BINDs — interleaved by the +/// executor, all riding the one registration. The producer's serving +/// side allocates the channel ids (the raw topology's direction). +#[tokio::test] +async fn concurrent_mixed_sessions_over_one_connection() { + let topo = full_raw_topology().await; + + const UDP_TAGS: [&[u8]; 3] = [b"udp-zero0", b"udp-one!1", b"udp-two!2"]; + const BIND_TAGS: [&[u8]; 3] = [b"bind-z0", b"bind-o1", b"bind-t2"]; + let mut futs: Vec = Vec::new(); + for i in 0..3usize { + futs.push(tcp_fut(&topo, format!("c{i}.test"), format!("tcp{i}"))); + futs.push(udp_fut(&topo, 10 + i as u8, UDP_TAGS[i])); + futs.push(bind_fut(&topo, format!("b{i}.test"), BIND_TAGS[i])); + } + let results = futures::future::join_all(futs).await; + for (i, result) in results.into_iter().enumerate() { + result.unwrap_or_else(|e| panic!("concurrent session {i} failed: {e}")); + } + + // The connection is healthy after the burst: one more of each. + connect_round_trip(&topo, "after.test".into(), "again".into()) + .await + .expect("fresh connect after the load"); + associate_round_trip(&topo, 20, b"again") + .await + .expect("fresh associate after the load"); + bind_round_trip(&topo, "after.test", b"again") + .await + .expect("fresh bind after the load"); +} + +/// Concurrency through the typed consumer sessions over one channels +/// connection: six concurrent sessions (two of each command) driven +/// by the real session types. The consumer dials here (the +/// consumer-side topology), so the consumer's side allocates the +/// channel ids — the mux/demux pinned in the other allocation +/// direction. No channel entry leaks on either side. +#[tokio::test] +async fn concurrent_consumer_sessions_over_one_connection() { + let topo = wire_consumer_side_with( + echo_dial(), + Some(echo_egress()), + Some(echo_accept()), + Some(both_scopes_identity()), + ) + .await; + + const UDP_TAGS: [&[u8]; 2] = [b"cu-zero00", b"cu-one!!1"]; + const BIND_TAGS: [&[u8]; 2] = [b"cb-zero", b"cb-one!"]; + let mut futs: Vec = Vec::new(); + for i in 0..2usize { + futs.push(consumer_tcp_fut( + &topo, + format!("cc{i}.test"), + format!("t{i}"), + )); + futs.push(consumer_udp_fut(&topo, 30 + i as u8, UDP_TAGS[i])); + futs.push(consumer_bind_fut( + &topo, + format!("cb{i}.test"), + BIND_TAGS[i], + )); + } + let results = futures::future::join_all(futs).await; + for (i, result) in results.into_iter().enumerate() { + result.unwrap_or_else(|e| panic!("consumer session {i} failed: {e}")); + } + + no_leaked_channels(&topo).await; +} + +/// The teardown matrix, CONNECT row (wire.md §Sentinels and +/// lifecycle): a client half-close ends the data plane cleanly (the +/// producer's pump exits) and a fresh session of every command +/// succeeds immediately after — no connection poisoning. +#[tokio::test] +async fn teardown_connect_half_close_then_every_command_serves_fresh() { + let topo = full_raw_topology().await; + + let (mut raw, pump) = open_adopt_pump(&topo).await; + let code = raw + .connect(&TargetAddr::Domain("drop.test".into(), 443)) + .await + .expect("connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + raw.stream.write_all(b"pre").await.expect("write"); + let mut buf = [0u8; 3]; + raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"pre"); + + // The client half-close: the producer's pump exits (the data + // plane is over; the connection stays healthy). + drop(raw); + within(pump) + .await + .expect("the pump ends on the client half-close"); + + // Fresh sessions of every command succeed immediately. + let (mut raw, _pump) = open_adopt_pump(&topo).await; + let code = raw + .connect(&TargetAddr::Domain("fresh.test".into(), 443)) + .await + .expect("fresh connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + raw.stream.write_all(b"again").await.expect("write"); + let mut buf = [0u8; 5]; + raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"again"); + + let (mut assoc, _pump) = raw_associate(&topo).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 2], 53))); + assoc + .stream + .write_all(&udp_frame(&target, b"fresh")) + .await + .expect("datagram"); + let mut frame_buf = vec![0u8; 2 + 7 + 5]; + assoc + .stream + .read_exact(&mut frame_buf) + .await + .expect("frame"); + let (_, payload) = udp_unframe(&frame_buf); + assert_eq!(&payload[..], b"fresh"); + + bind_round_trip(&topo, "fresh.test", b"fresh") + .await + .expect("fresh bind"); +} + +/// The teardown matrix, ASSOCIATE row: channel EOF ends the +/// association (the association lifetime = channel lifetime) and +/// fresh sessions of every command succeed immediately after. +#[tokio::test] +async fn teardown_associate_eof_then_every_command_serves_fresh() { + let topo = full_raw_topology().await; + + let (mut assoc, pump) = raw_associate(&topo).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 3], 53))); + assoc + .stream + .write_all(&udp_frame(&target, b"pre")) + .await + .expect("datagram"); + let mut frame_buf = vec![0u8; 2 + 7 + 3]; + assoc + .stream + .read_exact(&mut frame_buf) + .await + .expect("frame"); + let (_, payload) = udp_unframe(&frame_buf); + assert_eq!(&payload[..], b"pre"); + + // Channel EOF ends the association (the clean exit). + drop(assoc); + within(pump).await.expect("the stage ends on channel EOF"); + + // Fresh sessions of every command succeed immediately. + let (mut assoc, _pump) = raw_associate(&topo).await; + assoc + .stream + .write_all(&udp_frame(&target, b"fresh")) + .await + .expect("datagram"); + let mut frame_buf = vec![0u8; 2 + 7 + 5]; + assoc + .stream + .read_exact(&mut frame_buf) + .await + .expect("frame"); + let (_, payload) = udp_unframe(&frame_buf); + assert_eq!(&payload[..], b"fresh"); + + connect_round_trip(&topo, "fresh.test".into(), "again".into()) + .await + .expect("fresh connect"); + + bind_round_trip(&topo, "fresh.test", b"fresh") + .await + .expect("fresh bind"); +} + +/// The teardown matrix, BIND row: the client vanishing in the quiet +/// wait releases the accept (the parked accept future drops with the +/// `select!` scope — the client-gone teardown, ADR-004/ADR-007) and +/// fresh sessions succeed immediately after. +#[tokio::test] +async fn teardown_bind_client_gone_releases_the_accept_then_serves_fresh() { + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: Some(parked_accept()), + }, + Some(both_scopes_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // Reply#1 lands (the parked factory resolves); the quiet wait is + // then the session's only live phase. + let (mut raw, pump) = open_adopt_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::TCPBind, + &TargetAddr::Domain("gone.test".into(), ECHO_BIND_PORT), + ) + .await + .expect("request written"); + let (code1, _addr) = raw.read_reply().await.expect("reply#1"); + assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED); + + // The client vanishes mid-quiet-wait: the session ends cleanly + // (the accept future is released with it). + drop(raw); + within(pump) + .await + .expect("the session ends on the client-gone teardown"); + + // The accept re-arms: a fresh BIND reaches reply#1 again (and its + // quiet wait tears down cleanly the same way). + let (mut raw, pump) = open_adopt_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::TCPBind, + &TargetAddr::Domain("fresh.test".into(), ECHO_BIND_PORT), + ) + .await + .expect("request written"); + let (code1, _addr) = raw.read_reply().await.expect("reply#1"); + assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED); + drop(raw); + within(pump) + .await + .expect("the fresh BIND session ends cleanly"); + + // CONNECT serves the same topology (no poisoning). + connect_round_trip(&topo, "fresh.test".into(), "again".into()) + .await + .expect("fresh connect after the bind teardown"); +} + +/// The identity/scope matrix, `socks5:open`-only row (ADR-006): +/// CONNECT and ASSOCIATE serve; BIND is refused `ConnectionNotAllowed` +/// (0x02) at the command gate while CONNECT still serves the same +/// identity — ingress is never implied by egress. +#[tokio::test] +async fn identity_matrix_open_only_connect_associate_serve_bind_refused_0x02() { + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: Some(echo_accept()), + }, + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // CONNECT serves. + connect_round_trip(&topo, "open.test".into(), "tcp".into()) + .await + .expect("CONNECT serves the open-only identity"); + + // ASSOCIATE serves. + associate_round_trip(&topo, 40, b"udp") + .await + .expect("ASSOCIATE serves the open-only identity"); + + // BIND: the command gate refuses 0x02 — RFC-faithful, no data + // plane (the accept seam never ran). + let (mut raw, pump) = open_adopt_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::TCPBind, + &TargetAddr::Domain("gate.test".into(), ECHO_BIND_PORT), + ) + .await + .expect("request written"); + let (code, _addr) = raw.read_reply().await.expect("reply read"); + assert_eq!( + code, + consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED, + "the command gate's 0x02" + ); + drop(raw); + within(pump).await.expect("the gated session ends"); + + // CONNECT still serves the same identity (the gate distinguishes + // ingress from egress, not the other way around). + connect_round_trip(&topo, "still.test".into(), "again".into()) + .await + .expect("CONNECT still serves after the BIND refusal"); + + // The per-call opener identity the gates acted on (CF-006). + assert_eq!( + topo.identity_witness.lock().await.as_deref(), + Some("consumer") + ); +} + +/// The identity/scope matrix, `socks5:bind`-only row (ADR-006, as +/// landed): the open op's `required_scopes` is `socks5:open` +/// unconditionally — a bind-only identity fails closed at the open +/// gate for all three commands (FORBIDDEN; the establisher never +/// runs; no channel anywhere). `socks5:bind` alone grants nothing +/// here; the two grants are never implied, in either direction. +#[tokio::test] +async fn identity_matrix_bind_only_fails_closed_at_the_open_gate() { + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: Some(echo_accept()), + }, + Some(bind_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // The open itself is refused: fail-closed, no channel anywhere, + // the establisher never ran (the registry's ACL gate runs first). + let response = topo + .consumer_call + .call(alksocks::OP_SOCKS5_OPEN, Socks5Params::as_json()) + .await; + let err = response + .result + .expect_err("a bind-only identity fails closed"); + assert_eq!(err.code, "FORBIDDEN"); + assert!(topo + .consumer_manager + .channel_ids() + .into_iter() + .all(|id| id == 0)); + assert!(topo + .producer + .manager() + .channel_ids() + .into_iter() + .all(|id| id == 0)); + assert!( + topo.identity_witness.lock().await.is_none(), + "the establisher never ran — fail-closed" + ); + + // The same posture through the typed consumer sessions: the BIND + // session's open is a typed FORBIDDEN open error, no session, no + // phantom channel. + let ctopo = wire_consumer_side_with( + echo_dial(), + None, + Some(echo_accept()), + Some(bind_identity()), + ) + .await; + let err = BindSession::bind( + ctopo.consumer.as_ref(), + ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect_err("a bind-only identity fails closed"); + assert!( + expect_forbidden(&err), + "FORBIDDEN is the open error: {err:?}" + ); + no_leaked_channels(&ctopo).await; +} + +/// The identity/scope matrix, both-scopes row (ADR-006): both grants, +/// granted explicitly (never implied), serve every command on the one +/// registration. +#[tokio::test] +async fn identity_matrix_both_scopes_serve_every_command() { + let topo = full_raw_topology().await; + + connect_round_trip(&topo, "both.test".into(), "tcp".into()) + .await + .expect("CONNECT serves the both-scopes identity"); + associate_round_trip(&topo, 50, b"udp") + .await + .expect("ASSOCIATE serves the both-scopes identity"); + bind_round_trip(&topo, "both.test", b"bind") + .await + .expect("BIND serves the both-scopes identity"); + + assert_eq!( + topo.identity_witness.lock().await.as_deref(), + Some("consumer-both") + ); +} + +/// The readiness contract under concurrent load (the integration +/// level pin; the unit-level test is the producer-associate task's): +/// a hot `Ok(None)` egress across concurrent associations — with a +/// CONNECT interleaved — must not starve the datagram stage's channel +/// branch. Datagrams still egress and echo. +#[tokio::test] +async fn readiness_contract_holds_under_concurrent_load() { + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(hot_none_egress()), + accept: None, + }, + Some(both_scopes_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + const UDP_TAGS: [&[u8]; 2] = [b"hot-zero0", b"hot-one!!"]; + let mut futs: Vec = UDP_TAGS + .into_iter() + .enumerate() + .map(|(i, tag)| udp_fut(&topo, 40 + i as u8, tag)) + .collect(); + futs.push(tcp_fut(&topo, "hot.test".into(), "tcp".into())); + let results = futures::future::join_all(futs).await; + for (i, result) in results.into_iter().enumerate() { + result.unwrap_or_else(|e| panic!("the stage starved under load (session {i}): {e}")); + } + + // The stage is still healthy: another datagram round-trips. + associate_round_trip(&topo, 50, b"again") + .await + .expect("fresh associate after the hot-egress load"); +}