Unit 10: honest stubs, substrate mode docs, spec-doc renumbering
- C-10: channel/control returns channel:control_not_implemented (OQ-39) - C-11: channel/resources/subscribe returns channel:resources_not_implemented (OQ-40) - C-14: adapter.rs doc comments accurately describe in-line-only substrate mode; QUIC-native multi-stream deferred to alknet (OQ-41) - C-15: ChannelClient code already clean (stale open_channel_stream ref removed in prior unit); no code changes needed - C-26: spec docs updated to post-047 model (per-ALPN open ops, no ChannelDirection, no ResourceEntry.access); stale alknet ADR refs (071/075/076/078/079/080/093/094) renumbered to alkcall equivalents (034/035/039/040/041/042/043) - OQs 39-41 filed in open-questions.md Verification: cargo test (483 passed), cargo clippy (clean), cargo fmt (clean), cargo doc (no warnings)
This commit is contained in:
1 parent
49e01c1ff4
commit
6c5aa0c8e1
6 files changed
+163
-153
No files matched your search
+18
-9
@@ -1,9 +1,16 @@
|
||||
//! `ChannelsAdapter` — implements `ProtocolHandler` for
|
||||
//! `alknet/channels` (ADR-039). The accept path: receive one
|
||||
//! `Connection`, install channel 0, then run the demux loop — read
|
||||
//! 8-byte chunk headers off every bidi stream the transport yields,
|
||||
//! route each chunk's payload to the matching `channel_id`'s
|
||||
//! reassembly buffer.
|
||||
//! 8-byte chunk headers off the bidi stream and route each chunk's
|
||||
//! payload to the matching `channel_id`'s reassembly buffer.
|
||||
//!
|
||||
//! The current implementation handles the in-line substrate mode
|
||||
//! (single bidi stream, header-demuxed N channels — TCP+TLS,
|
||||
//! WebTransport session, SSH `direct-tcpip`). The QUIC-native
|
||||
//! multi-stream substrate (accept remaining bidi streams, read headers
|
||||
//! off each) is deferred to the downstream alknet crate (OQ-41).
|
||||
//! The wire format and demux loop are correct for both substrates;
|
||||
//! only the outer `accept_bi()` loop is missing.
|
||||
//!
|
||||
//! Channel 0 is pre-negotiated as `alknet/call` (ADR-036). The
|
||||
//! `install_channel_zero` hook (ADR-036 amendment — single-stream call
|
||||
@@ -92,12 +99,14 @@ impl ChannelsAdapter {
|
||||
}
|
||||
|
||||
/// The demux loop — reads 8-byte chunk headers off the bidi stream
|
||||
/// and routes payloads to the `ChannelManager`. On an in-line
|
||||
/// transport (TCP+TLS, WebTransport session), `accept_bi()` yields
|
||||
/// once and the header demuxes N channels from that stream. On
|
||||
/// QUIC, `accept_bi()` yields repeatedly — each stream carries one
|
||||
/// logical channel, and the header's `channel_id` correlates it.
|
||||
/// Same code path, same wire format (ADR-034 §substrate modes).
|
||||
/// and routes payloads to the `ChannelManager`. This handles the
|
||||
/// in-line substrate mode (single bidi stream, header-demuxed N
|
||||
/// channels — TCP+TLS, WebTransport session, SSH `direct-tcpip`).
|
||||
/// The QUIC-native multi-stream substrate (accept remaining bidi
|
||||
/// streams, read headers off each) is deferred to the downstream
|
||||
/// alknet crate (OQ-41). The wire format and demux loop are correct
|
||||
/// for both substrates; only the outer `accept_bi()` loop is missing.
|
||||
/// See ADR-034 §substrate modes.
|
||||
async fn run_demux_loop(
|
||||
manager: &ChannelManager,
|
||||
reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
|
||||
|
||||
+45
-58
@@ -232,6 +232,11 @@ fn make_close_handler(manager: ChannelManager, policy: Arc<dyn ChannelLifecycleP
|
||||
/// The `channel/control` handler. Routes `message` to the handler's
|
||||
/// control handle for `channel_id` (ALPN-specific; the channels layer
|
||||
/// does not interpret `message`).
|
||||
///
|
||||
/// Not yet implemented — the control-handle surface (per-channel
|
||||
/// control callbacks registered by ALPN crates) is deferred (OQ-39).
|
||||
/// Returns `channel:control_not_implemented` so callers fail loudly
|
||||
/// rather than receiving fake success.
|
||||
fn make_control_handler(manager: ChannelManager) -> Handler {
|
||||
Arc::new(move |input: Value, ctx: OperationContext| {
|
||||
let manager = manager.clone();
|
||||
@@ -252,12 +257,14 @@ fn make_control_handler(manager: ChannelManager) -> Handler {
|
||||
CallError::not_found("channel:unknown_channel"),
|
||||
);
|
||||
}
|
||||
// The control routing to the handler's control handle is
|
||||
// ALPN-specific and not implemented in the generic layer —
|
||||
// the ALPN crate registers a control callback. For now,
|
||||
// return ok; the ALPN crate overrides this behavior via
|
||||
// composition.
|
||||
ResponseEnvelope::ok(ctx.request_id, json!({ "ok": true }))
|
||||
ResponseEnvelope::error(
|
||||
ctx.request_id,
|
||||
CallError::new(
|
||||
"channel:control_not_implemented",
|
||||
"channel/control is not yet implemented (OQ-39)",
|
||||
false,
|
||||
),
|
||||
)
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -266,25 +273,23 @@ fn make_control_handler(manager: ChannelManager) -> Handler {
|
||||
/// 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 {
|
||||
Arc::new(move |_input: Value, ctx: OperationContext| {
|
||||
let manager = manager.clone();
|
||||
let _manager = manager.clone();
|
||||
Box::pin(futures::stream::once(async move {
|
||||
let mut resources: Vec<Value> = Vec::new();
|
||||
// The resource set is the set of open channels' ALPNs.
|
||||
// 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. 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 }));
|
||||
}
|
||||
}
|
||||
ResponseEnvelope::ok(ctx.request_id, json!({ "resources": resources }))
|
||||
ResponseEnvelope::error(
|
||||
ctx.request_id,
|
||||
CallError::new(
|
||||
"channel:resources_not_implemented",
|
||||
"channel/resources/subscribe is not yet implemented (OQ-40)",
|
||||
false,
|
||||
),
|
||||
)
|
||||
}))
|
||||
})
|
||||
}
|
||||
@@ -789,55 +794,37 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resources_subscribe_handler_emits_alpns_for_open_channels() {
|
||||
async fn resources_subscribe_handler_returns_not_implemented() {
|
||||
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()]
|
||||
);
|
||||
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"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn resources_subscribe_handler_empty_when_no_channels_open() {
|
||||
async fn resources_subscribe_handler_not_implemented_even_when_empty() {
|
||||
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"
|
||||
);
|
||||
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"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in new issue
Block a user