diff --git a/src/params.rs b/src/params.rs index fdf0d56..ea4216e 100644 --- a/src/params.rs +++ b/src/params.rs @@ -44,7 +44,7 @@ pub struct TunnelParams { /// The substrate discriminator (ADR-001). `Unix` ships in v1 (the /// `local` feature implements it — OQ-TN-14); a new substrate is a /// new value, not a format change. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum Substrate { Tcp, diff --git a/src/producer.rs b/src/producer.rs index 0ba0e3d..3f889d3 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -1,10 +1,238 @@ -//! The producer half: the `channels/tunnel/sub` open op — spec -//! builder, establisher shapes (dial + listen), the pump handler, and -//! registration (`register_tunnel_openable`) per `producer.md` / +//! The producer half: the `channels/tunnel/sub` open op — its +//! establisher (dial shape), its substrate-agnostic pump handler, and +//! its registration (`register_tunnel_openable`) per `producer.md` / //! ADR-004. //! -//! Skeleton module — filled by `tunnels/producer-open-op` (establisher -//! + pump handler + registration) and `tunnels/producer-listen` (the -//! listen establisher variant). +//! The establisher is the open op's awaited, bounded establishment +//! phase (alkcall ADR-049): parse params semantically, look the +//! resource up in the producer's registry, dial via the injected +//! `DialFn` (ADR-004 — "produce boxed halves for a resource" is a +//! function, not a trait), and return the handle in +//! `Establishment::new(plan)` (R-01 — typed-opaque `ChannelPlan`; its +//! `Send + Sync` bound is why `TargetHandle`'s halves carry `+ Sync`). +//! The dial closure closes over substrate code — the protocol crate +//! never sees a socket type. +//! +//! The pump handler downcasts the plan and awaits +//! `alkcall::channels::pump_bidi` **inline** inside its task (R-02: +//! the returned `JoinHandle` tracks the data-plane lifetime; early +//! return = teardown-at-birth). It cannot know what the halves came +//! from and must not (AGENTS.md convention 7). +//! +//! The listen establisher variant (the producer-side listener — the +//! `-R` far-side listener shape) lands with `tunnels/producer-listen`. -pub use crate::params::{establishment_reason, OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; +use std::sync::Arc; + +use alkcall::channels::operations::{ + ChannelCore, ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, +}; +use alkcall::channels::pump::pump_bidi; +use alkcall::core::auth::AuthContext; +use alkcall::core::types::Connection; +use alkcall::registry::registration::OperationRegistry; +use serde_json::Value; +use tokio::io::{AsyncRead, AsyncWrite}; + +use crate::params::{Substrate, TunnelParams}; + +pub use crate::params::{OP_TUNNEL_OPEN, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE}; + +/// The dialed target handle — the plan payload (typed-opaque +/// `ChannelPlan`; establisher and handler agree on the concrete +/// type). The boxed halves carry `+ Sync` to satisfy the plan's +/// `Send + Sync` bound (POC F-1, documented upstream). +pub struct TargetHandle { + pub read: Box, + pub write: Box, +} + +/// The typed establishment-failure vocabulary (ADR-049 §3's reason +/// codes as this crate's establisher errors; mapped 1:1 onto +/// `EstablishmentError`). +#[derive(Debug, thiserror::Error)] +pub enum TunnelEstablishError { + #[error("unknown resource: {0}")] + UnknownResource(String), + #[error("dial failed: {0}")] + DialFailed(String), + #[error("resource shortage: {0}")] + ResourceShortage(String), + #[error("handler error: {0}")] + HandlerError(String), +} + +impl From for EstablishmentError { + fn from(e: TunnelEstablishError) -> Self { + match e { + TunnelEstablishError::UnknownResource(m) => { + EstablishmentError::UnknownResource { message: m } + } + TunnelEstablishError::DialFailed(m) => EstablishmentError::DialFailed { message: m }, + TunnelEstablishError::ResourceShortage(m) => { + EstablishmentError::ResourceShortage { message: m } + } + TunnelEstablishError::HandlerError(m) => { + EstablishmentError::HandlerError { message: m } + } + } + } +} + +/// The producer's registry of produced resources: +/// `(resource, substrate)` → backing. An assembly-owned construct +/// (OQ-TN-11 — the collision domain is the producer's registry; two +/// producers on one connection may expose the same resource name +/// independently). The backing is opaque to the protocol — the +/// `DialFn` closes over whatever shape the assembly layer chose. +#[derive(Clone, Default)] +pub struct ResourceRegistry { + targets: Arc>>, +} + +impl ResourceRegistry { + pub fn new() -> Self { + Self::default() + } + + /// Produce a resource: map `(resource, substrate)` to its backing + /// (a local port, a socket path, an in-process address — the + /// dial closure's business, not the wire's). + pub async fn register(&self, resource: &str, substrate: Substrate, backing: &str) { + self.targets + .lock() + .await + .insert((resource.to_string(), substrate), backing.to_string()); + } + + /// Resolve a resource to its backing; `None` = the producer does + /// not produce it (`unknown_resource` at the establisher). + pub async fn lookup(&self, substrate: Substrate, resource: &str) -> Option { + self.targets + .lock() + .await + .get(&(resource.to_string(), substrate)) + .cloned() + } +} + +/// The substrate dial function the assembly layer injects (ADR-004): +/// given a substrate + a registry backing, produce the target's +/// boxed halves. Behind the `local` feature for real sockets, or +/// assembly-constructed (in-process pipes, test doubles). +pub type DialFn = Arc< + dyn Fn( + Substrate, + &str, + ) + -> futures::future::BoxFuture<'static, Result> + + Send + + Sync, +>; + +/// The witness capturing the establisher's per-call opener identity +/// (CF-006 probe — the establisher sees the END CALLER overlaid onto +/// the install-time context, not a synthetic install-time value). +pub type IdentityWitness = Arc>>; + +/// The establisher (dial shape): registry lookup → dial → +/// `Establishment::new(plan)`. The per-call `auth` is the dispatch- +/// resolved identity overlaid onto the install-time context (CF-006); +/// record it into `identity_witness` when provided. +pub fn tunnel_establisher(registry: ResourceRegistry, dial: DialFn) -> OpenEstablisher { + tunnel_establisher_with_witness(registry, dial, None) +} + +/// [`tunnel_establisher`] with an optional identity witness (the test +/// seam for the CF-006 contract). +pub fn tunnel_establisher_with_witness( + registry: ResourceRegistry, + dial: DialFn, + identity_witness: Option, +) -> OpenEstablisher { + Arc::new(move |input: Value, auth| { + let registry = registry.clone(); + let dial = Arc::clone(&dial); + let witness = identity_witness.clone(); + Box::pin(async move { + if let Some(w) = &witness { + *w.lock().await = auth.identity.as_ref().map(|i| i.id.clone()); + } + let params: TunnelParams = parse_params(&input)?; + let backing = registry + .lookup(params.substrate, ¶ms.resource) + .await + .ok_or_else(|| { + TunnelEstablishError::UnknownResource(format!( + "resource `{}` ({:?}) not produced by this producer", + params.resource, params.substrate + )) + })?; + let dialed: TargetHandle = dial(params.substrate, &backing).await?; + Ok(Establishment::new(Arc::new(dialed) as ChannelPlan)) + }) + }) +} + +fn parse_params(input: &Value) -> Result { + serde_json::from_value(input.clone()) + .map_err(|e| TunnelEstablishError::HandlerError(format!("invalid params: {e}"))) +} + +/// The pump handler: accept the channel's yield-once `BiStream`, +/// downcast the plan to `TargetHandle`, and await `pump_bidi` inline +/// (R-02 — the returned `JoinHandle` tracks the data-plane lifetime; +/// an early return is teardown-at-birth, and the birth-teardown +/// telemetry diagnoses the spawn-and-forget shape at runtime). The +/// handler is substrate-agnostic by construction — it sees boxed +/// halves, never a socket type. +pub fn make_tunnel_pump_handler() -> OpenHandler { + Arc::new( + move |_input: Value, plan: Option, conn: Connection, _auth: AuthContext| { + tokio::spawn(async move { + let Some(bidi) = conn.accept_bi().await.ok() else { + tracing::error!("tunnel pump: accept_bi failed"); + return; + }; + let Some(handle) = plan.and_then(|p| p.downcast::().ok()) else { + tracing::error!("tunnel pump: no TargetHandle in plan"); + return; + }; + let Ok(handle) = Arc::try_unwrap(handle) else { + tracing::error!("tunnel pump: TargetHandle still shared; cannot unwrap"); + return; + }; + let _counts = pump_bidi(bidi, handle.read, handle.write).await; + }) + }, + ) +} + +/// Register the tunnel open op on a `ChannelCore` + the session's +/// dispatch registry: spec + establisher + pump + injected dial, via +/// `register_openable_with_establisher` (`timeout: None` = the 10s +/// default bound). Post-hoc registration is supported (W2): the +/// dispatcher reads through the shared `Arc` per +/// dispatch. `identity_witness` is the optional CF-006 probe seam +/// (the establisher records the per-call opener identity it saw). +pub fn register_tunnel_openable( + core: &ChannelCore, + registry: &ResourceRegistry, + on_registry: &Arc, + auth: AuthContext, + dial: DialFn, + identity_witness: Option, +) -> Result<(), String> { + core.register_openable_with_establisher( + crate::params::tunnel_open_spec(), + Some(tunnel_establisher_with_witness( + registry.clone(), + dial, + identity_witness, + )), + make_tunnel_pump_handler(), + on_registry, + auth, + None, + ) +} diff --git a/tasks/tunnels/producer-open-op.md b/tasks/tunnels/producer-open-op.md index 43693ba..feb3ee5 100644 --- a/tasks/tunnels/producer-open-op.md +++ b/tasks/tunnels/producer-open-op.md @@ -1,7 +1,7 @@ --- id: tunnels/producer-open-op name: Producer half — establisher (dial shape) + pump handler + register_tunnel_openable -status: pending +status: completed depends_on: [tunnels/params, tunnels/wire-codec] scope: broad risk: high @@ -97,16 +97,16 @@ late-registration visibility, pump round-trip through `pump_bidi`. ## Acceptance Criteria -- [ ] Establisher: typed reason mapping exact (ADR-001's table) -- [ ] Pump handler: `pump_bidi` inline; no spawn-and-forget; no +- [x] Establisher: typed reason mapping exact (ADR-001's table) +- [x] Pump handler: `pump_bidi` inline; no spawn-and-forget; no substrate types leak into the handler -- [ ] `register_tunnel_openable` wires spec + establisher + pump + +- [x] `register_tunnel_openable` wires spec + establisher + pump + dial injection -- [ ] Integration tests: forward AND reverse topology (the harness +- [x] Integration tests: forward AND reverse topology (the harness shapes from both POCs), ≥8 tests covering the above -- [ ] Clippy/fmt clean; wasm32 check passes (dial closures are +- [x] Clippy/fmt clean; wasm32 check passes (dial closures are runtime-injected — the crate compiles without `local`) -- [ ] `cargo test` green +- [x] `cargo test` green ## References @@ -118,8 +118,49 @@ late-registration visibility, pump round-trip through `pump_bidi`. ## Notes -> Agent fills during implementation. +- Ported from the reverse POC's producer.rs (the 0.7.0-idiomatic plan + flow, R-01) with the crate generalizations: + - The dial closure is INJECTED (`DialFn`) — the POC had real socket + code inline (tokio TCP/UDP); the crate's establisher is + substrate-agnostic and the `local` feature (or the assembly layer) + supplies the closure (ADR-004's inversion point, tested via + in-process duplex echo + failing dials). + - `register_tunnel_openable` takes `dial` + `identity_witness` (the + CF-006 probe seam) — the POC's registration closure had no dial + injection. + - `ResourceRegistry` keys on `(String, Substrate)` (the enum, not a + &'static str — `Hash` added to `Substrate`). + - `TargetHandle` carries `+ Sync` halves (the F-1 plan-payload + bound). +- Harness (tests/harness.rs): the reverse POC's topology (producer = + `from_connection_with_serving` on the connect side, consumer = + ChannelsAdapter + capturing install hook) with the dial closure + standing in for sockets. The consumer drives the open op via + `call_with_payload` on channel 0 and adopts the producer-allocated + ID. +- 11 integration tests: round-trip through `pump_bidi` (wrapper reaps + the producer channel on pump completion — R-02), unknown_resource + + dial_failed typed errors (no phantom channel either side), FORBIDDEN + identity-less (fails closed), transport-identity open (CF-005 (b)), + token precedence (CF-005 (a) + the CF-006 witness), same-resource + concurrency (no plan race), late registration (W2), unknown + substrate schema rejection (INVALID_INPUT — the registry's + schema-validation runs before the establisher), outbound-calls- + resolve-while-serving (ADR-022 §2), substrate-keyed registry lookups. +- alkcall 0.7.0 note: the establisher's per-call `auth` carries the + dispatch-resolved identity (CF-005 corollary) — the witness proves + the overlay; `INVALID_INPUT` is the wire code for a schema failure + (distinct from `channel:open_failed` establishment failures). +- `futures` crate used for `BoxFuture` in `DialFn` (already a dep). ## Summary -> Agent fills this on completion. \ No newline at end of file +Producer half complete: dial-shape establisher (registry lookup → +injected dial → `Establishment::new(plan)`), the substrate-agnostic +pump handler (`pump_bidi` awaited inline, R-02), and +`register_tunnel_openable` (spec + establisher + pump + dial +injection + witness). 14 unit tests (params + codec) + 11 integration +tests over the duplex harness pass. Verified: cargo test, clippy +--all-targets -D warnings (native + wasm32), fmt --check, wasm32 +check — all clean. The listen establisher variant lands with +`tunnels/producer-listen`. \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs new file mode 100644 index 0000000..0492341 --- /dev/null +++ b/tests/harness.rs @@ -0,0 +1,275 @@ +//! Test harness — the producer↔consumer wiring over a +//! `tokio::io::duplex` carrying the channels 8-byte chunk format. +//! +//! Topology (the reverse POC's shape, mirrored for both directions): +//! - **producer** (connect side of the channels connection — it runs +//! `ChannelClient::from_connection_with_serving`): dials the +//! duplex, serves the tunnel open op + generic channel ops on its +//! serving registry. Its manager allocates the tunnel channel IDs +//! (`ChannelSide::Connect` — ADR-047 §5: the serving side +//! allocates; the initiator adopts). +//! - **consumer** (accept side): the `ChannelsAdapter` + capturing +//! install hook. The hook captures the consumer's channel-0 +//! `CallConnection` (for calling the producer's open op) and the +//! consumer's `ChannelManager` (for adopting the producer- +//! allocated IDs and tearing channels down). The consumer serves +//! no ops. +//! +//! Identity on the serving path follows alkcall 0.7.0's CF-005 +//! precedence: payload `auth_token` → `ServingConfig.identity` → the +//! transport connection's identity (`Connection::set_identity` before +//! dialing). CF-006 (the per-call opener identity on the establisher) +//! is probed via the `identity_witness` the establisher records. + +use std::sync::Arc; + +use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; +use alkcall::channels::client::{ChannelClient, ServingConfig}; +use alkcall::channels::manager::ChannelManager; +use alkcall::channels::operations::ChannelOperations; +use alkcall::core::auth::{AuthContext, AuthToken, Identity, IdentityProvider}; +use alkcall::core::types::Connection as CoreConnection; +use alkcall::protocol::connection::{split_single_stream, CallConnection}; +use alkcall::protocol::dispatch::Dispatcher; +use alkcall::registry::registration::OperationRegistry; + +use alktunnels::params::{Substrate, TUNNEL_OPEN_SCOPE}; +use alktunnels::producer::{register_tunnel_openable, ResourceRegistry}; + +/// The consumer identity the transport carries (the mTLS/QUIC +/// analogue — key-based identity resolved out-of-band, attached to +/// the dialing connection before `from_connection_with_serving`). +pub fn consumer_identity() -> Identity { + Identity { + id: "consumer".to_string(), + scopes: vec![TUNNEL_OPEN_SCOPE.to_string()], + resources: std::collections::HashMap::new(), + } +} + +/// The scoped token (the hub-forwarding path) — resolves to +/// [`consumer_identity`] via [`TestIdProvider`]. +pub const TEST_AUTH_TOKEN: &str = "test-token-with-tunnel-open-scope"; + +/// Resolve the scoped token to the consumer identity; anything else +/// resolves to none (falling through to the transport identity per +/// the 0.7.0 precedence order). +pub struct TestIdProvider; +impl IdentityProvider for TestIdProvider { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &AuthToken) -> Option { + if token.raw == TEST_AUTH_TOKEN.as_bytes() { + Some(consumer_identity()) + } else { + None + } + } +} + +/// The full producer↔consumer topology over one duplex transport. +pub struct Topology { + /// The producer's client (connect side, serving). Its manager + /// allocated the tunnel channels; its pumps are wrapper-managed. + pub producer: Arc, + /// The consumer's manager (accept side). `adopt` installs the + /// producer-allocated IDs here; reaping calls `teardown_channel`. + pub consumer_manager: ChannelManager, + /// The consumer's channel-0 call connection — its surface for + /// calling the producer's open op (and channel/close). + pub consumer_call: Arc, + /// The producer's serving registry (probes). + pub producer_registry: Arc, + /// What the establisher's per-call auth carried as the opener + /// identity id (the CF-006 probe — `None` until an open runs). + pub identity_witness: Arc>>, +} + +/// Wire the topology. `transport_identity` is the key-based identity +/// attached to the dialing connection (CF-005 remediation (b)); +/// `serving_identity` is the explicit `ServingConfig.identity` +/// override (CF-005 remediation (a)) — `None` in the primary path. +/// `provider` resolves payload tokens (the hub-forwarding path). +pub async fn wire_with( + registry: ResourceRegistry, + dial: alktunnels::producer::DialFn, + transport_identity: Option, + serving_identity: Option, + provider: Arc, +) -> Topology { + let producer_op_registry = Arc::new(OperationRegistry::new()); + let (consumer_call_tx, mut consumer_call_rx) = + tokio::sync::mpsc::channel::>(1); + let (consumer_manager_tx, mut consumer_manager_rx) = + tokio::sync::mpsc::channel::(1); + + // --- consumer (accept side): ChannelsAdapter + capturing hook ------ + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { + let call_tx = consumer_call_tx.clone(); + let manager_tx = consumer_manager_tx.clone(); + tokio::spawn(async move { + let channel0_bidi = match channel0_conn.accept_bi().await { + Ok(s) => s, + Err(_) => return, + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + let _ = call_tx.send(Arc::clone(&call_connection)).await; + let _ = manager_tx.send(manager).await; + // The consumer serves no ops — an empty registry. The + // dispatcher keeps the call loop alive (its read loop + // resolves the producer's responses to the consumer's + // pending entries). + let dp = Dispatcher::new( + Arc::new(OperationRegistry::new()), + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + // --- transport ------------------------------------------------------ + let (producer_end, consumer_end) = tokio::io::duplex(64 * 1024); + let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + + let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + let consumer_transport_auth = AuthContext::anonymous(b"alk/channels"); + let _adapter_task = tokio::spawn(async move { + let _ = alkcall::core::types::ProtocolHandler::handle( + &adapter, + consumer_conn, + &consumer_transport_auth, + ) + .await; + }); + + // The transport identity attaches to the dialing connection + // BEFORE from_connection_with_serving (the mTLS/QUIC posture). + if let Some(id) = transport_identity { + producer_conn + .set_identity(id) + .expect("transport identity set once"); + } + + // --- producer (connect side, serving) -------------------------------- + let producer_client = ChannelClient::from_connection_with_serving( + producer_conn, + Some(ServingConfig { + registry: Arc::clone(&producer_op_registry), + identity_provider: provider, + identity: serving_identity, + }), + ) + .await + .expect("producer channel client init"); + + // The generic channel ops ride the producer's serving registry + // (channel/close tears a channel down out-of-band — the review + // 007 recipe). + let ops = ChannelOperations::with_default_policy(producer_client.manager().clone()); + ops.register_on(&producer_op_registry) + .expect("register channel ops on producer"); + + // Post-hoc openable registration on the shared registry (W2). The + // identity witness captures what the establisher's per-call auth + // carried (the CF-006 probe). + let identity_witness = Arc::new(tokio::sync::Mutex::new(None::)); + let core = alkcall::channels::operations::ChannelCore::new( + producer_client.manager().clone(), + alkcall::channels::policy::default_policy(), + ); + register_tunnel_openable( + &core, + ®istry, + &producer_op_registry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + Some(Arc::clone(&identity_witness)), + ) + .expect("register tunnel openable"); + let producer_client = Arc::new(producer_client); + + // --- consumer pieces (captured from the hook) ------------------------- + let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn"); + let consumer_manager = consumer_manager_rx.recv().await.expect("consumer manager"); + + Topology { + producer: producer_client, + consumer_manager, + consumer_call, + producer_registry: producer_op_registry, + identity_witness, + } +} + +/// The default topology: the consumer's transport identity is the +/// primary caller-identity path (no token). The provider is +/// noop-shaped — only the identity chain authorizes. +pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn) -> Topology { + wire_with( + registry, + dial, + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +/// An echo dial closure (the ADR-004 injection point under test): an +/// in-process pipe per dial — the "target" echoes bytes until EOF. +/// The consumer side gets split `DuplexStream` halves; a spawned task +/// echoes the other end. No `local` feature, no sockets. +pub fn echo_dial() -> alktunnels::producer::DialFn { + Arc::new(move |substrate: Substrate, backing: &str| { + let backing = backing.to_string(); + Box::pin(async move { + match substrate { + Substrate::Tcp | Substrate::Unix => { + let (consumer_side, target_side) = tokio::io::duplex(64 * 1024); + let (mut t_read, mut t_write) = tokio::io::split(target_side); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buf = vec![0u8; 8192]; + loop { + match t_read.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if t_write.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + let _ = &backing; + let (c_read, c_write) = tokio::io::split(consumer_side); + Ok(alktunnels::producer::TargetHandle { + read: Box::new(c_read), + write: Box::new(c_write), + }) + } + Substrate::Udp => Err(alktunnels::producer::TunnelEstablishError::DialFailed( + format!("udp dial not wired in this harness: {backing}"), + )), + } + }) + }) +} + +/// A failing dial closure (the `dial_failed` stand-in). +pub fn failing_dial(message: &'static str) -> alktunnels::producer::DialFn { + Arc::new(move |_substrate: Substrate, _backing: &str| { + Box::pin(async move { + Err(alktunnels::producer::TunnelEstablishError::DialFailed( + message.to_string(), + )) + }) + }) +} diff --git a/tests/producer_open_op.rs b/tests/producer_open_op.rs new file mode 100644 index 0000000..2517599 --- /dev/null +++ b/tests/producer_open_op.rs @@ -0,0 +1,372 @@ +//! Integration tests for the producer half — the open-op path +//! (establisher + pump handler + registration) over the duplex +//! harness, ported from the reverse POC's suite (16 tests rode the +//! topology) with the dial closure standing in for real sockets. +//! +//! Covered: establishment success + the typed-error table +//! (`unknown_resource` / `dial_failed` / `FORBIDDEN`), the plan flow +//! under same-resource concurrency (R-01 — no handoff race), the +//! per-call opener identity (CF-006 witness), late-registration +//! visibility (W2), pump round-trip through `pump_bidi` (R-02 — the +//! wrapper-managed pump, no spawn-and-forget), and out-of-band +//! `channel/close`. + +mod harness; + +use std::sync::Arc; + +use alkcall::protocol::wire::CallError; +use alktunnels::params::{Substrate, TUNNEL_OPEN_SCOPE}; +use alktunnels::producer::ResourceRegistry; + +use harness::{echo_dial, failing_dial, wire, wire_with, Topology, TEST_AUTH_TOKEN}; + +fn params(resource: &str) -> alktunnels::params::TunnelParams { + alktunnels::params::TunnelParams { + resource: resource.to_string(), + substrate: Substrate::Tcp, + } +} + +/// Call the producer's open op on channel 0 (the consumer's surface), +/// returning the wire `CallError` on failure. +async fn call_open( + topo: &Topology, + input: serde_json::Value, + auth_token: Option<&str>, +) -> Result { + let mut payload = serde_json::json!({ + "operationId": "channels/tunnel/sub", + "input": input, + }); + if let Some(token) = auth_token { + payload["auth_token"] = serde_json::Value::String(token.to_string()); + } + let response = topo.consumer_call.call_with_payload(payload).await; + response.result +} + +async fn call_open_params( + topo: &Topology, + p: &alktunnels::params::TunnelParams, + auth_token: Option<&str>, +) -> Result { + call_open( + topo, + serde_json::json!({ + "resource": p.resource, + "substrate": match p.substrate { + Substrate::Tcp => "tcp", + Substrate::Udp => "udp", + Substrate::Unix => "unix", + }, + }), + auth_token, + ) + .await +} + +/// Adopt the producer-allocated channel and pump it against in-process +/// halves (the consumer-side data plane — the session's take_halves +/// shape, driven manually until the consumer session task lands). +async fn adopt_and_pump( + topo: &Topology, + channel_id: u32, +) -> ( + tokio::io::ReadHalf, + tokio::io::WriteHalf, + tokio::task::JoinHandle<(u64, u64)>, +) { + let (send, recv) = topo + .consumer_manager + .adopt_channel(channel_id, "alk/tunnel", None) + .await + .expect("adopt producer-allocated channel"); + let bidi = alkcall::core::types::BiStream::from_joined(recv, send); + let (channel_end, local_end) = tokio::io::duplex(64 * 1024); + let (c_read, c_write) = tokio::io::split(channel_end); + let pump = tokio::spawn(alkcall::channels::pump::pump_bidi(bidi, c_read, c_write)); + let (l_read, l_write) = tokio::io::split(local_end); + (l_read, l_write, pump) +} + +/// Only channel 0 (the pre-negotiated call channel) may remain — a +/// failed open leaves no data channel on either side (the +/// phantom-channel property). +fn no_data_channels(topo: &Topology) -> bool { + let consumer = topo.consumer_manager.channel_ids(); + let producer = topo.producer.manager().channel_ids(); + consumer.iter().all(|&id| id == 0) && producer.iter().all(|&id| id == 0) +} + +#[tokio::test] +async fn open_round_trips_through_pump_bidi() { + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let out = call_open_params(&topo, ¶ms("echo"), None) + .await + .expect("open must succeed"); + let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32; + assert!(topo.producer.manager().has_channel(channel_id)); + + let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + write.write_all(b"ping").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; 4]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip timed out") + .expect("read"); + assert_eq!(&buf, b"ping"); + + // EOF propagates: closing the local write half shuts the loop + // down through both pumps (the two-pump contract), and the + // wrapper reaps the producer-side channel on completion (R-02). + drop(write); + drop(read); + let (c2p, p2c) = tokio::time::timeout(std::time::Duration::from_secs(5), pump) + .await + .expect("pump completion timed out") + .expect("pump task"); + assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!( + !topo.producer.manager().has_channel(channel_id), + "wrapper reaped the producer-side channel after pump completion" + ); +} + +#[tokio::test] +async fn unknown_resource_is_typed_and_leaves_no_channel() { + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let err = call_open_params(&topo, ¶ms("ghost"), None) + .await + .expect_err("unknown resource must fail the open"); + assert_eq!(err.code, "channel:open_failed"); + assert_eq!(establishment_reason_of(&err), Some("unknown_resource")); + assert!(no_data_channels(&topo)); +} + +fn establishment_reason_of(err: &CallError) -> Option<&str> { + err.details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()) +} + +#[tokio::test] +async fn dial_failure_is_typed_and_leaves_no_channel() { + let registry = ResourceRegistry::new(); + registry + .register("dead", Substrate::Tcp, "unreachable") + .await; + let topo = wire(registry, failing_dial("no route to target")).await; + + let err = call_open_params(&topo, ¶ms("dead"), None) + .await + .expect_err("failed dial must fail the open"); + assert_eq!(err.code, "channel:open_failed"); + assert_eq!(establishment_reason_of(&err), Some("dial_failed")); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn open_denied_without_any_identity() { + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire_with( + registry, + echo_dial(), + None, + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + let err = call_open_params(&topo, ¶ms("echo"), None) + .await + .expect_err("identity-less open must be denied"); + assert_eq!(err.code, "FORBIDDEN"); + assert!(no_data_channels(&topo)); +} + +#[tokio::test] +async fn open_on_transport_identity_alone() { + // CF-005 (b): the transport identity (set before dialing) + // propagates to the serving dispatch and authorizes the scope + // gate. No token payload at all. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let out = call_open_params(&topo, ¶ms("echo"), None) + .await + .expect("transport-identity open must be authorized"); + assert!(out["channel_id"].is_u64()); + + // CF-006: the establisher's per-call auth carried the END CALLER's + // identity (the consumer), not a synthetic install-time value. + let witness = topo.identity_witness.lock().await.clone(); + assert_eq!( + witness.as_deref(), + Some("consumer"), + "establisher saw the per-call opener identity (CF-006)" + ); +} + +#[tokio::test] +async fn open_token_overrides_transport_identity() { + // CF-005 precedence: the payload token wins. The provider resolves + // the scoped token to the consumer identity — the witness proves + // the establisher saw it end-to-end. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire_with( + registry, + echo_dial(), + None, + None, + Arc::new(harness::TestIdProvider), + ) + .await; + + let out = call_open_params(&topo, ¶ms("echo"), Some(TEST_AUTH_TOKEN)) + .await + .expect("token open authorized"); + assert!(out["channel_id"].is_u64()); + let witness = topo.identity_witness.lock().await.clone(); + assert_eq!(witness.as_deref(), Some("consumer")); +} + +#[tokio::test] +async fn concurrent_opens_same_resource_no_plan_race() { + // The R-01 plan flow under concurrency: two opens of the SAME + // resource — the establisher returns each dialed handle via ITS + // plan; no handoff map to race. Each round-trips its own marker. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let mut sessions = Vec::new(); + for i in 0..2 { + let out = call_open_params(&topo, ¶ms("echo"), None) + .await + .expect("open must succeed"); + let channel_id = out["channel_id"].as_u64().unwrap() as u32; + let (mut read, mut write, _pump) = adopt_and_pump(&topo, channel_id).await; + let msg = format!("marker {i}"); + write.write_all(msg.as_bytes()).await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = vec![0u8; msg.len()]; + tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf)) + .await + .expect("round trip") + .expect("read"); + assert_eq!(buf, msg.as_bytes()); + sessions.push(channel_id); + } + assert_ne!(sessions[0], sessions[1]); +} + +#[tokio::test] +async fn late_registration_is_visible() { + // W2: the harness registers the openable AFTER + // from_connection_with_serving. If the dispatcher cached + // registrations, opens would fail with unknown-operation — the + // successful opens elsewhere prove it; assert registry state too. + let registry = ResourceRegistry::new(); + registry + .register("late", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + assert!(topo + .producer_registry + .registration("channels/tunnel/sub") + .is_some()); +} + +#[tokio::test] +async fn schema_rejects_unknown_substrate_loudly() { + // The registry schema-validates input before the establisher + // (ADR-001's loud posture): an unknown substrate value is a schema + // failure, not a silent pass-through. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let err = call_open( + &topo, + serde_json::json!({"resource": "echo", "substrate": "sctp"}), + None, + ) + .await + .expect_err("unknown substrate must fail the open"); + assert_eq!(err.code, "INVALID_INPUT"); +} + +#[tokio::test] +async fn producer_outbound_calls_still_resolve_while_serving() { + // ADR-022 §2 both-sides sanity: the producer is the connect side + // AND the serving side; its outbound calling still resolves while + // it serves tunnel opens. + let registry = ResourceRegistry::new(); + registry + .register("echo", Substrate::Tcp, "in-process") + .await; + let topo = wire(registry, echo_dial()).await; + + let response = topo + .producer + .call_open_op("consumer/serves/nothing", serde_json::json!({})) + .await; + assert!( + response.result.is_err(), + "consumer serves nothing; the point is the reply RESOLVED" + ); +} + +#[tokio::test] +async fn registry_lookup_respects_substrate_key() { + // The registry is keyed by (resource, substrate): a resource + // produced for TCP is unknown for UDP — the same name on a + // different substrate is a different resource (ADR-001). + let registry = ResourceRegistry::new(); + registry.register("svc", Substrate::Tcp, "in-process").await; + let topo = wire(registry, failing_dial("unreachable")).await; + + let mut p = params("svc"); + p.substrate = Substrate::Udp; + let err = call_open_params(&topo, &p, None) + .await + .expect_err("udp lookup of a tcp resource must miss"); + assert_eq!(establishment_reason_of(&err), Some("unknown_resource")); +} + +/// Type-shape assertions kept for harness parity. +#[allow(dead_code)] +fn type_assertions() { + let _dial = echo_dial(); + let _scope: &str = TUNNEL_OPEN_SCOPE; +}