diff --git a/docs/architecture/decisions/036-channel-0-pre-negotiated-call.md b/docs/architecture/decisions/036-channel-0-pre-negotiated-call.md index cd02893..614630d 100644 --- a/docs/architecture/decisions/036-channel-0-pre-negotiated-call.md +++ b/docs/architecture/decisions/036-channel-0-pre-negotiated-call.md @@ -5,7 +5,9 @@ Accepted (amended 2026-07-18 by ADR-035 — channel 0's `stream_types` field is removed; the channels layer has no `stream_type` concept; the call protocol's `EventEnvelope` framing is the channels payload, carried -transparently — see "Amendment (ADR-035, 2026-07-18)" below) +transparently — see "Amendment (ADR-035, 2026-07-18)" below; further +amended 2026-08-12 — channel 0 runs in single-stream call mode — see +"Amendment (single-stream call mode, 2026-08-12)" below) ## Amendment (ADR-035, 2026-07-18) @@ -24,6 +26,89 @@ The body below describes the **original** (with `stream_types`) shape; the amendment above is the operative decision. See ADR-035 for the resolution rationale and the cross-ADR impacts. +## Amendment (single-stream call mode, 2026-08-12) + +Channel 0 runs in **single-stream call mode**: all `EventEnvelope` +frames (requests, responses, published chunks, aborts) are multiplexed +on channel 0's one `BiStream`. The call protocol's stream-per-request +model (`CallConnection` opens a fresh `open_bi` per call; +`Dispatcher::run_loop` accepts fresh streams in a loop) does not apply +to channel 0 — channel 0's `ChannelBidiStreamSource` is yield-once +(ADR-038), so `open_bi` returns `StreamClosed` and `accept_bi` yields +the one `BiStream` once. Without single-stream mode, every +`call_open_op` on a `ChannelClient` fails with `StreamClosed` on the +connect side, and channel 0's `Connection` is a black hole on the +accept side (the dispatch loop has no stream to accept). + +### What this means concretely + +1. **`CallConnection` gains a single-stream mode.** + `CallConnection::new_single_stream(connection, shared_writer)` + constructs a `CallConnection` that holds a `SharedFrameWriter` (a + `tokio::sync::Mutex>`) instead of opening a + fresh `open_bi` per call. `call_with_payload`, + `subscribe_with_payload`, `publish_with_payload`, and `abort` + branch on `is_single_stream()`: in single-stream mode they write + frames through the shared writer; in stream-per-request mode they + open a fresh stream as before. The shared writer's mutex serializes + frames so concurrent calls do not interleave. + +2. **`Dispatcher` gains `run_loop_single_stream`.** The accept-side + dispatch loop reads `EventEnvelope` frames off channel 0's read + half (a `FrameFramedReader` backed by the reassembled `MpscRecvStream`), + dispatches `call.requested` events, and writes responses through the + same `SharedFrameWriter`. In-flight Pub (Sink) operations are tracked + by `request_id` in a local `HashMap`: + `call.published` / `call.completed` / `call.aborted` frames arriving + after the `call.requested` are routed to the matching sink's + `chunk_tx` while new `call.requested` frames for other requests + continue to be dispatched. Query/Mutation and Sub responses are + written through the shared writer immediately after dispatch. + +3. **`ChannelClient::from_connection` drives the client-side read + pump.** The client spawns a read pump + (`read_single_stream_until_closed`) that reads `EventEnvelope` + frames off channel 0's reassembled read half (fed by the demux) and + routes them into the `PendingRequestMap` via `dispatch_envelope`, + resolving pending calls. This is the single-stream analogue of + `read_stream_until_closed` in stream-per-request mode. + +4. **`ChannelsAdapter::handle`'s `InstallChannelZero` hook now + receives channel 0's `Connection`.** The adapter constructs + channel 0's `Connection` (from the reassembled read half + the mux + write half) and passes it to the hook; the hook runs + `Dispatcher::run_loop_single_stream` on it. This closes C-01's + accept-side black hole: channel 0's `Connection` is actually driven + by a call dispatch loop. + +5. **Top-level `alknet/call` connections are unchanged.** + `CallConnection::new` and `Dispatcher::run_loop` keep the + stream-per-request model. Single-stream mode is only for channel 0 + (and any future single-stream substrate that multiplexes call + frames on one `BiStream`). + +### Door type + +**Two-way (implementation detail).** Single-stream call mode is an +implementation strategy for channel 0's yield-once `BiStream`, not a +wire-format change. The wire format (length-prefixed `EventEnvelope` +frames carried in channel 0's 8-byte chunk header payloads) is +unchanged. A future QUIC-native channels substrate (ADR-039) that +yields multiple bidi streams could use stream-per-request mode on +channel 0 instead; the `CallConnection` and `Dispatcher` branching +supports both. The `SharedFrameWriter` / `run_loop_single_stream` +types are pub(crate) — not part of the public API surface. + +### References + +- C-01 in `docs/reviews/001-pub-and-channels-integration-review.md` + (the channel-0-is-dead-in-both-directions finding this amendment + resolves) +- ADR-038: `ChannelBidiStreamSource` (yield-once `accept_bi` — the + constraint that forces single-stream mode) +- ADR-035: channels pure channel multiplexing (no `stream_type` — the + amendment this builds on) + ## Context A channels connection carries N logical channels. One of them must carry the diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index a3e2c60..6512a68 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -5,9 +5,12 @@ //! route each chunk's payload to the matching `channel_id`'s //! reassembly buffer. //! -//! The `preinstall_channel_0` step (the hook for `channels-call` to -//! install the `CallAdapter` on channel 0) is exposed via a callback. -//! `channels-call` provides the implementation; this adapter calls it. +//! Channel 0 is pre-negotiated as `alknet/call` (ADR-036). The +//! `install_channel_zero` hook (ADR-036 amendment — single-stream call +//! mode) receives channel 0's `Connection` (already constructed by the +//! adapter from the reassembled read half + the mux write half) and +//! runs the call dispatch loop on it. `channels-call` provides the +//! implementation; this adapter calls it. //! //! See `docs/architecture/channels-adapter.md` for the full contract. @@ -29,17 +32,21 @@ use super::wire::CHUNK_HEADER_LEN; pub const CHANNELS_ALPN: &[u8] = b"alknet/channels"; /// The hook for `channels-call` to install the `CallAdapter` on -/// channel 0. The adapter calls this after allocating channel 0's -/// reassembly buffer; `channels-call` wraps it as a `Connection` via -/// `Connection::from_source(ChannelBidiStreamSource, alpn)` and hands -/// it to the `CallAdapter`. +/// channel 0 (ADR-036 amendment — single-stream call mode). The +/// adapter calls this after constructing channel 0's `Connection` +/// (from the reassembled read half + the mux write half); the hook +/// runs the call dispatch loop on it. /// -/// The callback receives the `ChannelManager` (so `channels-call` can -/// construct the `ChannelBidiStreamSource` from the reassembled read -/// half + the mux write half) and the `AuthContext` (so the -/// `CallAdapter` can resolve the peer's identity). -pub type InstallChannelZero = - Arc tokio::task::JoinHandle<()> + Send + Sync>; +/// The callback receives channel 0's `Connection` (carrying the +/// `alknet/call` ALPN) and the `AuthContext` (so the call dispatch +/// loop can resolve the peer's identity). The `ChannelManager` is +/// accessible via the connection's `BidiStreamSource` (the channel-0 +/// source closes over it), so the hook does not need it as a separate +/// argument; the manager is passed for callers that want to reach it +/// (e.g., to register channel lifecycle ops' `ChannelCore`). +pub type InstallChannelZero = Arc< + dyn Fn(ChannelManager, Connection, AuthContext) -> tokio::task::JoinHandle<()> + Send + Sync, +>; /// `ChannelsAdapter` — the `ProtocolHandler` for `alknet/channels`. /// Its `handle()` receives one `Connection`, installs channel 0 via @@ -174,33 +181,34 @@ impl ProtocolHandler for ChannelsAdapter { connection.remote_addr(), ); - // 3. Install channel 0 (pre-negotiated as `alknet/call`, - // ADR-036). The `install_channel_zero` hook (provided by - // `channels-call`) wraps channel 0's reassembly buffer as - // a `Connection` and hands it to the `CallAdapter`. - let (channel0_send, channel0_recv) = manager - .install_channel_zero(None) - .await - .map_err(|e| HandlerError::Internal(format!("channel 0 install failed: {e}").into()))?; - let channel0_source = - super::source::channel_source(channel0_recv, channel0_send, connection.remote_addr()); - let _channel0_conn = Connection::from_source(channel0_source, b"alknet/call".to_vec()); - let _handler_task = (self.install_channel_zero)(&manager, auth); - // The install hook owns the call-adapter task; the manager's - // clear_all on transport EOF will not abort it (it was passed - // as None). The task is tied to the connection lifetime via - // the manager's mux handle — when the mux runner ends, the - // call adapter's write half drops, and the call adapter's - // dispatch loop ends. - - // 4. Spawn the mux runner (write side). + // Spawn the mux runner BEFORE installing channel 0 — + // `install_channel_zero` calls `mux.register(0).await` which + // needs the runner to be draining `new_pumps`. let _mux_task = tokio::spawn(async move { if let Err(e) = mux_runner.run().await { warn!(error = %e, "mux runner ended with error"); } }); - // 5. Run the demux loop (read side). This blocks until + // 3. Install channel 0 (pre-negotiated as `alknet/call`, + // ADR-036). Channel 0's reassembly buffer + mux write half + // are joined into a `BiStream` and wrapped in a + // `Connection` (via `ChannelBidiStreamSource`). The + // `install_channel_zero` hook (provided by `channels-call`) + // runs the call dispatch loop on this `Connection` — the + // single-stream call mode (ADR-036 amendment) multiplexes + // all `EventEnvelope` frames on channel 0's one `BiStream`. + let (channel0_send, channel0_recv) = manager + .install_channel_zero(None) + .await + .map_err(|e| HandlerError::Internal(format!("channel 0 install failed: {e}").into()))?; + let channel0_source = + super::source::channel_source(channel0_recv, channel0_send, connection.remote_addr()); + let channel0_conn = Connection::from_source(channel0_source, b"alknet/call".to_vec()); + let _handler_task = + (self.install_channel_zero)(manager.clone(), channel0_conn, auth.clone()); + + // 4. Run the demux loop (read side). This blocks until // transport EOF, then clears the channel map (REQ-CH-02). Self::run_demux_loop(&manager, Box::new(reader)).await; diff --git a/src/channels/client.rs b/src/channels/client.rs index b771e95..126f23d 100644 --- a/src/channels/client.rs +++ b/src/channels/client.rs @@ -5,6 +5,14 @@ //! `open_channel(alpn, params)` to open data channels via the //! per-ALPN open ops on channel 0. //! +//! Channel 0 runs in single-stream call mode (ADR-036 amendment): +//! all `EventEnvelope` frames for call operations are multiplexed on +//! channel 0's one `BiStream`. `CallConnection::new_single_stream` +//! holds the framed write half; the client spawns a read pump that +//! reads `EventEnvelope` frames off channel 0's reassembled read half +//! (fed by the demux) and routes them into the `PendingRequestMap` +//! via `dispatch_envelope`. +//! //! The dial (TLS, QUIC, WebSocket) lives in the consumer — //! `ChannelClient` is transport-agnostic by construction (ADR-043). //! @@ -41,34 +49,72 @@ impl ChannelClient { /// Construct from an established `Connection` (the consumer dials /// the transport and establishes the `alknet/channels` ALPN). /// Installs channel 0 (pre-negotiated as `alknet/call`, - /// ADR-036), wraps it as a `CallConnection`, spawns the demux and - /// mux tasks, and returns the client. + /// ADR-036), wraps it as a `CallConnection` in single-stream call + /// mode (ADR-036 amendment), spawns the demux and mux tasks, and + /// returns the client. /// - /// The `CallAdapter`'s dispatch loop runs on channel 0; the - /// `ChannelClient` holds the `CallConnection` so the consumer can - /// call `open_channel` (which calls the per-ALPN open ops on - /// channel 0). + /// The call dispatch loop on channel 0 is driven by the client: + /// a read pump task reads `EventEnvelope` frames off channel 0's + /// reassembled read half (fed by the demux) and routes them into + /// the `PendingRequestMap` via `dispatch_envelope`, resolving + /// pending calls. The `CallConnection` holds the framed write + /// half so `call_open_op` can write `call.requested` frames. pub async fn from_connection(connection: Connection) -> Result { + let remote_addr = connection.remote_addr(); let bidi = connection.accept_bi().await?; let (reader, writer) = tokio::io::split(bidi); let (mux_handle, mux_runner) = MuxRunner::new(Box::new(writer)); - let manager = ChannelManager::with_defaults(mux_handle, connection.remote_addr()); + + // Spawn the mux runner BEFORE installing channel 0 — + // `install_channel_zero` calls `mux.register(0).await` which + // sends a `Registration` to the runner and waits for the + // response. The runner must be draining `new_pumps` for the + // registration to complete. + let _mux_task = tokio::spawn(async move { + if let Err(e) = mux_runner.run().await { + tracing::warn!(error = %e, "channel client: mux runner ended with error"); + } + }); + + let manager = ChannelManager::with_defaults(mux_handle, remote_addr); // Install channel 0 — the call adapter's read/write halves. let (channel0_send, channel0_recv) = manager .install_channel_zero(None) .await .map_err(|_| StreamError::StreamClosed)?; - let channel0_source = - super::source::channel_source(channel0_recv, channel0_send, connection.remote_addr()); - let channel0_conn = Connection::from_source(channel0_source, b"alknet/call".to_vec()); - let call_connection = CallConnection::new(channel0_conn); - // Spawn the mux runner (write side). - let _mux_task = tokio::spawn(async move { - if let Err(e) = mux_runner.run().await { - tracing::warn!(error = %e, "channel client: mux runner ended with error"); - } + // Join channel 0's reassembled read half + mux write half into + // a `BiStream` (via `channel_source`), wrap as a `Connection` + // (for ALPN/addr/identity), then yield the `BiStream` once via + // `accept_bi` and split it into a `SharedFrameWriter` (the + // framed write half for `call.requested` etc.) + a raw read + // half (for the read pump). This is the single-stream call + // mode (ADR-036 amendment): all `EventEnvelope` frames are + // multiplexed on channel 0's one `BiStream`. + let channel0_source = + super::source::channel_source(channel0_recv, channel0_send, remote_addr); + let channel0_conn = Connection::from_source(channel0_source, b"alknet/call".to_vec()); + let channel0_bidi = channel0_conn.accept_bi().await?; + let (single_stream_writer, single_stream_reader) = + crate::protocol::connection::split_single_stream(channel0_bidi); + + let call_connection = + CallConnection::new_single_stream(channel0_conn, single_stream_writer); + + // Spawn the read pump for channel 0's response frames. Reads + // `EventEnvelope` frames off channel 0's reassembled read half + // (fed by the demux) and routes them into the + // `PendingRequestMap` via `dispatch_envelope`, resolving + // pending calls. This is the single-stream analogue of + // `read_stream_until_closed` in stream-per-request mode. + let pending_map = Arc::clone(call_connection.pending()); + let _read_pump = tokio::spawn(async move { + crate::protocol::connection::read_single_stream_until_closed( + single_stream_reader, + &pending_map, + ) + .await; }); // Spawn the demux loop (read side). The loop reads 8-byte @@ -98,12 +144,10 @@ impl ChannelClient { /// `channels//pub`) on channel 0. Returns the /// `ResponseEnvelope` (which carries `channel_id` on success). /// - /// The consumer uses this to open a data channel: call - /// `channels/tty/sub` with the ALPN-specific params; the responder - /// allocates the `channel_id` and returns it; the consumer then - /// reads/writes on the channel's `BiStream` (obtained via - /// `manager.open_channel_stream(channel_id)` or the relay - /// machinery). + /// In single-stream call mode (ADR-036 amendment), this writes + /// `call.requested` through channel 0's shared frame writer and + /// awaits the response via the `PendingRequestMap` (resolved by + /// the read pump from the demux). pub async fn call_open_op(&self, operation_id: &str, input: Value) -> ResponseEnvelope { let guard = self.call_connection.lock().await; match guard.as_ref() { @@ -127,11 +171,278 @@ impl ChannelClient { #[cfg(test)] mod tests { use super::*; + use crate::channels::adapter::ChannelsAdapter; + use crate::core::auth::{AuthContext, IdentityProvider}; + use crate::core::types::Connection; + use crate::protocol::connection::split_single_stream; + use crate::protocol::dispatch::Dispatcher; + use crate::registry::registration::{ + make_handler, HandlerKind, HandlerRegistration, OperationProvenance, + }; + use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility}; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + use std::sync::Arc; + + const TEST_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321); + + fn external_query_spec(name: &str) -> OperationSpec { + OperationSpec::new( + name, + OperationType::Query, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ) + } + + struct NoopIdProvider; + impl IdentityProvider for NoopIdProvider { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token( + &self, + _: &crate::core::auth::AuthToken, + ) -> Option { + None + } + } + + /// Build the `install_channel_zero` hook for the accept side: it + /// yields channel 0's `BiStream` once, splits it into the shared + /// writer + reader, constructs a single-stream `CallConnection`, + /// and runs `Dispatcher::run_loop_single_stream` on it. Returns + /// the `JoinHandle` for the spawned dispatch loop. + fn make_install_channel_zero( + registry: Arc, + ) -> crate::channels::adapter::InstallChannelZero { + Arc::new(move |_manager, channel0_conn, _auth| { + let registry = Arc::clone(®istry); + let provider: Arc = Arc::new(NoopIdProvider); + 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 dp = Dispatcher::new(registry, provider); + dp.run_loop_single_stream(call_connection, reader, writer) + .await; + }) + }) + } #[test] fn channel_client_is_not_send_safe_across_await_by_default() { - // Smoke test: ChannelClient compiles. The actual send/sync - // bounds are exercised by the integration tests. let _ = std::marker::PhantomData::; } + + /// C-25 #1 — the end-to-end acceptance gate for Unit 2 (channel 0 + /// single-stream call mode, ADR-036 amendment). Wires + /// `ChannelClient` (connect side) ↔ `ChannelsAdapter` (accept + /// side) over a real `tokio::io::duplex` pair carrying the + /// channels 8-byte chunk header wire format. A `call.requested` + /// from the client is demuxed on channel 0, dispatched by the + /// server's single-stream `Dispatcher::run_loop_single_stream`, + /// and the `call.responded` is muxed back to the client's read + /// pump, resolving the pending call. This would have caught C-01 + /// immediately (every `call_open_op` failed with + /// `StreamClosed` on the connect side; channel 0 was a black hole + /// on the accept side). + #[tokio::test] + async fn channel_0_end_to_end_call_round_trip() { + let mut registry = crate::registry::registration::OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + external_query_spec("echo/run"), + HandlerKind::Once(make_handler(|input, ctx| async move { + ResponseEnvelope::ok(ctx.request_id, input) + })), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + )) + .unwrap(); + let registry = Arc::new(registry); + + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = + Connection::from_bidi(client_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + let server_conn = + Connection::from_bidi(server_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + + let adapter = ChannelsAdapter::new(make_install_channel_zero(Arc::clone(®istry))); + let auth = AuthContext::anonymous(b"alknet/channels"); + let _server_handle = tokio::spawn(async move { + let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await; + }); + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("channel client init"); + + let response = tokio::time::timeout( + std::time::Duration::from_secs(10), + client.call_open_op("echo/run", serde_json::json!({ "msg": "hi" })), + ) + .await + .expect("channel-0 round-trip timed out"); + + assert!( + response.result.is_ok(), + "channel-0 round-trip should succeed, got {:?}", + response.result + ); + assert_eq!(response.result.unwrap(), serde_json::json!({ "msg": "hi" })); + } + + /// C-25 #1 (negative case) — an unknown op over channel 0 returns + /// `NOT_FOUND`, proving the single-stream dispatch path reaches + /// the registry's not-found branch end-to-end (not just + /// `StreamClosed` from a dead channel 0 as in C-01). + #[tokio::test] + async fn channel_0_end_to_end_unknown_op_returns_not_found() { + let registry = Arc::new(crate::registry::registration::OperationRegistry::new()); + + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = + Connection::from_bidi(client_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + let server_conn = + Connection::from_bidi(server_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + + let adapter = ChannelsAdapter::new(make_install_channel_zero(Arc::clone(®istry))); + let auth = AuthContext::anonymous(b"alknet/channels"); + let _server_handle = tokio::spawn(async move { + let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await; + }); + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("channel client init"); + + let response = tokio::time::timeout( + std::time::Duration::from_secs(10), + client.call_open_op("no/such/op", serde_json::json!({})), + ) + .await + .expect("channel-0 unknown-op timed out"); + + match response.result { + Err(e) => assert_eq!(e.code, "NOT_FOUND", "unknown op over channel 0"), + other => panic!("expected NOT_FOUND, got {other:?}"), + } + } + + /// C-25 #1 (Pub over channel 0) — a `publish()` through channel + /// 0's single-stream mode delivers chunks to the responder's + /// `SinkHandler` and returns the handler's response. Proves the + /// single-stream dispatch loop's in-flight sink routing works + /// (`call.published` frames arriving after `call.requested` are + /// routed to the matching sink's `chunk_tx`). + #[tokio::test] + async fn channel_0_end_to_end_publish_delivers_chunks() { + use futures::stream::StreamExt; + + let mut registry = crate::registry::registration::OperationRegistry::new(); + let counting_sink = crate::registry::registration::make_sink_handler( + |_input, ctx, mut stream| async move { + let mut count = 0u32; + let mut last = serde_json::Value::Null; + while let Some(item) = stream.next().await { + match item { + Ok(v) => { + count += 1; + last = v; + } + Err(_) => break, + } + } + ResponseEnvelope::ok( + ctx.request_id, + serde_json::json!({ "count": count, "last": last }), + ) + }, + ); + registry + .register(HandlerRegistration::new( + OperationSpec::new( + "fs/upload", + OperationType::Pub, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ), + HandlerKind::Sink(counting_sink), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + )) + .unwrap(); + let registry = Arc::new(registry); + + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = + Connection::from_bidi(client_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + let server_conn = + Connection::from_bidi(server_end, b"alknet/channels".to_vec(), Some(TEST_ADDR)); + + let adapter = ChannelsAdapter::new(make_install_channel_zero(Arc::clone(®istry))); + let auth = AuthContext::anonymous(b"alknet/channels"); + let _server_handle = tokio::spawn(async move { + let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await; + }); + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("channel client init"); + + let call_conn = client + .take_call_connection() + .await + .expect("call connection present"); + + let chunks = vec![ + serde_json::json!({"chunk": 1}), + serde_json::json!({"chunk": 2}), + serde_json::json!({"chunk": 3}), + ]; + let stream: std::pin::Pin< + Box + Send>, + > = Box::pin(futures::stream::iter(chunks.clone())); + let response = tokio::time::timeout( + std::time::Duration::from_secs(10), + call_conn.publish("fs/upload", serde_json::json!({"path": "/x"}), stream), + ) + .await + .expect("channel-0 publish timed out"); + + assert!( + response.result.is_ok(), + "publish should succeed, got {:?}", + response.result + ); + let out = response.result.unwrap(); + assert_eq!( + out["count"], + serde_json::json!(3), + "responder saw all 3 chunks" + ); + assert_eq!( + out["last"], + serde_json::json!({"chunk": 3}), + "last chunk matches" + ); + } } diff --git a/src/protocol/connection.rs b/src/protocol/connection.rs index 2c03243..9743721 100644 --- a/src/protocol/connection.rs +++ b/src/protocol/connection.rs @@ -20,8 +20,8 @@ use tokio::sync::mpsc; use super::pending::PendingRequestMap; use super::wire::{ - CallError, EventEnvelope, FrameFramedReader, FrameFramedWriter, EVENT_ABORTED, EVENT_COMPLETED, - EVENT_ERROR, EVENT_RESPONDED, + CallError, EventEnvelope, FrameError, FrameFramedReader, FrameFramedWriter, EVENT_ABORTED, + EVENT_COMPLETED, EVENT_ERROR, EVENT_RESPONDED, }; use crate::protocol::wire::ResponseEnvelope; use crate::registry::context::{generate_request_id, AbortPolicy, OperationContext, ScopedPeerEnv}; @@ -31,11 +31,40 @@ use crate::registry::spec::AccessResult; const DEFAULT_CALL_TIMEOUT: Duration = Duration::from_secs(30); +/// A shared, thread-safe `EventEnvelope` frame writer for single-stream +/// call mode (ADR-036 amendment — channel 0 single-stream mode). Wraps a +/// write half in a `tokio::sync::Mutex` so multiple concurrent calls +/// (`call_with_payload`, `publish_with_payload`, `abort`) can serialize +/// frames on the one `BiStream` without interleaving. +/// +/// Each `write_frame` call is atomic under the mutex — the 4-byte length +/// prefix, the JSON body, and the `flush` all happen while the lock is +/// held, so a frame from one call is never partially written when +/// another call's frame begins. This is the single-stream analogue of +/// the stream-per-request model's "one `open_bi` per call" (which +/// naturally serializes per-call writes). +pub struct SharedFrameWriter { + inner: tokio::sync::Mutex>>, +} + +impl SharedFrameWriter { + pub fn new(writer: Box) -> Self { + Self { + inner: tokio::sync::Mutex::new(FrameFramedWriter::new(writer)), + } + } + + pub async fn write_frame(&self, envelope: &EventEnvelope) -> Result<(), FrameError> { + self.inner.lock().await.write_frame(envelope).await + } +} + pub struct CallConnection { connection: Option>, stored_identity: Option, imported_operations: Arc>>, pending: Arc>, + single_stream_writer: Option>, } impl Clone for CallConnection { @@ -45,6 +74,7 @@ impl Clone for CallConnection { stored_identity: self.stored_identity.clone(), imported_operations: Arc::clone(&self.imported_operations), pending: Arc::clone(&self.pending), + single_stream_writer: self.single_stream_writer.clone(), } } } @@ -56,6 +86,7 @@ impl CallConnection { stored_identity: None, imported_operations: Arc::new(RwLock::new(HashMap::new())), pending: Arc::new(Mutex::new(PendingRequestMap::new())), + single_stream_writer: None, } } @@ -65,9 +96,54 @@ impl CallConnection { stored_identity: Some(identity), imported_operations: Arc::new(RwLock::new(HashMap::new())), pending: Arc::new(Mutex::new(PendingRequestMap::new())), + single_stream_writer: None, } } + /// Construct a `CallConnection` in **single-stream call mode** + /// (ADR-036 amendment — channel 0 single-stream mode). Used by + /// `ChannelClient` (the connect side of a channels connection): + /// channel 0's `BiStream` is the one stream all `EventEnvelope`s + /// are multiplexed on — `open_bi` is never called (which would + /// return `StreamClosed` on channel 0's yield-once + /// `ChannelBidiStreamSource`, ADR-038). + /// + /// The caller splits channel 0's `BiStream` and hands the write + /// half here; the read half is pumped separately (the channels + /// layer's demux feeds it — channel 0's reassembly buffer is the + /// read side — so the client does NOT spawn its own read pump on + /// the single stream; the demux loop routes channel-0 chunks into + /// the `PendingRequestMap` via `dispatch_envelope`). + /// + /// `connection` is the channels-layer `Connection` (carrying + /// ALPN/addr/identity); `single_stream_writer` is channel 0's + /// framed write half. + pub fn new_single_stream( + connection: Connection, + single_stream_writer: Arc, + ) -> Self { + Self { + connection: Some(Arc::new(connection)), + stored_identity: None, + imported_operations: Arc::new(RwLock::new(HashMap::new())), + pending: Arc::new(Mutex::new(PendingRequestMap::new())), + single_stream_writer: Some(single_stream_writer), + } + } + + /// `true` if this `CallConnection` is in single-stream call mode + /// (channel 0 multiplexed framing, no `open_bi` per call). The + /// dispatch loop and the client methods branch on this. + pub fn is_single_stream(&self) -> bool { + self.single_stream_writer.is_some() + } + + /// The shared frame writer for single-stream mode. `None` in + /// stream-per-request mode. + pub fn single_stream_writer(&self) -> Option<&Arc> { + self.single_stream_writer.as_ref() + } + pub fn connection(&self) -> Option<&Arc> { self.connection.as_ref() } @@ -113,9 +189,20 @@ impl CallConnection { /// The payload MUST include `operationId` and `input`; the caller may add /// `forwarded_for` (ADR-032) and `auth_token` (ADR-017 §7) for the hub /// forwarding path used by `from_call`. + /// + /// In stream-per-request mode (top-level `alknet/call`), opens a fresh + /// bidi stream per call and pumps the read half in a spawned task. In + /// single-stream mode (channel 0, ADR-036 amendment), writes + /// `call.requested` through the shared frame writer and relies on the + /// channels-layer demux + `dispatch_envelope` to resolve the pending + /// entry from the response frame. pub async fn call_with_payload(&self, payload: Value) -> ResponseEnvelope { let request_id = generate_request_id(); + if let Some(writer) = self.single_stream_writer.clone() { + return self.call_single_stream(writer, request_id, payload).await; + } + let connection = match &self.connection { Some(c) => c, None => { @@ -167,6 +254,42 @@ impl CallConnection { } } + /// Single-stream call path (ADR-036 amendment). Writes `call.requested` + /// through the shared frame writer (no `open_bi`); the channels-layer + /// demux routes the response frame's `call.responded`/`call.error`/ + /// `call.completed` into the `PendingRequestMap` via + /// `dispatch_envelope`, resolving the `receiver`. + async fn call_single_stream( + &self, + writer: Arc, + request_id: String, + payload: Value, + ) -> ResponseEnvelope { + let receiver = { + let mut pending = self.pending.lock(); + pending.register_call( + request_id.clone(), + Some(Instant::now() + DEFAULT_CALL_TIMEOUT), + None, + ) + }; + + let envelope = EventEnvelope::requested(&request_id, payload); + if let Err(err) = writer.write_frame(&envelope).await { + let call_error = CallError::internal(format!("failed to write request frame: {err}")); + self.pending + .lock() + .handle_error(&request_id, call_error.clone()); + return ResponseEnvelope::error(request_id, call_error); + } + + match receiver.await { + Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value), + Ok(Err(error)) => ResponseEnvelope::error(request_id, error), + Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")), + } + } + pub async fn subscribe( &self, operation_id: &str, @@ -187,12 +310,24 @@ impl CallConnection { /// forwarding handler can populate `forwarded_for` + `auth_token` on the /// subscription payload (the plain [`subscribe`](Self::subscribe) builds /// the payload internally and omits those fields). + /// + /// In single-stream mode (channel 0, ADR-036 amendment), writes + /// `call.requested` through the shared frame writer; the + /// channels-layer demux routes `call.responded`/`call.error`/ + /// `call.completed` into the `PendingRequestMap`'s subscription + /// channel via `dispatch_envelope`. pub async fn subscribe_with_payload( &self, payload: Value, ) -> impl Stream { let request_id = generate_request_id(); + if let Some(writer) = self.single_stream_writer.clone() { + return self + .subscribe_single_stream(writer, request_id, payload) + .await; + } + let connection = match &self.connection { Some(c) => c, None => { @@ -233,8 +368,45 @@ impl CallConnection { SubscriptionStream::new(request_id, receiver) } + /// Single-stream subscribe path (ADR-036 amendment). Writes + /// `call.requested` through the shared frame writer; the + /// channels-layer demux routes events into the subscription + /// channel via `dispatch_envelope`. + async fn subscribe_single_stream( + &self, + writer: Arc, + request_id: String, + payload: Value, + ) -> SubscriptionStream { + let receiver = { + let mut pending = self.pending.lock(); + pending.register_subscribe(request_id.clone(), None, None) + }; + + let envelope = EventEnvelope::requested(&request_id, payload); + if let Err(err) = writer.write_frame(&envelope).await { + let call_error = CallError::internal(format!("failed to write request frame: {err}")); + self.pending + .lock() + .handle_error(&request_id, call_error.clone()); + return SubscriptionStream::closed(request_id, call_error); + } + + SubscriptionStream::new(request_id, receiver) + } + pub async fn abort(&self, request_id: &str) { let envelope = EventEnvelope::aborted(request_id); + + if let Some(writer) = self.single_stream_writer.as_ref() { + if let Err(err) = writer.write_frame(&envelope).await { + tracing::warn!(error = %err, request_id, "failed to send call.aborted"); + return; + } + self.pending.lock().handle_aborted(request_id); + return; + } + if let Err(err) = self.write_envelope(&envelope).await { tracing::warn!(error = %err, request_id, "failed to send call.aborted"); return; @@ -270,12 +442,17 @@ impl CallConnection { /// sink forwarding handler. Mirrors /// [`subscribe_with_payload`](Self::subscribe_with_payload). /// - /// All frames (`call.requested`, `call.published`, `call.completed`) - /// are written on the **same** bidirectional stream's write half — - /// the responder reads chunks from the same stream it received - /// `call.requested` on (ADR-046 §5 stream lifecycle). The read half - /// is pumped concurrently for the single `call.responded` / - /// `call.error` response. + /// In stream-per-request mode (top-level `alknet/call`), opens a + /// fresh bidi stream, writes `call.requested` + chunks + + /// `call.completed` on the same write half, and pumps the read half + /// concurrently for the single `call.responded` / `call.error` + /// response (ADR-046 §5 stream lifecycle). + /// + /// In single-stream mode (channel 0, ADR-036 amendment), writes + /// `call.requested` + chunks + `call.completed` through the shared + /// frame writer; the channels-layer demux routes the responder's + /// `call.responded` / `call.error` into the `PendingRequestMap` + /// via `dispatch_envelope`. pub async fn publish_with_payload( &self, payload: Value, @@ -283,6 +460,12 @@ impl CallConnection { ) -> ResponseEnvelope { let request_id = generate_request_id(); + if let Some(writer) = self.single_stream_writer.clone() { + return self + .publish_single_stream(writer, request_id, payload, stream) + .await; + } + let connection = match &self.connection { Some(c) => c, None => { @@ -331,6 +514,58 @@ impl CallConnection { } } + /// Single-stream publish path (ADR-036 amendment). Pumps + /// `call.requested` + `call.published` chunks + `call.completed` + /// through the shared frame writer; the channels-layer demux routes + /// the responder's `call.responded` / `call.error` into the + /// `PendingRequestMap` via `dispatch_envelope`. + async fn publish_single_stream( + &self, + writer: Arc, + request_id: String, + payload: Value, + mut stream: Pin + Send>>, + ) -> ResponseEnvelope { + let receiver = { + let mut pending = self.pending.lock(); + pending.register_call(request_id.clone(), None, None) + }; + + let requested = EventEnvelope::requested(&request_id, payload); + if let Err(err) = writer.write_frame(&requested).await { + let call_error = CallError::internal(format!("failed to write request frame: {err}")); + self.pending + .lock() + .handle_error(&request_id, call_error.clone()); + return ResponseEnvelope::error(request_id, call_error); + } + while let Some(chunk) = stream.next().await { + let envelope = EventEnvelope::published(&request_id, chunk); + if let Err(err) = writer.write_frame(&envelope).await { + let call_error = + CallError::internal(format!("failed to write published frame: {err}")); + self.pending + .lock() + .handle_error(&request_id, call_error.clone()); + return ResponseEnvelope::error(request_id, call_error); + } + } + let completed = EventEnvelope::completed(&request_id); + if let Err(err) = writer.write_frame(&completed).await { + let call_error = CallError::internal(format!("failed to write completed frame: {err}")); + self.pending + .lock() + .handle_error(&request_id, call_error.clone()); + return ResponseEnvelope::error(request_id, call_error); + } + + match receiver.await { + Ok(Ok(value)) => ResponseEnvelope::ok(request_id, value), + Ok(Err(error)) => ResponseEnvelope::error(request_id, error), + Err(_) => ResponseEnvelope::error(request_id, CallError::internal("request cancelled")), + } + } + async fn write_request( &self, send: W, @@ -408,6 +643,38 @@ where } } +/// Read `EventEnvelope` frames off a boxed read half and route them +/// into the `PendingRequestMap` via `dispatch_envelope` until the read +/// half closes. The single-stream analogue of `read_stream_until_closed` +/// (ADR-036 amendment — channel 0 single-stream mode). Used by +/// `ChannelClient::from_connection` to drive the client-side response +/// pump: the channels-layer demux feeds channel-0 chunks into the +/// reassembled read half, this pump reads `EventEnvelope` frames off +/// it, and `dispatch_envelope` resolves pending calls. +pub async fn read_single_stream_until_closed( + reader: Box, + pending: &Arc>, +) { + read_stream_until_closed(reader, pending).await; +} + +/// Split a `BiStream` into a `SharedFrameWriter` (the framed write +/// half, wrapped in a `tokio::sync::Mutex` for concurrent frame writes) +/// and the raw read half (for the single-stream dispatch loop to wrap +/// in a `FrameFramedReader`). Used by single-stream call mode +/// (ADR-036 amendment) on both the connect side (`ChannelClient`) and +/// the accept side (`ChannelsAdapter`'s `install_channel_zero` hook): +/// channel 0's `BiStream` is the one stream all `EventEnvelope`s are +/// multiplexed on. +pub fn split_single_stream( + stream: crate::core::types::BiStream, +) -> (Arc, Box) { + let (recv, send) = tokio::io::split(stream); + let writer = Arc::new(SharedFrameWriter::new(Box::new(send))); + let reader: Box = Box::new(recv); + (writer, reader) +} + fn dispatch_envelope(pending: &Arc>, envelope: EventEnvelope) { let request_id = envelope.id.clone(); match envelope.r#type.as_str() { diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index 59326d3..ca5bf52 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -30,7 +30,7 @@ use super::abort::AbortCascade; use super::connection::CallConnection; use super::wire::{ CallError, EventEnvelope, FrameFramedReader, FrameFramedWriter, ResponseEnvelope, - EVENT_ABORTED, EVENT_REQUESTED, + EVENT_ABORTED, EVENT_COMPLETED, EVENT_PUBLISHED, EVENT_REQUESTED, }; use crate::protocol::adapter::SessionOverlaySource; use crate::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv}; @@ -579,6 +579,192 @@ impl Dispatcher { sweeper_handle.abort(); } + + /// Run the single-stream dispatch loop (ADR-036 amendment — channel + /// 0 single-stream mode). Reads `EventEnvelope` frames off `reader` + /// (channel 0's read half) and dispatches them, writing responses + /// through `writer` (the shared frame writer). All frames — + /// requests, responses, published chunks, aborts — are multiplexed + /// on the one `BiStream` (channel 0). + /// + /// Unlike `run_loop` (stream-per-request), there is one read loop + /// for the whole connection. In-flight Pub (Sink) operations are + /// tracked by `request_id` in `in_flight_sinks`: their + /// `call.published` / `call.completed` / `call.aborted` frames + /// arrive *after* the `call.requested` and are routed to the + /// matching sink's `chunk_tx` while new `call.requested` frames for + /// other requests continue to be dispatched. Query/Mutation and Sub + /// responses are written through `writer` immediately after + /// dispatch. + /// + /// Returns when the read half closes (transport EOF). Outstanding + /// pending requests are failed with `connection closed`, and + /// in-flight sinks' `chunk_tx` are dropped (the handler's + /// `PublishStream` sees EOF). + pub async fn run_loop_single_stream( + self, + connection: Arc, + reader: Box, + writer: Arc, + ) { + let pending = Arc::clone(connection.pending()); + + let sweeper_pending = Arc::clone(&pending); + let sweeper_handle: JoinHandle<()> = tokio::spawn(async move { + let mut interval = tokio::time::interval(SWEEPER_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + let evicted = sweeper_pending.lock().evict_expired(); + if !evicted.is_empty() { + debug!( + count = evicted.len(), + "sweeper evicted expired pending entries" + ); + } + } + }); + + let mut reader = FrameFramedReader::new(reader); + let mut in_flight_sinks: HashMap>> = + HashMap::new(); + + loop { + let envelope = match reader.read_frame().await { + Ok(env) => env, + Err(super::wire::FrameError::ConnectionClosed) => break, + Err(err) => { + warn!(error = %err, "single-stream frame read error; closing loop"); + break; + } + }; + + match envelope.r#type.as_str() { + EVENT_REQUESTED => { + let request_id = envelope.id.clone(); + let payload = envelope.payload.clone(); + let dispatch_result = self + .dispatch(&connection, request_id.clone(), payload) + .await; + + match dispatch_result { + DispatchResult::Once(response) => { + let event: EventEnvelope = response.into(); + if let Err(err) = writer.write_frame(&event).await { + warn!( + error = %err, + "single-stream: failed to write Once response; closing loop" + ); + break; + } + } + DispatchResult::Stream(stream) => { + self.pump_stream_single_stream(&writer, &request_id, stream) + .await; + } + DispatchResult::Sink(sink) => { + let SinkDispatch { handler, chunk_tx } = sink; + in_flight_sinks.insert(request_id.clone(), chunk_tx); + let writer_clone = Arc::clone(&writer); + let request_id_for_handler = request_id.clone(); + tokio::spawn(async move { + let response = handler.await; + let event: EventEnvelope = response.into(); + if let Err(err) = writer_clone.write_frame(&event).await { + warn!( + error = %err, + request_id = %request_id_for_handler, + "single-stream: failed to write sink response frame" + ); + } + }); + } + } + } + EVENT_ABORTED => { + let request_id = envelope.id.clone(); + if let Some(mut chunk_tx) = in_flight_sinks.remove(&request_id) { + let _ = chunk_tx + .send(Err(CallError::internal("publish aborted by initiator"))) + .await; + } else { + self.handle_abort(&connection, &request_id).await; + } + } + EVENT_PUBLISHED => { + let request_id = envelope.id.clone(); + let chunk = envelope + .payload + .get("input") + .cloned() + .unwrap_or(Value::Null); + if let Some(mut chunk_tx) = in_flight_sinks.remove(&request_id) { + let _ = chunk_tx.send(Ok(chunk)).await; + in_flight_sinks.insert(request_id, chunk_tx); + } else { + debug!( + request_id = %request_id, + "single-stream: call.published for unknown in-flight sink; dropping" + ); + } + } + EVENT_COMPLETED => { + let request_id = envelope.id.clone(); + in_flight_sinks.remove(&request_id); + } + other => { + debug!( + event_type = %other, + id = %envelope.id, + "single-stream: ignoring non-requested/non-published/non-aborted/non-completed event" + ); + } + } + } + + in_flight_sinks.clear(); + + let failed = pending + .lock() + .fail_all(CallError::internal("connection closed")); + if !failed.is_empty() { + debug!( + count = failed.len(), + "single-stream: failed pending requests on connection close" + ); + } + + sweeper_handle.abort(); + } + + /// Pump a subscription's `ResponseStream` to the wire through the + /// shared single-stream writer (ADR-036 amendment). Each + /// `ResponseEnvelope` becomes a `call.responded` / `call.error` + /// frame; on natural end, a `call.completed` frame. The shared + /// writer serializes frames so this pump's frames do not interleave + /// with concurrent calls' frames. + async fn pump_stream_single_stream( + &self, + writer: &Arc, + request_id: &str, + mut stream: ResponseStream, + ) { + let mut last_was_error = false; + while let Some(envelope) = stream.next().await { + last_was_error = envelope.result.is_err(); + let event: EventEnvelope = envelope.into(); + if let Err(err) = writer.write_frame(&event).await { + warn!(error = %err, "single-stream: failed to write streaming frame"); + return; + } + } + if !last_was_error { + let completed = EventEnvelope::completed(request_id); + if let Err(err) = writer.write_frame(&completed).await { + warn!(error = %err, "single-stream: failed to write call.completed"); + } + } + } } impl Clone for Dispatcher {