alkdev/alkcall
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Unit 2 — channel 0 single-stream call mode (C-01, C-25 #1)

Browse Source 
Channel 0 was dead in both directions (C-01): the call protocol's
stream-per-request model (open_bi per call) is incompatible with
channel 0's single yield-once BiStream — every call_open_op failed
with StreamClosed on the connect side, and channel 0's Connection was
a black hole on the accept side.

The fix is single-stream call mode (ADR-036 amendment): all
EventEnvelope frames are multiplexed on channel 0's one BiStream.

Changes:
- CallConnection gains single_stream_writer: Option<Arc<SharedFrameWriter>>
  and new_single_stream() constructor. 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 (mutex-serialized) instead of opening a fresh
  open_bi per call.
- Dispatcher::run_loop_single_stream reads frames off channel 0's read
  half, dispatches call.requested, writes responses through the shared
  writer, and routes in-flight call.published/call.completed/
  call.aborted to the matching Pub sink's chunk_tx by request_id.
- ChannelsAdapter::handle's InstallChannelZero hook now receives
  channel 0's Connection (built by the adapter) and runs the
  single-stream dispatch loop on it — closing the accept-side black
  hole. Mux runner is spawned BEFORE install_channel_zero so
  mux.register(0) can complete.
- ChannelClient::from_connection uses CallConnection::new_single_stream
  and spawns a read pump (read_single_stream_until_closed) that routes
  channel-0 response frames into the PendingRequestMap via
  dispatch_envelope — closing the connect-side StreamClosed path.
- ADR-036 amendment records the single-stream-mode decision (two-way
  door: implementation detail, wire format unchanged).

Acceptance gate (C-25 #1): three end-to-end tests wire
ChannelClient ↔ ChannelsAdapter over a real tokio::io::duplex pair
carrying the channels 8-byte chunk header wire format:
- channel_0_end_to_end_call_round_trip: a Query op round-trip
- channel_0_end_to_end_unknown_op_returns_not_found: NOT_FOUND
- channel_0_end_to_end_publish_delivers_chunks: a Pub op with 3 chunks

Verification: 437 tests pass (was 434; +3 e2e), clippy clean, fmt
clean, doc warnings unchanged (4, pre-existing C-09).
This commit is contained in:
glm-5.2
2026-08-12 21:55:04 +00:00
parent 502488cc72
commit 495e04ed43
5 changed files with 925 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
docs/architecture/decisions/036-channel-0-pre-negotiated-call.md
View File

@@ -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<FrameFramedWriter<W>>`) 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<String, mpsc::Sender>`:
`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

76
src/channels/adapter.rs
View File

@@ -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<dyn Fn(&ChannelManager, &AuthContext) -> 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;

359
src/channels/client.rs
View File

@@ -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<Self, StreamError> {
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/<alpn>/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<crate::core::auth::Identity> {
None
}
fn resolve_from_token(
&self,
_: &crate::core::auth::AuthToken,
) -> Option<crate::core::auth::Identity> {
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::registry::registration::OperationRegistry>,
) -> crate::channels::adapter::InstallChannelZero {
Arc::new(move |_manager, channel0_conn, _auth| {
let registry = Arc::clone(&registry);
let provider: Arc<dyn IdentityProvider> = 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::<ChannelClient>;
}
/// 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(&registry)));
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(&registry)));
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(&registry)));
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<dyn futures::stream::Stream<Item = serde_json::Value> + 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"
);
}
}

283
src/protocol/connection.rs
View File

@@ -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<FrameFramedWriter<Box<dyn AsyncWrite + Send + Unpin>>>,
}
impl SharedFrameWriter {
pub fn new(writer: Box<dyn AsyncWrite + Send + Unpin>) -> 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<Arc<Connection>>,
stored_identity: Option<Identity>,
imported_operations: Arc<RwLock<HashMap<String, HandlerRegistration>>>,
pending: Arc<Mutex<PendingRequestMap>>,
single_stream_writer: Option<Arc<SharedFrameWriter>>,
}
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<SharedFrameWriter>,
) -> 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<SharedFrameWriter>> {
self.single_stream_writer.as_ref()
}
pub fn connection(&self) -> Option<&Arc<Connection>> {
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<SharedFrameWriter>,
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<Item = ResponseEnvelope> {
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<SharedFrameWriter>,
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<SharedFrameWriter>,
request_id: String,
payload: Value,
mut stream: Pin<Box<dyn Stream<Item = Value> + 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<W>(
&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<dyn AsyncRead + Send + Unpin>,
pending: &Arc<Mutex<PendingRequestMap>>,
) {
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<SharedFrameWriter>, Box<dyn AsyncRead + Send + Unpin>) {
let (recv, send) = tokio::io::split(stream);
let writer = Arc::new(SharedFrameWriter::new(Box::new(send)));
let reader: Box<dyn AsyncRead + Send + Unpin> = Box::new(recv);
(writer, reader)
}
fn dispatch_envelope(pending: &Arc<Mutex<PendingRequestMap>>, envelope: EventEnvelope) {
let request_id = envelope.id.clone();
match envelope.r#type.as_str() {

188
src/protocol/dispatch.rs
View File

@@ -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<CallConnection>,
reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
writer: Arc<super::connection::SharedFrameWriter>,
) {
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<String, mpsc::Sender<Result<Value, CallError>>> =
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<super::connection::SharedFrameWriter>,
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 {