diff --git a/Cargo.lock b/Cargo.lock index 1e161c0..a897736 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -27,9 +27,9 @@ dependencies = [ [[package]] name = "alkcall" -version = "0.4.1" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9badefe048a194c93eed09bc5326ebf092fc86817e561d88de2cd12b6302753a" +checksum = "0bbaeb718c370b74f9136d1d59b4608249bd7d1dd2332c0259f98351d709737f" dependencies = [ "async-trait", "bytes", @@ -1485,7 +1485,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1860,7 +1860,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2679,7 +2679,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 09eb397..0e7198a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,7 +42,7 @@ h2 = ["server", "dep:hyper", "hyper-util/http2", "hyper/http2"] http1 = ["server", "dep:hyper", "hyper-util/http1", "hyper/http1"] [dependencies] -alkcall = { version = "0.4", features = ["gateway"] } +alkcall = { version = "0.5", features = ["gateway"] } arc-swap = { version = "1", optional = true } axum = { version = "0.8", optional = true, features = ["ws"] } bytes = "1" diff --git a/docs/architecture/websocket.md b/docs/architecture/websocket.md index b22a54d..e8030c4 100644 --- a/docs/architecture/websocket.md +++ b/docs/architecture/websocket.md @@ -275,9 +275,18 @@ that motivated the alknet WebTransport track) workable over WS: the SSH byte stream rides a data channel. The openable set is declared with `HttpAdapter::with_ws_openable_alpns` -(each `OpenableAlpn { spec, open_handler }`; the ALPN-specific handlers -stay in the ALPN crates — alkhttp ferries the registrations), with the -`OpenableAlpns` request-extension fallback for bare-registry/custom +(each `OpenableAlpn { spec, open_handler, establisher, +establisher_timeout }`; the ALPN-specific handlers and establishers +stay in the ALPN crates — alkhttp ferries the registrations). The +establisher (alkcall 0.5.0 / ADR-049) is the awaited establishment +phase of the open op: `None` (the default) keeps the pre-0.5 shape — +the open replies as soon as the pump handler spawns; `Some` +semantically-validating hook dials/prepares the backend before the +reply, and a bounded failure resolves `channel:open_failed` with +`details.reason` (the channel never exists consumer-side). The +per-registration timeout override bounds the establisher when the +dispatch carries no deadline. The `OpenableAlpns` request-extension +fallback is available for bare-registry/custom upgrade routes. Cap policy is the `ChannelsPolicy` extension (one instance consulted by both the open wrappers and the demux teardown). Peer-announced ops (`op/register`) land in the connection-local diff --git a/src/server/adapter.rs b/src/server/adapter.rs index 24d99eb..de7b726 100644 --- a/src/server/adapter.rs +++ b/src/server/adapter.rs @@ -270,9 +270,10 @@ impl HttpAdapter { /// 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`](alkcall::channels::operations::OpenHandler). - /// Each WS + /// `channel_open` marker), the ALPN-specific + /// [`OpenHandler`](alkcall::channels::operations::OpenHandler), + /// and the optional establisher + per-registration timeout + /// (ADR-049, threaded through with `None` defaults). 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 diff --git a/src/websocket/upgrade.rs b/src/websocket/upgrade.rs index e900ac1..53574bc 100644 --- a/src/websocket/upgrade.rs +++ b/src/websocket/upgrade.rs @@ -46,7 +46,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use alkcall::channels::adapter::ChannelsAdapter; -use alkcall::channels::operations::{ChannelCore, OpenHandler}; +use alkcall::channels::operations::{ChannelCore, OpenEstablisher, OpenHandler}; use alkcall::channels::policy::{ChannelLifecyclePolicy, NoCap}; use alkcall::core::auth::{AuthContext, Identity}; use alkcall::core::types::{Connection, ProtocolHandler}; @@ -257,11 +257,13 @@ impl WsSessions { /// 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 +/// marker set via `OperationSpec::with_channel_open`), 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. +/// allocated channel's `Connection`, and the optional establishment +/// phase (ADR-049) the open-op wrapper awaits — bounded — before the +/// reply. The ALPN-specific handler and establisher stay 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` @@ -271,12 +273,50 @@ pub struct OpenableAlpn { /// The data-plane protocol handler spawned on the allocated /// channel's `Connection`. pub open_handler: OpenHandler, + /// The awaited establishment phase (ADR-049 §1): validate params + /// semantically, consult ownership, prepare/dial the backend — + /// before the open reply. `None` (the default) = an always-OK + /// establisher (the pre-ADR-049 shape; existing registrations + /// behave unchanged). + pub establisher: Option, + /// The per-registration bound on the establisher await + /// (ADR-049 §2). `None` = [`ESTABLISHMENT_TIMEOUT`] (10s) when the + /// dispatch carries no deadline; the effective bound is the + /// earlier of the dispatch deadline and this override. + /// + /// [`ESTABLISHMENT_TIMEOUT`]: alkcall::channels::operations::ESTABLISHMENT_TIMEOUT + pub establisher_timeout: Option, } impl OpenableAlpn { - /// Declare one openable ALPN. + /// Declare one openable ALPN (no establisher — the pre-ADR-049 + /// shape; the open op replies as soon as the pump handler is + /// spawned). pub fn new(spec: OperationSpec, open_handler: OpenHandler) -> Self { - Self { spec, open_handler } + Self { + spec, + open_handler, + establisher: None, + establisher_timeout: None, + } + } + + /// Attach an establishment phase (ADR-049 §1): awaited by the + /// open-op wrapper — bounded by the dispatch deadline, this + /// crate's [`ESTABLISHMENT_TIMEOUT`] default, or + /// `timeout` when set — before the reply; on failure the open op + /// resolves `channel:open_failed` with `details.reason` and the + /// channel never exists consumer-side. + /// + /// [`ESTABLISHMENT_TIMEOUT`]: alkcall::channels::operations::ESTABLISHMENT_TIMEOUT + pub fn with_establisher( + mut self, + establisher: OpenEstablisher, + timeout: Option, + ) -> Self { + self.establisher = Some(establisher); + self.establisher_timeout = timeout; + self } } @@ -373,7 +413,7 @@ impl Drop for ConnectionGuard { /// 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`), +/// deployment's openable ALPNs (`ChannelCore::register_openable_with_establisher`), /// 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` @@ -464,11 +504,13 @@ fn install_channel_zero( if let Some(openables) = openable_alpns.as_ref() { let core = ChannelCore::new(manager, Arc::clone(&policy)); for openable in openables.iter() { - core.register_openable( + core.register_openable_with_establisher( openable.spec.clone(), + openable.establisher.clone(), Arc::clone(&openable.open_handler), &fork, auth.clone(), + openable.establisher_timeout, )?; } }