From da12d65a0348d7c8c274ac49565639ae7d5f594c Mon Sep 17 00:00:00 2001 From: "glm-5.2" Date: Thu, 13 Aug 2026 09:15:20 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20Unit=207=20=E2=80=94=20small=20correctne?= =?UTF-8?q?ss=20fixes=20(C-19-misc,=20C-21,=20C-23,=20P-13)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No panic paths in library code, no misleading public fields, honest error envelopes, latent bugs in derive_alpn_from_op_name and the mux pump map fixed, per-discovery Box::leak removed, demux observability counter added, resources-snapshot unbounded loop replaced with an O(open channels) iterator. C-21 — write_header (wire.rs) sliced out[..CHUNK_HEADER_LEN] and panicked on short buffers; the doc comment declared the panic. Returns Result<_, ChunkError::HeaderTooShort> instead (the variant already existed for the parse side). write_chunk and all callers updated (they pass [0u8; 8] today so the Result is always Ok, but the API contract is typed, not panic-documented). Added a short-buffer test. C-23 — dispatch_requested's Sink and Stream error arms used String::new() as the request id (dispatch.rs), producing envelopes with an empty id. The request_id is now cloned before the move into dispatch and used in the error arms, so the envelope carries the real id (matching the Once arm, which already did this correctly). P-13 — SinkDispatch::chunk_tx was pub (dispatch.rs), exposing the futures::mpsc::Sender type and the 64-slot buffer size as effective public API. Made pub(crate); handle_stream and pump_sink (the only readers) are in the same module, and the dispatch tests are in the same module too. C-19 (mux pump map leak + duplicate registration) — MuxRunner::run inserted each pump's JoinHandle into self.pumps and never removed finished ones (they accumulated for the connection's lifetime). Duplicate register(channel_id) silently overwrote the old handle while the old pump kept running (two pumps, one channel id). Fixed: before insert, check for an existing entry; if it is_finished(), reap it and proceed (leak fix); if it is still running, reject the registration by dropping the responder without sending (the caller's receiver.await yields RecvError, which ChannelManager maps to ChannelExists). Added two tests: duplicate-rejected-while-running and reap-finished-then-reregister. C-19 (derive_alpn_from_op_name) — from_call.rs took only the first path segment (rest.split('/').next()), so channels/custom/proto/sub derived alknet/custom instead of the full ALPN custom/proto. The segment.starts_with("alknet/") branch was dead (a single segment cannot contain /), and segment == "alknet" yielded the nonsense ALPN "alknet". Fixed: strip the known /sub or /pub suffix from rest instead of taking the first segment, so multi-segment ALPNs survive. The rule (ADR-047 Negative): alknet/*-prefixed segments and multi-segment non-alknet/* ALPNs are returned as-is; single-segment non-alknet names get the alknet/ prefix prepended. Added tests for the multi-segment non-alknet case, the explicit alknet/ prefix case, the /pub suffix, and the no-suffix (non-channel-open-op) case. C-19 (Box::leak per discovery) — from_call.rs leaked a String to 'static on every leak_alpn call (bounded per unique ALPN in theory, but leaks on every rediscovery of every marked op). Refactored ChannelOpenSpec::alpn from &'static str to Cow<'static, str> (spec.rs); the common case (ALPN crates register at compile time with a &'static str literal) pays no allocation via Into, while from_call supplies an owned String without leaking. Removed leak_alpn. Amended ADR-047 §2 to record the Cow<'static, str> shape (two-way-door type detail; the wire format — a boolean channel_open marker — is unchanged). C-19 (REQ-CH-04 error counter) — the demux's lenient unknown-channel drop (manager.rs) only debug!-logged; there was no counter or stats surface. Added an AtomicU64 dropped_unknown_chunks counter to ChannelManager, incremented on the unknown-channel drop in route_payload, with a dropped_unknown_chunks() accessor for observability. Added two tests: counter increments per drop, and known-channel routing does not increment it. C-19 (resources-snapshot unbounded loop) — operations.rs iterated 0..u32::MAX with a per-iteration mutex lock, breaking when resources.len() >= manager.open_count(). With sparse ids (monotonic after churn) or a channel closing mid-iteration, this could iterate millions of times. Replaced with an iterator over ChannelManager:: channel_ids() — a new accessor that returns a point-in-time Vec of open channel ids — so the snapshot is O(open channels). Added channel_ids tests and behavioral tests for the resources/subscribe handler (open channels emit their ALPNs; no channels emit an empty set). Also added close/control handler tests (C-25 #8 coverage the review noted was missing): close rejects channel 0, close on unknown channel returns NOT_FOUND, control on unknown channel returns NOT_FOUND, control missing channel_id returns INVALID_INPUT. Verification: - cargo test → 465 passed, 0 failed (was 449; +16 new tests) - cargo clippy --all-targets -- -D warnings → clean - cargo fmt --check → clean - cargo doc --no-deps → 0 warnings --- .../047-openable-alpns-are-operations.md | 13 +- src/channels/adapter.rs | 8 +- src/channels/manager.rs | 86 +++++++++- src/channels/mux.rs | 58 +++++++ src/channels/operations.rs | 162 +++++++++++++++++- src/channels/wire.rs | 46 +++-- src/client/from_call.rs | 71 +++++--- src/protocol/dispatch.rs | 9 +- src/registry/spec.rs | 16 +- 9 files changed, 415 insertions(+), 54 deletions(-) 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() } } }