feat(websocket): data-channel + op/register wiring (review 006 Unit 2+3)

The WS path wires the alkcall 0.3 per-session mechanisms — the OQ-05
deferred half (review-003 WS-20/21/22/25/26, the decisions WS-24/WS-25
resolved upstream):

- install_channel_zero reworked per the ADR-047 §4 amendment #2 shape:
  fork the deployment's base registry, register the generic channel
  ops (channel/close, channel/control, channel/resources/subscribe —
  WS-21), the deployment's openable ALPNs (ChannelCore::register_openable,
  WS-22), the bootstrap discovery set closed over the fork
  (install_bootstrap_discovery, F-06), and op/register (WS-25; the
  collision set is the fork per review-005 G-03), then dispatch over
  the fork. The session's ChannelsPolicy rides the hook (one policy
  instance across open wrappers and the demux teardown path).
- OpenableAlpn { spec, open_handler } + HttpAdapter::with_ws_openable_alpns,
  threaded RouterState -> SessionState -> hook, with the OpenableAlpns
  request-extension fallback (mirroring ChannelsPolicy/WsTimeouts).
- WsSessions retains the channel-0 Arc<CallConnection> (WS-26) with a
  self-removing guard (ConnectionGuard); live_connections() is the
  deployment-visible surface.
- UP-01: ALREADY_EXISTS maps to 409 Conflict in the gateway error map.
- from_wss import excludes the protocol-session ops (bootstrap set +
  channel lifecycle ops): the fork serves them per session, and proxying
  session-scoped machinery (e.g. channel/close across sessions) would be
  nonsense. Discovery runs first, the filter is the listing minus those
  names.
- adapter_install_channel_zero cfg matches its caller (WS-27); it
  inherits the reworked hook (session ops now served in the from_wss
  test-server producer too).

Gates (Unit 3, tests/ws_upgrade_session.rs; the WS-23 e2e set):
open -> channel_id -> discoverable in services/list -> chunks both
ways -> handler sees bytes; channel/close resolves + ledger decrement;
cap denial (channel:-prefixed); mid-open disconnect teardown; TooLarge
demux resync through the WS path (16 MiB + 1 skip consumed);
op/register announce + overlay-collision + serving-registry-collision
ALREADY_EXISTS through the WS path.

