From 944c300c170d2fe7d74e266ce4ebf8d66faa1f8a Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Tue, 13 Oct 2026 09:00:21 +0000 Subject: [PATCH] =?UTF-8?q?feat(channels):=20implement=20the=20single-publ?= =?UTF-8?q?ish=20fixes=20=E2=80=94=20S-01..S-04=20(reviews=20010/011;=20AD?= =?UTF-8?q?R-047=20am.=204,=20ADR-052,=20ADR-053,=20ADR-022=20am.)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit S-01 — the marked-pub open contract (ADR-047 amendment 4): - ChannelCore::register_pub_openable + PubOpenHandler (the put path's handler returns its result for the merged single response; the wrapper awaits inline, teardown is EOF-driven, handler_task stays None); register_openable* refuses Pub-typed specs; the wrapper entry points refuse unmarked/non-Pub shapes - the reserved input key `channel_id` (the initiator-allocated id): stripped pre-validation by resolve_sink_handler (the single source of truth, P-08), handed to the wrapper via OperationContext.metadata, a missing key rejected INVALID_INPUT; adopt parity enforced (ManagerError::InvalidAdopt) and adopt failures answer channel:open_failed / adopt_failed - zero call.published on marked ops: the first publish chunk on a marked op is rejected loudly (channel:publish_chunks_rejected) in both dispatch loops; the handler is never fed - relay put-translation replaces RelayError::PubTypedOpen: hub allocates odd on the producer leg, opens the spoke op in the background (put-acks-in-result), one-direction pump, response-mirror with the producer's reserved key stripped - ChannelClient::open_pub_channel + PubOpenReply (the initiator- allocates put open capability) S-02 — channel/control implemented (ADR-052): establisher-contributed ChannelControl (Establishment::with_control), manager-owned registry, teardown removals, channel:no_control_handle capability answer. S-03 — channel/resources/subscribe implemented (ADR-053): the ResourceEnumerators registry (shared per peer), snapshot-then-changes enumerators, scan-once aggregation, the empty seed; the always-failing one-shot error stream removed. S-04 — op/register forward-through (ADR-022 amendment): announced unmarked Sub/Pub register the real streaming/sink forwarding shapes; marked announcements refused with MARKED_SPEC_UNSUPPORTED; marked specs refuse the generic registration path and register_imported (Result-returning now). Guard (review 010 action 3): tests/stub_guard.rs greps src/ for not_implemented; the four pinned stub tests are deleted and the marked-spec registration refusals are pinned by tests. Verified: cargo test (696 + guard), clippy --all-targets -D warnings, fmt --check, fuzz corpus replay. --- src/channels/client.rs | 132 +++- src/channels/gate2_tests.rs | 92 +++ src/channels/hub_leg_tests.rs | 26 +- src/channels/manager.rs | 76 +++ src/channels/mod.rs | 4 + src/channels/operations.rs | 1027 ++++++++++++++++++++++++-------- src/channels/pub_open_tests.rs | 447 ++++++++++++++ src/channels/relay.rs | 233 +++++++- src/channels/relay_tests.rs | 36 +- src/channels/resource.rs | 70 +++ src/client/mod.rs | 5 +- src/protocol/adapter.rs | 4 +- src/protocol/connection.rs | 58 +- src/protocol/dispatch.rs | 112 +++- src/registry/op_register.rs | 173 +++++- src/registry/registration.rs | 178 +++++- src/registry/spec.rs | 22 + tests/stub_guard.rs | 49 ++ 18 files changed, 2418 insertions(+), 326 deletions(-) create mode 100644 src/channels/pub_open_tests.rs create mode 100644 src/channels/resource.rs create mode 100644 tests/stub_guard.rs diff --git a/src/channels/client.rs b/src/channels/client.rs index b57c088..c09d849 100644 --- a/src/channels/client.rs +++ b/src/channels/client.rs @@ -70,8 +70,8 @@ impl ChannelOpenError { /// The establishment-failure reason code — `details.reason` of a /// `channel:open_failed` error (`dial_failed`, `unknown_resource`, - /// `resource_shortage`, `handler_error`, `timeout`). `None` for - /// any other failure shape. + /// `resource_shortage`, `handler_error`, `timeout`, `adopt_failed`). + /// `None` for any other failure shape. pub fn establishment_reason(&self) -> Option<&str> { match self { ChannelOpenError::CallFailed { error } @@ -88,6 +88,30 @@ impl ChannelOpenError { } } +/// The awaited single merged response of a put open +/// ([`ChannelClient::open_pub_channel`], ADR-047 amendment 4 §4.9 — +/// the put ack rides in the result, ADR-047 amendment 4 §4.5). The +/// putter writes bytes on the returned `send` half (or spawns pumps) +/// and awaits this for the put's wire contract (e.g. alkblobs' put +/// replying `{ token, digest }`). Await this after (or alongside) the +/// byte stream, never before it — the response cannot exist before the +/// byte stream completes. +pub struct PubOpenReply { + rx: tokio::sync::oneshot::Receiver>, +} + +impl PubOpenReply { + /// The putter-side await: the merged response output (`Ok`); + /// `Err` carries the wire `CallError` verbatim + /// (`channel:open_failed` + `details.reason` per ADR-049 §3). + pub async fn resolve(self) -> Result { + match self.rx.await { + Ok(result) => result, + Err(_) => Err(ChannelOpenError::MissingChannelId), + } + } +} + /// The client-side handle for a channels connection. Constructed via /// [`ChannelClient::from_connection`] from an established /// `Connection`. Holds the `ChannelManager` (for the demux/mux state) @@ -382,6 +406,110 @@ impl ChannelClient { .map(|(send, recv)| (channel_id, out, send, recv)) } + /// The putter-facing open (ADR-047 amendment 4 §4.9 — the + /// initiator-allocates capability `open_channel`/`open_channel_with_reply` + /// cannot provide, because those await the open reply before + /// returning the halves and a put's bytes flow before the reply + /// exists, review 011 §2.7): allocates the connect-side odd id + /// locally, inserts the reserved input key, sends the open op in a + /// background task, and returns **immediately** — the caller + /// writes bytes on `send` without waiting for anything (the + /// put ack rides in the merged response; ADR-047 amendment 4 §4.5). + /// + /// Returns `(channel_id, send, recv, reply)`; `reply` is the + /// awaited single merged response — `Ok(Value)` (the full output, + /// e.g. `{ token, digest }`) or the wire error. On a failed open + /// the local adopt entry is torn down and `send`/`recv` are + /// already EOF'd (teardown); the caller still holds them but every + /// write is EOF-shaped. The `recv` half is the (bidirectional) + /// channel's read side — a put that also receives reads it. + pub async fn open_pub_channel( + &self, + operation_id: &str, + input: Value, + alpn: &str, + ) -> Result<(u32, MpscSendStream, MpscRecvStream, PubOpenReply), ChannelOpenError> { + use crate::registry::spec::RESERVED_OPEN_INPUT_KEY; + + let (channel_id, send, recv) = self + .manager + .open_channel( + alpn, + // The opener ledger is inert on the connect side (it + // does not enforce the per-identity cap, ADR-041) — + // the record is observability only. + "channel-client", + None, + ) + .await + .map_err(|e| match e { + crate::channels::manager::ManagerError::TooManyChannels { count, max } => { + ChannelOpenError::CallFailed { + error: crate::protocol::wire::CallError::new( + "channel:too_many_channels", + format!("per-connection channel cap exceeded: {count}/{max}"), + false, + ) + .with_details(serde_json::json!({ "count": count, "max": max })), + } + } + other => ChannelOpenError::CallFailed { + error: crate::protocol::wire::CallError::new( + "channel:allocation_failed", + format!("channel allocation failed: {other}"), + false, + ), + }, + })?; + + let mut op_input = input; + match op_input.as_object_mut() { + Some(obj) => { + obj.insert( + RESERVED_OPEN_INPUT_KEY.to_string(), + serde_json::json!(channel_id), + ); + } + None => { + return Err(ChannelOpenError::CallFailed { + error: crate::protocol::wire::CallError::invalid_input( + "marked pub open op input must be an object (ADR-047 amendment 4 §4.3)", + ), + }); + } + } + + let call_conn = { + let guard = self.call_connection.lock().await; + guard.as_ref().map(Arc::clone) + }; + let manager = self.manager.clone(); + let op_id = operation_id.to_string(); + let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + let response = match &call_conn { + Some(conn) => conn.call(&op_id, op_input).await, + None => ResponseEnvelope::error( + "channel-client", + crate::protocol::wire::CallError::internal("channel client closed"), + ), + }; + // A failed open tears the allocator-side entry down: the + // responder refused, the channel is dead, and the local + // half must not linger in the connection's channel map. + if response.result.is_err() { + let _ = manager.teardown_channel(channel_id); + } + let reply = match response.result { + Ok(out) => Ok(out), + Err(error) => Err(ChannelOpenError::CallFailed { error }), + }; + let _ = reply_tx.send(reply); + }); + + Ok((channel_id, send, recv, PubOpenReply { rx: reply_rx })) + } + /// Take the `CallConnection` — used by the consumer to register /// imported ops (`from_call`) on the connection's overlay. After /// this, `call_open_op` returns an error (the connection is owned diff --git a/src/channels/gate2_tests.rs b/src/channels/gate2_tests.rs index 140e34a..264c46b 100644 --- a/src/channels/gate2_tests.rs +++ b/src/channels/gate2_tests.rs @@ -62,6 +62,7 @@ const TTY_OP: &str = "channels/tty/sub"; const TTY_ALPN: &str = "alk/tty"; const ALPN: &str = "alk/tunnel"; const PLAIN_OP: &str = "fs/read"; +const PUT_OP: &str = "channels/put/bulk"; fn identity(id: &str, scopes: &[&str]) -> Identity { Identity { @@ -256,6 +257,49 @@ fn make_spoke_install( ) .expect("spoke registers the tty open op"); + // The marked pub openable (ADR-047 amendment 4 §4.7's + // relay-translation gate): a put handler that consumes the + // adopted channel's data plane to EOF and replies with the + // digest over the streamed bytes. + let put_spec = OperationSpec::new( + PUT_OP, + OperationType::Pub, + Visibility::External, + json!({ "type": "object" }), + json!({ + "type": "object", + "properties": { + "channel_id": { "type": "integer" }, + "digest": { "type": "string" } + } + }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alk/blobs")); + let put_handler: crate::channels::operations::PubOpenHandler = + Arc::new(move |_input, _plan, channel_conn, _auth| { + Box::pin(async move { + let mut bidi = channel_conn.accept_bi().await.map_err(|_| { + crate::protocol::wire::CallError::internal( + "spoke put: the data plane never opened", + ) + })?; + let mut buf = Vec::new(); + tokio::io::AsyncReadExt::read_to_end(&mut bidi, &mut buf) + .await + .map_err(|e| { + crate::protocol::wire::CallError::internal(format!( + "spoke put read failed: {e}" + )) + })?; + Ok(json!({ "digest": format!("spoke-{}", buf.len()) })) + }) + }); + core.register_pub_openable(put_spec, None, put_handler, ®istry, auth.clone(), None) + .expect("spoke registers the marked pub openable"); + registry .register(HandlerRegistration::new( plain_spec(PLAIN_OP), @@ -740,3 +784,51 @@ async fn gate2_tty_standard_shape_relays_with_no_regression() { "the tty relay's ledger decremented" ); } + +// --- gate: the pub-openables relay translation (ADR-047 amendment 4 §4.7) --- + +/// A hub translating a pub openable end-to-end (review 011 §3): the +/// browser's put via `open_pub_channel` on the consumer leg; bytes +/// flow browser → hub → spoke (the hub never parses them); the +/// spoke's single response loses its reserved key and its result +/// fields ride the consumer reply; the consumer replies exactly once +/// with the hub-allocated consumer-leg id. +#[tokio::test] +async fn gate_pub_relay_put_end_to_end() { + let h = start().await; + + let (id, mut send, _recv, reply) = h + .consumer + .open_pub_channel(PUT_OP, json!({ "path": "g" }), "alk/blobs") + .await + .expect("the put opens through the relayed pub openable"); + send.write_all(b"payload!").await.expect("stream the bytes"); + drop(send); + + let merged = reply + .resolve() + .await + .expect("the put's single merged response"); + assert_eq!( + merged["digest"], + json!("spoke-8"), + "the spoke's result rode the relay (the producer's reserved key stripped)" + ); + assert_eq!( + merged["channel_id"], + json!(id), + "the reply carries the hub-allocated consumer-leg id" + ); + assert_eq!(id % 2, 1, "the browser (connect side) allocated odd"); + + wait_for( + || data_channels(&h.hub_consumer_manager) == 0, + "the consumer-leg put channel reclaimed at the wrapper's completion teardown", + ) + .await; + wait_for( + || data_channels(&h.hub_producer_manager) == 0, + "the producer-leg put channel reclaimed at the plan's drop guard", + ) + .await; +} diff --git a/src/channels/hub_leg_tests.rs b/src/channels/hub_leg_tests.rs index 6eca274..77866c7 100644 --- a/src/channels/hub_leg_tests.rs +++ b/src/channels/hub_leg_tests.rs @@ -668,7 +668,11 @@ async fn template_plain_bundle_registration_failure_is_loud_at_install() { /// call to any op — including the bootstrap discovery ops — does not /// resolve). Never a silent stub. #[tokio::test] -async fn template_pub_typed_marked_spec_is_loud_at_install() { +async fn template_pub_typed_marked_spec_registers_through_the_relay() { + // ADR-047 amendment 4 §4.7: the put-forwarding translation replaced + // the `PubTypedOpen` assembly error — a pub-typed marked spec in + // the stash composes through `register_pub_openable` and the op + // re-exposes on the consumer leg. let pub_bundle = HandlerRegistration::new( relay_spec( "channels/x/pub", @@ -685,15 +689,19 @@ async fn template_pub_typed_marked_spec_is_loud_at_install() { ); let imports = HubLegImports::from_bundles(vec![pub_bundle]); let consumer = start_hub(None, Some(imports), None, false).await; - let response = tokio::time::timeout( - std::time::Duration::from_secs(3), - consumer.call_open_op("services/list", json!({})), - ) - .await; + let response = consumer.call_open_op("services/list", json!({})).await; + let out = response + .result + .expect("the leg dispatches; the op re-exposed"); + let names: Vec<&str> = out["operations"] + .as_array() + .expect("operations") + .iter() + .filter_map(|o| o.get("name").and_then(|n| n.as_str())) + .collect(); assert!( - response.is_err(), - "the leg's channel 0 never dispatches after the loud assembly error, got {:?}", - response + names.contains(&"channels/x/pub"), + "the pub openable is re-exposed through the relay, got: {names:?}" ); } diff --git a/src/channels/manager.rs b/src/channels/manager.rs index 8e75f9c..c1492d7 100644 --- a/src/channels/manager.rs +++ b/src/channels/manager.rs @@ -18,6 +18,9 @@ use tokio::sync::mpsc; use tokio::task::JoinHandle; use tracing::debug; +use crate::protocol::wire::CallError; +use serde_json::Value; + use super::mux::{MuxHandle, OpenerLedger}; use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP}; @@ -38,6 +41,15 @@ pub enum ChannelSide { /// `ChannelLifecyclePolicy` (ADR-041, amended by ADR-047 §7). pub const DEFAULT_MAX_CHANNELS: usize = 256; +/// The per-channel out-of-band control callback (ADR-052): received +/// from the open-op establisher (`Establishment::with_control`), +/// installed by the open-op wrapper under the channel's id, routed to +/// by `channel/control`, removed by every teardown path. The manager +/// stays blind to what the closure does (ADR-039). +pub type ChannelControl = Arc< + dyn Fn(Value) -> futures::future::BoxFuture<'static, Result> + Send + Sync, +>; + /// The per-channel state — the reassembly sender (the demux feeds chunk /// payloads into it; the handler reads them out via `MpscRecvStream`), /// the handler task handle, and the ALPN (for observability). @@ -62,6 +74,13 @@ pub enum ManagerError { UnknownChannel(u32), #[error("channel already exists: {0}")] ChannelExists(u32), + #[error( + "channel {channel_id} is outside the remote side's allocation range (expected {expected})" + )] + InvalidAdopt { + channel_id: u32, + expected: &'static str, + }, } /// The shared state for a channels connection. `Clone` (cheap — `Arc` @@ -101,6 +120,10 @@ struct Inner { /// drains them into the new receiver. `clear_all` drops the map. early_arrivals: Mutex>>, early_arrival_count: AtomicU64, + /// Per-channel out-of-band control callbacks (ADR-052). Installed + /// by the open-op wrapper from the establisher's + /// `Establishment::with_control`; removed by every teardown path. + control_handles: Mutex>, } /// The per-channel cap on parked early-arrival chunks (`route_payload` @@ -151,6 +174,7 @@ impl ChannelManager { dropped_unknown_chunks: AtomicU64::new(0), early_arrivals: Mutex::new(HashMap::new()), early_arrival_count: AtomicU64::new(0), + control_handles: Mutex::new(HashMap::new()), }), } } @@ -317,6 +341,32 @@ impl ChannelManager { alpn: impl Into, handler_task: Option>, ) -> Result<(MpscSendStream, MpscRecvStream), ManagerError> { + // The adopt range is the remote side's allocation range (the + // odd/even split, ADR-047 §5): the connect side allocates odd, + // so the accept side adopts odd; the accept side allocates + // even, so the connect side adopts even. Adopting inside the + // local range would collide with this side's own monotonic + // allocator shortly after. The duplicate check comes first — + // adopting an id this side already holds returns + // `ChannelExists` and must never touch the live channel (the + // 0.8.1 regression pin). + let expected = match self.inner.side { + ChannelSide::Connect => "even (the accept side's allocated ids; ADR-047 §5)", + ChannelSide::Accept => "odd (the connect side's allocated ids; ADR-047 §5)", + }; + if self.inner.channels.lock().contains_key(&channel_id) { + return Err(ManagerError::ChannelExists(channel_id)); + } + let parity_ok = match self.inner.side { + ChannelSide::Connect => channel_id.is_multiple_of(2), + ChannelSide::Accept => !channel_id.is_multiple_of(2), + }; + if !parity_ok { + return Err(ManagerError::InvalidAdopt { + channel_id, + expected, + }); + } let alpn = alpn.into(); let send = self @@ -381,6 +431,30 @@ impl ChannelManager { } } + /// Install the out-of-band control callback for a channel + /// (ADR-052). Called by the open-op wrapper from the establisher's + /// `Establishment::with_control`. Replaces an existing callback for + /// the id (the establisher runs once per channel — replacement is + /// not an expected path; the last call wins). + pub fn set_control_handle(&self, channel_id: u32, control: ChannelControl) { + self.inner + .control_handles + .lock() + .insert(channel_id, control); + } + + /// The control callback for a channel, for routing (the + /// `channel/control` handler). `None` when the channel's ALPN + /// registered no control surface (the op answers + /// `channel:no_control_handle`). + pub fn control_handle(&self, channel_id: u32) -> Option { + self.inner.control_handles.lock().get(&channel_id).cloned() + } + + fn remove_control_handle(&self, channel_id: u32) { + self.inner.control_handles.lock().remove(&channel_id); + } + /// Install channel 0 (pre-negotiated as `alk/call`, ADR-036). /// Channel 0 is special only in that it's pre-allocated (by /// `channels-call`); the `ChannelsAdapter` hands the resulting @@ -530,6 +604,7 @@ impl ChannelManager { // Dropping `state.demux_sender` signals EOF to the // handler's read half (REQ-CH-02). drop(state.demux_sender); + self.remove_control_handle(channel_id); Ok(state.handler_task) } None => Err(ManagerError::UnknownChannel(channel_id)), @@ -550,6 +625,7 @@ impl ChannelManager { task.abort(); } } + self.inner.control_handles.lock().clear(); // The early-arrival buffers die with the connection too. self.inner.early_arrivals.lock().clear(); // The opener ledger has the opener PeerIds. diff --git a/src/channels/mod.rs b/src/channels/mod.rs index dff8c2d..b830d57 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -31,6 +31,9 @@ //! as amended by ADR-051): translate hop + `pump_bidi` byte-forward //! hop for from_call-imported marked specs re-exposed on a hub's //! consumer-leg channel-0 registry. +//! - [`resource`]: `ResourceEnumerators` — the enumerator registry +//! behind `channel/resources/subscribe`'s live subscription +//! (ADR-053). //! - [`hub_leg`]: `HubLegTemplate` — the hub-leg install template //! (ADR-051 §5): fork + generic channel ops + bootstrap discovery + //! plain bundles + relay openables + serving identity, composed into @@ -53,5 +56,6 @@ pub mod policy; pub mod pump; pub mod reassembly; pub mod relay; +pub mod resource; pub mod source; pub mod wire; diff --git a/src/channels/operations.rs b/src/channels/operations.rs index dd11b57..e75409e 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -31,6 +31,8 @@ use crate::registry::spec::{AccessControl, OperationSpec, OperationType, Visibil use super::manager::{ChannelManager, ManagerError}; use super::policy::{ChannelError, ChannelLifecyclePolicy}; +use super::resource::{ResourceEnumerator, ResourceEnumerators}; +use std::pin::Pin; /// The names of the generic channel lifecycle operations (ADR-037, /// amended by ADR-047). @@ -44,15 +46,20 @@ pub const OP_CHANNEL_RESOURCES_SUBSCRIBE: &str = "channel/resources/subscribe"; pub struct ChannelOperations { manager: ChannelManager, policy: Arc, + enumerators: Arc, } impl ChannelOperations { /// Construct with a `ChannelManager` and a /// `ChannelLifecyclePolicy`. The default policy is /// `PerIdentityChannelPolicy::new(256)` (via - /// [`default_policy`][super::policy::default_policy]). + /// [`default_policy`][super::policy::default_policy]). The + /// `channel/resources/subscribe` handler builds over an empty + /// enumerator registry (the op serves the truthful empty live + /// subscription) — pass one via + /// [`ChannelOperations::with_resource_enumerators`]. pub fn new(manager: ChannelManager, policy: Arc) -> Self { - Self { manager, policy } + Self::with_resource_enumerators(manager, policy, Arc::new(ResourceEnumerators::new())) } /// Construct with the default policy @@ -61,6 +68,23 @@ impl ChannelOperations { Self::new(manager, super::policy::default_policy()) } + /// Construct with the shared peer-level enumerator registry for + /// `channel/resources/subscribe` (ADR-053 §2 — registered at + /// assembly/install time by the composing crate; one registry + /// across every channels connection the peer accepts, like the + /// policy). + pub fn with_resource_enumerators( + manager: ChannelManager, + policy: Arc, + enumerators: Arc, + ) -> Self { + Self { + manager, + policy, + enumerators, + } + } + /// Register the three generic ops on the call `OperationRegistry`. /// The per-ALPN open ops are registered separately by the ALPN /// crates via [`ChannelCore::register_openable`] (ADR-047 §3). @@ -89,7 +113,10 @@ impl ChannelOperations { let manager = self.manager.clone(); registry.register(HandlerRegistration::new( channel_resources_subscribe_spec(), - HandlerKind::Stream(make_resources_subscribe_handler(manager)), + HandlerKind::Stream(make_resources_subscribe_handler( + manager, + Arc::clone(&self.enumerators), + )), OperationProvenance::Local, None, None, @@ -120,7 +147,9 @@ pub fn channel_close_spec() -> OperationSpec { ) } -/// `OperationSpec` for `channel/control` (ADR-037). +/// `OperationSpec` for `channel/control` (ADR-037, implemented per +/// ADR-052 — the reply carries the control callback's ALPN-specific +/// output under `control`). pub fn channel_control_spec() -> OperationSpec { OperationSpec::new( OP_CHANNEL_CONTROL, @@ -134,7 +163,12 @@ pub fn channel_control_spec() -> OperationSpec { }, "required": ["channel_id", "message"] }), - json!({ "type": "object", "properties": { "ok": { "type": "boolean" } } }), + json!({ + "type": "object", + "properties": { + "control": { "type": "object" } + } + }), vec![], AccessControl::default(), None, @@ -231,71 +265,141 @@ fn make_close_handler(manager: ChannelManager, policy: Arc Handler { Arc::new(move |input: Value, ctx: OperationContext| { let manager = manager.clone(); Box::pin(async move { - let channel_id = match input.get("channel_id").and_then(|v| v.as_u64()) { - Some(id) => id as u32, - None => { + let (channel_id, message) = match ( + input.get("channel_id").and_then(|v| v.as_u64()), + input.get("message").cloned(), + ) { + (Some(id), Some(message)) => (id as u32, message), + (Some(_), None) => { + return ResponseEnvelope::error( + ctx.request_id, + CallError::invalid_input( + "missing required field: message (the ALPN-specific control body)", + ), + ); + } + (None, _) => { return ResponseEnvelope::error( ctx.request_id, CallError::invalid_input("missing required field: channel_id"), ); } }; - let _message = input.get("message").cloned().unwrap_or(Value::Null); if !manager.has_channel(channel_id) { return ResponseEnvelope::error( ctx.request_id, CallError::not_found("channel:unknown_channel"), ); } - ResponseEnvelope::error( - ctx.request_id, - CallError::new( - "channel:control_not_implemented", - "channel/control is not yet implemented (OQ-39)", - false, - ), - ) + let Some(control) = manager.control_handle(channel_id) else { + let alpn = manager.channel_alpn(channel_id).unwrap_or_default(); + return ResponseEnvelope::error( + ctx.request_id, + CallError::new( + "channel:no_control_handle", + "this channel's ALPN registered no out-of-band control surface \ + (control that needs data-ordered delivery rides the data channel's \ + BiStream — channel-operations.md DP-4)", + false, + ) + .with_details(json!({ "alpn": alpn })), + ); + }; + match control(message).await { + Ok(reply) => ResponseEnvelope::ok(ctx.request_id, json!({ "control": reply })), + Err(e) => ResponseEnvelope::error(ctx.request_id, e), + } }) }) } -/// The `channel/resources/subscribe` handler. Emits an initial -/// snapshot of the open channels, then subsequent events on any change. -/// The `access` preview is dropped (ADR-047 §6 — it's on the op spec, -/// available via `services/schema`). -/// -/// Not yet implemented — the live subscription aggregated from -/// ALPN-crate resource enumerators (ADR-047 §6) is deferred (OQ-40). -/// Returns `channel:resources_not_implemented` so callers fail loudly -/// rather than receiving a fake one-shot snapshot. -fn make_resources_subscribe_handler(manager: ChannelManager) -> StreamingHandler { +/// The `channel/resources/subscribe` handler (ADR-053 — the +/// implemented ADR-037 §6 dynamic half). One `ResourceEnumerator` per +/// registered openable ALPN, provided by the producing crate +/// (snapshot-then-changes); the handler aggregates across the ALPNs +/// registered at subscription start (the scan-once pin): each +/// enumerator item becomes one `{ "resources": [{ "alpn": …, … }] }` +/// envelope; the enumerators' first items are the initial snapshot. +/// An empty registry emits exactly one `{ "resources": [] }` and +/// stays open. The `access` preview stays dropped (ADR-047 §6). +fn make_resources_subscribe_handler( + manager: ChannelManager, + enumerators: Arc, +) -> StreamingHandler { Arc::new(move |_input: Value, ctx: OperationContext| { - let _manager = manager.clone(); - Box::pin(futures::stream::once(async move { - ResponseEnvelope::error( - ctx.request_id, - CallError::new( - "channel:resources_not_implemented", - "channel/resources/subscribe is not yet implemented (OQ-40)", - false, - ), - ) - })) + let manager = manager.clone(); + let scanned = enumerators.snapshot(); + Box::pin(streams_from_scanned(scanned, ctx.request_id, manager)) }) } +/// Build the aggregate resource stream from the scan-once snapshot +/// (ADR-053 §2/§3): empty registry → one `{ "resources": [] }` seed +/// then the stream stays open; non-empty → `select_all` over every +/// enumerator's snapshot-then-changes stream, each item wrapped as a +/// one-entry resource envelope. +fn streams_from_scanned( + scanned: Vec<(String, Arc)>, + request_id: String, + _manager: ChannelManager, +) -> Pin + Send>> { + use futures::stream::{self, StreamExt}; + if scanned.is_empty() { + let request_id = request_id.clone(); + return Box::pin( + stream::once( + async move { ResponseEnvelope::ok(request_id, json!({ "resources": [] })) }, + ) + .chain(stream::pending()), + ); + } + let mut merged: Vec + Send>>> = + Vec::with_capacity(scanned.len()); + for (_alpn, enumerator) in scanned { + let request_id = request_id.clone(); + merged.push(Box::pin(enumerator.subscribe().filter_map( + move |(alpn, payload)| { + let request_id = request_id.clone(); + async move { + Some(ResponseEnvelope::ok( + request_id, + json!({ "resources": [resource_entry(alpn, payload)] }), + )) + } + }, + ))); + } + Box::pin(stream::select_all(merged)) +} + +/// Merge an enumerator payload with its ALPN into one wire resource +/// entry (ADR-053 §1): object payloads gain `"alpn"`; anything else +/// wraps as `{ "alpn", "data" }`. +fn resource_entry(alpn: String, payload: Value) -> Value { + match payload { + Value::Object(mut object) => { + object.insert("alpn".to_string(), json!(alpn)); + Value::Object(object) + } + other => json!({ "alpn": alpn, "data": other }), + } +} + /// `ChannelCore` — the channel machinery the ALPN crate's open-op /// wrapper uses (ADR-047 §3, as amended by ADR-049 — the wrapper /// gains an awaited, bounded establishment phase via @@ -386,7 +490,7 @@ pub type ChannelPlan = Arc; /// breaking release. Construct with [`Establishment::new`] (plan) or /// [`Establishment::default`] (no plan); chain /// [`Establishment::with_reply_field`] to contribute reply fields. -#[derive(Debug, Clone, Default)] +#[derive(Clone, Default)] #[non_exhaustive] pub struct Establishment { /// The channel plan — the payload the pump handler receives. The @@ -394,6 +498,10 @@ pub struct Establishment { /// ([`ChannelPlan`]); alkcall never inspects it. pub plan: Option, pub(crate) reply_fields: Option>, + /// The out-of-band control callback (ADR-052) — installed by the + /// wrapper under the allocated/adopted channel id and removed on + /// every teardown path. + pub(crate) control: Option, } /// The wrapper-owned success-reply key (ADR-049 amendment 3): the @@ -408,9 +516,24 @@ impl Establishment { Self { plan: Some(plan), reply_fields: None, + control: None, } } + /// Contribute the per-channel out-of-band control callback + /// (ADR-052). The wrapper installs it under the allocated/adopted + /// channel id (`channel/control` routes to it) and every teardown + /// path removes it. Builder-style. + pub fn with_control(mut self, control: crate::channels::manager::ChannelControl) -> Self { + self.control = Some(control); + self + } + + /// The contributed control callback, for inspection. + pub fn control(&self) -> Option<&crate::channels::manager::ChannelControl> { + self.control.as_ref() + } + /// Contribute one reply field (ADR-049 amendment 3 — the /// establisher → opener direction). Merged into the open-op /// success reply by the wrapper, after its `channel_id` @@ -594,11 +717,10 @@ impl ChannelCore { /// 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. + /// not in the call response stream). `Pub`-typed specs refuse at + /// this entry point — the put path is + /// [`ChannelCore::register_pub_openable`] (ADR-047 amendment 4 + /// §4.4). The `OpenHandler` is spawned in all allowed cases. pub fn register_openable( &self, spec: OperationSpec, @@ -646,6 +768,7 @@ impl ChannelCore { timeout: Option, ) -> Result<(), String> { let op_type = spec.op_type; + let op_name = spec.name.clone(); let channel_open = spec.channel_open.clone(); let alpn = channel_open .as_ref() @@ -679,24 +802,108 @@ impl ChannelCore { auth, alpn, )), - OperationType::Pub => HandlerKind::Sink(make_open_handler_sink( - manager, - policy, - establisher, - open_handler, - auth, - alpn, - )), + OperationType::Pub => { + return Err(format!( + "register_openable: spec `{op_name}` is Pub-typed — the initiator \ + allocates there (ADR-047 §5); marked pub openables register via \ + `ChannelCore::register_pub_openable` (ADR-047 amendment 4 §4.4) and \ + unmarked `Pub` ops are ordinary ADR-046 sinks registered with \ + `make_sink_handler`" + )); + } }; - registry.register(HandlerRegistration::new( - spec, - handler_kind, - OperationProvenance::Local, - None, - None, - Capabilities::new(), - )) + registry.register( + HandlerRegistration::new( + spec, + handler_kind, + OperationProvenance::Local, + None, + None, + Capabilities::new(), + ) + .with_channel_wrapper(), + ) + } + + /// Register a **marked `Pub` openable** (ADR-047 amendment 4 §4.4) — + /// the put path: the initiator allocates the channel id on its own + /// side (odd, per the odd/even split — ADR-047 §5) and streams bytes + /// on it before the open resolves; this side adopts and is the + /// responder of record (check_open → adopt → bounded establisher → + /// the [`PubOpenHandler`] runs on the channel's `Connection` → + /// awaited inline → one response merging `{ channel_id }`, the + /// establisher's ack fields, and the handler's result). + /// + /// Refusals (registration-time loudness): an unmarked spec, a non-`Pub` + /// type, or (via the registry, ADR-047 amendment 4 §4.2) a + /// `publish_schema` on a marked spec. + /// + /// `timeout` bounds the establisher exactly as + /// [`ChannelCore::register_openable_with_establisher`] (ADR-049 §2). + pub fn register_pub_openable( + &self, + spec: OperationSpec, + establisher: Option, + pub_open_handler: PubOpenHandler, + registry: &OperationRegistry, + auth: AuthContext, + timeout: Option, + ) -> Result<(), String> { + if spec.channel_open.is_none() { + return Err(format!( + "register_pub_openable: spec `{}` has no channel_open marker — the adopt \ + path is defined only for marked ops; an unmarked `Pub` op is an ordinary \ + ADR-046 sink registered with `make_sink_handler` (ADR-047 amendment 4 §4.4)", + spec.name + )); + } + if spec.op_type != OperationType::Pub { + return Err(format!( + "register_pub_openable: spec `{}` is {:?}-typed — only `Pub`-typed marked \ + ops ride the adopt path; `Sub`/`Query`/`Mutation` openables register via \ + `register_openable*` (the responder allocates there, ADR-047 §5)", + spec.name, spec.op_type + )); + } + + let alpn = spec + .channel_open + .as_ref() + .map(|mark| mark.alpn.to_string()) + .ok_or_else(|| { + format!( + "register_pub_openable: spec `{}` has no channel_open marker \ + (ADR-047 amendment 4 §4.4)", + spec.name + ) + })?; + let establisher = establisher.map(|establisher| EstablisherHook { + establisher, + timeout, + }); + let manager = self.manager.clone(); + let policy = Arc::clone(&self.policy); + let pub_open_handler = Arc::clone(&pub_open_handler); + + registry.register( + HandlerRegistration::new( + spec, + HandlerKind::Sink(make_open_handler_pub( + manager, + policy, + establisher, + pub_open_handler, + auth, + alpn, + )), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + ) + .with_channel_wrapper(), + ) } } @@ -881,72 +1088,26 @@ async fn run_open_wrapper( return ResponseEnvelope::error(request_id, map_channel_error_to_call_error(&channel_err)); } - let mut reply_fields: Option> = None; - let channel_id = match manager + let (channel_id, reply_fields) = match manager .open_channel(alpn, opener_identity.id.clone(), None) .await { Ok((id, send, recv)) => { - // The establishment phase (ADR-049 §1): awaited bounded, - // before the reply and before the pump handler is spawned. - // No establisher registered = an always-OK establisher - // (the pre-ADR-049 shape, unchanged) — its plan is `None`. - let plan = match establisher { - Some(hook) => { - let bound = establishment_bound(hook.timeout, deadline); - let establishment = tokio::time::timeout( - bound, - (hook.establisher)(input.clone(), per_call_auth.clone()), - ) - .await; - match establishment { - Err(_elapsed) => { - teardown_failed_channel(manager, policy, id, &opener_identity); - return ResponseEnvelope::error( - request_id, - establishment_timeout_call_error(), - ); - } - Ok(Err(e)) => { - tracing::debug!( - channel_id = id, - reason = e.reason(), - "open wrapper: establisher failed; tearing down channel" - ); - teardown_failed_channel(manager, policy, id, &opener_identity); - return ResponseEnvelope::error( - request_id, - establishment_error_to_call_error(&e), - ); - } - Ok(Ok(establishment)) => { - // The `channel_id` reservation (ADR-049 - // amendment 3): the wrapper owns the key. - // An establisher supplying it is a bug — - // fail loudly (same open_failed path) so a - // reserved-key collision never silently - // shadows the wrapper's value. - if establishment - .reply_fields - .as_ref() - .is_some_and(|m| m.contains_key(RESERVED_REPLY_KEY)) - { - tracing::warn!( - channel_id = id, - "open wrapper: establisher supplied the reserved reply key; tearing down channel" - ); - teardown_failed_channel(manager, policy, id, &opener_identity); - return ResponseEnvelope::error( - request_id, - reserved_reply_key_call_error(), - ); - } - reply_fields = establishment.reply_fields; - establishment.plan - } - } - } - None => None, + let (plan, reply_fields) = match run_establisher_phase( + manager, + policy, + establisher, + id, + &opener_identity, + input.clone(), + &per_call_auth, + deadline, + request_id.clone(), + ) + .await + { + Ok(pair) => pair, + Err(envelope) => return envelope, }; let remote_addr = manager.remote_addr(); @@ -1019,7 +1180,7 @@ async fn run_open_wrapper( "open wrapper: channel vanished between open and handler-task install" ); } - id + (id, reply_fields) } Err(ManagerError::TooManyChannels { count, max }) => { policy.on_close(&opener_identity); @@ -1052,6 +1213,131 @@ async fn run_open_wrapper( ) } +/// Run the establisher phase (ADR-049 §1/§2) shared by the three open +/// wrapper paths: awaited bounded (the earlier of the dispatch deadline +/// and the per-registration timeout). On success returns the plan and +/// the contributed reply fields — the contributed control callback +/// (ADR-052, `Establishment::with_control`) is installed under +/// `channel_id` here. On failure (error or deadline) the just-allocated +/// or just-adopted channel is torn down (demux sender drop, ledger +/// take, `policy.on_close` — the same atomic ledger-take gate every +/// teardown path runs, ADR-047 §7) and the `channel:open_failed` +/// reply envelope is returned. +/// +/// The `channel_id` reply-key reservation (ADR-049 amendment 3): an +/// establisher contributing the wrapper-owned key fails the open +/// loudly (`handler_error`) rather than shadowing the wrapper's value. +#[allow(clippy::too_many_arguments)] +async fn run_establisher_phase( + manager: &ChannelManager, + policy: &Arc, + establisher: Option<&EstablisherHook>, + channel_id: u32, + opener_identity: &Identity, + input: Value, + per_call_auth: &AuthContext, + deadline: Option, + request_id: String, +) -> Result<(Option, Option>), ResponseEnvelope> { + let Some(hook) = establisher else { + return Ok((None, None)); + }; + let bound = establishment_bound(hook.timeout, deadline); + let establishment = + tokio::time::timeout(bound, (hook.establisher)(input, per_call_auth.clone())).await; + match establishment { + Err(_elapsed) => { + teardown_failed_channel(manager, policy, channel_id, opener_identity); + Err(ResponseEnvelope::error( + request_id.clone(), + establishment_timeout_call_error(), + )) + } + Ok(Err(e)) => { + tracing::debug!( + channel_id, + reason = e.reason(), + "open wrapper: establisher failed; tearing down channel" + ); + teardown_failed_channel(manager, policy, channel_id, opener_identity); + Err(ResponseEnvelope::error( + request_id.clone(), + establishment_error_to_call_error(&e), + )) + } + Ok(Ok(establishment)) => { + if establishment + .reply_fields + .as_ref() + .is_some_and(|m| m.contains_key(RESERVED_REPLY_KEY)) + { + tracing::warn!( + channel_id, + "open wrapper: establisher supplied the reserved reply key; tearing down channel" + ); + teardown_failed_channel(manager, policy, channel_id, opener_identity); + Err(ResponseEnvelope::error( + request_id, + reserved_reply_key_call_error(), + )) + } else { + if let Some(control) = establishment.control { + manager.set_control_handle(channel_id, control); + } + Ok((establishment.plan, establishment.reply_fields)) + } + } + } +} + +/// Map an adopt failure (ADR-047 amendment 4 §4.4) to the +/// `channel:open_failed` reply with the new `adopt_failed` reason: +/// a collision (the initiator-allocated id is already in use on this +/// side) or the per-connection cap, or the id falling outside the +/// remote's allocation range. +fn adopt_failure_to_call_error(e: &ManagerError) -> CallError { + CallError::new( + CHANNEL_OPEN_FAILED, + format!("channel adoption failed: {e}"), + false, + ) + .with_details(json!({ "reason": "adopt_failed", "message": e.to_string() })) +} + +/// Map a marked-pub handler's terminal result to the single response +/// (ADR-047 amendment 4 §4.4): `Ok(result)` merges into the output +/// (object → flat fields, winning on key collision with the ack +/// fields — the handler's computed result postdates the ack; +/// non-object → the value rides under `"result"`); `Err(e)` is the +/// response directly. +pub(crate) fn handler_result_into_response( + request_id: String, + output: Value, + result: Result, +) -> ResponseEnvelope { + match result { + Ok(value) => { + let mut output = output; + match value { + Value::Object(fields) => { + if let Some(obj) = output.as_object_mut() { + for (key, v) in fields { + obj.insert(key, v); + } + } + } + non_object => { + if let Some(obj) = output.as_object_mut() { + obj.insert("result".to_string(), non_object); + } + } + } + ResponseEnvelope::ok(request_id, output) + } + Err(e) => ResponseEnvelope::error(request_id, e), + } +} + /// 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 @@ -1155,49 +1441,190 @@ fn make_open_handler_stream( }) } -/// Build the `Sink` handler for a `Pub`-typed open op (ADR-047 §1). +/// The marked-pub (`Pub`-typed, `channel_open`-marked) open handler +/// (ADR-047 amendment 4 §4.4). The put path's handler shape: the +/// ALPN-specific work consumes the data plane (the adopted channel's +/// `BiStream` — one direction, initiator → responder, the ADR-040 +/// bounded buffer as backpressure) and returns the result that merges +/// into the single response. The wrapper awaits the returned future +/// inline (R-02's lifetime contract is structural on this path — the +/// wrapper cannot respond before the handler completes), so a +/// spawn-and-forget return shape cannot exist here; a handler that +/// ends early on EOF produces whatever partial result it computes. /// -/// **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, - _establisher: Option, - _open_handler: OpenHandler, - _auth: AuthContext, - _alpn: String, +/// The `input` is the open op's input with the reserved `channel_id` +/// key already stripped (ADR-047 amendment 4 §4.3 — the wrapper +/// adopted the id). `Err` is the response directly (the wrapper's +/// single response contract; the channel tears down after it). +pub type PubOpenHandler = Arc< + dyn Fn( + Value, + Option, + Connection, + AuthContext, + ) -> BoxFuture<'static, Result> + + Send + + Sync, +>; + +/// The marked-pub wrapper (ADR-047 amendment 4 §4.4) — the responder +/// of record. The reserved input key was stripped pre-validation by +/// the sink resolve path ([`Self::register_pub_openable`]'s +/// dispatcher; ADR-047 amendment 4 §4.3) and handed over as +/// `ctx.metadata` under the reserved metadata key: the id exists +/// before this wrapper runs. +/// +/// Flow: `check_open` → adopt (the existing demux/mux install + +/// early-arrival drain) → bounded establisher (offer validation, ack +/// fields, control handle) → handler on the adopted channel's +/// `Connection`, awaited inline → teardown (demux sender drop, ledger +/// take, `policy.on_close`) → the single response (merge order +/// pinned: `channel_id` reserved → establisher ack fields → handler +/// result; handler `Err` is the response, per ADR-047 amendment 4 +/// §4.5's single-response semantics — a failed establishment or +/// handler tears the channel down and the only response carries the +/// reason). The adopted channel's `handler_task` stays `None`: +/// teardown is EOF-driven while the wrapper polls (a `channel/close` +/// during the put drops the demux sender, the handler's reads see +/// EOF, and the single response is still written). +/// +/// Adoption failure (`ChannelExists` — the initiator's id collides +/// with this side's own range, or cap/parity violations) replies +/// `channel:open_failed` with reason `adopt_failed` (ADR-047 +/// amendment 4 §4.4); nothing was created, so nothing is torn down. +#[allow(clippy::too_many_arguments)] +fn make_open_handler_pub( + manager: ChannelManager, + policy: Arc, + establisher: Option, + pub_open_handler: PubOpenHandler, + auth: AuthContext, + alpn: String, ) -> crate::registry::registration::SinkHandler { Arc::new( - move |_input: Value, + move |input: Value, ctx: OperationContext, _publish_stream: crate::registry::registration::PublishStream| { - let request_id = ctx.request_id.clone(); + let manager = manager.clone(); + let policy = Arc::clone(&policy); + let establisher = establisher.clone(); + let pub_open_handler = Arc::clone(&pub_open_handler); + let auth = auth.clone(); + let alpn = alpn.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)", - ), + let _request_id = ctx.request_id.clone(); + run_pub_open_wrapper( + &manager, + &policy, + establisher, + &pub_open_handler, + &auth, + &alpn, + input, + ctx, ) + .await }) }, ) } +/// The marked-pub wrapper body (ADR-047 amendment 4 §4.4). +#[allow(clippy::too_many_arguments)] +async fn run_pub_open_wrapper( + manager: &ChannelManager, + policy: &Arc, + establisher: Option, + pub_open_handler: &PubOpenHandler, + auth: &AuthContext, + alpn: &str, + input: Value, + ctx: OperationContext, +) -> ResponseEnvelope { + use crate::registry::spec::RESERVED_OPEN_INPUT_KEY; + use crate::registry::spec::RESERVED_OPEN_METADATA_KEY; + + let request_id = ctx.request_id.clone(); + let adopted_id = match ctx + .metadata + .get(RESERVED_OPEN_METADATA_KEY) + .and_then(|v| v.as_u64()) + { + Some(id) => id as u32, + None => { + // Unreachable through the wire path (the sink resolve path + // rejects a marked pub op without the reserved key before + // any handler runs) — an identity-less direct `invoke_sink` + // call against a marked spec. Stay loud, stay typed. + return ResponseEnvelope::error( + request_id, + CallError::invalid_input(format!( + "marked pub open op invoked without the reserved `{RESERVED_OPEN_INPUT_KEY}` \ + input key — invoke through the dispatch paths (ADR-047 amendment 4 §4.3)" + )), + ); + } + }; + + let mut per_call_auth = auth.clone(); + if let Some(identity) = &ctx.identity { + per_call_auth.identity = Some(identity.clone()); + } + let opener_identity = opener_identity_or_anonymous(ctx.identity.clone()); + + if let Err(channel_err) = policy.check_open(&opener_identity) { + return ResponseEnvelope::error(request_id, map_channel_error_to_call_error(&channel_err)); + } + + let (send, recv) = match manager.adopt_channel(adopted_id, alpn, None).await { + Ok((send, recv)) => (send, recv), + Err(e) => { + tracing::debug!( + channel_id = adopted_id, + error = %e, + "pub open wrapper: adopt failed" + ); + return ResponseEnvelope::error(request_id, adopt_failure_to_call_error(&e)); + } + }; + + let (plan, reply_fields) = match run_establisher_phase( + manager, + policy, + establisher.as_ref(), + adopted_id, + &opener_identity, + input.clone(), + &per_call_auth, + ctx.deadline, + request_id.clone(), + ) + .await + { + Ok(pair) => pair, + Err(envelope) => return envelope, + }; + + 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 handler_result = pub_open_handler(input, plan, channel_conn, per_call_auth).await; + if let Err(e) = manager.teardown_channel(adopted_id) { + tracing::debug!( + channel_id = adopted_id, + error = %e, + "pub open wrapper: completion teardown found no channel to remove (benign when racing channel/close)" + ); + } + if manager.opener_ledger().take(adopted_id).is_some() { + policy.on_close(&opener_identity); + } + + let output = merge_reply_fields(json!({ "channel_id": adopted_id }), reply_fields); + handler_result_into_response(request_id, output, handler_result) +} + #[cfg(test)] mod tests { use super::*; @@ -1299,38 +1726,61 @@ mod tests { assert_eq!(spec.visibility, Visibility::External); } + /// ADR-053 gate: initial snapshot — the registered enumerators' + /// first items arrive as the first envelopes, payload merged with + /// `"alpn"`. #[tokio::test] - async fn resources_subscribe_handler_returns_not_implemented() { - let manager = make_manager().await; - manager - .open_channel("alk/tty", "alice", None) - .await - .expect("open tty"); - let handler = make_resources_subscribe_handler(manager.clone()); - let mut stream = handler(Value::Null, test_context("res-1")); - let env = stream.next().await.expect("one envelope"); - match env.result { - Err(e) => assert_eq!( - e.code, "channel:resources_not_implemented", - "resources/subscribe should return not-implemented (OQ-40)" - ), - Ok(_) => panic!("resources/subscribe should return error"), + async fn resources_subscribe_aggregates_enumerators_initial_snapshot() { + use super::super::resource::{ResourceEnumerator, ResourceEnumerators, ResourceStream}; + + struct Snapshot(Vec); + impl ResourceEnumerator for Snapshot { + fn subscribe(&self) -> ResourceStream { + let items = self.0.clone(); + Box::pin(futures::stream::iter( + items.into_iter().map(|v| ("alk/tty".to_string(), v)), + )) + } } + let manager = make_manager().await; + let enumerators = Arc::new(ResourceEnumerators::new()); + enumerators.register( + "alk/tty", + Arc::new(Snapshot(vec![json!({ + "backends": ["docker", "local"], + })])), + ); + let handler = make_resources_subscribe_handler(manager, Arc::clone(&enumerators)); + let mut stream = handler(Value::Null, test_context("res-1")); + let env = stream.next().await.expect("the snapshot envelope"); + let out = env.result.expect("ok"); + assert_eq!( + out, + json!({ "resources": [ { "alpn": "alk/tty", "backends": ["docker", "local"] } ] }) + ); } + /// ADR-053 gate: empty registry — exactly one `{ "resources": [] }` + /// seed and the stream stays open (the poll use case reads the + /// first event, cancels). #[tokio::test] - async fn resources_subscribe_handler_not_implemented_even_when_empty() { + async fn resources_subscribe_empty_registry_seeds_and_stays_open() { + use super::super::resource::ResourceEnumerators; + let manager = make_manager().await; - let handler = make_resources_subscribe_handler(manager); + let handler = + make_resources_subscribe_handler(manager, Arc::new(ResourceEnumerators::new())); let mut stream = handler(Value::Null, test_context("res-2")); - let env = stream.next().await.expect("one envelope"); - match env.result { - Err(e) => assert_eq!( - e.code, "channel:resources_not_implemented", - "resources/subscribe should return not-implemented even with no channels (OQ-40)" - ), - Ok(_) => panic!("resources/subscribe should return error"), - } + let env = stream.next().await.expect("the empty seed"); + assert_eq!(env.result.expect("ok"), json!({ "resources": [] })); + // The stream stays open — the next value never arrives. + let second = tokio::time::timeout(std::time::Duration::from_millis(100), stream.next()) + .await + .is_err(); + assert!( + second, + "an empty registry emits exactly one seed and stays open" + ); } #[tokio::test] @@ -1435,23 +1885,74 @@ mod tests { } } + /// ADR-052 gate: established without a control callback → + /// `channel:no_control_handle`; with one → routed and the reply is + /// the callback's output under `"control"`. #[tokio::test] - async fn control_handler_success_path_returns_not_implemented() { + async fn control_handler_routes_to_the_establisher_contributed_callback() { + use futures::future::BoxFuture; + let manager = make_manager().await; let (id, _send, _recv) = manager .open_channel("alk/tty", "alice", None) .await .expect("open"); - let handler = make_control_handler(manager); + + // No control handle registered yet. + let handler = make_control_handler(manager.clone()); let env = handler( json!({ "channel_id": id, "message": {} }), + test_context("ctrl-nocb"), + ) + .await; + let e = env.result.expect_err("no_control_handle"); + assert_eq!(e.code, "channel:no_control_handle"); + + // Install one and route through it. + manager.set_control_handle( + id, + Arc::new( + move |message: Value| -> BoxFuture<'static, Result> { + Box::pin(async move { + Ok(json!({ "applied": message["type"].as_str() == Some("resize") })) + }) + }, + ), + ); + let env = handler( + json!({ "channel_id": id, "message": { "type": "resize" } }), test_context("ctrl-ok"), ) .await; - match env.result { - Err(e) => assert_eq!(e.code, "channel:control_not_implemented"), - Ok(_) => panic!("control should return not-implemented"), - } + let out = env.result.expect("control routed"); + assert_eq!(out, json!({ "control": { "applied": true } })); + + // A callback error passes through verbatim. + let (second, _s2, _r2) = manager + .open_channel("alk/tty", "alice", None) + .await + .expect("second open"); + manager.set_control_handle( + second, + Arc::new( + move |_message: Value| -> BoxFuture<'static, Result> { + Box::pin(async move { + Err(CallError::new( + "channel:target_gone", + "the controlling target is gone", + false, + )) + }) + }, + ), + ); + let env = handler( + json!({ "channel_id": second, "message": {} }), + test_context("ctrl-err"), + ) + .await; + let e = env.result.expect_err("callback error passes through"); + assert_eq!(e.code, "channel:target_gone"); } #[test] @@ -1691,28 +2192,105 @@ mod tests { ); } + /// ADR-047 amendment 4 §4.4 gate: the wrapper entry points refuse + /// wrong shapes at registration. #[tokio::test] - async fn make_open_handler_sink_returns_not_implemented() { + async fn marked_pub_entry_points_refuse_wrong_shapes() { + use crate::registry::spec::{ChannelOpenSpec, RESERVED_OPEN_INPUT_KEY}; let manager = make_manager().await; let policy = super::super::policy::default_policy(); - let open_handler: OpenHandler = - Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); + let core = ChannelCore::new(manager.clone(), Arc::clone(&policy)); + let registry = OperationRegistry::new(); let auth = AuthContext::anonymous(b"alk/call"); - let handler = make_open_handler_sink( - manager, - policy, + let pub_open_handler: PubOpenHandler = Arc::new(|_input, _plan, _conn, _auth| { + Box::pin(async { Err(CallError::internal("x")) }) + }); + + let unmarked = OperationSpec::new( + "channels/blobs/put", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), None, - open_handler, - auth, - "alk/tty".to_string(), ); - let publish_stream: crate::registry::registration::PublishStream = - Box::pin(futures::stream::empty()); - let env = handler(json!({}), test_context("open-sink-1"), publish_stream).await; - match env.result { - Err(e) => assert_eq!(e.code, "INTERNAL"), - Ok(_) => panic!("sink open should return not-implemented"), - } + let e = core + .register_pub_openable( + unmarked.clone(), + None, + Arc::clone(&pub_open_handler), + ®istry, + auth.clone(), + None, + ) + .expect_err("unmarked refused"); + assert!(e.contains("no channel_open marker"), "got: {e}"); + + let sub_spec = OperationSpec::new( + "channels/blobs/fetch", + OperationType::Sub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alk/blobs")); + let e = core + .register_pub_openable( + sub_spec.clone(), + None, + Arc::clone(&pub_open_handler), + ®istry, + auth.clone(), + None, + ) + .expect_err("Sub-typed refused"); + assert!(e.contains("only `Pub`-typed"), "got: {e}"); + + let pub_marked = OperationSpec::new( + "channels/blobs/put", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alk/blobs")); + let _open_handler: OpenHandler = + Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); + let e = core + .register_openable_with_establisher( + pub_marked.clone(), + None, + Arc::clone(&_open_handler), + ®istry, + auth.clone(), + None, + ) + .expect_err("Pub refused by register_openable*"); + assert!(e.contains("Pub-typed"), "got: {e}"); + + let pub_marked_schema = pub_marked + .clone() + .with_publish_schema(json!({ "type": "object" })); + let e = core + .register_pub_openable( + pub_marked_schema, + None, + Arc::clone(&pub_open_handler), + ®istry, + auth, + None, + ) + .expect_err("publish_schema refused on a marked spec (ADR-047 amendment 4 §4.2)"); + assert!(e.contains("channel_open"), "got: {e}"); + let _ = RESERVED_OPEN_INPUT_KEY; } #[tokio::test] @@ -1800,35 +2378,6 @@ mod tests { assert!(spawned.load(Ordering::SeqCst)); } - #[tokio::test] - async fn register_openable_pub_registers_stub() { - let manager = make_manager().await; - let policy = super::super::policy::default_policy(); - let core = ChannelCore::new(manager.clone(), policy); - let spec = OperationSpec::new( - "channels/tty/pub", - OperationType::Pub, - Visibility::External, - json!({}), - json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }), - vec![], - AccessControl::default(), - None, - ) - .with_channel_open(ChannelOpenSpec::new("alk/tty")); - let open_handler: OpenHandler = - Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {})); - let registry = OperationRegistry::new(); - core.register_openable( - spec, - open_handler, - ®istry, - AuthContext::anonymous(b"alk/call"), - ) - .expect("register"); - assert!(registry.registration("channels/tty/pub").is_some()); - } - #[test] fn register_openable_rejects_spec_without_channel_open_marker() { let (_client, server) = duplex(1024); @@ -2852,3 +3401,7 @@ mod tests { } } } + +#[cfg(test)] +#[path = "pub_open_tests.rs"] +mod pub_open_tests; diff --git a/src/channels/pub_open_tests.rs b/src/channels/pub_open_tests.rs new file mode 100644 index 0000000..5f631ad --- /dev/null +++ b/src/channels/pub_open_tests.rs @@ -0,0 +1,447 @@ +//! The marked-pub open contract gates (review 011 §3; ADR-047 +//! amendment 4). The put path end-to-end over a real channels +//! connection: response merge order, the digest-in-result wire +//! contract, publish-chunk rejection, establishment failure +//! (teardown + ledger + EOF-discarded racing bytes), the reserved-key +//! rejections, and adopt parity. + +use crate::channels::adapter::ChannelsAdapter; +use crate::channels::client::ChannelClient; +use crate::channels::manager::ChannelManager; +use crate::channels::operations::{ChannelCore, Establishment, OpenEstablisher, PubOpenHandler}; +use crate::channels::policy::NoCap; +use crate::core::auth::{AuthContext, Identity, IdentityProvider}; +use crate::core::types::Connection; +use crate::protocol::connection::{split_single_stream, CallConnection}; +use crate::protocol::dispatch::Dispatcher; +use crate::protocol::wire::CallError; +use crate::registry::discovery::install_bootstrap_discovery; +use crate::registry::registration::OperationRegistry; +use crate::registry::spec::{ + AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility, + RESERVED_OPEN_INPUT_KEY, +}; +use serde_json::{json, Value}; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; +use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; + +const TEST_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321); +const PUT_OP: &str = "channels/blobs/put"; + +struct NoopIdProvider; +impl IdentityProvider for NoopIdProvider { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, _: &crate::core::auth::AuthToken) -> Option { + None + } +} + +fn pub_put_spec() -> OperationSpec { + OperationSpec::new( + PUT_OP, + OperationType::Pub, + Visibility::External, + json!({ + "type": "object", + "properties": { "path": { "type": "string" } }, + "required": ["path"] + }), + json!({ + "type": "object", + "properties": { + "channel_id": { "type": "integer" }, + "token": { "type": "string" }, + "digest": { "type": "string" }, + "priority": { "type": "string" } + } + }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alk/blobs")) +} + +/// The put handler: consume the adopted channel's data plane to EOF, +/// return `{ digest, priority }` — the unknown-length put's wire +/// contract (the result computed from the streamed bytes). The +/// `priority` field exists to pin the merge order (the handler's +/// result wins the collision; ADR-047 amendment 4 §4.4). +fn digest_handler() -> PubOpenHandler { + Arc::new(move |_input, _plan, channel_conn, _auth| { + Box::pin(async move { + let mut bidi = channel_conn + .accept_bi() + .await + .map_err(|_| CallError::internal("put handler: the data plane never opened"))?; + let mut buf = Vec::new(); + bidi.read_to_end(&mut buf) + .await + .map_err(|e| CallError::internal(format!("put handler read failed: {e}")))?; + Ok(json!({ + "digest": format!("len-{}", buf.len()), + "priority": "handler", + })) + }) + }) +} + +/// The establisher: contributes an informative-in-result ack field +/// (`token`) and its own `priority` (the ack — which the handler's +/// result beats on collision). +fn token_establisher() -> OpenEstablisher { + Arc::new(|_input: Value, _auth| { + Box::pin(async { + Ok(Establishment::default().with_reply_field("token", json!("tok-1"))) + .map(|e: Establishment| e.with_reply_field("priority", json!("ack"))) + }) + }) +} + +/// Build the put-serving side over the duplex's server end: register +/// the marked pub openable (with the given establisher + handler) and +/// run the channel-0 dispatch loop. Returns the captured server +/// manager (the ledger/teardown observability point). +async fn serve_put_side( + server_end: DuplexStream, + establisher: OpenEstablisher, + handler: PubOpenHandler, +) -> Arc>> { + let captured = Arc::new(tokio::sync::Mutex::>::new(None)); + let captured_for_hook = Arc::clone(&captured); + let install_hook: crate::channels::adapter::InstallChannelZero = + Arc::new(move |manager, channel0_conn, auth| { + let establisher = establisher.clone(); + let handler = Arc::clone(&handler); + let manager = manager.clone(); + let captured = Arc::clone(&captured_for_hook); + tokio::spawn(async move { + let _ = captured.lock().await.replace(manager.clone()); + let Ok(channel0_bidi) = channel0_conn.accept_bi().await else { + return; + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let core = ChannelCore::new(manager, Arc::new(NoCap)); + let registry = Arc::new(OperationRegistry::new()); + install_bootstrap_discovery(®istry).expect("bootstrap discovery"); + core.register_pub_openable( + pub_put_spec(), + Some(establisher), + handler, + ®istry, + auth.clone(), + None, + ) + .expect("register the marked pub openable"); + let provider: Arc = Arc::new(NoopIdProvider); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + Dispatcher::new(registry, provider) + .run_loop_single_stream(call_connection, reader, writer) + .await; + }) + }); + let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap)); + let server_conn = Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let auth = AuthContext::anonymous(b"alk/channels"); + tokio::spawn(async move { + let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await; + }); + captured +} + +#[tokio::test] +async fn put_e2e_merge_order_and_digest() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let _captured = serve_put_side(server_end, token_establisher(), digest_handler()).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("channel client init"); + + { + // The known-length shape: offer + bytes → one response whose + // merge order is `{channel_id}` → ack fields → handler result. + let (_id, mut send, _recv, reply) = client + .open_pub_channel(PUT_OP, json!({ "path": "b/one" }), "alk/blobs") + .await + .expect("open_pub_channel"); + send.write_all(b"hello").await.expect("stream the bytes"); + drop(send); + + let merged = reply + .resolve() + .await + .expect("the put resolves with the merged response"); + let _ = _id; + assert_eq!( + merged["token"], + json!("tok-1"), + "the ack field rides the result" + ); + assert_eq!( + merged["digest"], + json!("len-5"), + "the handler result rides the result" + ); + assert_eq!( + merged["priority"], + json!("handler"), + "merge order: handler result wins the collision with the ack (§4.4)" + ); + assert!( + merged["channel_id"].as_u64().is_some(), + "channel_id reserved" + ); + } + + { + // The unknown-length shape: larger than the reassembly buffer + // (crosses chunk boundaries); the handler's digest is computed + // over the streamed bytes. + let (_id2, send2, _recv2, reply2) = client + .open_pub_channel(PUT_OP, json!({ "path": "b/two" }), "alk/blobs") + .await + .expect("open_pub_channel"); + let mut send2 = send2; + let chunk = vec![0xAB_u8; 4096]; + for _ in 0..8 { + send2.write_all(&chunk).await.expect("stream a chunk"); + } + drop(send2); + + let merged2 = reply2.resolve().await.expect("unknown-length put"); + assert_eq!(merged2["digest"], json!("len-32768")); + assert_eq!(merged2["token"], json!("tok-1")); + } +} + +/// The happy-path put's ids: the putter allocates odd (its side is +/// Connect); the responder adopts; the reply carries the putter's id. +#[tokio::test] +async fn put_e2e_client_parity_is_odd() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let _captured = serve_put_side(server_end, token_establisher(), digest_handler()).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("client"); + let (id, mut send, _recv, reply) = client + .open_pub_channel(PUT_OP, json!({ "path": "b/odd" }), "alk/blobs") + .await + .expect("open_pub_channel"); + send.write_all(b"zz").await.expect("stream"); + drop(send); + let merged = reply.resolve().await.expect("put"); + assert_eq!( + merged["channel_id"], + json!(id), + "the reply carries the putter's id" + ); + assert_eq!(id % 2, 1, "the connect side allocates odd (ADR-047 §5)"); +} + +/// Publish-chunk rejection (§4.1): a marked op fed via the call-plane +/// publish path rejects loudly in the initiator's direction and the +/// handler never consumes a chunk. +#[tokio::test] +async fn put_rejects_publish_chunks_loudly() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let _captured = serve_put_side(server_end, token_establisher(), digest_handler()).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("client"); + let conn = Arc::clone(&client.take_call_connection().await.expect("call conn")); + + let chunks = Box::pin(futures::stream::iter(vec![ + json!({ "chunk": 1 }), + json!({ "chunk": 2 }), + ])); + // The reserved key is present (this client misuses the wire + // contract — the open resolves, the chunks are what get rejected). + let response = conn + .publish_with_payload( + json!({ "operationId": PUT_OP, "input": { "path": "b/violating", RESERVED_OPEN_INPUT_KEY: 3 } }), + chunks, + ) + .await; + let e = response.result.expect_err("chunks rejected loudly"); + assert_eq!(e.code, crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED); + // The put side's own single response may also arrive (the initiator + // already resolved with the rejection); the rejection is what the + // pending call carries — assert it was NOT a success-shaped result + // (done above by expect_err). + let _ = conn; +} + +/// Establishment failure (ADR-049 §3 through the pub path): the reply +/// is `channel:open_failed` with the mapped reason; the channel is +/// torn down on both sides; racing bytes EOF-discard (the write +/// completes locally before the failure reply — the adopt entry is +/// gone afterward, ledger un-incremented). +#[tokio::test] +async fn put_establishment_failure_replies_open_failed_and_tears_down() { + use crate::channels::operations::EstablishmentError; + + let failing: OpenEstablisher = Arc::new(|_input, _auth| { + Box::pin(async { + Err(EstablishmentError::UnknownResource { + message: "container not found".to_string(), + }) + }) + }); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let captured = serve_put_side(server_end, failing, digest_handler()).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("client"); + let (id, mut send, _recv, reply) = client + .open_pub_channel(PUT_OP, json!({ "path": "b/gone" }), "alk/blobs") + .await + .expect("open_pub_channel"); + // Racing bytes: written before the failure reply arrives — the + // responder never adopted, so the bytes park and drop (EOF-shaped + // for the putter). + let _ = send.write_all(b"racing").await; + drop(send); + + let error = reply + .resolve() + .await + .expect_err("the put resolves with the establishment failure"); + assert_eq!( + error.establishment_reason(), + Some("unknown_resource"), + "the spoke establishment reason maps through (ADR-049 §3)" + ); + assert_eq!( + error.call_error().map(|e| e.code.as_str()), + Some("channel:open_failed") + ); + + // The responder's side: no channel lingers (open_failed tears + // down; nothing was adopted anyway — but the client-side + // open-failure teardown removed the allocator-side entry). + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let manager = captured.lock().await.clone().expect("server manager"); + assert!( + !manager.has_channel(id), + "a failed establishment leaves no channel (ADR-047 amendment 4 §4.4)" + ); + assert_eq!( + manager.opener_ledger().take(id), + None, + "ledger un-incremented" + ); +} + +/// A marked pub op whose input lacks the reserved `channel_id` key is +/// rejected (`INVALID_INPUT`) — the initiator-allocated id is +/// mandatory (ADR-047 amendment 4 §4.3). +#[tokio::test] +async fn put_missing_reserved_key_is_rejected() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let _captured = serve_put_side(server_end, token_establisher(), digest_handler()).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("client"); + let response = client + .call_open_op(PUT_OP, json!({ "path": "b/no-id" })) + .await; + let e = response.result.expect_err("the reserved key is mandatory"); + assert_eq!(e.code, "INVALID_INPUT"); + assert!( + e.message.contains(RESERVED_OPEN_INPUT_KEY), + "the error names the key: {}", + e.message + ); +} + +/// Adopt parity (§4.4 + ADR-047 §5): a reserved id inside the +/// responder's OWN allocation range (even — an id the accept side +/// would allocate itself, violating the split) fails the open with +/// `adopt_failed`; racing bytes are discarded. +#[tokio::test] +async fn put_out_of_range_reserved_id_replies_adopt_failed() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let _captured = serve_put_side(server_end, token_establisher(), digest_handler()).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("client"); + let response = client + .call_open_op( + PUT_OP, + json!({ "path": "b/bad", RESERVED_OPEN_INPUT_KEY: 2 }), + ) + .await; + let e = response + .result + .expect_err("the responder refuses its own range"); + assert_eq!(e.code, "channel:open_failed"); + assert_eq!( + e.details + .as_ref() + .and_then(|d| d.get("reason")) + .and_then(|r| r.as_str()), + Some("adopt_failed"), + "adoption failure → adopt_failed (ADR-047 amendment 4 §4.4): {e:?}" + ); +} + +/// The responder-side put handler that returns an error: the single +/// response IS that error (the wrapper passthrough; §4.4). +#[tokio::test] +async fn put_handler_error_becomes_the_response() { + let failing_handler: PubOpenHandler = Arc::new(|_input, _plan, channel_conn, _auth| { + Box::pin(async move { + let mut bidi = channel_conn + .accept_bi() + .await + .map_err(|_| CallError::internal("data plane"))?; + let mut buf = [0u8; 3]; + let _ = bidi.read_exact(&mut buf).await; + Err(CallError::new( + "blobs:read_only", + "the container is read-only", + false, + )) + }) + }); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR)); + let _captured = serve_put_side(server_end, token_establisher(), failing_handler).await; + + let client = ChannelClient::from_connection(client_conn) + .await + .expect("client"); + let (_id, mut send, _recv, reply) = client + .open_pub_channel(PUT_OP, json!({ "path": "b/ro" }), "alk/blobs") + .await + .expect("open_pub_channel"); + send.write_all(b"abc").await.expect("stream"); + drop(send); + + let error = reply + .resolve() + .await + .expect_err("the handler error IS the response"); + assert_eq!( + error.call_error().map(|e| e.code.clone()), + Some("blobs:read_only".to_string()) + ); +} diff --git a/src/channels/relay.rs b/src/channels/relay.rs index 447c953..f3a37c2 100644 --- a/src/channels/relay.rs +++ b/src/channels/relay.rs @@ -35,11 +35,13 @@ use crate::registry::spec::{OperationSpec, OperationType}; use super::manager::ChannelManager; use super::operations::{ - ChannelCore, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, - CHANNEL_OPEN_FAILED, RESERVED_REPLY_KEY, + ChannelCore, ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, + PubOpenHandler, CHANNEL_OPEN_FAILED, RESERVED_REPLY_KEY, }; use super::pump::pump_bidi; + use super::reassembly::{MpscRecvStream, MpscSendStream}; +use tokio::io::AsyncWriteExt; /// The producer-leg surface the relay holds (ADR-051 §2): the /// `CallConnection` for the translate hop's open call plus the @@ -81,8 +83,17 @@ struct RelayPlan { producer_manager: ChannelManager, spoke_send: Option, spoke_recv: Option, + /// The put path (ADR-047 amendment 4 §4.7): the producer leg's + /// open-op response, awaited by the relay pub handler after the + /// consumer EOF (put-acks-in-result — the spoke's response cannot + /// exist before its byte stream completes, so the establisher + /// never awaits it). + pub_reply: Option>, } +/// The producer leg's open-op response, carried on the put path's plan. +type CallConnectionResponse = crate::protocol::wire::ResponseEnvelope; + impl Drop for RelayPlan { fn drop(&mut self) { let _ = self.producer_manager.teardown_channel(self.spoke_id); @@ -120,9 +131,10 @@ impl ChannelRelay { /// produced open op. /// - `spec` must carry the `channel_open` marker (reconstructed /// WITH the marker by discovery, ADR-047 amendment 3). A - /// `Pub`-typed marked spec is a loud assembly error - /// ([`RelayError::PubTypedOpen`] — the C-08 blocker, ADR-051 - /// §6); an unmarked spec is likewise an error — branch on the + /// `Pub`-typed marked spec composes through the put-forwarding + /// translation (see [`Self::register_relay_pub_openable`]); an + /// unmarked spec is a + /// [`RelayError::UnmarkedSpec`] error — branch on the /// marker before calling. /// - The translate hop calls the producer leg with the spec's own /// op name; a hub that re-names the consumer-visible op composes @@ -145,14 +157,12 @@ impl ChannelRelay { }); } if spec.op_type == OperationType::Pub { - return Err(RelayError::PubTypedOpen { - message: format!( - "spec `{}` is a Pub-typed channel-open op — the Pub open path is \ - not implemented (C-08 blocker, `channel:pub_open_not_implemented` \ - class; ADR-051 §6)", - spec.name - ), - }); + return self.register_relay_pub_openable( + consumer_core, + consumer_registry, + spec, + consumer_auth, + ); } let producer_call = Arc::clone(&self.producer_leg.call); @@ -227,6 +237,76 @@ impl ChannelRelay { ) .map_err(|e| RelayError::Registration { message: e }) } + + /// Register a from_call-imported **marked `Pub`** spec as a relay + /// openable — the put-forwarding translation (ADR-047 amendment 4 + /// §4.7). The consumer leg runs the full marked-pub wrapper (adopt + /// → establish → pub handler awaited inline); this side's compose: + /// + /// - the establisher allocates the hub's connect-side id on the + /// producer leg (`open_channel` — an odd id, disjoint from the + /// spoke's even allocations), sends the open op **in the + /// background** (the spoke's put response cannot exist before + /// its byte stream completes; an establisher that awaited it + /// deadlocks the leg), and plans the spoke-leg streams + the + /// response's oneshot; the plan's Drop guard reclaims the + /// producer-leg entry on every handler path, as ADR-051 §6 pins. + /// - the handler pumps the consumer-leg bytes → the spoke leg + /// (one direction, pub), EOFs the spoke side on consumer EOF, + /// then awaits the spoke's single response, strips the + /// producer's reserved reply key (the consumer reply carries the + /// hub-allocated consumer-leg id; a spoke `channel_id` merged + /// through would shadow it — the same reserved-key class the + /// wrapper refuses), and returns the remainder — the consumer-leg + /// wrapper merges it with the pinned order (ack fields are none + /// here; the put acks in the spoke's result, §4.5). + /// + /// A spoke open-time establishment failure surfaces verbatim in + /// the final response (put-acks-in-result; the spoke's mapped + /// `channel:open_failed` rides the response, not the establisher). + fn register_relay_pub_openable( + &self, + consumer_core: &ChannelCore, + consumer_registry: &OperationRegistry, + spec: OperationSpec, + consumer_auth: AuthContext, + ) -> Result<(), RelayError> { + let producer_call = Arc::clone(&self.producer_leg.call); + let producer_manager = self.producer_leg.manager.clone(); + let op_name = spec.name.clone(); + + let establisher: OpenEstablisher = Arc::new(move |input, per_call_auth| { + let producer_call = Arc::clone(&producer_call); + let producer_manager = producer_manager.clone(); + let op_name = op_name.clone(); + Box::pin(async move { + open_pub_on_producer_leg( + &producer_call, + &producer_manager, + &op_name, + input, + &per_call_auth, + ) + .await + }) + }); + + let pub_open_handler: PubOpenHandler = + Arc::new(move |_input, plan, channel_conn, _auth| { + Box::pin(async move { pub_relay_forward(channel_conn, plan).await }) + }); + + consumer_core + .register_pub_openable( + spec, + Some(establisher), + pub_open_handler, + consumer_registry, + consumer_auth, + None, + ) + .map_err(|e| RelayError::Registration { message: e }) + } } /// The translate hop (ADR-051 §1 layer 1, §2): call the producer leg's @@ -288,6 +368,7 @@ async fn open_on_producer_leg( producer_manager: producer_manager.clone(), spoke_send: Some(spoke_send), spoke_recv: Some(spoke_recv), + pub_reply: None, }) as super::operations::ChannelPlan); Ok(if reply_fields.is_empty() { establishment @@ -334,6 +415,123 @@ fn map_spoke_error(spoke_error: &CallError) -> EstablishmentError { } } +/// The put path's translate hop (ADR-047 amendment 4 §4.7): allocate +/// the hub's connect-side channel id on the producer leg, send the +/// spoke's open op **in the background** with the reserved input key +/// set, and plan the spoke-leg streams + the response oneshot. The +/// establisher returns promptly; the spoke's response arrives when the +/// spoke's put completes (or fails) — put-acks-in-result. +async fn open_pub_on_producer_leg( + producer_call: &Arc, + producer_manager: &ChannelManager, + op_name: &str, + input: Value, + per_call_auth: &AuthContext, +) -> Result { + let (spoke_id, spoke_send, spoke_recv) = producer_manager + .open_channel( + "alk/relay", + // The opener ledger is inert on the hub's connect side + // (the connect side does not enforce the per-identity cap, + // ADR-041) — the record is observability only. + "hub-relay", + None, + ) + .await + .map_err(|e| EstablishmentError::ResourceShortage { + message: format!("producer leg: channel allocation failed: {e}"), + })?; + + let mut hop_input = input.clone(); + if let Some(obj) = hop_input.as_object_mut() { + obj.insert( + crate::registry::spec::RESERVED_OPEN_INPUT_KEY.to_string(), + serde_json::json!(spoke_id), + ); + } + let payload = build_relay_payload(op_name, hop_input, per_call_auth); + + let (reply_tx, reply_rx) = tokio::sync::oneshot::channel(); + let producer_call_task = Arc::clone(producer_call); + tokio::spawn(async move { + let _ = reply_tx.send(producer_call_task.call_with_payload(payload).await); + }); + + Ok(Establishment::new(Arc::new(RelayPlan { + spoke_id, + producer_manager: producer_manager.clone(), + spoke_send: Some(spoke_send), + spoke_recv: Some(spoke_recv), + pub_reply: Some(reply_rx), + }) as ChannelPlan)) +} + +/// The relay pub handler (ADR-047 amendment 4 §4.7): pump the +/// consumer-leg bytes → the spoke leg (one direction, pub — the +/// shutdown-on-completion contract, ADR-050's), then await the spoke's +/// single response and strip its reserved reply key. The plan's Drop +/// guard reclaims the producer-leg entry on every path. +async fn pub_relay_forward( + channel_conn: crate::core::types::Connection, + plan: Option, +) -> Result { + let Some(plan) = plan else { + return Err(CallError::internal( + "relay: pub establisher produced no plan", + )); + }; + let Ok(relay_plan_arc) = plan.downcast::() else { + return Err(CallError::internal( + "relay: establisher plan of unexpected type", + )); + }; + let Ok(mut relay_plan) = Arc::try_unwrap(relay_plan_arc) else { + return Err(CallError::internal( + "relay: establisher plan held elsewhere; skipping pump", + )); + }; + let Some(mut spoke_send) = relay_plan.spoke_send.take() else { + return Err(CallError::internal( + "relay: pub plan missing the spoke send half", + )); + }; + relay_plan.spoke_recv.take(); + + let Ok(bidi) = channel_conn.accept_bi().await else { + return Err(CallError::internal( + "relay: consumer leg data plane never opened", + )); + }; + let (mut read_half, _write_half) = tokio::io::split(bidi); + // The one-direction pump: consumer bytes → spoke; shutdown on + // consumer EOF (the pump contract, ADR-050). + let _copied = tokio::io::copy(&mut read_half, &mut spoke_send) + .await + .unwrap_or(0); + let _ = spoke_send.shutdown().await; + drop(spoke_send); + + match relay_plan.pub_reply.take() { + Some(rx) => match rx.await { + Ok(response) => match response.result { + Ok(mut out) => { + if let Some(obj) = out.as_object_mut() { + obj.remove(RESERVED_REPLY_KEY); + } + Ok(out) + } + Err(e) => Err(e), + }, + Err(_) => Err(CallError::internal( + "relay: producer-leg open resolved without a response", + )), + }, + None => Err(CallError::internal( + "relay: pub plan missing the response receiver", + )), + } +} + /// Build the `call.requested` payload for the translate hop: the hub /// as caller, the consumer as `forwarded_for` (ADR-026 §3, ADR-032 §3 /// — the same shape `build_forwarded_payload` produces for the plain @@ -357,18 +555,15 @@ fn build_relay_payload(op_name: &str, input: Value, per_call_auth: &AuthContext) /// The assembly-time errors `ChannelRelay::register_relay_openable` /// produces (ADR-051 §6 — the rejection postures are loud, never -/// silent stubs). +/// silent stubs; the `PubTypedOpen` variant was removed by ADR-047 +/// amendment 4 §4.7 — pub-typed marked specs translate now, never +/// reject-and-strand). #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] pub enum RelayError { /// The spec carries no `channel_open` marker — the relay only /// registers marked specs. #[error("unmarked spec: {message}")] UnmarkedSpec { message: String }, - /// A `Pub`-typed marked spec — the Pub open path is not - /// implemented (C-08 blocker; `channel:pub_open_not_implemented` - /// class). - #[error("pub-typed open op: {message}")] - PubTypedOpen { message: String }, /// The consumer-leg registration itself failed (schema compile, /// handler-kind mismatch). #[error("registration failed: {message}")] diff --git a/src/channels/relay_tests.rs b/src/channels/relay_tests.rs index 6d464ec..5018cdc 100644 --- a/src/channels/relay_tests.rs +++ b/src/channels/relay_tests.rs @@ -89,15 +89,21 @@ async fn unmarked_spec_is_rejected_at_registration() { } #[tokio::test] -async fn pub_typed_marked_spec_is_rejected_at_registration() { +async fn pub_typed_marked_spec_registers_through_the_relay() { + // ADR-047 amendment 4 §4.7: the put-forwarding translation replaced + // the `PubTypedOpen` rejection — a pub-typed marked spec composes + // through `register_pub_openable` (the full marked-pub wrapper + // machinery) and lands on the consumer-leg registry, discoverable. + let producer_manager = manager_for_test(); let relay = ChannelRelay::new(ProducerLeg { call: overlay_connection(), - manager: manager_for_test(), + manager: producer_manager, }); - let err = relay + let registry = OperationRegistry::new(); + relay .register_relay_openable( &core_for_test(), - &OperationRegistry::new(), + ®istry, relay_spec( "channels/x/pub", OperationType::Pub, @@ -105,12 +111,10 @@ async fn pub_typed_marked_spec_is_rejected_at_registration() { ), AuthContext::anonymous(b"alk/call"), ) - .expect_err("Pub-typed marked spec must be rejected"); - assert!( - matches!(err, RelayError::PubTypedOpen { .. }), - "got {err:?}" - ); - assert!(err.to_string().contains("pub_open_not_implemented")); + .expect("the pub openable registers through the relay"); + assert!(registry.registration("channels/x/pub").is_some()); + let registration = registry.registration("channels/x/pub").expect("registered"); + assert!(registration.spec.channel_open.is_some()); } // --- ADR-051 §3: the reason mapping -------------------------------------- @@ -228,6 +232,7 @@ async fn relay_plan_drop_reclaims_the_adopted_producer_leg_entry() { producer_manager: producer_manager.clone(), spoke_send: Some(spoke_send), spoke_recv: Some(spoke_recv), + pub_reply: None, }; drop(plan); @@ -256,7 +261,16 @@ fn manager_for_test() -> ChannelManager { tokio::spawn(async move { let _ = runner.run().await; }); - ChannelManager::with_defaults(handle, None) + // The producer leg's manager belongs to the hub, which is the + // connect side of the hub↔spoke channels connection (ADR-051 §4 — + // the spoke's ids are even, the hub's odd). + ChannelManager::new( + handle, + crate::channels::manager::DEFAULT_MAX_CHANNELS, + crate::channels::reassembly::DEFAULT_BUFFER_CAP, + None, + crate::channels::manager::ChannelSide::Connect, + ) } fn core_for_test() -> ChannelCore { diff --git a/src/channels/resource.rs b/src/channels/resource.rs new file mode 100644 index 0000000..7075ecf --- /dev/null +++ b/src/channels/resource.rs @@ -0,0 +1,70 @@ +//! `ResourceEnumerators` — the registry behind +//! `channel/resources/subscribe`'s live subscription (ADR-053). +//! +//! Each openable-ALPN crate registers one enumerator (its resource +//! truth: which containers are running, which sessions are active); +//! the subscribe handler aggregates them. The channels layer never +//! interprets the payloads — they ride opaquely, merged with the +//! producing ALPN's name (ADR-053 §1). + +use std::collections::HashMap; +use std::pin::Pin; +use std::sync::Arc; + +use futures::Stream; +use parking_lot::RwLock; +use serde_json::Value; + +/// A registered enumerator's item: `(alpn, payload)` — the payload is +/// the ALPN's full current resource state (snapshot-then-changes, +/// ADR-053 §1). The channels layer does not interpret it. +pub type ResourceItem = (String, Value); + +/// A registered enumerator's merged stream. +pub type ResourceStream = Pin + Send>>; + +/// The resource enumerator contract (ADR-053 §1). The producing crate +/// owns its change detection; the first item on the returned stream is +/// the current snapshot, and each subsequent item is a full snapshot +/// of the changed set. +pub trait ResourceEnumerator: Send + Sync + 'static { + fn subscribe(&self) -> ResourceStream; +} + +/// The registry of ALPN resource enumerators (ADR-053 §2). Shared per +/// accepting peer (like `PerIdentityChannelPolicy` — ADR-041): one +/// `Arc` across every channels connection that +/// peer accepts. Registered at assembly/install time; the subscribe +/// handler scans it once at subscription start (the scan-once pin — +/// late registrations are invisible to in-flight subscriptions; +/// re-subscribe to see them). +#[derive(Clone, Default)] +pub struct ResourceEnumerators { + inner: Arc>>>, +} + +impl ResourceEnumerators { + pub fn new() -> Self { + Self::default() + } + + /// Registered at assembly/install time by the composing crate. + pub fn register(&self, alpn: &str, enumerator: Arc) { + self.inner.write().insert(alpn.to_string(), enumerator); + } + + /// The registered ALPNs (observability). + pub fn alpns(&self) -> Vec { + self.inner.read().keys().cloned().collect() + } + + /// Snapshot the registered set (the subscribe handler's scan-once + /// read; ADR-053 §2). + pub(crate) fn snapshot(&self) -> Vec<(String, Arc)> { + self.inner + .read() + .iter() + .map(|(alpn, enumerator)| (alpn.clone(), Arc::clone(enumerator))) + .collect() + } +} diff --git a/src/client/mod.rs b/src/client/mod.rs index 8b6ec08..66f24f3 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -13,7 +13,10 @@ pub use from_call::{from_call, FromCallConfig}; // crate-internal surface for the `op/register` bootstrap path (review // 004 F-05): the forwarding-handler constructor and the spec wire // parser are shared with `registry::op_register`. -pub(crate) use from_call::{make_forwarding_handler, rebuild_spec_for}; +pub(crate) use from_call::{ + make_forwarding_handler, make_sink_forwarding_handler, make_streaming_forwarding_handler, + rebuild_spec_for, +}; use crate::registry::registration::HandlerRegistration; diff --git a/src/protocol/adapter.rs b/src/protocol/adapter.rs index 507c0d4..2a736aa 100644 --- a/src/protocol/adapter.rs +++ b/src/protocol/adapter.rs @@ -538,7 +538,7 @@ mod tests { None, Capabilities::new(), ); - conn.register_imported(imported); + let _error = conn.register_imported(imported); let peer_identity = Identity { id: "worker-a".to_string(), scopes: vec![], @@ -607,7 +607,7 @@ mod tests { None, Capabilities::new(), ); - conn.register_imported(imported); + let _error = conn.register_imported(imported); let context = adapter.build_root_context("req-6".to_string(), "fs/readFile", None, None, &conn); diff --git a/src/protocol/connection.rs b/src/protocol/connection.rs index 3249dd3..d3bcedc 100644 --- a/src/protocol/connection.rs +++ b/src/protocol/connection.rs @@ -159,9 +159,27 @@ impl CallConnection { &self.pending } - pub fn register_imported(&self, registration: HandlerRegistration) { + /// Land a peer-announced/discovered bundle on the connection + /// overlay. Marked (`channel_open`) specs refuse here: a marked op + /// is unreachable by composition (`OperationEnv::invoke` is + /// request/response-only; ADR-046 §6) and its data plane has no + /// overlay path — registration-time loudness, never an advertised + /// dead body (ADR-047 amendment 4 §4.8; ADR-022's 2026-10-13 + /// amendment). Relay composition + /// (`ChannelRelay::register_relay_openable`) is the sanctioned + /// marked-op import. + pub fn register_imported(&self, registration: HandlerRegistration) -> Result<(), String> { + if registration.spec.channel_open.is_some() { + return Err(format!( + "register_imported: spec `{}` carries the `channel_open` marker — marked \ + specs are not overlay-importable; compose `ChannelRelay` or consume via \ + `ChannelClient` (ADR-047 amendment 4 §4.8)", + registration.spec.name + )); + } let name = registration.spec.name.clone(); self.imported_operations.write().insert(name, registration); + Ok(()) } /// `true` when the connection overlay holds a registration for @@ -176,11 +194,24 @@ impl CallConnection { self.imported_operations.read().get(name).cloned() } - pub fn register_imported_all(&self, registrations: Vec) { + pub fn register_imported_all( + &self, + registrations: Vec, + ) -> Result<(), String> { let mut overlay = self.imported_operations.write(); for reg in registrations { - overlay.insert(reg.spec.name.clone(), reg); + if reg.spec.channel_open.is_some() { + return Err(format!( + "register_imported_all: spec `{}` carries the `channel_open` marker — \ + marked specs are not overlay-importable; compose `ChannelRelay` or \ + consume via `ChannelClient` (ADR-047 amendment 4 §4.8)", + reg.spec.name + )); + } + let name = reg.spec.name.clone(); + overlay.insert(name, reg); } + Ok(()) } pub fn overlay_env(&self) -> Arc { @@ -975,7 +1006,7 @@ mod tests { assert!(!env.contains("worker/exec")); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); assert!(env.contains("worker/exec")); assert!(!env.contains("worker/missing")); @@ -986,11 +1017,12 @@ mod tests { let conn = CallConnection::new(stub_connection()); let env = conn.overlay_env(); - conn.register_imported_all(vec![ + let registrations_result = conn.register_imported_all(vec![ imported_registration("worker/exec"), imported_registration("worker/status"), imported_registration("fs/readFile"), ]); + let _error = registrations_result; assert!(env.contains("worker/exec")); assert!(env.contains("worker/status")); @@ -1001,7 +1033,7 @@ mod tests { #[tokio::test] async fn overlay_env_dispatches_to_imported_op() { let conn = CallConnection::new(stub_connection()); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); let env = conn.overlay_env(); let scoped = ScopedPeerEnv::new(["worker/exec"]); @@ -1018,7 +1050,7 @@ mod tests { #[tokio::test] async fn overlay_env_contains_returns_false_for_non_imported_op() { let conn = CallConnection::new(stub_connection()); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); let env = conn.overlay_env(); assert!(!env.contains("worker/missing")); @@ -1039,7 +1071,7 @@ mod tests { #[tokio::test] async fn overlay_env_reachability_check_returns_not_found_for_disallowed_op() { let conn = CallConnection::new(stub_connection()); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); let env = conn.overlay_env(); let scoped = ScopedPeerEnv::empty(); @@ -1069,7 +1101,7 @@ mod tests { }), ) }); - conn.register_imported(HandlerRegistration::new( + let _error = conn.register_imported(HandlerRegistration::new( external_spec("worker/exec"), HandlerKind::Once(inspect_handler), OperationProvenance::FromCall, @@ -1115,7 +1147,7 @@ mod tests { Arc::new(RwLock::new(HashMap::new())); { let conn = CallConnection::new(stub_connection()); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); assert!(conn.overlay_env().contains("worker/exec")); std::mem::swap( &mut *captured.write(), @@ -1380,14 +1412,14 @@ mod tests { let conn = CallConnection::new_overlay_only(sample_identity("ws-peer")); let env = conn.overlay_env(); assert!(!env.contains("worker/exec")); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); assert!(env.contains("worker/exec")); } #[tokio::test] async fn overlay_only_overlay_env_dispatches_imported_op() { let conn = CallConnection::new_overlay_only(sample_identity("ws-peer")); - conn.register_imported(imported_registration("worker/exec")); + let _error = conn.register_imported(imported_registration("worker/exec")); let env = conn.overlay_env(); let scoped = ScopedPeerEnv::new(["worker/exec"]); @@ -1433,7 +1465,7 @@ mod tests { let streaming_handler = make_streaming_handler(|input, ctx| { futures::stream::iter(vec![ResponseEnvelope::ok(ctx.request_id, input)]) }); - conn.register_imported(HandlerRegistration::new( + let _error = conn.register_imported(HandlerRegistration::new( OperationSpec::new( "events/stream", OperationType::Sub, diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index eaf7868..5019589 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -86,6 +86,11 @@ pub struct SinkDispatch { pub(crate) handler: Pin + Send>>, pub(crate) chunk_tx: mpsc::Sender>, pub(crate) publish_validator: Option, + /// Marked (`channel_open`) `Pub` op (ADR-047 amendment 4 §4.1): + /// publish chunks never feed the handler — the first + /// `call.published` chunk is rejected loudly in the initiator's + /// direction and the rest are ignored. + pub(crate) reject_publish: bool, } /// An in-flight Pub (Sink) on the single-stream dispatch loop @@ -100,6 +105,11 @@ struct InFlightSink { chunk_tx: mpsc::Sender>, publish_validator: Option, handler_handle: JoinHandle<()>, + /// Marked (`channel_open`) `Pub` op (ADR-047 amendment 4 §4.1): + /// the first `call.published` chunk is rejected loudly in the + /// initiator's direction and never fed to the handler. + reject_publish: bool, + reject_written: bool, } impl std::fmt::Debug for DispatchResult { @@ -356,7 +366,7 @@ impl Dispatcher { .get("forwarded_for") .and_then(|v| serde_json::from_value::(v.clone()).ok()); - let input = payload.get("input").cloned().unwrap_or(Value::Null); + let mut input = payload.get("input").cloned().unwrap_or(Value::Null); let op_type = self .registry @@ -391,9 +401,13 @@ impl Dispatcher { context.deadline = None; match self .registry - .resolve_sink_handler(&operation_name, &input, &context) + .resolve_sink_handler(&operation_name, &mut input, &mut context) { Ok(sink_handler) => { + let reject_publish = self + .registry + .registration(&operation_name) + .is_some_and(|r| r.spec.channel_open.is_some()); let publish_validator = self.registry.publish_validator(&operation_name); let (chunk_tx, chunk_rx) = mpsc::channel::>(PUBLISH_CHANNEL_BUFFER); @@ -403,6 +417,7 @@ impl Dispatcher { handler: Box::pin(handler), chunk_tx, publish_validator, + reject_publish, }) } Err(envelope) => StartedDispatch::Once(Box::pin(std::future::ready(envelope))), @@ -657,6 +672,7 @@ impl Dispatcher { handler, mut chunk_tx, publish_validator, + reject_publish, } = sink; let (abort_tx, mut abort_rx) = oneshot::channel::<()>(); @@ -668,6 +684,7 @@ impl Dispatcher { let mut handler = handler.fuse(); let mut response: Option = None; let mut aborted = false; + let mut reject_written = false; loop { tokio::select! { @@ -698,6 +715,31 @@ impl Dispatcher { } match envelope.r#type.as_str() { EVENT_PUBLISHED => { + // Marked ops: the data plane is the adopted + // channel's `BiStream` — publish chunks are + // rejected loudly, never consumed (ADR-047 + // amendment 4 §4.1). The handler continues on + // the channel plane; the initiator's pending + // call resolves with this rejection. + if reject_publish { + if !reject_written { + reject_written = true; + let rejection: EventEnvelope = ResponseEnvelope::error( + request_id.clone(), + CallError::new( + crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED, + "the data plane of a marked pub op is the adopted \ + channel's BiStream — call.published chunks are \ + rejected (ADR-047 amendment 4 §4.1)", + false, + ), + ).into(); + if let Err(err) = writer.write_frame(&rejection).await { + warn!(error = %err, request_id = %request_id, "failed to write marked-op publish rejection"); + } + } + continue; + } let chunk = envelope .payload .get("input") @@ -923,6 +965,7 @@ impl Dispatcher { handler, chunk_tx, publish_validator, + reject_publish, } = sink; let handle = Self::spawn_sink_response_writer( &writer, @@ -935,6 +978,8 @@ impl Dispatcher { chunk_tx, publish_validator, handler_handle: handle, + reject_publish, + reject_written: false, }, ); } @@ -977,6 +1022,36 @@ impl Dispatcher { .unwrap_or(Value::Null); let entry = in_flight_sinks.lock().remove(&request_id); if let Some(mut entry) = entry { + if entry.reject_publish { + // Marked ops: the data plane is the adopted + // channel's `BiStream` — publish chunks are + // rejected loudly, never consumed (ADR-047 + // amendment 4 §4.1). The loud error is written + // once per request; the entry stays registered + // so the handler's own final response still gets + // written. The handler continues on the channel + // plane; the initiator's pending call resolves + // with this rejection immediately. + if !entry.reject_written { + entry.reject_written = true; + let rejection: EventEnvelope = ResponseEnvelope::error( + request_id.clone(), + CallError::new( + crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED, + "the data plane of a marked pub op is the \ + adopted channel's BiStream — call.published \ + chunks are rejected (ADR-047 amendment 4 §4.1)", + false, + ), + ) + .into(); + if let Err(err) = writer.write_frame(&rejection).await { + warn!(error = %err, request_id = %request_id, "failed to write marked-op publish rejection"); + } + } + in_flight_sinks.lock().insert(request_id, entry); + continue; + } let validated = match &entry.publish_validator { Some(validator) => { if validator.is_valid(&chunk) { @@ -1146,6 +1221,7 @@ impl Dispatcher { handler, chunk_tx, publish_validator, + reject_publish, } = sink; let handle = Self::spawn_sink_response_writer( &writer, @@ -1158,6 +1234,8 @@ impl Dispatcher { chunk_tx, publish_validator, handler_handle: handle, + reject_publish, + reject_written: false, }, ); } @@ -1200,6 +1278,36 @@ impl Dispatcher { .unwrap_or(Value::Null); let entry = in_flight_sinks.lock().remove(&request_id); if let Some(mut entry) = entry { + if entry.reject_publish { + // Marked ops: the data plane is the adopted + // channel's `BiStream` — publish chunks are + // rejected loudly, never consumed (ADR-047 + // amendment 4 §4.1). The loud error is written + // once per request; the entry stays registered + // so the handler's own final response still gets + // written. The handler continues on the channel + // plane; the initiator's pending call resolves + // with this rejection immediately. + if !entry.reject_written { + entry.reject_written = true; + let rejection: EventEnvelope = ResponseEnvelope::error( + request_id.clone(), + CallError::new( + crate::registry::spec::MARKED_PUB_CHUNKS_REJECTED, + "the data plane of a marked pub op is the \ + adopted channel's BiStream — call.published \ + chunks are rejected (ADR-047 amendment 4 §4.1)", + false, + ), + ) + .into(); + if let Err(err) = writer.write_frame(&rejection).await { + warn!(error = %err, request_id = %request_id, "failed to write marked-op publish rejection"); + } + } + in_flight_sinks.lock().insert(request_id, entry); + continue; + } let validated = match &entry.publish_validator { Some(validator) => { if validator.is_valid(&chunk) { diff --git a/src/registry/op_register.rs b/src/registry/op_register.rs index 2c36bec..f4d39ee 100644 --- a/src/registry/op_register.rs +++ b/src/registry/op_register.rs @@ -178,16 +178,37 @@ pub fn op_register_handler( } let mut spec = request.spec; + if spec.channel_open.is_some() { + return ResponseEnvelope::error( + context.request_id, + CallError::new( + "MARKED_SPEC_UNSUPPORTED", + format!( + "op/register: `{}` carries the `channel_open` marker — marked ops \ + are channel-machinery ops (the data plane is the allocated \ + channel's `BiStream`, ADR-047 amendment 4); forward them with \ + `ChannelRelay::register_relay_openable` (a relaying hub) or \ + consume them directly (`ChannelClient`), never as an \ + announcement", + spec.name + ), + false, + ), + ); + } spec.visibility = Visibility::Internal; let remote_name = spec.name.clone(); - let handler = forwarding_stub_for_announced_op( + let handler = forwarding_kind_for_announced_op( Arc::clone(&connection), remote_name, spec.op_type, ); - connection.register_imported(HandlerRegistration::new( + // The marked-refusal arm is unreachable here (marked + // announcements were refused above); an invariant break + // maps to the wire rather than a panic. + let registered = connection.register_imported(HandlerRegistration::new( spec.clone(), handler, OperationProvenance::FromCall, @@ -195,6 +216,12 @@ pub fn op_register_handler( None, Capabilities::new(), )); + if let Err(e) = registered { + return ResponseEnvelope::error( + context.request_id, + CallError::internal(format!("op/register: overlay landing refused: {e}")), + ); + } ResponseEnvelope::ok( context.request_id, @@ -204,15 +231,14 @@ pub fn op_register_handler( }) } -/// The forwarding stub for a peer-announced op. Query/Mutation ops get -/// the `from_call` forwarding shape (nested `call.requested` back over -/// channel 0, `forwarded_for` populated per ADR-032 §3). Announced -/// Sub/Pub ops are registered as stubs that return -/// `INVALID_OPERATION_TYPE` on invocation — nested composition is -/// request/response-only (`OverlayOperationEnv`'s contract); the -/// streaming/sink forwarding shapes ride on the `from_call` import -/// path, which the announcing side can use in the other direction. -fn forwarding_stub_for_announced_op( +/// The forwarding handler kind for a peer-announced op. Query/Mutation +/// get the `from_call` forwarding shape (nested `call.requested` back +/// over channel 0, `forwarded_for` populated per ADR-032 §3); unmarked +/// `Sub`/`Pub` get the real streaming/sink forwarding shapes (review +/// 010 S-04's remedy: never accept-then-fail — the handler actually +/// forwards). Marked specs refuse at this boundary (ADR-047 amendment +/// 4 — checked before this function is reached). +fn forwarding_kind_for_announced_op( connection: Arc, remote_name: String, op_type: OperationType, @@ -221,17 +247,13 @@ fn forwarding_stub_for_announced_op( OperationType::Query | OperationType::Mutation => HandlerKind::Once( crate::client::make_forwarding_handler(connection, remote_name), ), - OperationType::Sub | OperationType::Pub => { - HandlerKind::Once(make_handler(|_input, context| async move { - ResponseEnvelope::error( - context.request_id, - CallError::invalid_operation_type( - "peer-announced Sub/Pub ops are not invocable over nested \ - composition (request/response only)", - ), - ) - })) - } + OperationType::Sub => HandlerKind::Stream( + crate::client::make_streaming_forwarding_handler(connection, remote_name), + ), + OperationType::Pub => HandlerKind::Sink(crate::client::make_sink_forwarding_handler( + connection, + remote_name, + )), } } @@ -757,4 +779,111 @@ mod tests { .expect("marker survives the announce"); assert_eq!(marker.alpn, "alk/tunnel"); } + + // --- review 010 S-04's remedy (ADR-022, 2026-10-13 amendment) -------- + + /// Forward-through: an announced unmarked `Sub` op registers the + /// real streaming forwarding shape, not an always-failing stub. + #[tokio::test] + async fn announced_unmarked_sub_registers_real_streaming_forward() { + let conn = Arc::new(CallConnection::new(stub_connection())); + let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new())); + + let input = OpRegisterRequest { + spec: OperationSpec::new( + "worker/events", + OperationType::Sub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ), + replace: false, + } + .to_json(); + let response = handler(input, test_context("req-s04-1")).await; + assert!(response.result.is_ok(), "got {:?}", response.result); + + let registration = conn + .overlay_registration("worker/events") + .expect("registered"); + assert!( + matches!(registration.handler, HandlerKind::Stream(_)), + "the announced sub op is a real forwarding handler, not a stub" + ); + } + + /// Forward-through: an announced unmarked `Pub` op registers the + /// real sink forwarding shape. + #[tokio::test] + async fn announced_unmarked_pub_registers_real_sink_forward() { + let conn = Arc::new(CallConnection::new(stub_connection())); + let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new())); + + let input = OpRegisterRequest { + spec: OperationSpec::new( + "worker/ingest", + OperationType::Pub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ), + replace: false, + } + .to_json(); + let response = handler(input, test_context("req-s04-2")).await; + assert!(response.result.is_ok(), "got {:?}", response.result); + + let registration = conn + .overlay_registration("worker/ingest") + .expect("registered"); + assert!( + matches!(registration.handler, HandlerKind::Sink(_)), + "the announced pub op is a real forwarding handler, not a stub" + ); + } + + /// Marked-spec refusal (ADR-047 amendment 4 §4.8's wire side): a + /// marked announcement is rejected with the typed code, no overlay + /// entry, and the message names the sanctioned compositions. + #[tokio::test] + async fn announced_marked_spec_is_refused() { + use crate::registry::spec::ChannelOpenSpec; + + let conn = Arc::new(CallConnection::new(stub_connection())); + let handler = op_register_handler(Arc::clone(&conn), Arc::new(OperationRegistry::new())); + + let input = OpRegisterRequest { + spec: OperationSpec::new( + "channels/tunnel/direct", + OperationType::Sub, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alk/tunnel")), + replace: false, + } + .to_json(); + let response = handler(input, test_context("req-s04-3")).await; + let e = response.result.expect_err("marked announcement refused"); + assert_eq!(e.code, "MARKED_SPEC_UNSUPPORTED"); + assert!( + e.message.contains("ChannelRelay"), + "the refusal names the sanctioned composition: {}", + e.message + ); + assert!( + !conn.overlay_contains("channels/tunnel/direct"), + "the refused announcement lands nowhere" + ); + } } diff --git a/src/registry/registration.rs b/src/registry/registration.rs index f964088..3e7c0ba 100644 --- a/src/registry/registration.rs +++ b/src/registry/registration.rs @@ -84,6 +84,12 @@ pub struct HandlerRegistration { pub composition_authority: Option, pub scoped_env: Option, pub capabilities: Capabilities, + /// Set only by the channels-layer wrapper builders — a marked + /// (`channel_open`) spec is registered only through the + /// channel-machinery seam (`register_openable*`, `register_pub_openable`, + /// the relay). Generic registrations of marked specs are refused + /// (ADR-047 amendment 4 §4.8 — the registered-but-incapable class). + pub(crate) channel_wrapper: bool, } impl HandlerRegistration { @@ -102,8 +108,17 @@ impl HandlerRegistration { composition_authority, scoped_env, capabilities, + channel_wrapper: false, } } + + /// The channel-machinery wrapper family's constructor seam + /// (crate-internal — the only sanctioned registrations of marked + /// specs). + pub(crate) fn with_channel_wrapper(mut self) -> Self { + self.channel_wrapper = true; + self + } } pub struct OperationRegistry { @@ -215,6 +230,23 @@ impl OperationRegistry { registration.spec.op_type, expected, actual )); } + if registration.spec.channel_open.is_some() && !registration.channel_wrapper { + return Err(format!( + "spec `{}` carries the `channel_open` marker — marked ops are registered \ + only through the channel-machinery seam (`ChannelCore::register_openable*`, \ + `ChannelCore::register_pub_openable`, `ChannelRelay::register_relay_openable`); \ + a generic registration cannot serve the marker's data plane (ADR-047 amendment 4 §4.8)", + registration.spec.name + )); + } + if registration.spec.channel_open.is_some() && registration.spec.publish_schema.is_some() { + return Err(format!( + "spec `{}` carries both the `channel_open` marker and a `publish_schema` — \ + publish chunks never exist on a marked op (the data plane is the adopted \ + channel's `BiStream`); the field is dead machinery here (ADR-047 amendment 4 §4.2)", + registration.spec.name + )); + } Self::validate_and_cache_publish_schema(&mut self.validators.write(), ®istration.spec)?; Self::validate_and_cache_input_schema( &mut self.input_validators.write(), @@ -436,12 +468,21 @@ impl OperationRegistry { /// dispatch path (`Dispatcher::dispatch` Pub branch) call it, so the /// two paths cannot diverge (P-08: the `invoke_sink` path IS the /// wire path). + /// + /// Marked `Pub` ops (ADR-047 amendment 4 §4.3): the + /// wrapper-reserved input key ([`crate::registry::spec:: + /// RESERVED_OPEN_INPUT_KEY`]) is stripped **before** + /// `input_schema` validation — the producing crate's schema never + /// contains it — and the initiator-allocated channel id is handed + /// to the wrapper via `context.metadata`. A marked pub op whose + /// input lacks the reserved key is rejected here (`INVALID_INPUT`) + /// — the id is mandatory, not optional plumbing. #[allow(clippy::result_large_err)] pub(crate) fn resolve_sink_handler( &self, name: &str, - input: &Value, - context: &OperationContext, + input: &mut Value, + context: &mut OperationContext, ) -> Result { let request_id = context.request_id.clone(); let registration = match self.registration(name) { @@ -477,6 +518,43 @@ impl OperationRegistry { return Err(ResponseEnvelope::forbidden(request_id, message)); } + if registration.spec.channel_open.is_some() + && registration.spec.op_type == OperationType::Pub + { + let reserved_id = match input + .as_object_mut() + .and_then(|obj| obj.remove(crate::registry::spec::RESERVED_OPEN_INPUT_KEY)) + { + Some(value) => match value.as_u64() { + Some(id) if id <= u32::MAX as u64 => id as u32, + _ => { + return Err(ResponseEnvelope::error( + request_id, + CallError::invalid_input(format!( + "marked pub open op: the reserved `{}` key must be a u32 \ + channel id (the initiator-allocated id; ADR-047 amendment 4 §4.3)", + crate::registry::spec::RESERVED_OPEN_INPUT_KEY + )), + )); + } + }, + None => { + return Err(ResponseEnvelope::error( + request_id, + CallError::invalid_input(format!( + "marked pub open op input missing the reserved `{}` key \ + (the initiator-allocated channel id; ADR-047 amendment 4 §4.3)", + crate::registry::spec::RESERVED_OPEN_INPUT_KEY + )), + )); + } + }; + context.metadata.insert( + crate::registry::spec::RESERVED_OPEN_METADATA_KEY.to_string(), + serde_json::json!(reserved_id), + ); + } + self.check_input_schema(®istration, input, &request_id)?; match ®istration.handler { @@ -505,14 +583,16 @@ impl OperationRegistry { pub async fn invoke_sink( &self, name: &str, - input: Value, + mut input: Value, publish_stream: PublishStream, - context: OperationContext, + mut context: OperationContext, ) -> ResponseEnvelope { - match self.resolve_sink_handler(name, &input, &context) { - Err(envelope) => envelope, - Ok(sink_handler) => sink_handler(input, context, publish_stream).await, - } + let publish_stream_resolved = + match self.resolve_sink_handler(name, &mut input, &mut context) { + Err(envelope) => return envelope, + Ok(sink_handler) => sink_handler, + }; + publish_stream_resolved(input, context, publish_stream).await } } @@ -2942,3 +3022,85 @@ mod tests { assert_eq!(err.code, "INVALID_INPUT"); } } + +#[cfg(test)] +mod marked_spec_gates { + use super::*; + + fn marked_spec(name: &str, op_type: OperationType) -> OperationSpec { + OperationSpec::new( + name, + op_type, + crate::registry::spec::Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + crate::registry::spec::AccessControl::default(), + None, + ) + .with_channel_open(crate::registry::spec::ChannelOpenSpec::new("alk/x")) + } + + /// ADR-047 amendment 4 §4.2 gate: `publish_schema` on a marked + /// spec refuses at registration — publish chunks never exist on a + /// marked op. (Registered through the sanctioned wrapper seam, so + /// the §4.8 invariant is satisfied and this refusal is the one + /// that fires.) + #[test] + fn marked_spec_with_publish_schema_refused() { + let registry = OperationRegistry::new(); + let spec = marked_spec("channels/x/pub", OperationType::Pub) + .with_publish_schema(serde_json::json!({ "type": "object" })); + let registration = HandlerRegistration::new( + spec, + HandlerKind::Sink(make_sink_handler(|_input, _ctx, _stream| async { + ResponseEnvelope::ok("r", serde_json::json!({})) + })), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + ) + .with_channel_wrapper(); + let e = registry.register(registration).expect_err("refused"); + assert!(e.contains("publish_schema"), "got: {e}"); + } + + /// ADR-047 amendment 4 §4.8 gate: a marked spec through the + /// generic registration path is refused — the registered-but- + /// incapable class review 010 outlaws; the sanctioned seam is the + /// channel wrapper family. + #[test] + fn marked_spec_refused_outside_the_wrapper_seam() { + let registry = OperationRegistry::new(); + let spec = marked_spec("channels/x/pub", OperationType::Pub); + let registration = HandlerRegistration::new( + spec, + HandlerKind::Sink(make_sink_handler(|_input, _ctx, _stream| async { + ResponseEnvelope::ok("r", serde_json::json!({})) + })), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + ); + let e = registry.register(registration).expect_err("refused"); + assert!(e.contains("channel-machinery seam"), "got: {e}"); + + // A marked `Sub` spec — the same class. + let registry2 = OperationRegistry::new(); + let spec2 = marked_spec("channels/x/sub", OperationType::Sub); + let registration2 = HandlerRegistration::new( + spec2, + HandlerKind::Stream(make_streaming_handler(|_input, _ctx| { + Box::pin(futures::stream::empty()) + })), + OperationProvenance::Local, + None, + None, + crate::core::types::Capabilities::new(), + ); + let e2 = registry2.register(registration2).expect_err("refused"); + assert!(e2.contains("channel-machinery seam"), "got: {e2}"); + } +} diff --git a/src/registry/spec.rs b/src/registry/spec.rs index a8485e8..277a523 100644 --- a/src/registry/spec.rs +++ b/src/registry/spec.rs @@ -49,6 +49,28 @@ pub struct ChannelOpenSpec { pub alpn: Cow<'static, str>, } +/// The wrapper-reserved input key on a **marked `Pub` op's** input +/// (ADR-047 amendment 4 §4.3): the initiator-allocated `channel_id` the +/// responder adopts. The same key the open wrapper owns on the reply +/// side (the channels layer's `RESERVED_REPLY_KEY`); the wrapper owns +/// `channel_id` on both sides of a marked open op. Stripped +/// pre-validation in the sink resolve path and delivered to the +/// wrapper via `OperationContext.metadata` +/// ([`RESERVED_OPEN_METADATA_KEY`]). +pub const RESERVED_OPEN_INPUT_KEY: &str = "channel_id"; + +/// The `OperationContext.metadata` key the sink resolve path uses to +/// hand the stripped reserved input value (the initiator-allocated +/// channel id) to the marked-pub open wrapper (ADR-047 amendment 4 +/// §4.3). +pub const RESERVED_OPEN_METADATA_KEY: &str = "channel_open_channel_id"; + +/// The wire error code for a `call.published` chunk arriving on a +/// marked (`channel_open`) `Pub` op (ADR-047 amendment 4 §4.1): the +/// marked op's data plane is the adopted channel's `BiStream`; publish +/// chunks are rejected loudly and never consumed. +pub const MARKED_PUB_CHUNKS_REJECTED: &str = "channel:publish_chunks_rejected"; + impl ChannelOpenSpec { pub fn new(alpn: impl Into>) -> Self { Self { alpn: alpn.into() } diff --git a/tests/stub_guard.rs b/tests/stub_guard.rs new file mode 100644 index 0000000..e0e4ec2 --- /dev/null +++ b/tests/stub_guard.rs @@ -0,0 +1,49 @@ +//! The stub guard (review 010 action 3): no `*:not_implemented` path +//! may exist in `src/`. A handler that registers successfully and +//! returns an error of this form on every invocation is a bug, not a +//! documented limitation — shipping one is a hard failure per the +//! repo ruling (reviews 010/011; four stubs shipped across nine +//! published versions before this guard existed). + +#[test] +fn source_tree_has_no_not_implemented_path() { + let manifest = env!("CARGO_MANIFEST_DIR"); + let src = std::path::Path::new(manifest).join("src"); + let mut offenders = Vec::new(); + walk(&src, &mut offenders); + assert!( + offenders.is_empty(), + "no `not_implemented` string may appear in src/ — a registered surface \ + that always fails is the stub class (review 010 S-01..S-04); offended \ + by: {offenders:?}" + ); +} + +fn walk(dir: &std::path::Path, offenders: &mut Vec) { + let entries = match std::fs::read_dir(dir) { + Ok(entries) => entries, + Err(_) => return, + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + walk(&path, offenders); + } else if path.extension().is_some_and(|e| e == "rs") { + let contents = match std::fs::read_to_string(&path) { + Ok(c) => c, + Err(_) => continue, + }; + for (line_no, line) in contents.lines().enumerate() { + if line.contains("not_implemented") { + offenders.push(format!( + "{}:{}", + path.strip_prefix(env!("CARGO_MANIFEST_DIR")) + .unwrap_or(&path) + .display(), + line_no + 1 + )); + } + } + } + } +}