alkdev/alkcall
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Unit 7 — small correctness fixes (C-19-misc, C-21, C-23, P-13)

Browse Source 
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<Cow>, 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<u32>
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
This commit is contained in:
glm-5.2
2026-08-13 09:15:20 +00:00
parent bfb265e31b
commit da12d65a03
9 changed files with 415 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
docs/architecture/decisions/047-openable-alpns-are-operations.md
View File

@@ -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:

8
src/channels/adapter.rs
View File

@@ -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

86
src/channels/manager.rs
View File

@@ -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<SocketAddr>,
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<u32> {
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;

58
src/channels/mux.rs
View File

@@ -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(&registration.channel_id) {
if existing.is_finished() {
self.pumps.remove(&registration.channel_id);
} else {
continue;
}
}
let (send, mut recv) = futures::channel::mpsc::channel::<Bytes>(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();

162
src/channels/operations.rs
View File

@@ -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<String> = 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"),
}
}
}

46
src/channels/wire.rs
View File

@@ -103,14 +103,23 @@ pub fn parse_header(buf: &[u8]) -> Result<ChunkHeader, ChunkError> {
/// 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);

71
src/client/from_call.rs
View File

@@ -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<str>` / `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<String> {
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<str>` 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<OperationType, AdapterError> {
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()

9
src/protocol/dispatch.rs
View File

@@ -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<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
pub chunk_tx: mpsc::Sender<Result<Value, CallError>>,
pub(crate) handler: Pin<Box<dyn Future<Output = ResponseEnvelope> + Send>>,
pub(crate) chunk_tx: mpsc::Sender<Result<Value, CallError>>,
}
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"),
),
}

16
src/registry/spec.rs
View File

@@ -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<Cow<'static, str>>) -> Self {
Self { alpn: alpn.into() }
}
}