call_and_await now filters by request id and tolerates data-channel
chunks (a prior Sub's trailing call.completed may interleave).

Verification: cargo test 454 (default) / 582 (all-features), clippy
both sides -D warnings clean, fmt clean, doc clean.
This commit is contained in:
glm-5.3-flash committed 2026-09-04 15:47:15 +00:00
1 parent 90790374fa
commit 030c5efa51
8 files changed
+891 -28

No files matched your search

+57
View File
@@ -323,6 +323,28 @@ impl WssSession {
}
}
/// The per-session protocol ops a channels session serves on channel 0
/// (alkcall ADR-022 amendment's bootstrap set + the channel lifecycle
/// set): a `from_wss` import must not proxy these — they are
/// session-scoped machinery, not domain operations. Importing
/// `channel/close` would forward a close for *this* session's channel
/// ids to the remote (nonsense: ids are per-session), and
/// `op/register`/`services/list-peers` are registration/discovery
/// machinery whose proxy duplicates the import itself. The
/// `services/list` + `services/schema` discovery pair is excluded
/// from import for the same reason (the importer calls them on every
/// import; a proxy of them is dead weight), matching what the pure
/// `from_call` path does — its own discovery dials the real ones.
fn is_protocol_session_op(name: &str) -> bool {
name == "services/list"
|| name == "services/schema"
|| name == "services/list-peers"
|| name == alkcall::registry::op_register::OP_REGISTER_NAME
|| name == alkcall::channels::operations::OP_CHANNEL_CLOSE
|| name == alkcall::channels::operations::OP_CHANNEL_CONTROL
|| name == alkcall::channels::operations::OP_CHANNEL_RESOURCES_SUBSCRIBE
}
#[async_trait::async_trait]
impl OperationAdapter for FromWss {
async fn import(
@@ -334,10 +356,15 @@ impl OperationAdapter for FromWss {
self.allow_plaintext,
)
.await?;
// The protocol-session ops ride the listing (the session fork
// serves them) but are excluded: the importer builds an
// operation_filter from the listing minus those names.
let config = match &self.namespace {
Some(ns) => FromCallConfig::new().with_namespace_prefix(ns),
None => FromCallConfig::new(),
};
let config = config
.with_operation_filter(protocol_session_ops_from(&session.call_connection).await?);
let bundles = import_from_call(&session.call_connection, config).await;
// Fire-and-forget: the imported handlers keep working off the
// session's Arc'd CallConnection; the session's tasks are
@@ -348,6 +375,36 @@ impl OperationAdapter for FromWss {
}
}
/// The remote's `services/list` names minus the protocol-session ops —
/// the `operation_filter` the import runs with. A listing failure is
/// `DiscoveryFailed` from `from_call` itself (the real error path); a
/// parse failure of the names array is a transport-level parse error.
async fn protocol_session_ops_from(
connection: &CallConnection,
) -> Result<std::collections::HashSet<String>, AdapterError> {
let response = connection
.call("services/list", serde_json::json!({}))
.await;
let output = response.result.map_err(|e| AdapterError::DiscoveryFailed {
message: format!("services/list failed: {} ({})", e.code, e.message),
})?;
let ops = output
.get("operations")
.and_then(|v| v.as_array())
.ok_or_else(|| AdapterError::SchemaParse {
message: "services/list response missing 'operations' array".to_string(),
})?;
let mut filter = std::collections::HashSet::new();
for op in ops {
if let Some(name) = op.get("name").and_then(|v| v.as_str()) {
if !is_protocol_session_op(name) {
filter.insert(name.to_string());
}
}
}
Ok(filter)
}
#[cfg(all(test, feature = "server"))]
mod tests {
use super::*;
+14 -3
View File
@@ -2,9 +2,11 @@
//!
//! Protocol-level vs operation-level code distinction: protocol codes
//! (`NOT_FOUND`, `FORBIDDEN`, `INVALID_INPUT`, `INVALID_OPERATION_TYPE`,
//! `TIMEOUT`, `INTERNAL`) map to fixed statuses; operation-level codes
//! imported from external HTTP APIs are prefixed `HTTP_<status>` and map
//! to their declared status.
//! `TIMEOUT`, `ALREADY_EXISTS` — the seventh code, alkcall ADR-022's
//! `op/register` collision policy, mapped 409 Conflict, review-006
//! UP-01) map to fixed statuses; operation-level codes imported from
//! external HTTP APIs are prefixed `HTTP_<status>` and map to their
//! declared status.
//!
//! The identity-aware variant maps the ambiguous protocol codes
//! (`FORBIDDEN`, `INVALID_OPERATION_TYPE`) to `401` when no token
@@ -27,6 +29,7 @@ const PROTOCOL_CODE_FORBIDDEN: &str = "FORBIDDEN";
const PROTOCOL_CODE_INVALID_INPUT: &str = "INVALID_INPUT";
const PROTOCOL_CODE_INVALID_OPERATION_TYPE: &str = "INVALID_OPERATION_TYPE";
const PROTOCOL_CODE_TIMEOUT: &str = "TIMEOUT";
const PROTOCOL_CODE_ALREADY_EXISTS: &str = "ALREADY_EXISTS";
const PROTOCOL_CODE_INTERNAL: &str = "INTERNAL";
const HTTP_PREFIX: &str = "HTTP_";
@@ -36,6 +39,7 @@ const STATUS_UNAUTHORIZED: u16 = 401;
const STATUS_FORBIDDEN: u16 = 403;
const STATUS_UNPROCESSABLE: u16 = 422;
const STATUS_TIMEOUT: u16 = 504;
const STATUS_CONFLICT: u16 = 409;
const STATUS_INTERNAL: u16 = 500;
const RETRY_AFTER_STATUSES: &[u16] = &[429, 503];
@@ -72,6 +76,7 @@ pub fn call_error_to_http_status_with_identity(
}
}
PROTOCOL_CODE_TIMEOUT => STATUS_TIMEOUT,
PROTOCOL_CODE_ALREADY_EXISTS => STATUS_CONFLICT,
PROTOCOL_CODE_INTERNAL => STATUS_INTERNAL,
code if code.starts_with(HTTP_PREFIX) => code[HTTP_PREFIX.len()..]
.parse::<u16>()
@@ -182,6 +187,12 @@ mod tests {
assert_eq!(call_error_to_http_status(&error), 504);
}
#[test]
fn already_exists_maps_to_409() {
let error = CallError::already_exists("name taken");
assert_eq!(call_error_to_http_status(&error), 409);
}
#[test]
fn internal_maps_to_500() {
let error = CallError::internal("boom");
+1
View File
@@ -1137,6 +1137,7 @@ mod tests {
crate::websocket::DEFAULT_WS_MAX_SESSIONS,
)),
ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
ws_openable_alpns: None,
};
let auth_state = Arc::clone(&provider);
gateway_router()
+46
View File
@@ -98,6 +98,7 @@ pub struct HttpAdapter {
ws_max_sessions: usize,
ws_session_slots: Arc<tokio::sync::Semaphore>,
ws_idle_timeout: Option<Duration>,
ws_openable_alpns: Option<Arc<[crate::websocket::OpenableAlpn]>>,
}
impl HttpAdapter {
@@ -136,6 +137,7 @@ impl HttpAdapter {
ws_sessions: Arc::clone(&ws_sessions),
ws_session_slots: Arc::clone(&ws_session_slots),
ws_idle_timeout,
ws_openable_alpns: None,
};
let router = build_router(state, None);
Self {
@@ -150,6 +152,7 @@ impl HttpAdapter {
ws_max_sessions,
ws_session_slots,
ws_idle_timeout,
ws_openable_alpns: None,
}
}
@@ -165,6 +168,7 @@ impl HttpAdapter {
ws_sessions: Arc::clone(&self.ws_sessions),
ws_session_slots: Arc::clone(&self.ws_session_slots),
ws_idle_timeout: self.ws_idle_timeout,
ws_openable_alpns: self.ws_openable_alpns.clone(),
};
// `extra_routes` is borrowed, not consumed (SRV-05): a builder
// call after `with_extra_routes` must keep the custom routes in
@@ -186,6 +190,7 @@ impl HttpAdapter {
ws_sessions: Arc::clone(&self.ws_sessions),
ws_session_slots: Arc::clone(&self.ws_session_slots),
ws_idle_timeout: self.ws_idle_timeout,
ws_openable_alpns: self.ws_openable_alpns.clone(),
};
self.router = build_router(state, Some(routes.clone()));
self.extra_routes = Some(routes);
@@ -209,6 +214,7 @@ impl HttpAdapter {
ws_sessions: Arc::clone(&self.ws_sessions),
ws_session_slots: Self::rebuild_session_slots(max_sessions),
ws_idle_timeout: self.ws_idle_timeout,
ws_openable_alpns: self.ws_openable_alpns.clone(),
};
self.router = build_router(state, self.extra_routes.clone());
self
@@ -240,6 +246,7 @@ impl HttpAdapter {
ws_sessions: Arc::clone(&self.ws_sessions),
ws_session_slots: Arc::clone(&self.ws_session_slots),
ws_idle_timeout: self.ws_idle_timeout,
ws_openable_alpns: self.ws_openable_alpns.clone(),
};
self.router = build_router(state, self.extra_routes.clone());
self
@@ -253,6 +260,44 @@ impl HttpAdapter {
Arc::clone(&self.ws_sessions)
}
/// The openable-ALPN set for WS sessions (WS-22, review 006
/// Unit 2): one [`OpenableAlpn`](crate::websocket::OpenableAlpn)
/// per openable data-channel ALPN — the open-op spec (with the
/// `channel_open` marker) and the ALPN-specific
/// [`OpenHandler`](crate::websocket::OpenHandler). Each WS
/// session's per-session fork registers the set (plus the generic
/// channel ops, bootstrap discovery, and `op/register`), so a WS
/// client can open data channels exactly as any channels consumer
/// (ADR-067: the browser reaches what a Rust consumer on an
/// in-line connection would reach).
///
/// The ALPN-specific `OpenHandler` implementations stay in the
/// ALPN crates (alktty et al.); this adapter only ferries the
/// registrations onto the session fork. Bare-registry / custom
/// upgrade routes pass their own set via the
/// [`OpenableAlpns`](crate::websocket::OpenableAlpns) request
/// extension instead.
///
/// Default: no openables (channel 0 only — the pre-Unit-2 shape).
pub fn with_ws_openable_alpns(
mut self,
openables: Vec<crate::websocket::OpenableAlpn>,
) -> Self {
self.ws_openable_alpns = Some(openables.into());
let state = RouterState {
registry: Arc::clone(&self.registry),
identity_provider: Arc::clone(&self.identity_provider),
decoy: self.decoy.clone(),
openapi_doc: self.openapi_doc.clone(),
ws_sessions: Arc::clone(&self.ws_sessions),
ws_session_slots: Arc::clone(&self.ws_session_slots),
ws_idle_timeout: self.ws_idle_timeout,
ws_openable_alpns: self.ws_openable_alpns.clone(),
};
self.router = build_router(state, self.extra_routes.clone());
self
}
/// A fresh semaphore for the new cap; the retained handles of
/// already-open sessions are unaffected (they hold their permits).
fn rebuild_session_slots(max_sessions: usize) -> Arc<tokio::sync::Semaphore> {
@@ -1153,6 +1198,7 @@ mod tests {
crate::websocket::DEFAULT_WS_MAX_SESSIONS,
)),
ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
ws_openable_alpns: None,
}
}
+7
View File
@@ -48,6 +48,11 @@ pub(crate) struct RouterState {
pub(crate) ws_session_slots: Arc<tokio::sync::Semaphore>,
/// Idle-read timeout for the WS pumps (WS-01); `None` disables.
pub(crate) ws_idle_timeout: Option<std::time::Duration>,
/// The openable-ALPN set for WS sessions (WS-22): the per-ALPN
/// open-op specs + handlers registered on each session's fork.
/// `None` (the default) declares no openables — a WS session then
/// carries no data-channel open ops (the pre-Unit-2 shape).
pub(crate) ws_openable_alpns: Option<Arc<[crate::websocket::OpenableAlpn]>>,
}
impl axum::extract::FromRef<RouterState> for crate::websocket::SessionState {
@@ -57,6 +62,7 @@ impl axum::extract::FromRef<RouterState> for crate::websocket::SessionState {
Arc::clone(&state.ws_sessions),
Arc::clone(&state.ws_session_slots),
state.ws_idle_timeout,
state.ws_openable_alpns.clone(),
)
}
}
@@ -96,6 +102,7 @@ mod tests {
crate::websocket::DEFAULT_WS_MAX_SESSIONS,
)),
ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
ws_openable_alpns: None,
};
let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state);
assert!(matches!(extracted, DecoyConfig::Redirect { .. }));
+5 -4
View File
@@ -30,15 +30,16 @@ pub use byte_adapter::{
split_ws_to_bytes, split_ws_to_bytes_idle, split_ws_to_bytes_idle_with_write,
};
#[cfg(feature = "server")]
pub use alkcall::channels::operations::ChannelCore;
#[cfg(any(test, feature = "wss"))]
pub use byte_adapter::split_tungstenite_to_bytes;
#[cfg(feature = "server")]
#[allow(unused_imports)]
#[cfg(all(test, feature = "server", feature = "wss"))]
pub(crate) use upgrade::adapter_install_channel_zero;
#[cfg(feature = "server")]
pub use upgrade::{
run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy, SessionState,
WsSessions, WsTimeouts, DEFAULT_WS_MAX_SESSIONS,
run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy, OpenableAlpn,
OpenableAlpns, SessionState, WsSessions, WsTimeouts, DEFAULT_WS_MAX_SESSIONS,
};
#[cfg(all(any(test, feature = "test-support"), feature = "server"))]
+222 -17
View File
@@ -6,9 +6,14 @@
//! `resolve_from_token` path as any HTTP request), then upgrade →
//! WS↔byte-stream adapter → `Connection::from_bidi(ws_stream,
//! b"alk/channels")` → alkcall `ChannelsAdapter` (the channels accept
//! path). The `install_channel_zero` hook constructs channel 0's
//! path). The `install_channel_zero` hook forks the deployment's base
//! registry, registers the per-session ops on the fork (the generic
//! channel lifecycle ops, the deployment's openable ALPNs, the
//! bootstrap discovery set, and `op/register` — alkcall ADR-047 §4
//! amendment #2 + ADR-022 amendment), constructs channel 0's
//! `CallConnection` (the identity rides the channels-layer
//! `Connection`) and runs the shared `Dispatcher::run_loop_single_stream`.
//! `Connection`), retains it in the session registry (WS-26), and
//! runs the shared `Dispatcher::run_loop_single_stream` over the fork.
//!
//! ## Detached task lifetime semantics (WS-10)
//!
@@ -41,10 +46,12 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use alkcall::channels::adapter::ChannelsAdapter;
use alkcall::channels::operations::{ChannelCore, OpenHandler};
use alkcall::channels::policy::{ChannelLifecyclePolicy, NoCap};
use alkcall::core::auth::{AuthContext, Identity};
use alkcall::core::types::{Connection, ProtocolHandler};
use alkcall::registry::registration::OperationRegistry;
use alkcall::registry::spec::OperationSpec;
use axum::extract::ws::WebSocketUpgrade;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
@@ -71,6 +78,7 @@ use super::byte_adapter::{
pub struct WsSessions {
counter: Arc<AtomicU64>,
sessions: Arc<Mutex<HashMap<u64, Arc<WsPumps>>>>,
connections: Arc<Mutex<HashMap<u64, Arc<alkcall::protocol::connection::CallConnection>>>>,
}
/// Default bound on concurrent WS sessions (WS-09): the semaphore
@@ -94,6 +102,8 @@ pub struct SessionState {
session_slots: Arc<tokio::sync::Semaphore>,
/// Idle-read timeout (WS-01): `None` disables the knob.
idle_timeout: Option<std::time::Duration>,
/// The openable-ALPN set (WS-22): `None` declares no openables.
openable_alpns: Option<Arc<[OpenableAlpn]>>,
}
impl SessionState {
@@ -107,24 +117,27 @@ impl SessionState {
sessions: Arc::new(WsSessions::new()),
session_slots: Arc::new(tokio::sync::Semaphore::new(DEFAULT_WS_MAX_SESSIONS)),
idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
openable_alpns: None,
}
}
/// From the adapter's router state: the shared [`WsSessions`]
/// instance the assembly layer can hold for eviction, and the
/// instance the assembly layer can hold for eviction, the
/// pre-built session-cap semaphore (one per `HttpAdapter`, shared
/// across requests).
/// across requests), and the deployment's openable-ALPN set.
pub(crate) fn new(
registry: Arc<OperationRegistry>,
sessions: Arc<WsSessions>,
session_slots: Arc<tokio::sync::Semaphore>,
idle_timeout: Option<std::time::Duration>,
openable_alpns: Option<Arc<[OpenableAlpn]>>,
) -> Self {
Self {
registry,
sessions,
session_slots,
idle_timeout,
openable_alpns,
}
}
@@ -139,6 +152,10 @@ impl SessionState {
pub(crate) fn idle_timeout(&self) -> Option<std::time::Duration> {
self.idle_timeout
}
pub(crate) fn openable_alpns(&self) -> Option<Arc<[OpenableAlpn]>> {
self.openable_alpns.clone()
}
}
/// `FromRef` chain: a bare `Arc<OperationRegistry>` router state lifts
@@ -173,6 +190,26 @@ impl WsSessions {
self.sessions.lock().is_empty()
}
/// The live session's channel-0 [`CallConnection`] handles (WS-26):
/// the deployment-visible surface the assembly layer (or the hub)
/// reaches browser/session-side ops through — connection-local
/// overlay composition, `announce_op`-style pushes, session
/// enumeration. The handle is registered when the session's
/// channel-0 connection is built and removed when the channel-0
/// task ends (session teardown of any path), so the registry only
/// holds live sessions.
///
/// Un-registered (default) — no handles are retained; the upgrade
/// runs exactly as the pre-WS-26 shape.
pub fn live_connections(&self) -> Vec<Arc<alkcall::protocol::connection::CallConnection>> {
self.connections.lock().values().cloned().collect()
}
/// Number of live channel-0 connection handles currently tracked.
pub fn live_connection_count(&self) -> usize {
self.connections.lock().len()
}
fn insert(&self, pumps: Arc<WsPumps>) -> u64 {
let id = self.counter.fetch_add(1, Ordering::Relaxed);
self.sessions.lock().insert(id, pumps);
@@ -182,6 +219,41 @@ impl WsSessions {
fn remove(&self, id: u64) {
self.sessions.lock().remove(&id);
}
fn insert_connection(&self, conn: Arc<alkcall::protocol::connection::CallConnection>) -> u64 {
let id = self.counter.fetch_add(1, Ordering::Relaxed);
self.connections.lock().insert(id, conn);
id
}
fn remove_connection(&self, id: u64) {
self.connections.lock().remove(&id);
}
}
/// One deployment-declared openable ALPN for the WS path (WS-22):
/// the per-ALPN open-op `OperationSpec` (with the `channel_open`
/// marker set via `OperationSpec::with_channel_open`) and the
/// ALPN-specific [`OpenHandler`] the data-plane protocol runs on the
/// allocated channel's `Connection`. The ALPN-specific handler stays
/// in the ALPN crates (alktty et al.); this crate only ferries the
/// registration onto each session's fork.
#[derive(Clone)]
pub struct OpenableAlpn {
/// The open-op spec (Query/Mutation/Sub with the `channel_open`
/// marker; a `Pub` open op resolves `channel:pub_open_not_implemented`
/// per the upstream C-08 stub until channel adoption lands).
pub spec: OperationSpec,
/// The data-plane protocol handler spawned on the allocated
/// channel's `Connection`.
pub open_handler: OpenHandler,
}
impl OpenableAlpn {
/// Declare one openable ALPN.
pub fn new(spec: OperationSpec, open_handler: OpenHandler) -> Self {
Self { spec, open_handler }
}
}
/// The channels session for an upgraded socket: adapt → `Connection`
@@ -193,7 +265,8 @@ impl WsSessions {
/// closes the read with 1001 after `d` without inbound chunk
/// progress. `write_timeout` bounds one outbound WS send (WS-18):
/// `None` = the crate default window, `Some(d)` a deployment-set
/// window.
/// window. `openable_alpns` (WS-22) is the deployment's openable-ALPN
/// set registered per session (see [`install_channel_zero`]).
#[allow(clippy::too_many_arguments)]
pub async fn run_channels_session(
socket: axum::extract::ws::WebSocket,
@@ -203,20 +276,27 @@ pub async fn run_channels_session(
sessions: Option<WsSessions>,
idle_timeout: Option<std::time::Duration>,
write_timeout: Option<std::time::Duration>,
openable_alpns: Option<Arc<[OpenableAlpn]>>,
) {
let (byte_stream, pumps) =
split_ws_to_bytes_idle_with_write(socket, idle_timeout, write_timeout);
let pumps = Arc::new(pumps);
let _guard = sessions.map(|sessions| {
let id = sessions.insert(Arc::clone(&pumps));
SessionGuard { sessions, id }
let _guard = sessions.as_ref().map(|s| {
let id = s.insert(Arc::clone(&pumps));
SessionGuard {
sessions: s.clone(),
id,
}
});
let conn = Connection::from_bidi(byte_stream, b"alk/channels".to_vec(), None);
let _ = conn.set_identity(identity.clone());
let adapter = ChannelsAdapter::new(install_channel_zero(registry), policy);
let adapter = ChannelsAdapter::new(
install_channel_zero(registry, sessions, Arc::clone(&policy), openable_alpns),
policy,
);
let auth = AuthContext {
identity: Some(identity),
alpn: b"alk/channels".to_vec(),
@@ -239,10 +319,45 @@ impl Drop for SessionGuard {
}
}
/// The `install_channel_zero` hook: split channel 0's `BiStream` into
/// the shared writer + reader, construct the single-stream
/// `CallConnection`, and run the shared `Dispatcher`'s single-stream
/// loop (the alkcall `channels/client.rs` accept-side wiring pattern).
/// The channel-0 connection handle's self-removing guard (WS-26): the
/// drop removes the retained [`CallConnection`] from the session
/// registry, so the registry only holds live sessions' handles.
struct ConnectionGuard {
sessions: WsSessions,
id: u64,
}
impl Drop for ConnectionGuard {
fn drop(&mut self) {
self.sessions.remove_connection(self.id);
}
}
/// The `install_channel_zero` hook (WS-20/21/22/25/26, review 006
/// Unit 2): fork the deployment's base registry, register the
/// per-session ops on the fork — the generic channel lifecycle ops
/// (`ChannelOperations::register_on`: `channel/close`,
/// `channel/control`, `channel/resources/subscribe`), the
/// deployment's openable ALPNs (`ChannelCore::register_openable`),
/// the bootstrap discovery set closed over the fork
/// (`install_bootstrap_discovery`, so `services/list` sees the
/// session's own openables — the F-06 shape), and `op/register`
/// (alkcall ADR-022 amendment; the collision set is the fork) — then
/// run the dispatcher over the fork (alkcall ADR-047 §4 amendment #2,
/// the per-session-fork mechanism; the alkcall e2e reference shape).
///
/// `policy` is the session's channel cap policy (the `ChannelsPolicy`
/// extension resolution, `NoCap` default): the open-op wrappers the
/// hook registers consult it (`check_open` per identity, ledger
/// decrement on teardown), and the adapter's demux loop decrements it
/// on connection-drop teardown — one policy instance across both
/// halves.
///
/// The hook also retains the channel-0 `CallConnection` in the
/// session registry (WS-26) when a shared [`WsSessions`] instance is
/// in play: the deployment reaches the session's connection-local
/// overlay and can hub→session-call announced or imported ops. The
/// handle is removed when the channel-0 task ends (any teardown path).
///
/// The dispatcher's token resolver is a no-op: the WS identity is
/// attached to the connection at upgrade time and
@@ -251,9 +366,15 @@ impl Drop for SessionGuard {
/// identity was established at upgrade time, not per-call).
fn install_channel_zero(
registry: Arc<OperationRegistry>,
sessions: Option<WsSessions>,
policy: Arc<dyn ChannelLifecyclePolicy>,
openable_alpns: Option<Arc<[OpenableAlpn]>>,
) -> alkcall::channels::adapter::InstallChannelZero {
Arc::new(move |_manager, channel0_conn, auth| {
Arc::new(move |manager, channel0_conn, auth| {
let registry = Arc::clone(&registry);
let sessions = sessions.clone();
let policy = Arc::clone(&policy);
let openable_alpns = openable_alpns.clone();
tokio::spawn(async move {
// The WS identity rides the upgrade request; propagate it to
// channel 0's `CallConnection` so the dispatcher's
@@ -276,8 +397,69 @@ fn install_channel_zero(
Arc::clone(&writer),
),
);
// The per-session dispatch registry (WS-24's decided
// mechanism): fork the base, register the session's ops on
// the fork, dispatch over the fork. Registration failures
// (a spec whose publish_schema fails to compile — the only
// error class the registry can produce here, since the
// openable specs were accepted at declaration) kill the
// session before the loop starts: a session that would
// mis-discover must not silently lose the failing op.
let fork = Arc::new(registry.fork());
let register_result: Result<(), String> = (|| {
alkcall::channels::operations::ChannelOperations::new(
manager.clone(),
Arc::clone(&policy),
)
.register_on(&fork)?;
if let Some(openables) = openable_alpns.as_ref() {
let core = ChannelCore::new(manager, Arc::clone(&policy));
for openable in openables.iter() {
core.register_openable(
openable.spec.clone(),
Arc::clone(&openable.open_handler),
&fork,
auth.clone(),
)?;
}
}
alkcall::registry::discovery::install_bootstrap_discovery(&fork)?;
fork.register(alkcall::registry::registration::HandlerRegistration::new(
alkcall::registry::op_register::op_register_spec(
alkcall::registry::spec::AccessControl::default(),
),
alkcall::registry::registration::HandlerKind::Once(
alkcall::registry::op_register::op_register_handler(
Arc::clone(&call_connection),
Arc::clone(&fork),
),
),
alkcall::registry::registration::OperationProvenance::Local,
None,
None,
alkcall::core::types::Capabilities::new(),
))?;
Ok(())
})();
if let Err(e) = register_result {
tracing::error!(error = %e, "channel-0 session registry setup failed");
return;
}
if let Some(sessions) = &sessions {
let id = sessions.insert_connection(Arc::clone(&call_connection));
let _conn_guard = ConnectionGuard {
sessions: sessions.clone(),
id,
};
// The guard lives for this task — its drop removes the
// handle when the dispatcher loop returns.
}
let dispatcher = alkcall::protocol::dispatch::Dispatcher::new(
registry,
fork,
std::sync::Arc::new(NoopProvider),
);
dispatcher
@@ -287,11 +469,14 @@ fn install_channel_zero(
})
}
#[allow(dead_code)]
/// The bare-registry variant of the hook for the from_wss test
/// server: no session registry (no WS-26 handle retention — the test
/// server's pump slot is its own teardown lever) and no openables.
#[cfg(all(test, feature = "server", feature = "wss"))]
pub(crate) fn adapter_install_channel_zero(
registry: Arc<OperationRegistry>,
) -> alkcall::channels::adapter::InstallChannelZero {
install_channel_zero(registry)
install_channel_zero(registry, None, Arc::new(NoCap), None)
}
struct NoopProvider;
@@ -341,6 +526,16 @@ pub struct WsTimeouts {
pub write: Option<std::time::Duration>,
}
/// Per-request openable-ALPN override (WS-22): a deployment inserts
/// `OpenableAlpns` into the request extensions (a route layer on the WS
/// route, mirroring [`ChannelsPolicy`] / [`WsTimeouts`]) to set the
/// session's openable set per route instead of the router-state
/// default. Without the extension the [`SessionState`] value applies —
/// [`HttpAdapter::with_ws_openable_alpns`](crate::server::HttpAdapter::with_ws_openable_alpns)
/// for the built-in surface; bare-registry routes carry none.
#[derive(Clone)]
pub struct OpenableAlpns(pub Arc<[OpenableAlpn]>);
/// The upgrade handler. Requires the resolved identity in request
/// extensions (stashed by [`ws_bearer_auth`]) — a WS session without
/// an identity cannot run `AccessControl::check`.
@@ -362,6 +557,11 @@ pub struct WsTimeouts {
/// above; a deployment overriding them inserts the `WsTimeouts`
/// extension on its upgrade route.
///
/// The openable-ALPN set (WS-22) comes from the [`OpenableAlpns`]
/// request extension when present, else from the router state
/// (`HttpAdapter::with_ws_openable_alpns`); the default is no
/// openables (channel 0 only).
///
/// Session cap (WS-09): one semaphore permit is acquired per upgrade,
/// post-auth and pre-upgrade; when the configured cap
/// (`HttpAdapter::with_ws_max_sessions`, default
@@ -374,6 +574,7 @@ pub async fn ws_upgrade_handler(
axum::Extension(identity): axum::Extension<Identity>,
policy: Option<axum::Extension<ChannelsPolicy>>,
timeouts: Option<axum::Extension<WsTimeouts>>,
openables: Option<axum::Extension<OpenableAlpns>>,
ws_upgrade: WebSocketUpgrade,
) -> Response {
let Ok(permit) = state.session_slots.clone().try_acquire_owned() else {
@@ -397,6 +598,9 @@ pub async fn ws_upgrade_handler(
None => state.idle_timeout(),
};
let write_timeout = timeouts.and_then(|t| t.write);
let openable_alpns = openables
.map(|axum::Extension(o)| o.0)
.or_else(|| state.openable_alpns());
ws_upgrade
.max_frame_size(INBOUND_WS_FRAME_CAP)
.max_message_size(INBOUND_WS_MESSAGE_CAP)
@@ -410,6 +614,7 @@ pub async fn ws_upgrade_handler(
sessions,
idle_timeout,
Some(write_timeout.unwrap_or(crate::websocket::DEFAULT_WS_WRITE_TIMEOUT)),
openable_alpns,
)
.await
})