fix: Unit 6 — convention + doc cleanup (C-09, C-20-rem, C-22-rem, C-24, P-10, P-11)
Conventions satisfied, `cargo doc` clean, no actively-wrong comments, producer/consumer naming consistent in the call/registry docs. P-10 — `pump_sink` matched the string literals "call.published", "call.completed", "call.aborted" (dispatch.rs) instead of the EVENT_PUBLISHED / EVENT_COMPLETED / EVENT_ABORTED constants the rest of the file imports. Replaced with the constants — pure refactor hazard, no behavior change. C-20 remainder — the wrong "SAFETY:" comment at from_call.rs:271 marked no `unsafe` block and was factually wrong (described a `'static` return that isn't what `derive_alpn_from_op_name` does — it returns `Option<String>`; the leak happens in `leak_alpn`). Reworded to a plain note about the `'static` lifetime requirement. Also reworded the abort-cancels claims in the `pump_sink` and `pump_stream` doc comments (dispatch.rs): both claimed `call.aborted` "cancels the task and drops the handler future" — the handler is actually `join!`-ed to completion and not yet cancelled (the abort-cancels-Pub mechanism is review 001 Unit 9). Trimmed step-numbered narration comments in adapter.rs / client.rs that restated what the code does, keeping the ordering-constraint and REQ-CH comments. The big reassembly.rs deliberation landed with Unit 4; this finishes the remainder. C-09 — fixed the 2 remaining `cargo doc` warnings (was 4; the register_openable links were fixed in Unit 3): - `unresolved link to default_policy` (operations.rs:50) — the [`default_policy`] intra-doc link resolves to super::policy::default_policy; used the full path. - `env is both a module and a macro` (channels/mod.rs:30) — the [`env`] link collided with the std `env!` macro; qualified as [`self::env`]. `cargo doc --no-deps` now emits 0 warnings. C-22 remainder — removed the filler `PhantomData` test at client.rs (`let _ = std::marker::PhantomData::<ChannelClient>;` — asserts nothing). The env.rs tautology was already removed in Unit 3. C-24 — replaced "client→server streaming" with "producer→consumer streaming" in call-protocol.md, operation-registry.md, README.md, and open-questions.md (4 occurrences). Per AGENTS.md §8 the convention is producer/consumer, not server/client. The remaining "client→server" references in channels ADRs 034/037 are in stream_type table contexts that Unit 10 (C-26) will handle as part of the spec-doc renumbering. P-11 — amended ADR-046 §3's SinkHandler type so the stream item type matches §6. §3 declared `Pin<Box<dyn Stream<Item = Value> + Send>>`; §6 declared `Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>`. The code uses §6's shape uniformly (registration.rs:32-40, aliased as PublishStream). §3's text and the Door-type section are amended to match §6; the 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. Added an amendment note dated 2026-08-13. Verification: - cargo test --lib → 449 passed, 0 failed (was 450; -1 removed filler test) - cargo clippy --all-targets -- -D warnings → clean - cargo fmt --check → clean - cargo doc --no-deps → 0 warnings (was 2)
This commit is contained in:
1 parent
48564a8f49
commit
bfb265e31b
11 files changed
+65
-84
No files matched your search
@@ -98,7 +98,7 @@ are wire-stable and unchanged — see ADR-004.
|
|||||||
| [043](decisions/043-channelclient.md) | ChannelClient | Transport-agnostic from_connection |
|
| [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) |
|
| [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 |
|
| [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/<alpn>/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 |
|
| [047](decisions/047-openable-alpns-are-operations.md) | Openable ALPNs Are Operations | `channel/open` dissolves into per-ALPN ops `channels/<alpn>/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
|
## Relevant Open Questions
|
||||||
@@ -107,7 +107,7 @@ See [open-questions.md](open-questions.md) for the full tracker. Key
|
|||||||
questions affecting this crate:
|
questions affecting this crate:
|
||||||
|
|
||||||
- **OQ-01**: Call protocol pub/sub primitive (partially resolved) —
|
- **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
|
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).
|
out-of-scope for alkcall; the hub composes the broker on top).
|
||||||
- **OQ-02**: Full channel-level flow-control windowing (deferred(scope))
|
- **OQ-02**: Full channel-level flow-control windowing (deferred(scope))
|
||||||
|
|||||||
@@ -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 |
|
| 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` |
|
| 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` |
|
| 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
|
## Open Questions
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
/// Sink handler — `Pub` operations. Receives the initiator's data
|
||||||
/// stream and returns a single `ResponseEnvelope` (the result of
|
/// stream and returns a single `ResponseEnvelope` (the result of
|
||||||
/// consuming the stream). Each `Ok(value)` published by the initiator
|
/// consuming the stream). Each `Ok(value)` published by the initiator
|
||||||
/// arrives as an item in the `RecvStream`; the handler processes them
|
/// arrives as an item in the stream; the handler processes them and
|
||||||
/// and returns one result (success or error).
|
/// returns one result (success or error). An `Err` item (an
|
||||||
|
/// initiator-side `call.error`) terminates the stream early.
|
||||||
pub type SinkHandler = Arc<
|
pub type SinkHandler = Arc<
|
||||||
dyn Fn(
|
dyn Fn(
|
||||||
Value,
|
Value,
|
||||||
OperationContext,
|
OperationContext,
|
||||||
Pin<Box<dyn Stream<Item = Value> + Send>>,
|
Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>,
|
||||||
) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
|
) -> Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>
|
||||||
+ Send
|
+ Send
|
||||||
+ Sync,
|
+ Sync,
|
||||||
@@ -156,13 +157,25 @@ pub enum HandlerKind {
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
The `Pin<Box<dyn Stream<Item = Value> + Send>>` is the initiator's
|
The `Pin<Box<dyn Stream<Item = Result<Value, CallError>> + Send>>` is
|
||||||
published stream — each `call.published` event's `payload.input` yields
|
the initiator's published stream — each `call.published` event's
|
||||||
one `Value` item. The handler consumes the stream to completion, then
|
`payload.input` yields one `Ok(Value)` item. The handler consumes the
|
||||||
returns a single `ResponseEnvelope`. An `Err` in the stream (a
|
stream to completion, then returns a single `ResponseEnvelope`. An
|
||||||
`call.error` from the initiator) terminates the stream early; the
|
`Err(CallError)` item (a `call.error` from the initiator) terminates
|
||||||
handler may produce its `ResponseEnvelope` from the partial input or
|
the stream early; the handler may produce its `ResponseEnvelope` from
|
||||||
return the initiator's error.
|
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<Value, CallError>`. The code resolves this in favor of §6
|
||||||
|
> uniformly (`SinkHandler` takes `Item = Result<Value, CallError>`,
|
||||||
|
> 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
|
Registration validates: `Pub` → `HandlerKind::Sink`. Mismatch is a
|
||||||
startup error (same pattern as ADR-021's `Subscription` → `Stream`
|
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)
|
The `HandlerKind` enum shape (`Once(Handler) | Stream(StreamingHandler)
|
||||||
| Sink(SinkHandler)`) is the one-way commitment: three handler variants,
|
| Sink(SinkHandler)`) is the one-way commitment: three handler variants,
|
||||||
validated against `op_type`. The concrete `Pin<Box<dyn Stream<Item =
|
validated against `op_type`. The concrete stream item type
|
||||||
Value> + Send>>` choice for the sink's input stream is a two-way-door
|
(`Result<Value, CallError>`, per §3/§6) is a two-way-door
|
||||||
implementation detail within the one-way decision.
|
implementation detail within the one-way decision.
|
||||||
|
|
||||||
The `publish_schema: Option<Value>` field on `OperationSpec` is a
|
The `publish_schema: Option<Value>` field on `OperationSpec` is a
|
||||||
|
|||||||
@@ -49,10 +49,10 @@ The call protocol's `StreamingHandler` / `invoke_streaming()` path
|
|||||||
(ADR-021) is point-to-point: a `call.requested` arrives, the handler
|
(ADR-021) is point-to-point: a `call.requested` arrives, the handler
|
||||||
produces a stream of `call.responded` events back to that one caller.
|
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
|
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
|
**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
|
renamed for symmetry). `HandlerKind::Sink` is the consuming handler
|
||||||
type. `call.published` is the wire event carrying stream chunks.
|
type. `call.published` is the wire event carrying stream chunks.
|
||||||
`invoke_sink()` is the dispatch path. `CallConnection::publish()` is
|
`invoke_sink()` is the dispatch path. `CallConnection::publish()` is
|
||||||
|
|||||||
@@ -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 |
|
| 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 |
|
| 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 |
|
| 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
|
## Open Questions
|
||||||
|
|
||||||
|
|||||||
+6
-20
@@ -197,21 +197,13 @@ impl ProtocolHandler for ChannelsAdapter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> {
|
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 {
|
let bidi = connection.accept_bi().await.map_err(|e| match e {
|
||||||
StreamError::ConnectionClosed => HandlerError::ConnectionClosed,
|
StreamError::ConnectionClosed => HandlerError::ConnectionClosed,
|
||||||
other => HandlerError::StreamError(std::io::Error::other(format!("{other:?}"))),
|
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);
|
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 (mux_handle, mux_runner) = MuxRunner::new(Box::new(writer));
|
||||||
let manager = ChannelManager::new(
|
let manager = ChannelManager::new(
|
||||||
mux_handle,
|
mux_handle,
|
||||||
@@ -230,14 +222,9 @@ impl ProtocolHandler for ChannelsAdapter {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
// 3. Install channel 0 (pre-negotiated as `alknet/call`,
|
// Channel 0 is pre-negotiated as `alknet/call` (ADR-036); the
|
||||||
// ADR-036). Channel 0's reassembly buffer + mux write half
|
// `install_channel_zero` hook runs the single-stream call
|
||||||
// are joined into a `BiStream` and wrapped in a
|
// dispatch loop on channel 0's `Connection`.
|
||||||
// `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
|
let (channel0_send, channel0_recv) = manager
|
||||||
.install_channel_zero(None)
|
.install_channel_zero(None)
|
||||||
.await
|
.await
|
||||||
@@ -248,10 +235,9 @@ impl ProtocolHandler for ChannelsAdapter {
|
|||||||
let _handler_task =
|
let _handler_task =
|
||||||
(self.install_channel_zero)(manager.clone(), channel0_conn, auth.clone());
|
(self.install_channel_zero)(manager.clone(), channel0_conn, auth.clone());
|
||||||
|
|
||||||
// 4. Run the demux loop (read side). This blocks until
|
// The demux loop blocks until transport EOF, then clears the
|
||||||
// transport EOF, then clears the channel map (REQ-CH-02)
|
// channel map (REQ-CH-02) and decrements the per-identity
|
||||||
// and decrements the per-identity policy for each drained
|
// policy for each drained channel (ADR-047 §7).
|
||||||
// channel (ADR-047 §7 — connection-drop teardown path).
|
|
||||||
Self::run_demux_loop(&manager, Box::new(reader), &self.policy).await;
|
Self::run_demux_loop(&manager, Box::new(reader), &self.policy).await;
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
+7
-26
@@ -68,9 +68,7 @@ impl ChannelClient {
|
|||||||
|
|
||||||
// Spawn the mux runner BEFORE installing channel 0 —
|
// Spawn the mux runner BEFORE installing channel 0 —
|
||||||
// `install_channel_zero` calls `mux.register(0).await` which
|
// `install_channel_zero` calls `mux.register(0).await` which
|
||||||
// sends a `Registration` to the runner and waits for the
|
// needs the runner to be draining `new_pumps`.
|
||||||
// response. The runner must be draining `new_pumps` for the
|
|
||||||
// registration to complete.
|
|
||||||
let _mux_task = tokio::spawn(async move {
|
let _mux_task = tokio::spawn(async move {
|
||||||
if let Err(e) = mux_runner.run().await {
|
if let Err(e) = mux_runner.run().await {
|
||||||
tracing::warn!(error = %e, "channel client: mux runner ended with error");
|
tracing::warn!(error = %e, "channel client: mux runner ended with error");
|
||||||
@@ -85,20 +83,14 @@ impl ChannelClient {
|
|||||||
ChannelSide::Connect,
|
ChannelSide::Connect,
|
||||||
);
|
);
|
||||||
|
|
||||||
// Install channel 0 — the call adapter's read/write halves.
|
|
||||||
let (channel0_send, channel0_recv) = manager
|
let (channel0_send, channel0_recv) = manager
|
||||||
.install_channel_zero(None)
|
.install_channel_zero(None)
|
||||||
.await
|
.await
|
||||||
.map_err(|_| StreamError::StreamClosed)?;
|
.map_err(|_| StreamError::StreamClosed)?;
|
||||||
|
|
||||||
// Join channel 0's reassembled read half + mux write half into
|
// Single-stream call mode (ADR-036 amendment): channel 0's
|
||||||
// a `BiStream` (via `channel_source`), wrap as a `Connection`
|
// read+write halves are split into a `SharedFrameWriter`
|
||||||
// (for ALPN/addr/identity), then yield the `BiStream` once via
|
// (for `call.requested`) and a read pump (for responses).
|
||||||
// `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 =
|
let channel0_source =
|
||||||
super::source::channel_source(channel0_recv, channel0_send, remote_addr);
|
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_conn = Connection::from_source(channel0_source, b"alknet/call".to_vec());
|
||||||
@@ -109,12 +101,6 @@ impl ChannelClient {
|
|||||||
let call_connection =
|
let call_connection =
|
||||||
CallConnection::new_single_stream(channel0_conn, single_stream_writer);
|
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 pending_map = Arc::clone(call_connection.pending());
|
||||||
let _read_pump = tokio::spawn(async move {
|
let _read_pump = tokio::spawn(async move {
|
||||||
crate::protocol::connection::read_single_stream_until_closed(
|
crate::protocol::connection::read_single_stream_until_closed(
|
||||||
@@ -124,9 +110,9 @@ impl ChannelClient {
|
|||||||
.await;
|
.await;
|
||||||
});
|
});
|
||||||
|
|
||||||
// Spawn the demux loop (read side). The loop reads 8-byte
|
// The demux loop ends on transport EOF, clearing the channel
|
||||||
// chunk headers and routes payloads to the manager. It ends
|
// map (REQ-CH-02). The connect side passes `policy: None` (it
|
||||||
// on transport EOF, clearing the channel map (REQ-CH-02).
|
// does not enforce the per-identity cap).
|
||||||
let demux_manager = manager.clone();
|
let demux_manager = manager.clone();
|
||||||
let _demux_task = tokio::spawn(async move {
|
let _demux_task = tokio::spawn(async move {
|
||||||
super::adapter::ChannelsAdapter::run_demux_loop_for_client(
|
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::<ChannelClient>;
|
|
||||||
}
|
|
||||||
|
|
||||||
/// C-25 #1 — the end-to-end acceptance gate for Unit 2 (channel 0
|
/// C-25 #1 — the end-to-end acceptance gate for Unit 2 (channel 0
|
||||||
/// single-stream call mode, ADR-036 amendment). Wires
|
/// single-stream call mode, ADR-036 amendment). Wires
|
||||||
/// `ChannelClient` (connect side) ↔ `ChannelsAdapter` (accept
|
/// `ChannelClient` (connect side) ↔ `ChannelsAdapter` (accept
|
||||||
|
|||||||
+1
-1
@@ -27,7 +27,7 @@
|
|||||||
//! (ADR-041, amended by ADR-047 §7 — opener ledger).
|
//! (ADR-041, amended by ADR-047 §7 — opener ledger).
|
||||||
//! - [`client`]: `ChannelClient` — transport-agnostic
|
//! - [`client`]: `ChannelClient` — transport-agnostic
|
||||||
//! `from_connection` (ADR-043).
|
//! `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).
|
//! keeps the call crate free of channels types).
|
||||||
//!
|
//!
|
||||||
//! See `docs/architecture/` for the full specification.
|
//! See `docs/architecture/` for the full specification.
|
||||||
|
|||||||
@@ -47,7 +47,8 @@ pub struct ChannelOperations {
|
|||||||
impl ChannelOperations {
|
impl ChannelOperations {
|
||||||
/// Construct with a `ChannelManager` and a
|
/// Construct with a `ChannelManager` and a
|
||||||
/// `ChannelLifecyclePolicy`. The default policy is
|
/// `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<dyn ChannelLifecyclePolicy>) -> Self {
|
pub fn new(manager: ChannelManager, policy: Arc<dyn ChannelLifecyclePolicy>) -> Self {
|
||||||
Self { manager, policy }
|
Self { manager, policy }
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -268,12 +268,12 @@ fn rebuild_spec_for(
|
|||||||
.unwrap_or(false)
|
.unwrap_or(false)
|
||||||
{
|
{
|
||||||
if let Some(alpn) = derive_alpn_from_op_name(remote_name) {
|
if let Some(alpn) = derive_alpn_from_op_name(remote_name) {
|
||||||
// SAFETY: `derive_alpn_from_op_name` returns a `'static str`
|
// `ChannelOpenSpec::alpn` is `&'static str` (ADR-047 §2). The
|
||||||
// only when the ALPN is a known `alknet/*` ALPN baked into
|
// discovered ALPN is a runtime `String`, so `leak_alpn` boxes
|
||||||
// the binary at compile time. For dynamically-discovered
|
// and leaks it to satisfy the `'static` bound. The leak is
|
||||||
// ALPNs we'd need a `String`-backed `ChannelOpenSpec`; that
|
// bounded per unique ALPN but accumulates on rediscovery;
|
||||||
// is a two-way-door extension deferred until a non-
|
// the `Arc<str>` / `Cow<'static, str>` refactor that removes
|
||||||
// `alknet/*` openable ALPN actually exists.
|
// the leak is deferred (see review 001 Unit 7).
|
||||||
spec = spec.with_channel_open(ChannelOpenSpec::new(leak_alpn(alpn)));
|
spec = spec.with_channel_open(ChannelOpenSpec::new(leak_alpn(alpn)));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+12
-12
@@ -403,11 +403,11 @@ impl Dispatcher {
|
|||||||
/// ends after it and we do NOT write `call.completed` (ADR-049 §6).
|
/// 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
|
/// 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
|
/// return, releasing the handler's resources via `Drop` (ADR-016). A
|
||||||
/// pump is cancellable: it runs inside the `handle_stream` task, so a
|
/// `call.aborted` for this request ID mutates the `PendingRequestMap`
|
||||||
/// `call.aborted` for this request ID (handled by `handle_abort` on
|
/// (via `handle_abort` on another stream) but does not yet cancel this
|
||||||
/// another stream) or connection close cancels the task and drops the
|
/// task — the pump ends when the stream ends naturally or the I/O fails
|
||||||
/// stream.
|
/// (the abort-cancels mechanism for Sub/Pub is review 001 Unit 9).
|
||||||
pub(crate) async fn pump_stream<W: tokio::io::AsyncWrite + Unpin>(
|
pub(crate) async fn pump_stream<W: tokio::io::AsyncWrite + Unpin>(
|
||||||
&self,
|
&self,
|
||||||
writer: &mut super::wire::FrameFramedWriter<W>,
|
writer: &mut super::wire::FrameFramedWriter<W>,
|
||||||
@@ -440,10 +440,10 @@ impl Dispatcher {
|
|||||||
/// is written to the wire as `call.responded` (or `call.error`).
|
/// is written to the wire as `call.responded` (or `call.error`).
|
||||||
///
|
///
|
||||||
/// The handler runs concurrently — it consumes the
|
/// The handler runs concurrently — it consumes the
|
||||||
/// `PublishStream` while we feed it. `call.aborted` for this request
|
/// `PublishStream` while we feed it. On `call.aborted` for this
|
||||||
/// ID (handled by `handle_abort` on another stream) or connection
|
/// request ID, the feed injects an `Err` into `chunk_tx` and ends;
|
||||||
/// close cancels the task and drops the handler future, releasing
|
/// the handler future is awaited to completion (it is not yet
|
||||||
/// the handler's resources via `Drop` (ADR-020).
|
/// cancelled — the abort-cancels-Pub mechanism is review 001 Unit 9).
|
||||||
pub(crate) async fn pump_sink<R, W>(
|
pub(crate) async fn pump_sink<R, W>(
|
||||||
&self,
|
&self,
|
||||||
reader: &mut super::wire::FrameFramedReader<R>,
|
reader: &mut super::wire::FrameFramedReader<R>,
|
||||||
@@ -472,7 +472,7 @@ impl Dispatcher {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
match envelope.r#type.as_str() {
|
match envelope.r#type.as_str() {
|
||||||
"call.published" => {
|
EVENT_PUBLISHED => {
|
||||||
let chunk = envelope
|
let chunk = envelope
|
||||||
.payload
|
.payload
|
||||||
.get("input")
|
.get("input")
|
||||||
@@ -482,8 +482,8 @@ impl Dispatcher {
|
|||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
"call.completed" => break,
|
EVENT_COMPLETED => break,
|
||||||
"call.aborted" => {
|
EVENT_ABORTED => {
|
||||||
let _ = chunk_tx
|
let _ = chunk_tx
|
||||||
.send(Err(CallError::internal("publish aborted by initiator")))
|
.send(Err(CallError::internal("publish aborted by initiator")))
|
||||||
.await;
|
.await;
|
||||||
|
|||||||
Reference in new issue
Block a user