diff --git a/docs/architecture/README.md b/docs/architecture/README.md index bde5b86..863a1e8 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -98,7 +98,7 @@ are wire-stable and unchanged — see ADR-004. | [043](decisions/043-channelclient.md) | ChannelClient | Transport-agnostic from_connection | | [044](decisions/044-channels-subcrate-decomposition.md) | Channels Sub-Crate Decomposition | channels-core / channels-call (modules in alkcall) | | [045](decisions/045-alknetclient-native-dial-seam.md) | AlknetClient Dial Seam | spawn_dispatch / from_connection take-over; dial in consumer | -| [046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | Publish Operation Type and HandlerKind::Sink | `OperationType::Pub` (client→server streaming); `SinkHandler` + `HandlerKind::Sink`; `call.published` wire event; `invoke_sink()` dispatch; `Subscription` renamed to `Sub` | +| [046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | Publish Operation Type and HandlerKind::Sink | `OperationType::Pub` (producer→consumer streaming); `SinkHandler` + `HandlerKind::Sink`; `call.published` wire event; `invoke_sink()` dispatch; `Subscription` renamed to `Sub` | | [047](decisions/047-openable-alpns-are-operations.md) | Openable ALPNs Are Operations | `channel/open` dissolves into per-ALPN ops `channels//sub`/`pub`; `channel_open` marker on `OperationSpec`; `ChannelCore` wrapper; extension-trait `ChannelOperationEnv`; connection-owner allocates `channel_id`; opener ledger (Gap 2 fix); ALPNs are call apps | ## Relevant Open Questions @@ -107,7 +107,7 @@ See [open-questions.md](open-questions.md) for the full tracker. Key questions affecting this crate: - **OQ-01**: Call protocol pub/sub primitive (partially resolved) — - ADR-046 adds the `Pub` primitive (client→server streaming). The + ADR-046 adds the `Pub` primitive (producer→consumer streaming). The fan-out/broker is deferred to channels (Gap B in ADR-047 is named out-of-scope for alkcall; the hub composes the broker on top). - **OQ-02**: Full channel-level flow-control windowing (deferred(scope)) diff --git a/docs/architecture/call-protocol.md b/docs/architecture/call-protocol.md index a48891f..26ab99d 100644 --- a/docs/architecture/call-protocol.md +++ b/docs/architecture/call-protocol.md @@ -581,7 +581,7 @@ Handlers clean up resources when their call is cancelled (in Rust, the future is | Forwarded-for identity | [ADR-026](decisions/032-forwarded-for-identity.md) | `forwarded_for` field on `call.requested` and `OperationContext`; metadata only — `AccessControl::check` never reads it; the `from_call` handler populates it | | Operation error schemas | [ADR-016](decisions/023-operation-error-schemas.md) | Operations declare domain errors; `call.error` carries typed `details` | | Streaming handler for subscriptions | [ADR-021](decisions/021-streaming-handler-for-subscriptions.md) | `StreamingHandler` type, `invoke_streaming()` dispatch path, `INVALID_OPERATION_TYPE` protocol code; the server-side streaming branch in `handle_stream` | -| Publish operation type and HandlerKind::Sink | [ADR-046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | `OperationType::Pub` (client→server streaming); `SinkHandler` + `HandlerKind::Sink`; `call.published` wire event; `invoke_sink()` dispatch; `Subscription` renamed to `Sub` | +| Publish operation type and HandlerKind::Sink | [ADR-046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | `OperationType::Pub` (producer→consumer streaming); `SinkHandler` + `HandlerKind::Sink`; `call.published` wire event; `invoke_sink()` dispatch; `Subscription` renamed to `Sub` | ## Open Questions diff --git a/docs/architecture/decisions/046-publish-operation-type-and-handler-kind-sink.md b/docs/architecture/decisions/046-publish-operation-type-and-handler-kind-sink.md index a947df5..827a190 100644 --- a/docs/architecture/decisions/046-publish-operation-type-and-handler-kind-sink.md +++ b/docs/architecture/decisions/046-publish-operation-type-and-handler-kind-sink.md @@ -137,13 +137,14 @@ A new handler kind consumes the initiator's stream and returns a single /// Sink handler — `Pub` operations. Receives the initiator's data /// stream and returns a single `ResponseEnvelope` (the result of /// consuming the stream). Each `Ok(value)` published by the initiator -/// arrives as an item in the `RecvStream`; the handler processes them -/// and returns one result (success or error). +/// arrives as an item in the stream; the handler processes them and +/// returns one result (success or error). An `Err` item (an +/// initiator-side `call.error`) terminates the stream early. pub type SinkHandler = Arc< dyn Fn( Value, OperationContext, - Pin + Send>>, + Pin> + Send>>, ) -> Pin + Send>> + Send + Sync, @@ -156,13 +157,25 @@ pub enum HandlerKind { } ``` -The `Pin + Send>>` is the initiator's -published stream — each `call.published` event's `payload.input` yields -one `Value` item. The handler consumes the stream to completion, then -returns a single `ResponseEnvelope`. An `Err` in the stream (a -`call.error` from the initiator) terminates the stream early; the -handler may produce its `ResponseEnvelope` from the partial input or -return the initiator's error. +The `Pin> + Send>>` is +the initiator's published stream — each `call.published` event's +`payload.input` yields one `Ok(Value)` item. The handler consumes the +stream to completion, then returns a single `ResponseEnvelope`. An +`Err(CallError)` item (a `call.error` from the initiator) terminates +the stream early; the handler may produce its `ResponseEnvelope` from +the partial input or propagate the error. The `Result`-carrying item +type matches §6's `publish_stream` shape so the handler sees initiator +errors; the concrete stream item type is a two-way-door detail (see +§"Door-type decisions" below). + +> **Amendment (2026-08-13, review 001 P-11):** §3 originally declared +> the stream item type as `Value`, while §6 declared +> `Result`. The code resolves this in favor of §6 +> uniformly (`SinkHandler` takes `Item = Result`, +> aliased as `PublishStream`). §3's text is amended here to match §6; +> the ADR's Door-type section already marked the concrete stream item +> type a two-way-door detail, so this is a text correction, not a +> design change. Registration validates: `Pub` → `HandlerKind::Sink`. Mismatch is a startup error (same pattern as ADR-021's `Subscription` → `Stream` @@ -418,8 +431,8 @@ the canonical form; `"subscription"` is gone). The `HandlerKind` enum shape (`Once(Handler) | Stream(StreamingHandler) | Sink(SinkHandler)`) is the one-way commitment: three handler variants, -validated against `op_type`. The concrete `Pin + Send>>` choice for the sink's input stream is a two-way-door +validated against `op_type`. The concrete stream item type +(`Result`, per §3/§6) is a two-way-door implementation detail within the one-way decision. The `publish_schema: Option` field on `OperationSpec` is a diff --git a/docs/architecture/open-questions.md b/docs/architecture/open-questions.md index 2d9edd1..fe360cc 100644 --- a/docs/architecture/open-questions.md +++ b/docs/architecture/open-questions.md @@ -49,10 +49,10 @@ The call protocol's `StreamingHandler` / `invoke_streaming()` path (ADR-021) is point-to-point: a `call.requested` arrives, the handler produces a stream of `call.responded` events back to that one caller. There was no mechanism for a producer to stream data *to* a responder -(client→server streaming), and no fan-out (one producer, N consumers). +(producer→consumer streaming), and no fan-out (one producer, N consumers). **ADR-046** resolves the directional gap: `OperationType::Pub` is the -client→server streaming complement to `Sub` (was `Subscription`, +producer→consumer streaming complement to `Sub` (was `Subscription`, renamed for symmetry). `HandlerKind::Sink` is the consuming handler type. `call.published` is the wire event carrying stream chunks. `invoke_sink()` is the dispatch path. `CallConnection::publish()` is diff --git a/docs/architecture/operation-registry.md b/docs/architecture/operation-registry.md index d6cda1a..14c05c8 100644 --- a/docs/architecture/operation-registry.md +++ b/docs/architecture/operation-registry.md @@ -972,7 +972,7 @@ The `Capabilities` type holds non-serializable, zeroized secret material. It doe | Forwarded-for identity | [ADR-026](decisions/032-forwarded-for-identity.md) | `forwarded_for` field on `OperationContext` and `call.requested`; metadata only — `AccessControl::check` never reads it; the `from_call` handler populates it | | Streaming handler for subscriptions | [ADR-021](decisions/049-streaming-handler-for-subscriptions.md) | `StreamingHandler` type alongside `Handler`; `HandlerKind` enum on `HandlerRegistration` validated against `op_type`; `invoke_streaming()` on `OperationRegistry`; `invoke()` and `OperationEnv::invoke()` error with `INVALID_OPERATION_TYPE` on `Subscription` ops; composition stays request/response-only, stream composition is handler-level | | Dynamic resource ownership for runtime-spawned resources | [ADR-011](decisions/050-dynamic-resource-ownership-for-runtime-spawned-resources.md) | `AccessControl::check` consults an `OwnershipProvider` (sync read trait, ADR-033 repo/adapter pattern); `OperationSpec` gains `resource_id_path` (JSON pointer into the input); proxy-only access pattern (spawner owns, proxy to share, teardown revokes); `list` = scope-gate + result-filter; teardown = automatic, handler-driven; composition = two orthogonal checks, ADR-017/022 unchanged | -| Publish operation type and HandlerKind::Sink | [ADR-046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | `OperationType::Pub` (client→server streaming); `SinkHandler` + `HandlerKind::Sink` + `PublishStream`; `invoke_sink()` dispatch path; `call.published` wire event; `OperationSpec.publish_schema`; `Subscription` renamed to `Sub`; fan-out/broker deferred to channels | +| Publish operation type and HandlerKind::Sink | [ADR-046](decisions/046-publish-operation-type-and-handler-kind-sink.md) | `OperationType::Pub` (producer→consumer streaming); `SinkHandler` + `HandlerKind::Sink` + `PublishStream`; `invoke_sink()` dispatch path; `call.published` wire event; `OperationSpec.publish_schema`; `Subscription` renamed to `Sub`; fan-out/broker deferred to channels | ## Open Questions diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index fd4cc58..918aeda 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -197,21 +197,13 @@ impl ProtocolHandler for ChannelsAdapter { } async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> { - // 1. Get the bidi stream(s) from the transport. On an in-line - // transport, `accept_bi()` yields once (the single stream - // carries all channels via the header). On QUIC, it yields - // repeatedly. We take the first stream for the demux loop - // and use its write half for the mux. let bidi = connection.accept_bi().await.map_err(|e| match e { StreamError::ConnectionClosed => HandlerError::ConnectionClosed, other => HandlerError::StreamError(std::io::Error::other(format!("{other:?}"))), })?; - // Split the bidi stream into read and write halves. The read - // half feeds the demux; the write half feeds the mux. let (reader, writer) = tokio::io::split(bidi); - // 2. Construct the mux (write side) and the manager. let (mux_handle, mux_runner) = MuxRunner::new(Box::new(writer)); let manager = ChannelManager::new( mux_handle, @@ -230,14 +222,9 @@ impl ProtocolHandler for ChannelsAdapter { } }); - // 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`. + // Channel 0 is pre-negotiated as `alknet/call` (ADR-036); the + // `install_channel_zero` hook runs the single-stream call + // dispatch loop on channel 0's `Connection`. let (channel0_send, channel0_recv) = manager .install_channel_zero(None) .await @@ -248,10 +235,9 @@ impl ProtocolHandler for ChannelsAdapter { 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) - // and decrements the per-identity policy for each drained - // channel (ADR-047 §7 — connection-drop teardown path). + // The demux loop blocks until transport EOF, then clears the + // channel map (REQ-CH-02) and decrements the per-identity + // policy for each drained channel (ADR-047 §7). Self::run_demux_loop(&manager, Box::new(reader), &self.policy).await; Ok(()) diff --git a/src/channels/client.rs b/src/channels/client.rs index bc1f5a4..1ef453a 100644 --- a/src/channels/client.rs +++ b/src/channels/client.rs @@ -68,9 +68,7 @@ impl ChannelClient { // 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. + // needs the runner to be draining `new_pumps`. 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"); @@ -85,20 +83,14 @@ impl ChannelClient { ChannelSide::Connect, ); - // 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)?; - // 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`. + // Single-stream call mode (ADR-036 amendment): channel 0's + // read+write halves are split into a `SharedFrameWriter` + // (for `call.requested`) and a read pump (for responses). 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()); @@ -109,12 +101,6 @@ impl ChannelClient { 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( @@ -124,9 +110,9 @@ impl ChannelClient { .await; }); - // Spawn the demux loop (read side). The loop reads 8-byte - // chunk headers and routes payloads to the manager. It ends - // on transport EOF, clearing the channel map (REQ-CH-02). + // The demux loop ends on transport EOF, clearing the channel + // map (REQ-CH-02). The connect side passes `policy: None` (it + // does not enforce the per-identity cap). let demux_manager = manager.clone(); let _demux_task = tokio::spawn(async move { super::adapter::ChannelsAdapter::run_demux_loop_for_client( @@ -288,11 +274,6 @@ mod tests { }) } - #[test] - fn channel_client_is_not_send_safe_across_await_by_default() { - 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 diff --git a/src/channels/mod.rs b/src/channels/mod.rs index d0123ec..4c81737 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -27,7 +27,7 @@ //! (ADR-041, amended by ADR-047 §7 — opener ledger). //! - [`client`]: `ChannelClient` — transport-agnostic //! `from_connection` (ADR-043). -//! - [`env`]: `ChannelOperationEnv` extension trait (ADR-047 §4 — +//! - [`self::env`]: `ChannelOperationEnv` extension trait (ADR-047 §4 — //! keeps the call crate free of channels types). //! //! See `docs/architecture/` for the full specification. diff --git a/src/channels/operations.rs b/src/channels/operations.rs index 0df506c..f77e256 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -47,7 +47,8 @@ pub struct ChannelOperations { impl ChannelOperations { /// Construct with a `ChannelManager` and a /// `ChannelLifecyclePolicy`. The default policy is - /// `PerIdentityChannelPolicy::new(256)` (via [`default_policy`]). + /// `PerIdentityChannelPolicy::new(256)` (via + /// [`default_policy`][super::policy::default_policy]). pub fn new(manager: ChannelManager, policy: Arc) -> Self { Self { manager, policy } } diff --git a/src/client/from_call.rs b/src/client/from_call.rs index 7a3ca73..611a935 100644 --- a/src/client/from_call.rs +++ b/src/client/from_call.rs @@ -268,12 +268,12 @@ fn rebuild_spec_for( .unwrap_or(false) { if let Some(alpn) = derive_alpn_from_op_name(remote_name) { - // SAFETY: `derive_alpn_from_op_name` returns a `'static str` - // only when the ALPN is a known `alknet/*` ALPN baked into - // the binary at compile time. For dynamically-discovered - // ALPNs we'd need a `String`-backed `ChannelOpenSpec`; that - // is a two-way-door extension deferred until a non- - // `alknet/*` openable ALPN actually exists. + // `ChannelOpenSpec::alpn` is `&'static str` (ADR-047 §2). The + // discovered ALPN is a runtime `String`, so `leak_alpn` boxes + // and leaks it to satisfy the `'static` bound. The leak is + // bounded per unique ALPN but accumulates on rediscovery; + // the `Arc` / `Cow<'static, str>` refactor that removes + // the leak is deferred (see review 001 Unit 7). spec = spec.with_channel_open(ChannelOpenSpec::new(leak_alpn(alpn))); } } diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index ca5bf52..1aaf071 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -403,11 +403,11 @@ impl Dispatcher { /// ends after it and we do NOT write `call.completed` (ADR-049 §6). /// /// If a frame write fails the pump stops early; the stream is dropped on - /// return, releasing the handler's resources via `Drop` (ADR-016). The - /// pump is cancellable: it runs inside the `handle_stream` task, so a - /// `call.aborted` for this request ID (handled by `handle_abort` on - /// another stream) or connection close cancels the task and drops the - /// stream. + /// return, releasing the handler's resources via `Drop` (ADR-016). A + /// `call.aborted` for this request ID mutates the `PendingRequestMap` + /// (via `handle_abort` on another stream) but does not yet cancel this + /// task — the pump ends when the stream ends naturally or the I/O fails + /// (the abort-cancels mechanism for Sub/Pub is review 001 Unit 9). pub(crate) async fn pump_stream( &self, writer: &mut super::wire::FrameFramedWriter, @@ -440,10 +440,10 @@ impl Dispatcher { /// is written to the wire as `call.responded` (or `call.error`). /// /// The handler runs concurrently — it consumes the - /// `PublishStream` while we feed it. `call.aborted` for this request - /// ID (handled by `handle_abort` on another stream) or connection - /// close cancels the task and drops the handler future, releasing - /// the handler's resources via `Drop` (ADR-020). + /// `PublishStream` while we feed it. On `call.aborted` for this + /// request ID, the feed injects an `Err` into `chunk_tx` and ends; + /// the handler future is awaited to completion (it is not yet + /// cancelled — the abort-cancels-Pub mechanism is review 001 Unit 9). pub(crate) async fn pump_sink( &self, reader: &mut super::wire::FrameFramedReader, @@ -472,7 +472,7 @@ impl Dispatcher { continue; } match envelope.r#type.as_str() { - "call.published" => { + EVENT_PUBLISHED => { let chunk = envelope .payload .get("input") @@ -482,8 +482,8 @@ impl Dispatcher { break; } } - "call.completed" => break, - "call.aborted" => { + EVENT_COMPLETED => break, + EVENT_ABORTED => { let _ = chunk_tx .send(Err(CallError::internal("publish aborted by initiator"))) .await;