diff --git a/docs/architecture/decisions/047-openable-alpns-are-operations.md b/docs/architecture/decisions/047-openable-alpns-are-operations.md index 13d7315..761df69 100644 --- a/docs/architecture/decisions/047-openable-alpns-are-operations.md +++ b/docs/architecture/decisions/047-openable-alpns-are-operations.md @@ -215,7 +215,7 @@ pub struct OperationSpec { } pub struct ChannelOpenSpec { - pub alpn: &'static str, // e.g., b"alknet/tty" as str + pub alpn: Cow<'static, str>, // e.g., "alknet/tty"; Cow so from_call can supply an owned String without Box::leak } ``` @@ -239,6 +239,17 @@ on a bare top-level `alknet/call` connection, where there is no invocation time via the extension trait (Gap E); if it returns `None`, the wrapper returns `channel:no_channels_session`. +> **Amended 2026-08-13** — `ChannelOpenSpec::alpn` is +> `Cow<'static, str>` (not `&'static str` as originally decided). The +> `&'static str` shape forced `from_call`'s discovery path to +> `Box::leak` each runtime-parsed ALPN `String` to satisfy the +> `'static` bound, accumulating a leak on every rediscovery. The +> `Cow<'static, str>` keeps the common case (ALPN crates register at +> compile time with a `&'static str` literal — zero allocation) cheap +> while letting `from_call` supply an owned `String` without leaking. +> This is a two-way-door type detail (the wire format — a boolean +> `channel_open` marker — is unchanged). + ### 3. `ChannelCore` wrapper (the open-op composition seam) The ALPN crate provides: diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index 918aeda..a189cd1 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -283,7 +283,7 @@ mod tests { let oversized_len = super::super::wire::MAX_CHUNK_LEN + 1; let mut header = [0u8; 8]; - super::super::wire::write_header(id, oversized_len, &mut header); + super::super::wire::write_header(id, oversized_len, &mut header).expect("write header"); let mut client_write = client; client_write .write_all(&header) @@ -295,7 +295,7 @@ mod tests { .await .expect("write oversized payload"); - super::super::wire::write_header(id, 5, &mut header); + super::super::wire::write_header(id, 5, &mut header).expect("write header"); client_write .write_all(&header) .await @@ -344,7 +344,7 @@ mod tests { let mut client_write = client; for i in 0..10u8 { let payload = [i; 4]; - super::super::wire::write_header(id_a, 4, &mut header); + super::super::wire::write_header(id_a, 4, &mut header).expect("write header"); client_write .write_all(&header) .await @@ -354,7 +354,7 @@ mod tests { .await .expect("write payload a"); } - super::super::wire::write_header(id_b, 4, &mut header); + super::super::wire::write_header(id_b, 4, &mut header).expect("write header"); client_write .write_all(&header) .await diff --git a/src/channels/manager.rs b/src/channels/manager.rs index 03a04cf..109efae 100644 --- a/src/channels/manager.rs +++ b/src/channels/manager.rs @@ -9,7 +9,7 @@ use std::collections::HashMap; use std::net::SocketAddr; -use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::Arc; use bytes::Bytes; @@ -92,6 +92,7 @@ struct Inner { opener_ledger: OpenerLedger, remote_addr: Option, side: ChannelSide, + dropped_unknown_chunks: AtomicU64, } impl ChannelManager { @@ -122,6 +123,7 @@ impl ChannelManager { opener_ledger: OpenerLedger::new(), remote_addr, side, + dropped_unknown_chunks: AtomicU64::new(0), }), } } @@ -162,6 +164,17 @@ impl ChannelManager { self.inner.channels.lock().len() } + /// The number of chunks the demux dropped because the `channel_id` + /// was unknown (REQ-CH-04 — lenient handling: the demux drops and + /// continues rather than tearing down the connection). A non-zero + /// value indicates either a peer sending chunks for channels this + /// side hasn't opened/adopted yet, or a desync that the lenient + /// drop is masking. Observability only — the counter does not + /// drive any control flow. + pub fn dropped_unknown_chunks(&self) -> u64 { + self.inner.dropped_unknown_chunks.load(Ordering::Relaxed) + } + /// The mux handle — for registering new channels' write halves. pub fn mux(&self) -> &MuxHandle { &self.inner.mux @@ -387,6 +400,9 @@ impl ChannelManager { } } None => { + self.inner + .dropped_unknown_chunks + .fetch_add(1, Ordering::Relaxed); debug!( channel_id, "demux: unknown channel_id, dropping chunk (lenient)" @@ -448,6 +464,17 @@ impl ChannelManager { .get(&channel_id) .map(|s| s.alpn.clone()) } + + /// The `channel_id`s of all currently-open channels (observability + /// and snapshot iterators). The returned `Vec` is a point-in-time + /// snapshot — channels may open or close concurrently with the + /// caller iterating it. Used by `channel/resources/subscribe`'s + /// initial snapshot so the loop is O(open channels) rather than + /// O(max id) (the previous `0..u32::MAX` iteration could spin + /// millions of times on sparse ids after churn). + pub fn channel_ids(&self) -> Vec { + self.inner.channels.lock().keys().copied().collect() + } } #[cfg(test)] @@ -511,6 +538,26 @@ mod tests { assert!(!manager.has_channel(id), "channel removed"); } + #[tokio::test] + async fn channel_ids_returns_open_channel_ids() { + let manager = make_manager_with_runner().await; + assert!(manager.channel_ids().is_empty(), "no channels open yet"); + let (id1, _send1, _recv1) = manager + .open_channel("alknet/tty", "alice", None) + .await + .expect("open 1"); + let (id2, _send2, _recv2) = manager + .open_channel("alknet/tunnel", "bob", None) + .await + .expect("open 2"); + let mut ids = manager.channel_ids(); + ids.sort(); + assert_eq!(ids, vec![id1, id2], "channel_ids returns all open ids"); + manager.teardown_channel(id1).expect("teardown 1"); + let ids = manager.channel_ids(); + assert_eq!(ids, vec![id2], "channel_ids reflects teardown"); + } + #[tokio::test] async fn route_payload_to_open_channel_succeeds() { let manager = make_manager_with_runner().await; @@ -535,6 +582,43 @@ mod tests { .await; } + #[tokio::test] + async fn route_payload_to_unknown_channel_increments_dropped_counter() { + let manager = make_manager_with_runner().await; + assert_eq!(manager.dropped_unknown_chunks(), 0); + manager + .route_payload(999, Bytes::from_static(b"data")) + .await; + manager + .route_payload(998, Bytes::from_static(b"more")) + .await; + assert_eq!( + manager.dropped_unknown_chunks(), + 2, + "dropped_unknown_chunks counter increments per unknown-channel drop" + ); + } + + #[tokio::test] + async fn route_payload_to_known_channel_does_not_increment_dropped_counter() { + let manager = make_manager_with_runner().await; + let (id, _send, mut recv) = manager + .open_channel("alknet/tty", "alice", None) + .await + .expect("open"); + manager + .route_payload(id, Bytes::from_static(b"hello")) + .await; + assert_eq!( + manager.dropped_unknown_chunks(), + 0, + "known channel does not increment the dropped counter" + ); + use tokio::io::AsyncReadExt; + let mut buf = [0u8; 5]; + recv.read_exact(&mut buf).await.expect("read"); + } + #[tokio::test] async fn teardown_unknown_channel_returns_error() { let manager = make_manager_with_runner().await; diff --git a/src/channels/mux.rs b/src/channels/mux.rs index b8ff6cb..07c8716 100644 --- a/src/channels/mux.rs +++ b/src/channels/mux.rs @@ -110,8 +110,25 @@ impl MuxRunner { /// `MpscSendStream` without calling `shutdown`), the pump writes an /// EOF chunk for that `channel_id` before exiting (REQ-CH-01 /// implicit-EOF path). + /// + /// **Duplicate registration:** if `register(channel_id)` is called + /// for a `channel_id` that already has a live pump, the registration + /// is rejected — the responder is dropped without sending, so the + /// caller's `receiver.await` yields `RecvError`, which + /// `ChannelManager` maps to `ChannelExists`. If the existing pump + /// has finished (`is_finished()`), it is reaped from the map and the + /// new registration proceeds (this is the leak fix: finished + /// `JoinHandle`s don't accumulate for the connection's lifetime). pub async fn run(mut self) -> io::Result<()> { while let Some(registration) = self.new_pumps.recv().await { + if let Some(existing) = self.pumps.get(®istration.channel_id) { + if existing.is_finished() { + self.pumps.remove(®istration.channel_id); + } else { + continue; + } + } + let (send, mut recv) = futures::channel::mpsc::channel::(64); let stream = MpscSendStream::new(send); let _ = registration.responder.send(stream); @@ -296,6 +313,47 @@ mod tests { let _ = runner_task.await; } + #[tokio::test] + async fn mux_register_duplicate_channel_id_rejected_while_pump_running() { + let (_client, server) = tokio::io::duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, runner) = MuxRunner::new(Box::new(writer)); + let runner_task = tokio::spawn(async move { runner.run().await }); + + let _first = handle.register(5).await.expect("first register succeeds"); + match handle.register(5).await { + Err(e) => assert!( + e.kind() == io::ErrorKind::ConnectionReset, + "duplicate register while pump running should error, got {e:?}" + ), + Ok(_) => panic!("expected duplicate register to fail while pump running"), + } + + drop(handle); + let _ = runner_task.await; + } + + #[tokio::test] + async fn mux_register_reaps_finished_pump_and_accepts_reregistration() { + let (_client, server) = tokio::io::duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, runner) = MuxRunner::new(Box::new(writer)); + let runner_task = tokio::spawn(async move { runner.run().await }); + + let first = handle.register(9).await.expect("first register"); + drop(first); + tokio::task::yield_now().await; + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + + let _second = handle + .register(9) + .await + .expect("reregister after pump finished should succeed"); + + drop(handle); + let _ = runner_task.await; + } + #[tokio::test] async fn opener_ledger_record_and_take() { let ledger = OpenerLedger::new(); diff --git a/src/channels/operations.rs b/src/channels/operations.rs index f77e256..58305b9 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -275,14 +275,14 @@ fn make_resources_subscribe_handler(manager: ChannelManager) -> StreamingHandler // A real implementation would aggregate across all // registered openable ALPNs (each ALPN crate provides a // resource enumerator, ADR-047 §6). For now, emit the - // currently-open channels. - for channel_id in 0..u32::MAX { + // currently-open channels. Iterating `channel_ids()` is + // O(open channels) — the previous `0..u32::MAX` loop was + // O(max id) and could spin millions of times on sparse ids + // after churn. + for channel_id in manager.channel_ids() { if let Some(alpn) = manager.channel_alpn(channel_id) { resources.push(json!({ "alpn": alpn })); } - if resources.len() >= manager.open_count() { - break; - } } ResponseEnvelope::ok(ctx.request_id, json!({ "resources": resources })) })) @@ -693,6 +693,12 @@ fn make_open_handler_sink( mod tests { use super::*; use crate::channels::mux::MuxRunner; + use crate::core::auth::Identity; + use crate::registry::context::{AbortPolicy, ScopedPeerEnv}; + use crate::registry::env::OperationEnv; + use futures::stream::StreamExt; + use std::collections::HashMap; + use std::sync::Arc; use tokio::io::duplex; async fn make_manager() -> ChannelManager { @@ -705,6 +711,46 @@ mod tests { ChannelManager::with_defaults(handle, None) } + struct NoopEnv; + #[async_trait::async_trait] + impl OperationEnv for NoopEnv { + async fn invoke_with_policy( + &self, + _ns: &str, + _op: &str, + _input: Value, + parent: &OperationContext, + _policy: AbortPolicy, + ) -> ResponseEnvelope { + ResponseEnvelope::ok(parent.request_id.clone(), Value::Null) + } + fn contains(&self, _name: &str) -> bool { + false + } + } + + fn test_context(request_id: &str) -> OperationContext { + OperationContext { + request_id: request_id.to_string(), + parent_request_id: None, + identity: Some(Identity { + id: "alice".to_string(), + scopes: vec![], + resources: HashMap::new(), + }), + handler_identity: None, + forwarded_for: None, + capabilities: Capabilities::new(), + metadata: HashMap::new(), + scoped_env: ScopedPeerEnv::empty(), + env: Arc::new(NoopEnv), + abort_policy: AbortPolicy::default(), + deadline: None, + internal: false, + ownership: None, + } + } + #[tokio::test] async fn register_on_registers_three_ops() { let manager = make_manager().await; @@ -741,4 +787,110 @@ mod tests { assert_eq!(spec.op_type, OperationType::Sub); assert_eq!(spec.visibility, Visibility::External); } + + #[tokio::test] + async fn resources_subscribe_handler_emits_alpns_for_open_channels() { + let manager = make_manager().await; + manager + .open_channel("alknet/tty", "alice", None) + .await + .expect("open tty"); + manager + .open_channel("alknet/tunnel", "alice", None) + .await + .expect("open tunnel"); + 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 snapshot envelope"); + let out = env.result.expect("ok"); + let resources = out + .get("resources") + .and_then(|v| v.as_array()) + .expect("resources array"); + let mut alpns: Vec = resources + .iter() + .map(|r| { + r.get("alpn") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string() + }) + .collect(); + alpns.sort(); + assert_eq!( + alpns, + vec!["alknet/tty".to_string(), "alknet/tunnel".to_string()] + ); + } + + #[tokio::test] + async fn resources_subscribe_handler_empty_when_no_channels_open() { + let manager = make_manager().await; + let handler = make_resources_subscribe_handler(manager); + let mut stream = handler(Value::Null, test_context("res-2")); + let env = stream.next().await.expect("snapshot envelope"); + let out = env.result.expect("ok"); + let resources = out + .get("resources") + .and_then(|v| v.as_array()) + .expect("resources array"); + assert!( + resources.is_empty(), + "no channels open → empty resource set" + ); + } + + #[tokio::test] + async fn close_handler_rejects_channel_zero() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let handler = make_close_handler(manager, policy); + let env = handler(json!({ "channel_id": 0 }), test_context("close-0")).await; + match env.result { + Err(e) => assert!( + e.message.contains("channel 0"), + "error should mention channel 0, got: {}", + e.message + ), + Ok(_) => panic!("closing channel 0 should be rejected"), + } + } + + #[tokio::test] + async fn close_handler_unknown_channel_returns_not_found() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let handler = make_close_handler(manager, policy); + let env = handler(json!({ "channel_id": 999 }), test_context("close-unknown")).await; + match env.result { + Err(e) => assert_eq!(e.code, "NOT_FOUND"), + Ok(_) => panic!("closing unknown channel should error"), + } + } + + #[tokio::test] + async fn control_handler_unknown_channel_returns_not_found() { + let manager = make_manager().await; + let handler = make_control_handler(manager); + let env = handler( + json!({ "channel_id": 999, "message": {} }), + test_context("ctrl-unknown"), + ) + .await; + match env.result { + Err(e) => assert_eq!(e.code, "NOT_FOUND"), + Ok(_) => panic!("control on unknown channel should error"), + } + } + + #[tokio::test] + async fn control_handler_missing_channel_id_returns_invalid_input() { + let manager = make_manager().await; + let handler = make_control_handler(manager); + let env = handler(json!({ "message": {} }), test_context("ctrl-missing")).await; + match env.result { + Err(e) => assert_eq!(e.code, "INVALID_INPUT"), + Ok(_) => panic!("missing channel_id should error"), + } + } } diff --git a/src/channels/wire.rs b/src/channels/wire.rs index 37650eb..2f9f974 100644 --- a/src/channels/wire.rs +++ b/src/channels/wire.rs @@ -103,14 +103,23 @@ pub fn parse_header(buf: &[u8]) -> Result { /// Write an 8-byte chunk header into `out`. Pure function — no /// allocation, no async, WASM-clean. /// -/// `out` must be at least 8 bytes; panics if not (the caller — the mux -/// write path — always provides an 8-byte buffer). The -/// `[channel_id: u32 BE][length: u32 BE]` layout is the wire format -/// (ADR-034, amended by ADR-035 — no `stream_type` byte). -pub fn write_header(channel_id: u32, length: u32, out: &mut [u8]) { +/// Returns [`ChunkError::HeaderTooShort`] if `out` is shorter than 8 +/// bytes. The `[channel_id: u32 BE][length: u32 BE]` layout is the wire +/// format (ADR-034, amended by ADR-035 — no `stream_type` byte). All +/// current callers pass an 8-byte buffer, but the contract is typed +/// rather than panic-documented (AGENTS.md §2: no panics in library +/// code). +pub fn write_header(channel_id: u32, length: u32, out: &mut [u8]) -> Result<(), ChunkError> { + if out.len() < CHUNK_HEADER_LEN { + return Err(ChunkError::HeaderTooShort { + need: CHUNK_HEADER_LEN, + have: out.len(), + }); + } let header = &mut out[..CHUNK_HEADER_LEN]; header[0..4].copy_from_slice(&channel_id.to_be_bytes()); header[4..8].copy_from_slice(&length.to_be_bytes()); + Ok(()) } /// Read an 8-byte chunk header from `reader`. Async convenience wrapper @@ -141,7 +150,8 @@ where { use tokio::io::AsyncWriteExt; let mut header = [0u8; CHUNK_HEADER_LEN]; - write_header(channel_id, payload.len() as u32, &mut header); + write_header(channel_id, payload.len() as u32, &mut header) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?; writer.write_all(&header).await?; if !payload.is_empty() { writer.write_all(payload).await?; @@ -166,7 +176,7 @@ mod tests { #[test] fn parse_header_round_trips_channel_id_and_length() { let mut buf = [0u8; 8]; - write_header(7, 1024, &mut buf); + write_header(7, 1024, &mut buf).expect("write"); let header = parse_header(&buf).expect("parse"); assert_eq!(header, ChunkHeader::new(7, 1024)); assert!(!header.is_eof()); @@ -175,7 +185,7 @@ mod tests { #[test] fn parse_header_eof_sentinel() { let mut buf = [0u8; 8]; - write_header(3, 0, &mut buf); + write_header(3, 0, &mut buf).expect("write"); let header = parse_header(&buf).expect("parse"); assert_eq!(header.length, 0); assert!(header.is_eof()); @@ -184,7 +194,7 @@ mod tests { #[test] fn parse_header_channel_zero() { let mut buf = [0u8; 8]; - write_header(CHANNEL_ID_ZERO, 512, &mut buf); + write_header(CHANNEL_ID_ZERO, 512, &mut buf).expect("write"); let header = parse_header(&buf).expect("parse"); assert_eq!(header.channel_id, CHANNEL_ID_ZERO); } @@ -192,7 +202,7 @@ mod tests { #[test] fn parse_header_max_length_accepted() { let mut buf = [0u8; 8]; - write_header(1, MAX_CHUNK_LEN, &mut buf); + write_header(1, MAX_CHUNK_LEN, &mut buf).expect("write"); let header = parse_header(&buf).expect("parse"); assert_eq!(header.length, MAX_CHUNK_LEN); } @@ -200,7 +210,7 @@ mod tests { #[test] fn parse_header_too_large_returns_error() { let mut buf = [0u8; 8]; - write_header(1, MAX_CHUNK_LEN + 1, &mut buf); + write_header(1, MAX_CHUNK_LEN + 1, &mut buf).expect("write"); match parse_header(&buf) { Err(ChunkError::TooLarge { length, max }) => { assert_eq!(length, MAX_CHUNK_LEN + 1); @@ -225,10 +235,22 @@ mod tests { #[test] fn write_header_writes_be_bytes() { let mut buf = [0u8; 8]; - write_header(0x01020304, 0x05060708, &mut buf); + write_header(0x01020304, 0x05060708, &mut buf).expect("write"); assert_eq!(buf, [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08]); } + #[test] + fn write_header_short_buffer_returns_error() { + let mut buf = [0u8; 4]; + match write_header(1, 0, &mut buf) { + Err(ChunkError::HeaderTooShort { need, have }) => { + assert_eq!(need, CHUNK_HEADER_LEN); + assert_eq!(have, 4); + } + other => panic!("expected HeaderTooShort, got {other:?}"), + } + } + #[tokio::test] async fn read_header_round_trips_through_duplex() { let (mut reader, mut writer) = tokio::io::duplex(64); diff --git a/src/client/from_call.rs b/src/client/from_call.rs index 611a935..b8ba41f 100644 --- a/src/client/from_call.rs +++ b/src/client/from_call.rs @@ -268,13 +268,7 @@ fn rebuild_spec_for( .unwrap_or(false) { if let Some(alpn) = derive_alpn_from_op_name(remote_name) { - // `ChannelOpenSpec::alpn` is `&'static str` (ADR-047 §2). The - // discovered ALPN is a runtime `String`, so `leak_alpn` boxes - // and leaks it to satisfy the `'static` bound. The leak is - // bounded per unique ALPN but accumulates on rediscovery; - // the `Arc` / `Cow<'static, str>` refactor that removes - // the leak is deferred (see review 001 Unit 7). - spec = spec.with_channel_open(ChannelOpenSpec::new(leak_alpn(alpn))); + spec = spec.with_channel_open(ChannelOpenSpec::new(alpn)); } } @@ -287,34 +281,25 @@ fn rebuild_spec_for( /// the op is not a channel-open op, and the marker (if present) is /// ignored. ADR-047 §"Negative": the path segment is the ALPN with the /// `alknet/` prefix stripped; ALPNs without that prefix use their full -/// ALPN string as the path segment (rare case). +/// ALPN string as the path segment (rare case). Multi-segment non-`alknet/*` +/// ALPNs (e.g. `custom/proto`) survive because the `/sub` or `/pub` +/// suffix is stripped from `rest` rather than taking only the first +/// path segment. fn derive_alpn_from_op_name(op_name: &str) -> Option { let rest = op_name.strip_prefix("channels/")?; - let segment = rest.split('/').next()?; + let segment = rest + .strip_suffix("/sub") + .or_else(|| rest.strip_suffix("/pub"))?; if segment.is_empty() { return None; } - if segment.starts_with("alknet/") || segment == "alknet" { + if segment.starts_with("alknet/") || segment == "alknet" || segment.contains('/') { Some(segment.to_string()) } else { - // Non-`alknet/*` ALPN: the path segment IS the full ALPN. Some(format!("alknet/{segment}")) } } -/// Leak a `String` to a `'static str` for `ChannelOpenSpec::alpn`. -/// -/// This is a small, bounded leak: the set of ALPNs is fixed at -/// discovery time per peer, and `ChannelOpenSpec` is held in the -/// `OperationRegistry` for the connection's lifetime. A future -/// two-way-door refactor would make `ChannelOpenSpec::alpn` a -/// `Cow<'static, str>` or `Arc` to avoid the leak; for now the -/// `'static str` keeps the type simple and matches ADR-047 §2's -/// `&'static str` shape. -fn leak_alpn(alpn: String) -> &'static str { - Box::leak(alpn.into_boxed_str()) -} - fn parse_op_type(s: &str) -> Result { match s { "query" => Ok(OperationType::Query), @@ -662,6 +647,44 @@ mod tests { assert_eq!(derive_alpn_from_op_name("channels/"), None); } + #[test] + fn derive_alpn_from_op_name_multi_segment_non_alknet_alpn_survives() { + assert_eq!( + derive_alpn_from_op_name("channels/custom/proto/sub"), + Some("custom/proto".to_string()), + "multi-segment non-alknet/* ALPN uses full ALPN as the path segment" + ); + assert_eq!( + derive_alpn_from_op_name("channels/vendor/service/run/pub"), + Some("vendor/service/run".to_string()) + ); + } + + #[test] + fn derive_alpn_from_op_name_explicit_alknet_prefix_returned_as_is() { + assert_eq!( + derive_alpn_from_op_name("channels/alknet/tty/sub"), + Some("alknet/tty".to_string()) + ); + } + + #[test] + fn derive_alpn_from_op_name_strips_pub_suffix() { + assert_eq!( + derive_alpn_from_op_name("channels/tty/pub"), + Some("alknet/tty".to_string()) + ); + } + + #[test] + fn derive_alpn_from_op_name_no_suffix_returns_none() { + assert_eq!( + derive_alpn_from_op_name("channels/tty/query"), + None, + "op name without /sub or /pub suffix is not a channel-open op" + ); + } + #[test] fn from_call_config_builder_methods() { let config = FromCallConfig::new() diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index 1aaf071..9af89bb 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -74,8 +74,8 @@ pub enum DispatchResult { /// (`call.completed` or wire read end), drops `chunk_tx` and awaits /// `handler` to get the single `ResponseEnvelope` to write to the wire. pub struct SinkDispatch { - pub handler: Pin + Send>>, - pub chunk_tx: mpsc::Sender>, + pub(crate) handler: Pin + Send>>, + pub(crate) chunk_tx: mpsc::Sender>, } impl std::fmt::Debug for DispatchResult { @@ -229,18 +229,19 @@ impl Dispatcher { request_id: String, payload: Value, ) -> ResponseEnvelope { + let request_id_for_error = request_id.clone(); match self.dispatch(connection, request_id, payload).await { DispatchResult::Once(envelope) => envelope, DispatchResult::Stream(mut stream) => stream.next().await.unwrap_or_else(|| { ResponseEnvelope::error( - String::new(), + request_id_for_error, CallError::internal( "dispatch_requested called on a Sub op; use the streaming path", ), ) }), DispatchResult::Sink(_) => ResponseEnvelope::error( - String::new(), + request_id_for_error, CallError::internal("dispatch_requested called on a Pub op; use the sink path"), ), } diff --git a/src/registry/spec.rs b/src/registry/spec.rs index fc412e3..518adf7 100644 --- a/src/registry/spec.rs +++ b/src/registry/spec.rs @@ -3,6 +3,8 @@ //! //! See `docs/architecture/` for the full specification. +use std::borrow::Cow; + use crate::core::auth::Identity; use crate::core::ownership::OwnershipProvider; use serde_json::Value; @@ -34,14 +36,22 @@ pub enum Visibility { /// channels layer doesn't have to parse the op name. On the wire /// (`services/schema`), the marker is a boolean `"channel_open": true`; /// the ALPN is not serialized (it's derivable). +/// +/// The `alpn` is a `Cow<'static, str>` so that statically-registered +/// ops (the common case — ALPN crates register at compile time with a +/// `&'static str`) pay no allocation, while `from_call`-discovered ops +/// (whose ALPN is a runtime `String` parsed from the op name) can +/// supply an owned value without `Box::leak`-ing it to `'static` +/// (ADR-047 §2 said `&'static str` was "for now"; the `Cow` is the +/// deferred refactor that removes the per-discovery leak). #[derive(Debug, Clone, PartialEq, Eq)] pub struct ChannelOpenSpec { - pub alpn: &'static str, + pub alpn: Cow<'static, str>, } impl ChannelOpenSpec { - pub fn new(alpn: &'static str) -> Self { - Self { alpn } + pub fn new(alpn: impl Into>) -> Self { + Self { alpn: alpn.into() } } }