From ae372c0c7eb9667837d040fb13383aaf3815cb66 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 31 Aug 2026 09:56:07 +0000 Subject: [PATCH] test+docs: prepublish hardening from review (failure paths, doc hygiene) Coverage: - CF-002: demux skipped-bytes budget teardown test (268 MiB skip in-memory; a budgetless demux wedges in the 17th skip, the real one tears down) and the skip-hits-EOF arm (truncated oversized payload ends the loop). - CF-001: retryable CONNECTION_CLOSED pinned for subscribe write failures in both stream modes and single-stream publish request-frame failures; non-retryable INTERNAL pinned for mid-publish failures in both modes (deterministic FailOnFlushN write half). - Gateway: invoke_sink with with_deadline(None) completes a slow sink. Docs: - Fix broken intra-doc link on lib.rs's feature-gated gateway mention (rustdoc warned on default-feature builds). - Re-point 40 src/ references from the old alknet mono-repo ADR numbering (049/050/052/065/070/074/092) to this crate's numbering (021/011/034/007/008/009/005); drop into_sub_streams references removed by ADR-035. Verification: 565 default / 582 all-features (8 new), clippy -D warnings on default/gateway/all-features/wasm32, fmt clean, rustdoc warning-free on default and all-features, publish dry-run clean. Consumer-facing API continuity 0.1.1 -> 0.2.0 verified by compiling an API-surface probe against both versions. --- src/channels/adapter.rs | 138 +++++++++++++++++++++++ src/channels/source.rs | 10 +- src/channels/wire.rs | 2 +- src/client/from_call.rs | 2 +- src/core/types.rs | 54 ++++----- src/gateway/dispatch.rs | 32 ++++++ src/lib.rs | 2 +- src/protocol/connection.rs | 208 ++++++++++++++++++++++++++++++++++- src/protocol/dispatch.rs | 6 +- src/protocol/test_support.rs | 4 +- src/registry/context.rs | 2 +- src/registry/registration.rs | 2 +- 12 files changed, 418 insertions(+), 44 deletions(-) diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index 4c9ccbf..126380d 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -552,4 +552,142 @@ mod tests { assert_eq!(buf, [i; 4], "channel A chunk {i} intact — no data loss"); } } + + /// CF-002 #1 — the skipped-bytes budget tears down a dribbling peer. + /// Sixteen back-to-back `TooLarge` headers each declare 16 MiB + 1 and + /// their payloads are genuinely on the wire (the demux must consume + /// every byte to reach the next header, so the test cannot avoid + /// moving ~268 MiB through the duplex). The 16th skip pushes the + /// cumulative total past the 256 MiB budget: a working demux tears + /// down and *stops reading there*, while a peer that keeps dribbling + /// (a 17th header parked mid-payload) must not wedge it. Without the + /// budget the demux blocks skipping the 17th payload forever and the + /// timeout fires. + #[tokio::test] + async fn demux_tears_down_after_cumulative_skipped_bytes_budget_exceeded() { + let oversized_len = super::super::wire::MAX_CHUNK_LEN as usize + 1; + let burst_count = 16; + let (client, server) = tokio::io::duplex(1024 * 1024); + let (server_read, server_write) = tokio::io::split(server); + let (mux_handle, mux_runner) = MuxRunner::new(Box::new(server_write)); + let _mux_task = tokio::spawn(async move { + let _ = mux_runner.run().await; + }); + let manager = ChannelManager::with_defaults(mux_handle, None); + + let (_id, _send, _recv) = manager + .open_channel("alk/tty", "alice", None) + .await + .expect("open"); + + let demux_manager = manager.clone(); + let demux_task = tokio::spawn(async move { + ChannelsAdapter::run_demux_loop_for_client(&demux_manager, Box::new(server_read), None) + .await; + }); + + // Writer task: 16 full oversized payloads (15 × ~16 MiB ≈ 240 MiB + // skip cleanly under the budget; the 16th crosses it), then a 17th + // header + partial payload — and park instead of finishing (the + // peer keeps dribbling, never EOFs). + let writer_task = tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + let mut client_write = client; + let mut header = [0u8; 8]; + let chunk = vec![0u8; 64 * 1024]; + let full = oversized_len / chunk.len(); + let tail = oversized_len % chunk.len(); + for _ in 0..burst_count { + super::super::wire::write_header(1, oversized_len as u32, &mut header) + .expect("write header"); + client_write.write_all(&header).await.expect("write header"); + for _ in 0..full { + client_write.write_all(&chunk).await.expect("write payload"); + } + if tail > 0 { + client_write + .write_all(&chunk[..tail]) + .await + .expect("write payload tail"); + } + } + // The budget is already blown by the 16th skip; this header is + // the dribble that a budgetless demux would park on forever. + super::super::wire::write_header(1, oversized_len as u32, &mut header) + .expect("write header"); + client_write.write_all(&header).await.expect("write header"); + client_write + .write_all(&chunk[..8192]) + .await + .expect("write partial payload"); + std::future::pending::<()>().await; + }); + + let demux_done = + tokio::time::timeout(std::time::Duration::from_secs(120), demux_task).await; + assert!( + demux_done.is_ok(), + "demux must tear down on the skipped-bytes budget, not block in the skip loop" + ); + writer_task.abort(); + } + + /// CF-002 #2 — a skip that hits EOF mid-payload (peer declared more + /// bytes than it sent) ends the demux loop; nothing routes to the + /// channel and the loop does not spin back to header parsing on a + /// truncated stream. + #[tokio::test] + async fn demux_ends_when_oversized_payload_skip_hits_eof() { + let (client, server) = tokio::io::duplex(1024 * 1024); + let (server_read, server_write) = tokio::io::split(server); + let (mux_handle, mux_runner) = MuxRunner::new(Box::new(server_write)); + let _mux_task = tokio::spawn(async move { + let _ = mux_runner.run().await; + }); + let manager = ChannelManager::with_defaults(mux_handle, None); + + let (_id, _send, mut recv) = manager + .open_channel("alk/tty", "alice", None) + .await + .expect("open"); + + let demux_manager = manager.clone(); + let demux_task = tokio::spawn(async move { + ChannelsAdapter::run_demux_loop_for_client(&demux_manager, Box::new(server_read), None) + .await; + }); + + use tokio::io::AsyncWriteExt; + let oversized_len = super::super::wire::MAX_CHUNK_LEN as usize + 1; + let mut header = [0u8; 8]; + super::super::wire::write_header(1, oversized_len as u32, &mut header).expect("header"); + let mut client_write = client; + client_write.write_all(&header).await.expect("write header"); + client_write + .write_all(&[0u8; 4096]) + .await + .expect("write partial payload"); + drop(client_write); + + let demux_done = tokio::time::timeout(std::time::Duration::from_secs(10), demux_task).await; + assert!( + demux_done.is_ok(), + "demux must end when the skip hits EOF instead of looping" + ); + + // The oversized chunk was never routed — nothing to assert on the + // channel beyond the loop ending; reading would surface EOF since + // the demux dropped its sender without delivering a payload. + use tokio::io::AsyncReadExt; + let mut buf = [0u8; 4]; + let read = tokio::time::timeout( + std::time::Duration::from_millis(500), + recv.read_exact(&mut buf), + ) + .await; + assert!( + read.is_err() || read.unwrap().is_err(), + "no payload must arrive after the truncated skip" + ); + } } diff --git a/src/channels/source.rs b/src/channels/source.rs index d381f8d..66d6eee 100644 --- a/src/channels/source.rs +++ b/src/channels/source.rs @@ -4,7 +4,7 @@ //! //! The handler for a channel receives a `Connection` constructed via //! `Connection::from_source(ChannelBidiStreamSource, alpn)`. The -//! handler calls `accept_bi()` once (yield-once per channel, ADR-065) +//! handler calls `accept_bi()` once (yield-once per channel, ADR-007) //! and gets a `BiStream` — identical to how it works on a top-level //! QUIC connection. The `BiStream` is the reassembled read half joined //! to the mux write half via `BiStream::from_joined`. @@ -25,8 +25,8 @@ use crate::core::types::{BiStream, BidiStreamSource, StreamError}; /// /// The `BiStream` is constructed once, in [`ChannelBidiStreamSource::new`], /// and held in an `Option` — `accept_bi` takes it. This preserves the -/// yield-once contract (ADR-065) and the "split never crosses a crate -/// boundary as part of a constructor" rule (ADR-092) — the join happens +/// yield-once contract (ADR-007) and the "split never crosses a crate +/// boundary as part of a constructor" rule (ADR-009) — the join happens /// here, in the `BidiStreamSource` impl, not per-handler. pub struct ChannelBidiStreamSource { stream: Mutex>, @@ -65,9 +65,9 @@ impl BidiStreamSource for ChannelBidiStreamSource { /// `code`/`reason` are ignored: a single channel has no /// QUIC-shaped application-level close codes. The drop is the - /// close (ADR-065 §"Negative"). The `_` prefix is intentional — + /// close (ADR-007 §"Negative"). The `_` prefix is intentional — /// the signature matches the public `Connection::close` API - /// (ADR-070 §"REQ-CORE-02"). + /// (ADR-008 §"REQ-CORE-02"). fn close(&self, _code: u32, _reason: &str) { let _ = self.stream.lock().take(); } diff --git a/src/channels/wire.rs b/src/channels/wire.rs index 6cd356d..66daf13 100644 --- a/src/channels/wire.rs +++ b/src/channels/wire.rs @@ -23,7 +23,7 @@ use thiserror::Error; pub const CHUNK_HEADER_LEN: usize = 8; /// The maximum chunk payload length (16 MiB, matching TTY's cap — -/// ADR-052 §5). A chunk with `length > MAX_CHUNK_LEN` returns +/// ADR-034). A chunk with `length > MAX_CHUNK_LEN` returns /// [`ChunkError::TooLarge`] and does not corrupt the stream — the demux /// drops the chunk and continues. The header is always exactly 8 bytes, /// so the demux can always resync by reading the next 8-byte header. diff --git a/src/client/from_call.rs b/src/client/from_call.rs index 402b819..79063ac 100644 --- a/src/client/from_call.rs +++ b/src/client/from_call.rs @@ -411,7 +411,7 @@ fn make_forwarding_handler(connection: Arc, remote_name: String) /// leaf: on invocation, calls `CallConnection::subscribe_with_payload()` and /// forwards the remote stream end-to-end. Each `call.responded` from the /// remote becomes a stream item, `call.completed` ends the stream, and -/// `call.aborted` drops it (ADR-049 §8). No truncation, no first-value +/// `call.aborted` drops it (ADR-021 §8). No truncation, no first-value /// fallback. /// /// `forwarded_for` is populated from `context.identity` (ADR-032 §3), exactly diff --git a/src/core/types.rs b/src/core/types.rs index fc78e24..6bb18a3 100644 --- a/src/core/types.rs +++ b/src/core/types.rs @@ -227,14 +227,14 @@ pub trait ProtocolHandler: Send + Sync + 'static { async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError>; } -// --- BiStream: the handler leaf (ADR-092) --------------------------------- +// --- BiStream: the handler leaf (ADR-009) --------------------------------- /// Internal helper trait — the union of `AsyncRead + AsyncWrite + Send`. /// Not public; exists only to give `BiStream` a single boxed field. trait AsyncReadWrite: AsyncRead + AsyncWrite + Send {} impl AsyncReadWrite for T {} -/// The handler leaf — a bidirectional byte stream (ADR-092). +/// The handler leaf — a bidirectional byte stream (ADR-005, ADR-009). /// /// `accept_bi`/`open_bi` return a `BiStream`, not a split /// `(SendStream, RecvStream)` pair. Handlers that want the split halves call @@ -316,7 +316,7 @@ impl AsyncWrite for BiStream { } } -// --- SendStream / RecvStream: thin newtypes (ADR-092) --------------------- +// --- SendStream / RecvStream: thin newtypes (ADR-009) --------------------- pub struct SendStream { inner: Box, @@ -327,10 +327,11 @@ pub struct RecvStream { } impl SendStream { - /// Box a write half into the thin `SendStream` newtype. Used by - /// `into_sub_streams()` (ADR-074) and the channels reassembly path. - /// Not a constructor that feeds `Connection` — the split never crosses - /// a crate boundary as part of a constructor (ADR-092). + /// Box a write half into the thin `SendStream` newtype. Used by tests + /// and any consumer that holds pre-split halves (the channels layer + /// has its own `MpscSendStream`; `into_sub_streams()` was removed by + /// ADR-035). Not a constructor that feeds `Connection` — the split + /// never crosses a crate boundary as part of a constructor (ADR-009). pub fn from_stream(stream: impl AsyncWrite + Send + Unpin + 'static) -> Self { Self { inner: Box::new(stream), @@ -339,10 +340,11 @@ impl SendStream { } impl RecvStream { - /// Box a read half into the thin `RecvStream` newtype. Used by - /// `into_sub_streams()` (ADR-074) and the channels reassembly path. - /// Not a constructor that feeds `Connection` — the split never crosses - /// a crate boundary as part of a constructor (ADR-092). + /// Box a read half into the thin `RecvStream` newtype. Used by tests + /// and any consumer that holds pre-split halves (the channels layer + /// has its own `MpscRecvStream`; `into_sub_streams()` was removed by + /// ADR-035). Not a constructor that feeds `Connection` — the split + /// never crosses a crate boundary as part of a constructor (ADR-009). pub fn from_stream(stream: impl AsyncRead + Send + Unpin + 'static) -> Self { Self { inner: Box::new(stream), @@ -386,15 +388,15 @@ impl AsyncRead for RecvStream { /// Yield bidirectional streams to a `Connection`. Downstream crates implement /// this trait to add connection shapes (channels, a future transport, a test -/// double beyond the single-stream case) without editing core. See ADR-070 -/// for the full rationale and ADR-065 for the yield-once contract the +/// double beyond the single-stream case) without editing core. See ADR-008 +/// for the full rationale and ADR-007 for the yield-once contract the /// `StreamBidiStreamSource` impl preserves. The return type is `BiStream` -/// (ADR-092) — the join happens once, in the impl, not per-handler. +/// (ADR-009) — the join happens once, in the impl, not per-handler. #[async_trait] pub trait BidiStreamSource: Send + Sync + 'static { /// Yield the next bidirectional stream this connection provides. /// - /// Transport semantics (carried from ADR-065): + /// Transport semantics (carried from ADR-007): /// - QUIC (quinn/iroh): returns a new bidi stream on each call, /// `ConnectionClosed` when the underlying connection closes. /// - Single-stream (TCP+TLS, SSH channel, WebTransport stream, wasm): @@ -407,7 +409,7 @@ pub trait BidiStreamSource: Send + Sync + 'static { /// Open a bidirectional stream to the peer. /// /// Single-stream sources return `StreamClosed` (a single stream cannot - /// open new application streams — ADR-065). QUIC and channels sources + /// open new application streams — ADR-007). QUIC and channels sources /// open new streams. async fn open_bi(&self) -> Result; @@ -416,13 +418,13 @@ pub trait BidiStreamSource: Send + Sync + 'static { /// Close the connection. The `code`/`reason` args are QUIC application- /// level close codes; non-QUIC sources ignore them (the drop is the - /// close — ADR-065 §"Negative"). See ADR-070 §"REQ-CORE-02" for the + /// close — ADR-007 §"Negative"). See ADR-008 §"REQ-CORE-02" for the /// rationale for keeping the QUIC-shaped signature on the trait. fn close(&self, code: u32, reason: &str); } /// Single-stream `BidiStreamSource` (TCP+TLS, SSH channel, WebTransport -/// stream, wasm stream — ADR-065). Crate-private; constructed via +/// stream, wasm stream — ADR-007). Crate-private; constructed via /// `Connection::from_bidi`. `accept_bi` yields the underlying `BiStream` /// once, then `ConnectionClosed`; `open_bi` returns `StreamClosed`. struct StreamBidiStreamSource { @@ -449,9 +451,9 @@ impl BidiStreamSource for StreamBidiStreamSource { } /// `code`/`reason` are ignored: a single stream has no QUIC-shaped - /// application-level close codes. The drop is the close (ADR-065 + /// application-level close codes. The drop is the close (ADR-007 /// §"Negative"). The `_` prefix is intentional — the signature matches - /// the public `Connection::close` API (ADR-070 §"REQ-CORE-02"). + /// the public `Connection::close` API (ADR-008 §"REQ-CORE-02"). fn close(&self, _code: u32, _reason: &str) { let _ = self.stream.lock().unwrap_or_else(|e| e.into_inner()).take(); } @@ -467,11 +469,11 @@ impl Connection { /// Construct a `Connection` from a single bidirectional stream (e.g. /// `tokio::io::DuplexStream`, `TlsStream`, /// `russh::Channel::into_stream()`). The stream is wrapped in a - /// `BiStream` (ADR-092) and yielded by `accept_bi` once, then + /// `BiStream` (ADR-009) and yielded by `accept_bi` once, then /// `ConnectionClosed`. `open_bi` returns `StreamClosed` (a single - /// stream can't open new application streams — ADR-065). + /// stream can't open new application streams — ADR-007). /// - /// This is the only public stream constructor (ADR-092): the split + /// This is the only public stream constructor (ADR-009): the split /// never crosses a crate boundary as part of a constructor. Handlers /// that want the split halves call `tokio::io::split(&mut *stream)` on /// the `BiStream` they receive from `accept_bi`. @@ -492,7 +494,7 @@ impl Connection { /// Construct from a caller-supplied `BidiStreamSource` impl. The /// extension point for downstream crates — implement the trait and - /// construct a `Connection` from it without editing core. See ADR-070. + /// construct a `Connection` from it without editing core. See ADR-008. pub fn from_source(source: impl BidiStreamSource, alpn: Vec) -> Self { Self { source: Box::new(source), @@ -514,7 +516,7 @@ impl Connection { /// Handlers that loop `accept_bi` (e.g. `TtyAdapter`) get one session /// per single-stream connection; handlers that call once (e.g. /// `HttpAdapter`) get the stream directly. Both are correct. The - /// return type is `BiStream` (ADR-092); handlers that want the split + /// return type is `BiStream` (ADR-009); handlers that want the split /// halves call `tokio::io::split` on the `BiStream`. pub async fn accept_bi(&self) -> Result { self.source.accept_bi().await @@ -656,7 +658,7 @@ mod tests { /// `tokio::io::sink()` + `tokio::io::empty()`: reads yield EOF /// immediately (zero bytes), writes discard. Exists because /// `Connection::from_bidi` requires a single value that implements - /// both traits (ADR-092). Used only to construct a `Connection` for + /// both traits (ADR-009). Used only to construct a `Connection` for /// tests that exercise `Connection`-level state (alpn, addr, identity) /// without ever reading or writing the stream. pub(crate) struct SinkEmpty; diff --git a/src/gateway/dispatch.rs b/src/gateway/dispatch.rs index 58ad4d5..281eee9 100644 --- a/src/gateway/dispatch.rs +++ b/src/gateway/dispatch.rs @@ -527,6 +527,38 @@ mod tests { ); } + #[tokio::test] + async fn invoke_sink_with_no_deadline_completes_a_slow_sink_handler() { + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + spec("slow/sink", Visibility::External, OperationType::Pub), + HandlerKind::Sink(make_sink_handler(|_input, ctx, mut chunks| async move { + while let Some(chunk) = chunks.next().await { + let _ = chunk; + } + tokio::time::sleep(Duration::from_millis(80)).await; + ResponseEnvelope::ok(ctx.request_id, serde_json::json!({})) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + let dispatch = GatewayDispatch::new(Arc::new(registry)).with_deadline(None); + let chunks: crate::registry::registration::PublishStream = Box::pin(futures::stream::iter( + vec![Ok(serde_json::json!({ "n": 1 }))], + )); + let envelope = dispatch + .invoke_sink(None, "/slow/sink", serde_json::json!({}), chunks) + .await; + assert!( + envelope.result.is_ok(), + "a no-deadline spine must not time out a slow sink, got {envelope:?}" + ); + } + #[tokio::test] async fn streaming_sub_op_streams_envelopes() { let mut registry = OperationRegistry::new(); diff --git a/src/lib.rs b/src/lib.rs index a68dce9..7c4447b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,7 +17,7 @@ //! going forward. //! - **Registry** ([`registry`]): operation specs, context, dispatch, and //! the operation registry — the call half's dispatch core. -//! - **Gateway** ([`gateway`], feature `gateway`): the transport-neutral +//! - **Gateway** (module `gateway`, feature `gateway`): the transport-neutral //! dispatch spine and `services/schema` disclosure guard — the //! deadline-bounded, re-rooted-context invoke surface for HTTP //! gateways, hub relays, and other transport front-ends (ADR-048). diff --git a/src/protocol/connection.rs b/src/protocol/connection.rs index 8c54e92..0895a0d 100644 --- a/src/protocol/connection.rs +++ b/src/protocol/connection.rs @@ -213,7 +213,7 @@ impl CallConnection { } }; - // `open_bi` returns a `BiStream` (ADR-092); split it into halves for + // `open_bi` returns a `BiStream` (ADR-009); split it into halves for // the call protocol's separate write (request) and read (response) // pumps. The split is the stdlib idiom; no per-handler wrapper. let stream = match connection.open_bi().await { @@ -337,7 +337,7 @@ impl CallConnection { } }; - // `open_bi` returns a `BiStream` (ADR-092); split for the separate + // `open_bi` returns a `BiStream` (ADR-009); split for the separate // write (request) and read (subscription events) pumps. let stream = match connection.open_bi().await { Ok(s) => s, @@ -600,7 +600,7 @@ impl CallConnection { .connection .as_ref() .ok_or_else(|| "no underlying connection (overlay-only)".to_string())?; - // `open_bi` returns a `BiStream` (ADR-092). We only need the write + // `open_bi` returns a `BiStream` (ADR-009). We only need the write // half to send the envelope; split and drop the read half. let stream = connection .open_bi() @@ -1938,6 +1938,39 @@ mod tests { crate::core::types::BiStream::from_joined(read_end, BrokenWrite) } + /// AsyncWrite whose writes succeed but whose Nth `poll_flush` fails. + /// `FrameFramedWriter::write_frame` flushes exactly once per frame, so + /// `remaining_ok_flushes = k` lets frames 1..=k land and fails inside + /// frame k+1 — a deterministic mid-stream write failure. + struct FailOnFlushN { + remaining_ok_flushes: std::sync::atomic::AtomicUsize, + } + impl AsyncWrite for FailOnFlushN { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Poll::Ready(Ok(buf.len())) + } + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + let prev = self + .remaining_ok_flushes + .fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + if prev == 0 { + Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "mux dead", + ))) + } else { + Poll::Ready(Ok(())) + } + } + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + #[tokio::test] async fn single_stream_call_write_failure_is_retryable_connection_closed() { let bi_stream = broken_write_bi_stream(); @@ -2045,6 +2078,175 @@ mod tests { assert!(err.retryable); } + #[tokio::test] + async fn single_stream_subscribe_write_failure_is_retryable_connection_closed() { + use futures::stream::StreamExt; + + let bi_stream = broken_write_bi_stream(); + let (writer, reader) = split_single_stream(bi_stream); + let conn = CallConnection::new_single_stream(stub_connection(), writer); + drop(reader); + + let mut stream = conn.subscribe("test/events", serde_json::json!({})).await; + let item = stream.next().await.unwrap(); + match item.result { + Err(e) => { + assert_eq!(e.code, "CONNECTION_CLOSED"); + assert!(e.retryable); + } + other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"), + } + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn multi_stream_subscribe_write_failure_is_retryable_connection_closed() { + use crate::core::types::StreamError; + use crate::core::types::{BiStream, BidiStreamSource, Connection}; + use futures::stream::StreamExt; + use std::net::SocketAddr; + + struct OneBiStream(std::sync::Mutex>); + #[async_trait::async_trait] + impl BidiStreamSource for OneBiStream { + async fn accept_bi(&self) -> Result { + Err(StreamError::ConnectionClosed) + } + async fn open_bi(&self) -> Result { + let stream = self.0.lock().expect("open once").take(); + stream.ok_or(StreamError::StreamClosed) + } + fn remote_addr(&self) -> Option { + None + } + fn close(&self, _code: u32, _reason: &str) {} + } + + let conn = Connection::from_source( + OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))), + b"alk/call".to_vec(), + ); + let call_conn = CallConnection::new(conn); + + let mut stream = call_conn + .subscribe("test/events", serde_json::json!({})) + .await; + let item = stream.next().await.unwrap(); + match item.result { + Err(e) => { + assert_eq!(e.code, "CONNECTION_CLOSED"); + assert!(e.retryable); + } + other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"), + } + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn single_stream_publish_request_frame_write_failure_is_retryable() { + let bi_stream = broken_write_bi_stream(); + let (writer, reader) = split_single_stream(bi_stream); + let conn = CallConnection::new_single_stream(stub_connection(), writer); + drop(reader); + + let response = conn + .publish( + "fs/upload", + serde_json::json!({}), + Box::pin(futures::stream::iter(vec![serde_json::json!({"c": 1})])), + ) + .await; + match response.result { + Err(e) => { + // The request frame never landed — nothing delivered, retry safe. + assert_eq!(e.code, "CONNECTION_CLOSED"); + assert!(e.retryable); + } + other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"), + } + } + + #[tokio::test] + async fn single_stream_publish_midpublish_write_failure_stays_internal() { + // Flush succeeds for the request frame (1) + first chunk (2), then + // fails inside the second chunk — delivery is ambiguous from there. + let bi_stream = crate::core::types::BiStream::from_joined( + tokio::io::empty(), + FailOnFlushN { + remaining_ok_flushes: std::sync::atomic::AtomicUsize::new(2), + }, + ); + let (writer, reader) = split_single_stream(bi_stream); + let conn = CallConnection::new_single_stream(stub_connection(), writer); + drop(reader); + + let chunks = vec![serde_json::json!({"c": 1}), serde_json::json!({"c": 2})]; + let stream: Pin + Send>> = + Box::pin(futures::stream::iter(chunks)); + let response = conn + .publish("fs/upload", serde_json::json!({}), stream) + .await; + match response.result { + Err(e) => { + assert_eq!(e.code, "INTERNAL"); + assert!(!e.retryable, "mid-publish failure must not be retryable"); + } + other => panic!("expected non-retryable INTERNAL, got {other:?}"), + } + } + + #[tokio::test] + async fn multi_stream_publish_midpublish_write_failure_stays_internal() { + use crate::core::types::StreamError; + use crate::core::types::{BiStream, BidiStreamSource, Connection}; + use std::net::SocketAddr; + + struct OneBiStream(std::sync::Mutex>); + #[async_trait::async_trait] + impl BidiStreamSource for OneBiStream { + async fn accept_bi(&self) -> Result { + Err(StreamError::ConnectionClosed) + } + async fn open_bi(&self) -> Result { + let stream = self.0.lock().expect("open once").take(); + stream.ok_or(StreamError::StreamClosed) + } + fn remote_addr(&self) -> Option { + None + } + fn close(&self, _code: u32, _reason: &str) {} + } + + let conn = Connection::from_source( + OneBiStream(std::sync::Mutex::new(Some( + crate::core::types::BiStream::from_joined( + tokio::io::empty(), + FailOnFlushN { + remaining_ok_flushes: std::sync::atomic::AtomicUsize::new(2), + }, + ), + ))), + b"alk/call".to_vec(), + ); + let call_conn = CallConnection::new(conn); + + let chunks = vec![serde_json::json!({"c": 1}), serde_json::json!({"c": 2})]; + let stream: Pin + Send>> = + Box::pin(futures::stream::iter(chunks)); + let response = call_conn + .publish("fs/upload", serde_json::json!({}), stream) + .await; + match response.result { + Err(e) => { + // The pump died on a *chunk* frame — the request landed, + // delivery is ambiguous, INTERNAL (non-retryable) is correct. + assert_eq!(e.code, "INTERNAL"); + assert!(!e.retryable); + } + other => panic!("expected non-retryable INTERNAL, got {other:?}"), + } + } + #[tokio::test] async fn shared_frame_writer_concurrent_writes_do_not_interleave() { let (client_end, server_end) = tokio::io::duplex(64 * 1024); diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index cc43de5..30b7e0a 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -382,7 +382,7 @@ impl Dispatcher { connection: Arc, stream: crate::core::types::BiStream, ) { - // `stream` is a `BiStream` (ADR-092) — `AsyncRead + AsyncWrite + Send + // `stream` is a `BiStream` (ADR-009) — `AsyncRead + AsyncWrite + Send // + Unpin`. Split into the read and write halves the call protocol's // frame reader/writer consume. The split is the stdlib idiom; no // per-handler wrapper. @@ -441,7 +441,7 @@ impl Dispatcher { /// for `Ok`, `call.error` for `Err`). On natural stream end (the stream /// returned `None` without the last item being an `Err`), write a /// `call.completed` frame. An `Err` envelope is terminal — the stream - /// ends after it and we do NOT write `call.completed` (ADR-049 §6). + /// ends after it and we do NOT write `call.completed` (ADR-021 §6). /// /// If a frame write fails the pump stops early; the stream is dropped on /// return, releasing the handler's resources via `Drop` (ADR-016). A @@ -1477,7 +1477,7 @@ mod tests { ); } - // --- streaming dispatch branch (ADR-049 §6) --------------------------- + // --- streaming dispatch branch (ADR-021 §6) --------------------------- fn subscription_spec(name: &str, acl: AccessControl) -> OperationSpec { OperationSpec::new( diff --git a/src/protocol/test_support.rs b/src/protocol/test_support.rs index 39682be..65ce323 100644 --- a/src/protocol/test_support.rs +++ b/src/protocol/test_support.rs @@ -1,6 +1,6 @@ //! Shared test helpers for the call protocol's inline `#[cfg(test)]` //! modules. Kept here (not in each test module) so the `stub_connection()` -//! shape is defined once — `Connection::from_stream` was removed (ADR-092) +//! shape is defined once — `Connection::from_stream` was removed (ADR-009) //! and every test stub that previously called it now calls //! `Connection::from_bidi(SinkEmpty, ...)` via `sink_empty_connection()`. @@ -15,7 +15,7 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; /// A test-only `AsyncRead + AsyncWrite` pair equivalent to /// `tokio::io::sink() + tokio::io::empty()`: reads yield EOF immediately /// (zero bytes), writes discard. Exists because `Connection::from_bidi` -/// (ADR-092 — the only public stream constructor, replacing +/// (ADR-009 — the only public stream constructor, replacing /// `from_stream`) requires a single value that implements both traits. /// Used only to construct a `Connection` for tests that exercise /// `Connection`-level state (alpn, addr, identity, dispatcher run loop diff --git a/src/registry/context.rs b/src/registry/context.rs index 8d3cc57..4c6c5e1 100644 --- a/src/registry/context.rs +++ b/src/registry/context.rs @@ -33,7 +33,7 @@ pub struct OperationContext { pub internal: bool, /// `None` when no ownership provider is wired (backward compat — /// `check` falls back to static `Identity.resources` path). Wired by - /// the assembly layer via `CallAdapter`/`Dispatcher` (ADR-050). + /// the assembly layer via `CallAdapter`/`Dispatcher` (ADR-011). pub ownership: Option>, } diff --git a/src/registry/registration.rs b/src/registry/registration.rs index e3e73b3..7b25666 100644 --- a/src/registry/registration.rs +++ b/src/registry/registration.rs @@ -662,7 +662,7 @@ where /// shallow nested lookups (translated to `/a/b`). /// /// Returns `None` if the path is malformed, the field is missing, or the -/// value at the path is not a string. Graceful — never panics (ADR-050 §2a): +/// value at the path is not a string. Graceful — never panics (ADR-011 §2a): /// a missing field simply yields `resource_id: None`, and the caller's /// `AccessControl::check` decides whether to deny based on whether the spec /// requires a specific resource ID.