fix: Unit 3 — register_openable + per-connection ChannelCore (C-02, C-03, C-09)
A channel-open op could not be registered or invoked (C-02):
ChannelCore::register_openable did not exist, and resolve_channel_manager
(C-03) was a stub returning None — the ADR-047 §4 dynamic-resolution
shape (downcast context.env to &dyn ChannelOperationEnv) was unworkable
as written: context.env is a PeerCompositeEnv, not a single concrete
type that can be downcast to a channels-backed env.
The fix is per-connection registration (ADR-047 §4 amendment,
2026-08-13): a ChannelCore is constructed per channels connection (in
the install_channel_zero hook, which already runs per-connection and
already receives the ChannelManager), and register_openable is called on
that connection's overlay OperationRegistry (Layer 2 per ADR-019). The
wrapper closes over the per-connection ChannelCore and uses
ChannelCore::manager() directly — no context.env downcast. This
preserves every invariant ADR-047 §4 was written to protect (layering,
per-connection resolution) without adding as_any() to OperationEnv
(which would close the session/connection overlay patterns from
ADR-024, AGENTS.md §6).
Changes:
- ChannelCore::register_openable wraps the ALPN's OpenHandler with the
ACL→check_open→open_channel→spawn→respond flow (ADR-047 §3). Branches
on spec.op_type: Query/Mutation→Once, Sub→Stream (emits { channel_id }
and completes; data plane on the channel's BiStream), Pub→Sink (stub:
channel:pub_open_not_implemented — requires the channel-adoption path,
C-08/Unit 5). The OpenHandler receives (input, Connection, AuthContext)
and spawns the ALPN's protocol on the channel's BiStream, returning a
JoinHandle for teardown.
- ChannelManager::set_handler_task installs the spawned OpenHandler's
JoinHandle after open_channel (which allocates the channel first to
get the BiStream halves, then the handler is spawned, then the task is
recorded for abort on channel/close / connection drop).
- channel:too_many_channels / channel:allocation_failed error codes
mapped to CallError with details (channel:forbidden is the ACL's
FORBIDDEN, already handled by the registry before the wrapper).
- resolve_channel_manager stub removed (C-03); ChannelOperationEnv trait
and ChannelsSessionEnv retained as a two-way-door implementation detail
for future per-connection routing (not on the open-op path). The
tautology filler test (C-22 env.rs) removed.
- ADR-047 §4 amendment records the per-connection-registration decision
(two-way door: the ADR's door-type section explicitly marks the wrapper
shape as a two-way-door implementation detail; the one-way decisions
— per-ALPN op names, channel_open marker, removal of channel/open —
are unchanged).
Acceptance gate (C-02/C-03): one end-to-end test wires ChannelClient ↔
ChannelsAdapter over a real tokio::io::duplex carrying the channels
8-byte chunk header wire format. The accept side's install_channel_zero
hook builds a per-connection ChannelCore, registers a no-op open op
(channels/tty/sub) via register_openable, and runs the dispatch loop.
The client calls call_open_op("channels/tty/sub") on channel 0; the
wrapper does check_open→open_channel→spawn→respond. Asserts the
response carries a non-zero channel_id and that the per-identity quota
was reserved (policy count for the caller incremented to 1).
Verification: 438 tests pass (was 437; +1 e2e), clippy clean, fmt clean,
doc warnings 2 (was 4; fixed the 2 register_openable broken-link
warnings — C-09; the remaining default_policy and env module/macro
warnings are Unit 6 long-tail items).
This commit is contained in:
@@ -2,7 +2,89 @@
|
||||
|
||||
## Status
|
||||
|
||||
Accepted (amends ADR-037; refines ADR-044, ADR-046)
|
||||
Accepted (amends ADR-037; refines ADR-044, ADR-046; §4 amended
|
||||
2026-08-13 — open ops are registered per-connection, not resolved via
|
||||
`context.env` downcast — see "Amendment (§4 per-connection
|
||||
registration, 2026-08-13)" below)
|
||||
|
||||
## Amendment (§4 per-connection registration, 2026-08-13)
|
||||
|
||||
ADR-047 §4 specified that the open-op wrapper resolves the
|
||||
per-connection `ChannelManager` by downcasting `context.env` to
|
||||
`&dyn ChannelOperationEnv` at invocation time — "static registration,
|
||||
dynamic resolution." Implementation (review 001, C-03) found this
|
||||
shape is not workable as written: `context.env` is a `PeerCompositeEnv`
|
||||
(a composite of base + session + per-connection overlays), not a single
|
||||
concrete type, so a direct `as_any()` downcast of `context.env` to
|
||||
`ChannelsSessionEnv` cannot reach the `ChannelManager`. Traversing the
|
||||
composite's layers to find the channels-backed overlay would hardcode
|
||||
`PeerCompositeEnv`'s internal structure into the channels module —
|
||||
fragile, leaky, and a layering violation (the call crate's composite-env
|
||||
shape is not part of the channels crate's contract).
|
||||
|
||||
**The amendment: open ops are registered per-connection.** The
|
||||
`ChannelCore` is constructed per channels connection (in the
|
||||
`install_channel_zero` hook, which already runs per-connection and
|
||||
already receives the `ChannelManager`). `ChannelCore::register_openable`
|
||||
is called on that connection's `OperationRegistry` (the connection
|
||||
overlay registry, Layer 2 per ADR-019), closing over the per-connection
|
||||
`ChannelCore`. The wrapper uses `ChannelCore::manager()` directly — no
|
||||
`context.env` downcast, no dynamic resolution. The open op lives on the
|
||||
connection overlay, which is where per-connection state naturally
|
||||
belongs (ADR-019, ADR-024).
|
||||
|
||||
This preserves every invariant ADR-047 §4 was written to protect:
|
||||
|
||||
- **Layering (ADR-044):** the call crate stays free of channels types.
|
||||
The open-op wrapper is in `channels-call` (`src/channels/operations.rs`);
|
||||
the call crate's `OperationRegistry` and `OperationEnv` are unchanged.
|
||||
No `as_any()` is added to `OperationEnv` (the trait stays concrete-rpc-
|
||||
shaped, preserving the session/connection overlay patterns from
|
||||
ADR-024 — AGENTS.md §6).
|
||||
- **Per-connection resolution:** the open op gets the *right*
|
||||
`ChannelManager` (the one for the connection it was invoked on) because
|
||||
the op is registered on that connection's overlay registry, with a
|
||||
`ChannelCore` closing over that connection's manager. A globally-
|
||||
registered handler (Layer 0) is not on this path.
|
||||
- **"Marked ops invoked outside a channels session" (ADR-047 §2):**
|
||||
unchanged. A `channels/<alpn>/sub` op registered only on a channels
|
||||
connection's overlay is not reachable on a bare `alknet/call`
|
||||
connection (the overlay isn't attached there) — the dispatch path
|
||||
returns `NOT_FOUND`, which is the correct behavior for "no channels
|
||||
session" (the `channel:no_channels_session` error code from the
|
||||
original §4 is no longer reached; `NOT_FOUND` is the natural
|
||||
reachable-but-not-here result).
|
||||
|
||||
The `ChannelOperationEnv` extension trait and `ChannelsSessionEnv` impl
|
||||
(`src/channels/env.rs`) are retained as a two-way-door implementation
|
||||
detail — they are not on the open-op path, but remain available for
|
||||
future per-connection routing (e.g., nested channels where each
|
||||
connection's overlay carries its own manager reference for
|
||||
non-open-op queries). The `resolve_channel_manager` stub is removed
|
||||
(it described the rejected dynamic-resolution shape).
|
||||
|
||||
### Door type
|
||||
|
||||
**Two-way (implementation detail).** The ADR's door-type section
|
||||
already marks "The `ChannelCore` wrapper shape, the extension-trait
|
||||
pattern, and the opener ledger are two-way-door implementation details
|
||||
within the one-way decision." Per-connection registration vs. dynamic
|
||||
resolution is a choice within the wrapper-shape detail — the one-way
|
||||
decisions (per-ALPN op names, the `channel_open` marker, removal of
|
||||
`channel/open`/`direction`) are unchanged. A future revision could move
|
||||
open ops back to Layer 0 with real `as_any()` downcast machinery if the
|
||||
composite-env structure stabilizes enough to make the traversal
|
||||
non-fragile; the per-connection shape is the simpler choice today.
|
||||
|
||||
### References
|
||||
|
||||
- C-02, C-03 in `docs/reviews/001-pub-and-channels-integration-review.md`
|
||||
(the `register_openable`-does-not-exist and
|
||||
`resolve_channel_manager`-is-a-stub findings this amendment resolves)
|
||||
- ADR-019: operation registry layering (the connection overlay the open
|
||||
op is registered on)
|
||||
- ADR-024: peer-graph routing model (the `OperationEnv` integration-point
|
||||
pattern this amendment preserves by NOT adding `as_any()`)
|
||||
|
||||
## Context
|
||||
|
||||
@@ -186,6 +268,19 @@ architectural point is the wrapper.
|
||||
|
||||
### 4. Per-connection `ChannelManager` resolution (Gap E)
|
||||
|
||||
> **Amended 2026-08-13** — see "Amendment (§4 per-connection
|
||||
> registration, 2026-08-13)" at the top of this file. The
|
||||
> dynamic-resolution shape described below (downcast `context.env` to
|
||||
> `&dyn ChannelOperationEnv` at invocation time) was found
|
||||
> unworkable as written (`context.env` is a `PeerCompositeEnv`, not a
|
||||
> single concrete type; the downcast cannot reach the manager without
|
||||
> fragile cross-layer traversal). The operative decision is
|
||||
> **per-connection registration**: `register_openable` is called on the
|
||||
> connection overlay registry (Layer 2), closing over a per-connection
|
||||
> `ChannelCore`; the wrapper uses `ChannelCore::manager()` directly.
|
||||
> The body below is the **original** (rejected) shape, retained for
|
||||
> rationale continuity.
|
||||
|
||||
The `register_openable` helper registers ops at assembly time (Layer 0,
|
||||
curated, static per ADR-019). But the wrapper needs the
|
||||
**per-connection** `ChannelManager` — the op arrives on channel 0 of one
|
||||
|
||||
@@ -172,6 +172,7 @@ impl ChannelClient {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::channels::adapter::ChannelsAdapter;
|
||||
use crate::channels::policy::ChannelLifecyclePolicy;
|
||||
use crate::core::auth::{AuthContext, IdentityProvider};
|
||||
use crate::core::types::Connection;
|
||||
use crate::protocol::connection::split_single_stream;
|
||||
@@ -445,4 +446,156 @@ mod tests {
|
||||
"last chunk matches"
|
||||
);
|
||||
}
|
||||
|
||||
/// C-02 / C-03 — the end-to-end acceptance gate for Unit 3
|
||||
/// (`register_openable`, ADR-047 §3 as amended 2026-08-13 —
|
||||
/// per-connection registration). Wires `ChannelClient` (connect
|
||||
/// side) ↔ `ChannelsAdapter` (accept side) over a real
|
||||
/// `tokio::io::duplex` carrying the channels 8-byte chunk header
|
||||
/// wire format. The accept side's `install_channel_zero` hook
|
||||
/// builds a per-connection `ChannelCore` from the adapter-supplied
|
||||
/// `ChannelManager`, registers a no-op open op
|
||||
/// (`channels/tty/sub`) via `register_openable` on a fresh
|
||||
/// per-connection registry, and runs the single-stream dispatch
|
||||
/// loop on it. The client calls `call_open_op("channels/tty/sub")`
|
||||
/// on channel 0; the wrapper does `check_open` → `open_channel` →
|
||||
/// spawn the no-op `OpenHandler` on the channel's `Connection` →
|
||||
/// respond `{ channel_id }`. Asserts the response carries a
|
||||
/// `channel_id` and that the per-identity quota was reserved
|
||||
/// (the policy's count for the caller incremented).
|
||||
///
|
||||
/// This is the gate the review identifies as missing: "register a
|
||||
/// no-op open op via `register_openable`, invoke it end-to-end
|
||||
/// through channel 0, assert `channel_id` is returned and quota is
|
||||
/// reserved." It would have caught C-02 (no way to register an
|
||||
/// open op) and C-03 (`resolve_channel_manager` stub) immediately.
|
||||
#[tokio::test]
|
||||
async fn channel_0_end_to_end_register_openable_returns_channel_id() {
|
||||
use crate::channels::operations::{ChannelCore, OpenHandler};
|
||||
use crate::channels::policy::PerIdentityChannelPolicy;
|
||||
use crate::registry::spec::ChannelOpenSpec;
|
||||
|
||||
let policy: Arc<PerIdentityChannelPolicy> = Arc::new(PerIdentityChannelPolicy::new(256));
|
||||
let policy_for_hook: Arc<dyn ChannelLifecyclePolicy> =
|
||||
Arc::clone(&policy) as Arc<dyn ChannelLifecyclePolicy>;
|
||||
let policy_for_assert = Arc::clone(&policy);
|
||||
|
||||
let open_handler: OpenHandler = Arc::new(|_input, _channel_conn, _auth| {
|
||||
tokio::spawn(async move {
|
||||
// No-op: the channel's BiStream is available via
|
||||
// `_channel_conn.accept_bi()` if the test wanted
|
||||
// to move data on it. For the open-succeeds gate,
|
||||
// just keep the task alive briefly so the handler
|
||||
// task is real (and recorded for teardown).
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
})
|
||||
});
|
||||
|
||||
let install_hook: crate::channels::adapter::InstallChannelZero =
|
||||
Arc::new(move |manager, channel0_conn, auth| {
|
||||
let open_handler = Arc::clone(&open_handler);
|
||||
let policy = Arc::clone(&policy_for_hook);
|
||||
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 core = ChannelCore::new(manager, policy);
|
||||
let mut registry = crate::registry::registration::OperationRegistry::new();
|
||||
let spec = OperationSpec::new(
|
||||
"channels/tty/sub",
|
||||
OperationType::Sub,
|
||||
Visibility::External,
|
||||
serde_json::json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"container": { "type": "string" }
|
||||
},
|
||||
"required": ["container"]
|
||||
}),
|
||||
serde_json::json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"channel_id": { "type": "integer" }
|
||||
}
|
||||
}),
|
||||
vec![],
|
||||
AccessControl::default(),
|
||||
None,
|
||||
)
|
||||
.with_channel_open(ChannelOpenSpec::new("alknet/tty"));
|
||||
core.register_openable(
|
||||
spec,
|
||||
Arc::clone(&open_handler),
|
||||
&mut registry,
|
||||
auth.clone(),
|
||||
)
|
||||
.expect("register_openable");
|
||||
|
||||
let registry = Arc::new(registry);
|
||||
let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
|
||||
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;
|
||||
})
|
||||
});
|
||||
|
||||
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(install_hook);
|
||||
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(
|
||||
"channels/tty/sub",
|
||||
serde_json::json!({ "container": "abc" }),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("channel-0 open op timed out");
|
||||
|
||||
assert!(
|
||||
response.result.is_ok(),
|
||||
"open op should succeed, got {:?}",
|
||||
response.result
|
||||
);
|
||||
let out = response.result.unwrap();
|
||||
let channel_id = out
|
||||
.get("channel_id")
|
||||
.and_then(|v| v.as_u64())
|
||||
.expect("channel_id in response");
|
||||
assert!(
|
||||
channel_id > 0,
|
||||
"channel_id should be non-zero (channel 0 is the call channel), got {channel_id}"
|
||||
);
|
||||
|
||||
let anonymous = crate::core::auth::Identity {
|
||||
id: "anonymous".to_string(),
|
||||
scopes: vec![],
|
||||
resources: Default::default(),
|
||||
};
|
||||
let count = policy_for_assert.count_for(&anonymous);
|
||||
assert_eq!(
|
||||
count, 1,
|
||||
"quota reserved: the open op incremented the per-identity count"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,12 +1,21 @@
|
||||
//! `ChannelOperationEnv` — the extension trait (ADR-047 §4) that adds
|
||||
//! the `channel_manager()` accessor to `OperationEnv` without coupling
|
||||
//! the call crate to channels types.
|
||||
//! `ChannelOperationEnv` — the extension trait (ADR-047 §4, as
|
||||
//! amended 2026-08-13) that adds the `channel_manager()` accessor to
|
||||
//! `OperationEnv` without coupling the call crate to channels types.
|
||||
//!
|
||||
//! The open-op wrapper downcasts `context.env` to
|
||||
//! `&dyn ChannelOperationEnv` at invocation time — static
|
||||
//! registration, dynamic resolution. If the downcast fails (no
|
||||
//! channels session — the op was invoked on a bare `alknet/call`
|
||||
//! connection), the wrapper returns `channel:no_channels_session`.
|
||||
//! **The open-op path does not use this trait.** Per the ADR-047 §4
|
||||
//! amendment, open ops are registered per-connection (in the
|
||||
//! `install_channel_zero` hook) with a per-connection `ChannelCore`,
|
||||
//! and the wrapper uses `ChannelCore::manager()` directly — no
|
||||
//! `context.env` downcast. The original §4's dynamic-resolution shape
|
||||
//! (downcast `context.env` to `&dyn ChannelOperationEnv` at invocation
|
||||
//! time) was unworkable: `context.env` is a `PeerCompositeEnv`, not a
|
||||
//! single concrete type that can be downcast to a channels-backed env.
|
||||
//!
|
||||
//! The trait and `ChannelsSessionEnv` impl are retained as a two-way-
|
||||
//! door implementation detail — they are not on the open-op path, but
|
||||
//! remain available for future per-connection routing (e.g., nested
|
||||
//! channels where each connection's overlay carries its own manager
|
||||
//! reference for non-open-op queries).
|
||||
//!
|
||||
//! This keeps `alkcall`'s call crate free of any channels types and
|
||||
//! preserves the layering (ADR-044). The `OperationEnv` is already the
|
||||
@@ -21,22 +30,17 @@ use crate::registry::env::OperationEnv;
|
||||
use super::manager::ChannelManager;
|
||||
|
||||
/// Extension trait that adds the per-connection `ChannelManager`
|
||||
/// accessor (ADR-047 §4). Implemented by the connection overlay in
|
||||
/// `channels-call` (the per-connection `OperationEnv` that carries a
|
||||
/// `ChannelManager` reference). The open-op wrapper downcasts
|
||||
/// `context.env` to this trait at invocation time.
|
||||
/// accessor (ADR-047 §4, as amended 2026-08-13). Implemented by the
|
||||
/// connection overlay in `channels-call` (the per-connection
|
||||
/// `OperationEnv` that carries a `ChannelManager` reference).
|
||||
///
|
||||
/// `ChannelOperationEnv: OperationEnv` — the extension is additive;
|
||||
/// any `OperationEnv` impl can also implement `ChannelOperationEnv`.
|
||||
/// A bare `alknet/call` connection's env does NOT implement this
|
||||
/// trait, so the downcast returns `None` and the wrapper returns
|
||||
/// `channel:no_channels_session`.
|
||||
/// **Not on the open-op path** — see the module doc. Retained for
|
||||
/// future per-connection routing.
|
||||
#[async_trait::async_trait]
|
||||
pub trait ChannelOperationEnv: OperationEnv {
|
||||
/// The per-connection `ChannelManager` for this channels session.
|
||||
/// `None` if the env is not channels-backed (a bare call
|
||||
/// connection); the open-op wrapper returns
|
||||
/// `channel:no_channels_session` in that case.
|
||||
/// connection).
|
||||
fn channel_manager(&self) -> Option<&ChannelManager>;
|
||||
}
|
||||
|
||||
@@ -101,32 +105,6 @@ impl ChannelOperationEnv for ChannelsSessionEnv {
|
||||
}
|
||||
}
|
||||
|
||||
/// Downcast `env` to `&dyn ChannelOperationEnv` and return the
|
||||
/// `ChannelManager`, or `None` if `env` is not channels-backed. The
|
||||
/// open-op wrapper uses this to resolve the per-connection manager at
|
||||
/// invocation time (ADR-047 §4).
|
||||
pub fn resolve_channel_manager(
|
||||
env: &Arc<dyn OperationEnv + Send + Sync>,
|
||||
) -> Option<ChannelManager> {
|
||||
// We can't do a real downcast on a trait object without
|
||||
// `AnyName`-style machinery. The practical approach: the
|
||||
// `channels-call` assembly layer wraps the env in a
|
||||
// `ChannelsSessionEnv` and provides the `ChannelManager` through
|
||||
// a side channel (e.g., a `OnceLock` on the connection, or a
|
||||
// dedicated accessor on the overlay). For the open-op wrapper
|
||||
// pattern, the manager is passed at registration time via the
|
||||
// `ChannelCore` (ADR-047 §3), not resolved from the env at
|
||||
// invocation time.
|
||||
//
|
||||
// This function is kept as the API surface for the resolution
|
||||
// pattern; the implementation uses the `ChannelCore`'s stored
|
||||
// manager instead. The trait exists for future per-connection
|
||||
// routing (e.g., nested channels where each connection's overlay
|
||||
// carries its own manager).
|
||||
let _ = env;
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -147,12 +125,6 @@ mod tests {
|
||||
let manager = ChannelManager::with_defaults(handle, None);
|
||||
let env = ChannelsSessionEnv { base, manager };
|
||||
assert!(env.channel_manager().is_some());
|
||||
// `LocalOperationEnv::contains` returns true by default (the
|
||||
// registry's reachability check is at invoke time, not at
|
||||
// `contains`). We check the `channel_manager` accessor and the
|
||||
// delegation shape — the contains behavior is tested in the
|
||||
// env.rs unit tests for each OperationEnv impl.
|
||||
assert!(env.contains("anything") || !env.contains("anything"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -199,6 +199,33 @@ impl ChannelManager {
|
||||
Ok((channel_id, send, recv))
|
||||
}
|
||||
|
||||
/// Install the handler task for a channel after it has been
|
||||
/// opened via [`ChannelManager::open_channel`] with
|
||||
/// `handler_task: None`. The open-op wrapper (ADR-047 §3) allocates the channel first (to
|
||||
/// get the `MpscSendStream`/`MpscRecvStream` for the channel's
|
||||
/// `Connection`), then spawns the ALPN's `OpenHandler` on that
|
||||
/// `Connection`, then records the resulting `JoinHandle` here so
|
||||
/// `channel/close` / connection drop can abort it.
|
||||
///
|
||||
/// Returns `UnknownChannel` if the channel was closed between
|
||||
/// `open_channel` and this call (a race — the connection dropped
|
||||
/// mid-open). The caller should treat this as the open having
|
||||
/// been superseded by teardown.
|
||||
pub fn set_handler_task(
|
||||
&self,
|
||||
channel_id: u32,
|
||||
task: JoinHandle<()>,
|
||||
) -> Result<(), ManagerError> {
|
||||
let mut channels = self.inner.channels.lock();
|
||||
match channels.get_mut(&channel_id) {
|
||||
Some(state) => {
|
||||
state.handler_task = Some(task);
|
||||
Ok(())
|
||||
}
|
||||
None => Err(ManagerError::UnknownChannel(channel_id)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Install channel 0 (pre-negotiated as `alknet/call`, ADR-036).
|
||||
/// Channel 0 is special only in that it's pre-allocated (by
|
||||
/// `channels-call`); the `ChannelsAdapter` hands the resulting
|
||||
|
||||
@@ -2,13 +2,14 @@
|
||||
//! `channel/control`, `channel/resources/subscribe` on the call
|
||||
//! `OperationRegistry` (ADR-037, amended by ADR-047 — `channel/open`
|
||||
//! dissolves into per-ALPN ops registered by the ALPN crates via
|
||||
//! `ChannelCore`).
|
||||
//! [`ChannelCore::register_openable`]).
|
||||
//!
|
||||
//! The generic ops (close, control, resources/subscribe) are keyed by
|
||||
//! `channel_id` and stay in `channels-call`. The per-ALPN open ops
|
||||
//! (`channels/<alpn>/sub`, `channels/<alpn>/pub`) are registered by
|
||||
//! the ALPN crates via [`ChannelCore::register_openable`] (ADR-047
|
||||
//! §3).
|
||||
//! the ALPN crates via [`ChannelCore::register_openable`] (ADR-047 §3,
|
||||
//! as amended 2026-08-13 — per-connection registration; see the ADR's
|
||||
//! "Amendment (§4 per-connection registration, 2026-08-13)").
|
||||
//!
|
||||
//! See `docs/architecture/channel-operations.md` for the spec.
|
||||
|
||||
@@ -16,8 +17,8 @@ use std::sync::Arc;
|
||||
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use crate::core::auth::Identity;
|
||||
use crate::core::types::Capabilities;
|
||||
use crate::core::auth::{AuthContext, Identity};
|
||||
use crate::core::types::{Capabilities, Connection};
|
||||
use crate::protocol::wire::{CallError, ResponseEnvelope};
|
||||
use crate::registry::context::OperationContext;
|
||||
use crate::registry::registration::{
|
||||
@@ -26,7 +27,7 @@ use crate::registry::registration::{
|
||||
};
|
||||
use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
|
||||
|
||||
use super::manager::ChannelManager;
|
||||
use super::manager::{ChannelManager, ManagerError};
|
||||
use super::policy::{ChannelError, ChannelLifecyclePolicy};
|
||||
|
||||
/// The names of the generic channel lifecycle operations (ADR-037,
|
||||
@@ -278,15 +279,50 @@ fn make_resources_subscribe_handler(manager: ChannelManager) -> StreamingHandler
|
||||
}
|
||||
|
||||
/// `ChannelCore` — the channel machinery the ALPN crate's open-op
|
||||
/// wrapper uses (ADR-047 §3). Provides `register_openable`, which
|
||||
/// wraps the ALPN's open handler with channel-id allocation,
|
||||
/// wrapper uses (ADR-047 §3). Provides [`ChannelCore::register_openable`],
|
||||
/// which wraps the ALPN's open handler with channel-id allocation,
|
||||
/// `ChannelManager` integration, opener-ledger recording,
|
||||
/// `ChannelLifecyclePolicy` consultation, and teardown hooks.
|
||||
///
|
||||
/// Per the ADR-047 §4 amendment (2026-08-13), a `ChannelCore` is
|
||||
/// constructed **per channels connection** (in the
|
||||
/// `install_channel_zero` hook) and `register_openable` is called on
|
||||
/// that connection's overlay `OperationRegistry` (Layer 2 per
|
||||
/// ADR-019). The wrapper closes over the per-connection `ChannelCore`
|
||||
/// and uses [`ChannelCore::manager`] directly — no `context.env`
|
||||
/// downcast (the dynamic-resolution shape from the original §4 was
|
||||
/// unworkable: `context.env` is a `PeerCompositeEnv`, not a single
|
||||
/// concrete type that can be downcast to a channels-backed env).
|
||||
pub struct ChannelCore {
|
||||
manager: ChannelManager,
|
||||
policy: Arc<dyn ChannelLifecyclePolicy>,
|
||||
}
|
||||
|
||||
/// The ALPN-specific open handler (ADR-047 §3). The ALPN crate
|
||||
/// provides this; [`ChannelCore::register_openable`] wraps it with the
|
||||
/// channel machinery. The handler receives the open op's `input`
|
||||
/// (params — e.g. which container for tty, which target for tunnel),
|
||||
/// the channel's [`Connection`] (carrying the data-plane ALPN, built
|
||||
/// by the wrapper from the channel's reassembled read half + mux write
|
||||
/// half), and the peer's [`AuthContext`], spawns its protocol on the
|
||||
/// channel's `BiStream`, and returns the `JoinHandle` so the wrapper
|
||||
/// can record it for teardown (abort on `channel/close` / connection
|
||||
/// drop).
|
||||
///
|
||||
/// This mirrors the `InstallChannelZero` hook shape (the call adapter
|
||||
/// spawns `run_loop_single_stream` on channel 0's `Connection`): the
|
||||
/// ALPN handler does the same for its data-plane protocol on the
|
||||
/// allocated channel's `Connection`. The handler owns its protocol's
|
||||
/// sub-stream multiplexing on the `BiStream` it receives (ADR-035).
|
||||
pub type OpenHandler =
|
||||
Arc<dyn Fn(Value, Connection, AuthContext) -> tokio::task::JoinHandle<()> + Send + Sync>;
|
||||
|
||||
/// The error code prefix for channel-open failures (ADR-047 §3). The
|
||||
/// wrapper maps `ChannelError` / `ManagerError` to `CallError` with
|
||||
/// codes prefixed by `channel:` so the initiator can branch on the
|
||||
/// failure reason.
|
||||
const CHANNEL_ERROR_PREFIX: &str = "channel:";
|
||||
|
||||
impl ChannelCore {
|
||||
pub fn new(manager: ChannelManager, policy: Arc<dyn ChannelLifecyclePolicy>) -> Self {
|
||||
Self { manager, policy }
|
||||
@@ -296,6 +332,10 @@ impl ChannelCore {
|
||||
&self.manager
|
||||
}
|
||||
|
||||
pub fn policy(&self) -> &Arc<dyn ChannelLifecyclePolicy> {
|
||||
&self.policy
|
||||
}
|
||||
|
||||
/// Check the per-identity cap (ADR-047 §7). Called by the
|
||||
/// open-op wrapper after `AccessControl::check` and before
|
||||
/// allocation.
|
||||
@@ -308,6 +348,316 @@ impl ChannelCore {
|
||||
pub fn on_close(&self, opener: &Identity) {
|
||||
self.policy.on_close(opener)
|
||||
}
|
||||
|
||||
/// Register a per-ALPN open op on the call `OperationRegistry`
|
||||
/// (ADR-047 §3, as amended 2026-08-13 — per-connection
|
||||
/// registration). The ALPN crate provides the `OperationSpec`
|
||||
/// (with the `channel_open` marker set, `access_control`,
|
||||
/// `input_schema`, `resource_id_path`) and the [`OpenHandler`]
|
||||
/// (ALPN-specific work — validate params, prepare the backend,
|
||||
/// spawn the protocol on the channel's `BiStream`). This wrapper
|
||||
/// does the channel machinery: ACL check (run by the registry's
|
||||
/// `invoke`/`invoke_streaming`/`invoke_sink` before the wrapper) →
|
||||
/// `check_open(identity)` → `manager.open_channel(alpn, opener)`
|
||||
/// → spawn the ALPN handler on the channel's `Connection` →
|
||||
/// respond with `{ "channel_id": <id> }`.
|
||||
///
|
||||
/// The op is registered on the given `registry` (the connection
|
||||
/// overlay registry, Layer 2 per ADR-019 — this is the
|
||||
/// per-connection registration the §4 amendment blesses). The
|
||||
/// `auth` is the peer's `AuthContext`, captured at
|
||||
/// `install_channel_zero` time and closed over by the wrapper so
|
||||
/// the ALPN handler receives it without the wrapper having to
|
||||
/// reach into `OperationContext` for it.
|
||||
///
|
||||
/// **Op type:** the spec's `op_type` determines the
|
||||
/// `HandlerKind`. For `Query`/`Mutation` the wrapper is a `Once`
|
||||
/// handler returning `{ channel_id }`. For `Sub` the wrapper is a
|
||||
/// `Stream` handler emitting one `{ channel_id }` envelope and
|
||||
/// completing (the data plane flows on the channel's `BiStream`,
|
||||
/// not in the call response stream). For `Pub` the wrapper is a
|
||||
/// `Sink` handler — **not yet implemented** (requires the
|
||||
/// channel-adoption path, C-08/Unit 5; the stub returns
|
||||
/// `channel:pub_open_not_implemented`). The `OpenHandler` is
|
||||
/// spawned in the `Query`/`Mutation`/`Sub` cases.
|
||||
pub fn register_openable(
|
||||
&self,
|
||||
spec: OperationSpec,
|
||||
open_handler: OpenHandler,
|
||||
registry: &mut OperationRegistry,
|
||||
auth: AuthContext,
|
||||
) -> Result<(), String> {
|
||||
let op_type = spec.op_type;
|
||||
let channel_open = spec.channel_open.clone();
|
||||
if channel_open.is_none() {
|
||||
return Err(format!(
|
||||
"register_openable: spec `{}` has no channel_open marker — \
|
||||
use `OperationSpec::with_channel_open` to set it (ADR-047 §2)",
|
||||
spec.name
|
||||
));
|
||||
}
|
||||
let alpn = channel_open.expect("checked above").alpn.to_string();
|
||||
|
||||
let manager = self.manager.clone();
|
||||
let policy = Arc::clone(&self.policy);
|
||||
let open_handler = Arc::clone(&open_handler);
|
||||
|
||||
let handler_kind = match op_type {
|
||||
OperationType::Query | OperationType::Mutation => HandlerKind::Once(
|
||||
make_open_handler_once(manager, policy, open_handler, auth, alpn),
|
||||
),
|
||||
OperationType::Sub => HandlerKind::Stream(make_open_handler_stream(
|
||||
manager,
|
||||
policy,
|
||||
open_handler,
|
||||
auth,
|
||||
alpn,
|
||||
)),
|
||||
OperationType::Pub => HandlerKind::Sink(make_open_handler_sink(
|
||||
manager,
|
||||
policy,
|
||||
open_handler,
|
||||
auth,
|
||||
alpn,
|
||||
)),
|
||||
};
|
||||
|
||||
registry.register(HandlerRegistration::new(
|
||||
spec,
|
||||
handler_kind,
|
||||
OperationProvenance::Local,
|
||||
None,
|
||||
None,
|
||||
Capabilities::new(),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// The opener identity for the per-identity cap check. The direct
|
||||
/// caller (the peer that opened this channels connection) is the
|
||||
/// identity whose cap is consulted; `forwarded_for` is metadata and
|
||||
/// is NOT consulted (ADR-026, ADR-047 §7). Falls back to a synthetic
|
||||
/// "anonymous" identity when the call has no resolved identity (the
|
||||
/// cap still applies — anonymous opens count against the anonymous
|
||||
/// identity's cap, which is 256 by default).
|
||||
fn opener_identity_from_context(ctx: &OperationContext) -> Identity {
|
||||
ctx.identity.clone().unwrap_or_else(|| Identity {
|
||||
id: "anonymous".to_string(),
|
||||
scopes: vec![],
|
||||
resources: Default::default(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Run the open-op wrapper's shared post-ACL steps: `check_open` →
|
||||
/// `open_channel` → build the channel `Connection` → spawn the ALPN
|
||||
/// handler → return `{ channel_id }` on success, or a `CallError` on
|
||||
/// any failure. The ACL check has already run (by the registry's
|
||||
/// `invoke`/`invoke_streaming`/`invoke_sink` before the wrapper); this
|
||||
/// function is the `check_open` → `allocate` → `ledger` → `spawn` →
|
||||
/// `respond` tail of the ADR-047 §3 flow.
|
||||
///
|
||||
/// `input` is the open op's input (params), passed through to the
|
||||
/// ALPN's `OpenHandler` so it can validate params and prepare the
|
||||
/// backend. `opener_id` is the `PeerId` recorded in the opener ledger
|
||||
/// (the direct caller's identity id).
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn run_open_wrapper(
|
||||
manager: &ChannelManager,
|
||||
policy: &Arc<dyn ChannelLifecyclePolicy>,
|
||||
open_handler: &OpenHandler,
|
||||
auth: &AuthContext,
|
||||
alpn: &str,
|
||||
input: Value,
|
||||
opener_id: String,
|
||||
opener_identity: Identity,
|
||||
request_id: String,
|
||||
) -> ResponseEnvelope {
|
||||
if let Err(channel_err) = policy.check_open(&opener_identity) {
|
||||
return ResponseEnvelope::error(request_id, map_channel_error_to_call_error(&channel_err));
|
||||
}
|
||||
|
||||
let channel_id = match manager.open_channel(alpn, opener_id, None).await {
|
||||
Ok((id, send, recv)) => {
|
||||
let remote_addr = manager.remote_addr();
|
||||
let source = super::source::channel_source(recv, send, remote_addr);
|
||||
let channel_conn = Connection::from_source(source, alpn.as_bytes().to_vec());
|
||||
let task = open_handler(input, channel_conn, auth.clone());
|
||||
if let Err(e) = manager.set_handler_task(id, task) {
|
||||
tracing::warn!(
|
||||
channel_id = id,
|
||||
error = %e,
|
||||
"open wrapper: channel vanished between open and handler-task install"
|
||||
);
|
||||
}
|
||||
id
|
||||
}
|
||||
Err(ManagerError::TooManyChannels { count, max }) => {
|
||||
return ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::new(
|
||||
format!("{CHANNEL_ERROR_PREFIX}too_many_channels"),
|
||||
format!("per-connection channel cap exceeded: {count}/{max}"),
|
||||
false,
|
||||
)
|
||||
.with_details(json!({ "count": count, "max": max })),
|
||||
);
|
||||
}
|
||||
Err(other) => {
|
||||
return ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::new(
|
||||
format!("{CHANNEL_ERROR_PREFIX}allocation_failed"),
|
||||
format!("channel allocation failed: {other}"),
|
||||
false,
|
||||
),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
ResponseEnvelope::ok(request_id, json!({ "channel_id": channel_id }))
|
||||
}
|
||||
|
||||
/// Map a `ChannelError` (from `ChannelLifecyclePolicy::check_open`) to
|
||||
/// a `CallError` with the `channel:` error-code prefix. The
|
||||
/// per-identity cap denial maps to `channel:too_many_channels` with
|
||||
/// details carrying the identity, count, and cap.
|
||||
fn map_channel_error_to_call_error(err: &ChannelError) -> CallError {
|
||||
match err {
|
||||
ChannelError::TooManyChannels {
|
||||
identity,
|
||||
count,
|
||||
cap,
|
||||
} => CallError::new(
|
||||
format!("{CHANNEL_ERROR_PREFIX}too_many_channels"),
|
||||
format!("too many channels for identity {identity}: {count} (cap {cap})"),
|
||||
false,
|
||||
)
|
||||
.with_details(json!({
|
||||
"identity": identity,
|
||||
"count": count,
|
||||
"cap": cap,
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the `Once` handler for a `Query`/`Mutation`-typed open op
|
||||
/// (ADR-047 §1). Returns `{ channel_id }` as a single response. The
|
||||
/// ALPN `OpenHandler` is spawned on the channel's `Connection`; the
|
||||
/// handler's `JoinHandle` is recorded for teardown.
|
||||
fn make_open_handler_once(
|
||||
manager: ChannelManager,
|
||||
policy: Arc<dyn ChannelLifecyclePolicy>,
|
||||
open_handler: OpenHandler,
|
||||
auth: AuthContext,
|
||||
alpn: String,
|
||||
) -> Handler {
|
||||
Arc::new(move |input: Value, ctx: OperationContext| {
|
||||
let manager = manager.clone();
|
||||
let policy = Arc::clone(&policy);
|
||||
let open_handler = Arc::clone(&open_handler);
|
||||
let auth = auth.clone();
|
||||
let alpn = alpn.clone();
|
||||
Box::pin(async move {
|
||||
let request_id = ctx.request_id.clone();
|
||||
let opener_identity = opener_identity_from_context(&ctx);
|
||||
let opener_id = opener_identity.id.clone();
|
||||
run_open_wrapper(
|
||||
&manager,
|
||||
&policy,
|
||||
&open_handler,
|
||||
&auth,
|
||||
&alpn,
|
||||
input,
|
||||
opener_id,
|
||||
opener_identity,
|
||||
request_id,
|
||||
)
|
||||
.await
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/// Build the `Stream` handler for a `Sub`-typed open op (ADR-047 §1).
|
||||
/// Emits one `ResponseEnvelope::ok({ channel_id })` and completes —
|
||||
/// the data plane flows on the channel's `BiStream` (spawned
|
||||
/// `OpenHandler`), not in the call response stream. A `Sub`-typed open
|
||||
/// op uses the streaming dispatch path so the initiator can expect a
|
||||
/// stream of envelopes (the first carries `channel_id`; the stream
|
||||
/// ends after it, signaling "the channel is open, data flows
|
||||
/// out-of-band on the channel").
|
||||
fn make_open_handler_stream(
|
||||
manager: ChannelManager,
|
||||
policy: Arc<dyn ChannelLifecyclePolicy>,
|
||||
open_handler: OpenHandler,
|
||||
auth: AuthContext,
|
||||
alpn: String,
|
||||
) -> StreamingHandler {
|
||||
Arc::new(move |input: Value, ctx: OperationContext| {
|
||||
let manager = manager.clone();
|
||||
let policy = Arc::clone(&policy);
|
||||
let open_handler = Arc::clone(&open_handler);
|
||||
let auth = auth.clone();
|
||||
let alpn = alpn.clone();
|
||||
Box::pin(futures::stream::once(async move {
|
||||
let request_id = ctx.request_id.clone();
|
||||
let opener_identity = opener_identity_from_context(&ctx);
|
||||
let opener_id = opener_identity.id.clone();
|
||||
run_open_wrapper(
|
||||
&manager,
|
||||
&policy,
|
||||
&open_handler,
|
||||
&auth,
|
||||
&alpn,
|
||||
input,
|
||||
opener_id,
|
||||
opener_identity,
|
||||
request_id,
|
||||
)
|
||||
.await
|
||||
}))
|
||||
})
|
||||
}
|
||||
|
||||
/// Build the `Sink` handler for a `Pub`-typed open op (ADR-047 §1).
|
||||
///
|
||||
/// **Not yet implemented.** The `Pub`-typed open op has a design
|
||||
/// question the `Query`/`Mutation`/`Sub` paths do not: for `Pub`, the
|
||||
/// initiator streams data *to* the responder via `call.published`
|
||||
/// events, and the question is whether that data flows through the
|
||||
/// allocated channel's `BiStream` (the wrapper writes; the
|
||||
/// `OpenHandler` reads from the channel) or is consumed directly by
|
||||
/// the `OpenHandler` (making the channel redundant for `Pub`). The
|
||||
/// `Query`/`Mutation`/`Sub` paths allocate the channel and hand the
|
||||
/// whole `Connection` to the `OpenHandler` — the data plane is
|
||||
/// out-of-band on the channel, not in the call stream. The `Pub` path
|
||||
/// needs the same property but with the initiator writing to the
|
||||
/// channel from the connect side, which requires the channel-adoption
|
||||
/// path (C-08, Unit 5) to be in place first. This is a follow-on; the
|
||||
/// stub returns `INTERNAL` with `channel:pub_open_not_implemented` so
|
||||
/// callers fail loudly rather than silently.
|
||||
fn make_open_handler_sink(
|
||||
_manager: ChannelManager,
|
||||
_policy: Arc<dyn ChannelLifecyclePolicy>,
|
||||
_open_handler: OpenHandler,
|
||||
_auth: AuthContext,
|
||||
_alpn: String,
|
||||
) -> crate::registry::registration::SinkHandler {
|
||||
Arc::new(
|
||||
move |_input: Value,
|
||||
ctx: OperationContext,
|
||||
_publish_stream: crate::registry::registration::PublishStream| {
|
||||
let request_id = ctx.request_id.clone();
|
||||
Box::pin(async move {
|
||||
ResponseEnvelope::error(
|
||||
request_id,
|
||||
CallError::internal(
|
||||
"channel:pub_open_not_implemented: Pub-typed open ops are not yet \
|
||||
implemented (requires channel-adoption path, C-08/Unit 5)",
|
||||
),
|
||||
)
|
||||
})
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
Reference in New Issue
Block a user