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.
This commit is contained in:
glm-5.3-flash committed 2026-08-31 09:56:07 +00:00
1 parent d5fd548b8d
commit ae372c0c7e
12 files changed
+418 -44

No files matched your search

+138
View File
@@ -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"
);
}
}
+5 -5
View File
@@ -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<Option<BiStream>>,
@@ -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();
}
+1 -1
View File
@@ -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.