diff --git a/docs/architecture/websocket.md b/docs/architecture/websocket.md index 8b5da9a..c8db82e 100644 --- a/docs/architecture/websocket.md +++ b/docs/architecture/websocket.md @@ -1,6 +1,6 @@ --- status: draft -last_updated: 2026-08-27 +last_updated: 2026-08-30 --- # WebSocket — the Browser Bidirectional Path (Channels over WS) @@ -176,6 +176,49 @@ The adapter is shared with the `from_wss` consumer path ([ADR-070](decisions/070-from-wss-consumer-adapter.md)) — one implementation, both directions. +#### Idle-read timeout: progress semantics + the no-keepalive decision (WS-01, WS-13) + +The WS pumps carry an **idle-read eviction knob** +(`HttpAdapter::with_ws_idle_timeout`, default +`DEFAULT_WS_IDLE_TIMEOUT` = 60 s, disable with `None`). Its semantics, +decided in review-002 (WS-13), are **progress-based, deliberately +strict**: + +- The deadline resets on **demux progress** — bytes actually forwarded + into the byte stream that *complete* an inbound chunk (a full + 8-byte header + its declared payload). WS message arrival resets + nothing. +- Therefore the dribble stall (declare a chunk, deliver its payload a + byte per message) hits the deadline and is evicted with a `1001` + (GoingAway) close — even though messages keep arriving — while a + peer delivering complete chunks, however slowly per message, re-arms + the window with each chunk and survives. +- **There is no WS ping/pong keepalive, on purpose.** A keepalive + rescues app-silence only by re-arming the deadline — and a pong is + indistinguishable from the dribble's almost-invisible arrivals, so + adding one would reopen the stall it exists to seal. The recorded + decision (module doc of `src/websocket/byte_adapter.rs`, option (b) + of the two WS-13 alternatives): **60 s of no chunk progress is an + intentional eviction line, even for a silent subscription.** +- A deployment running long-lived silent-but-alive sessions (quiet + subscriptions that outlast the window) disables the knob with + `with_ws_idle_timeout(None)` and leans on the remaining levers: + the session registry's forced-eviction + ([`WsSessions::abort`](#connection-local-overlay)), the + inbound caps (WS-06), and the write-side caps (WS-04/05/06). This is + the same deployment posture as `from_wss`'s drop monitor + ([ADR-070](decisions/070-from-wss-consumer-adapter.md)): the idle + knob bounds *demux parking*, not app liveness. +- Layered note (the WS-13/FWD-15 interaction recorded here): the + *HTTP* SSE path (`/subscribe`) sends server-side keep-alive comment + frames every 15 s — see + [http-server.md](http-server.md)) — because its idle enemy is + LB/proxy timeouts, and its keep-alive does not reset any + progress deadline, it cannot reopen the WS-13 hole. The two live at + different layers: SSE keep-alive fights transport fires; the WS idle + knob bounds demux parking. Both documented in the same pass per + review-002 Unit-2 sequencing. + ### Dispatch: channel 0 = the shared `Dispatcher`, unchanged Channel 0's session is the alknet design's native session, verbatim: diff --git a/src/server/adapter.rs b/src/server/adapter.rs index 57b1480..838499f 100644 --- a/src/server/adapter.rs +++ b/src/server/adapter.rs @@ -218,11 +218,20 @@ impl HttpAdapter { self } - /// The WS idle-read timeout (WS-01): the read pump closes the - /// connection with a 1001 (GoingAway) close frame after this long - /// without an inbound WS message — bounding the demux stall a - /// dribbling (or silently-stalled) peer can pin. `None` disables - /// the knob (not recommended: the stall window is then unbounded). + /// The WS idle-read timeout (WS-01, WS-13 semantics): the read + /// pump closes the connection with a 1001 (GoingAway) close frame + /// after this long producing **no completed inbound chunk** — the + /// deadline resets on demux progress (complete chunks forwarded + /// into the byte stream), not on WS message arrival, so a + /// dribbling peer (slow message arrivals inside a declared chunk) + /// is bounded while productive-but-slow peers survive. This is an + /// intentional no-progress eviction line, not a transport-idle + /// bound: there is no WS ping/pong keepalive, and app-silence that + /// outlasts the window (a quiet subscription) is evicted by design + /// — see `websocket::byte_adapter`'s module doc. `None` disables + /// the knob (not recommended: the stall window is then unbounded; + /// long-lived silent subscriptions are the intended `None` case, + /// leaning on `WsSessions::abort` and the write-side caps). /// /// Default: [`crate::websocket::DEFAULT_WS_IDLE_TIMEOUT`] (60 s). pub fn with_ws_idle_timeout(mut self, idle_timeout: Option) -> Self { diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index 1b23123..0ac4bb8 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -32,6 +32,32 @@ //! Text WS messages are rejected with a protocol-level close (code //! 1002); all frames are binary (websocket.md §Framing). //! +//! Idle-read timeout (WS-01, WS-13 semantics): the knob +//! (`DEFAULT_WS_IDLE_TIMEOUT`, deployment-adjustable via +//! `HttpAdapter::with_ws_idle_timeout`, disable via `None`) evicts a +//! connection whose inbound stream produces **no completed chunk for +//! the whole window** — the deadline resets on demux progress (bytes +//! forwarded into `read_tx` that complete 8-byte-header-framed chunks), +//! never on WS message arrival, so a forever-dribble inside a declared +//! chunk hits the deadline even though messages keep arriving, while a +//! peer delivering complete chunks — however slowly per-message — +//! re-arms the window with each one. +//! +//! Legitimate silence (WS-13 decision, recorded — option (b)): there +//! is deliberately **no WS ping/pong keepalive**. A keepalive can only +//! rescue app-silence by re-arming the deadline, which would reopen the +//! dribble hole it exists to seal (pong = traffic from the attacker's +//! point of view); instead, 60 s of *no chunk progress* is an +//! intentional eviction line even for a silent subscription — a +//! long-lived quiet subscription that must survive past the window +//! (with server-side pushes; see the keep-alive discussion in +//! `websocket.md`) is exactly the deployment that dials +//! `with_ws_idle_timeout(None)` and leans on the other bounds +//! (`WsSessions::abort` eviction, the write-side caps). Read eviction +//! closes with 1001 (Going Away) — a normal connection end from the +//! demux's point of view (EOF → channels cleared, pendings failed), +//! not a protocol error. +//! //! Close mapping: WS close (either side) → read EOF → the demux clears //! all channels (REQ-CH-02) and the dispatch loop fails outstanding //! pendings. `AsyncWrite::shutdown` closes the WS sink after the queued @@ -107,18 +133,81 @@ pub const WS_PROTOCOL_ERROR: u16 = 1002; /// violation it cannot recover from (WS-04/HY-09, WS-05). pub const WS_INTERNAL_ERROR: u16 = 1011; -/// Idle-read close code (WS-01): a read side that stays silent past the -/// configured idle timeout is closed with 1001 (Going Away) — a normal -/// connection end from the demux's point of view (EOF → channels -/// cleared, pendings failed), not a protocol error. +/// Idle-read close code (WS-01/WS-13): a read side that produces no +/// demux progress (no completed inbound chunk) past the configured +/// window is closed with 1001 (Going Away) — a normal connection end +/// from the demux's point of view (EOF → channels cleared, pendings +/// failed), not a protocol error. pub const WS_GOING_AWAY: u16 = 1001; -/// Default idle-read timeout for the WS pumps (WS-01): a peer that -/// drips bytes (or a stalled partial chunk header) cannot park the -/// single demux loop longer than this without traffic. Zero disables -/// the timeout. Deployment knob (`HttpAdapter::with_ws_idle_timeout`). +/// Default idle-read timeout for the WS pumps (WS-01, WS-13): a +/// connection whose inbound stream completes **no chunk** within this +/// window is evicted — the deadline resets on demux progress (complete +/// chunks forwarded into `read_tx`), not on WS message arrival, so a +/// forever-dribble inside a declared chunk still hits it. +/// +/// This is an intentional no-progress eviction line, *not* a +/// transport-idle bound: there is no WS ping/pong keepalive, and +/// app-silence that outlasts the window (a quiet subscription) is +/// evicted with 1001 by design — see the module doc's "Legitimate +/// silence" decision. A deployment running long-lived silent +/// subscriptions disables the knob with +/// `HttpAdapter::with_ws_idle_timeout(None)` (`None`, not zero — zero +/// is not a meaningful window). pub const DEFAULT_WS_IDLE_TIMEOUT: Duration = Duration::from_secs(60); +/// Observes the inbound byte stream for chunk framing (WS-13): counts +/// complete chunks whose bytes have been forwarded into `read_tx` — +/// the progress signal the idle-read deadline resets on. The parse +/// walk mirrors the demux loop's byte-for-byte (8-byte header → +/// payload skip; an over-`MAX_CHUNK_LEN` length is skipped like the +/// demux's TooLarge arm and parsing continues), with O(1) state: no +/// byte copying, and the forwarded byte stream is unmodified. +struct InboundChunkProgress { + header: [u8; 8], + header_fill: u8, + payload_remaining: u64, +} + +impl InboundChunkProgress { + fn new() -> Self { + Self { + header: [0u8; 8], + header_fill: 0, + payload_remaining: 0, + } + } + + fn observe(&mut self, bytes: &[u8]) -> u32 { + let mut completed = 0u32; + for &byte in bytes { + if self.payload_remaining > 0 { + self.payload_remaining -= 1; + if self.payload_remaining == 0 { + completed += 1; + } + continue; + } + self.header[self.header_fill as usize] = byte; + self.header_fill += 1; + if self.header_fill == 8 { + self.header_fill = 0; + let len = u32::from_be_bytes([ + self.header[4], + self.header[5], + self.header[6], + self.header[7], + ]); + self.payload_remaining = len as u64; + if len == 0 { + completed += 1; + } + } + } + completed + } +} + /// Maximum chunk payload length — the channels protocol's 16 MiB wire /// bound (ADR-052 §5), re-exported from alkcall. The write-side chunk /// parser rejects a header claiming a longer payload instead of waiting @@ -204,15 +293,19 @@ where /// from_wss drop monitor's input; its `Sender` semantics are preserved /// EXACTLY: single `send` after the read loop terminates). /// -/// Idle-read timeout (WS-01): when `idle_timeout` is non-zero, the -/// next-message await is wrapped in a `tokio::time::timeout` — a peer -/// that dribbles (or a stalled header) cannot park the single demux -/// loop beyond the knob. Staleness sends the 1001 GoingAway close to -/// the peer (a normal connection end for the demux/from_wss EOF -/// machinery, not a protocol error) and ends the loop, so the -/// adapter's read half sees EOF and the channels teardown runs. -/// Messages that *do* arrive reset the window; a partial chunk header -/// alone does not (the boundary is a complete WS message). +/// Idle-read timeout (WS-01, WS-13 semantics): when `idle_timeout` is +/// non-zero, each next-message await is bounded by the *remaining* +/// budget — the window minus the time since the last **demux +/// progress** event (bytes actually forwarded into `read_tx` that +/// complete inbound chunks, i.e. frames the demux will route). WS +/// message arrival resets nothing: a peer dribbling bytes into a +/// declared chunk forever hits the deadline even though messages keep +/// arriving; a peer delivering complete chunks (even slowly, one per +/// message) refreshes the window with each chunk. Budget exhaustion +/// sends the 1001 GoingAway close to the peer (a normal connection +/// end for the demux/from_wss EOF machinery, not a protocol error) +/// and ends the loop, so the adapter's read half sees EOF and the +/// channels teardown runs. async fn run_read_pump( mut ws_stream: S, read_tx: mpsc::Sender>, @@ -226,15 +319,18 @@ async fn run_read_pump( S::Item: IntoWsResult, F: FnOnce(), { + let mut progress = InboundChunkProgress::new(); + let mut last_progress_at = tokio::time::Instant::now(); loop { - let msg = match idle_timeout { + let budget = idle_timeout.map(|window| window.saturating_sub(last_progress_at.elapsed())); + let msg = match budget { None => ws_stream.next().await, - Some(timeout) => match tokio::time::timeout(timeout, ws_stream.next()).await { + Some(budget) => match tokio::time::timeout(budget, ws_stream.next()).await { Ok(msg) => msg, Err(_elapsed) => { send_write_error( &write_error_slot_for_read, - "connection idle past the read timeout", + "connection made no inbound chunk progress past the read timeout", ); let _ = write_tx_for_read .clone() @@ -250,9 +346,13 @@ async fn run_read_pump( match msg.into_ws_result() { Ok(m) => { if let Some(b) = M::binary(&m) { + let completed = progress.observe(b.as_ref()); if read_tx.send(b.as_ref().to_vec()).await.is_err() { break; } + if completed > 0 { + last_progress_at = tokio::time::Instant::now(); + } } else if M::is_text(&m) { let _ = write_tx_for_read .clone() @@ -1159,11 +1259,15 @@ mod tests { assert!(eof_fired, "EOF signal fired for the from_wss machinery"); } - /// The idle knob resets on traffic: a peer dribbling one message - /// per window (each inside the knob) keeps the connection open — - /// only the *stall* (no message within one full window) closes. + /// The forever-dribble stall is bounded by *progress*, not + /// message arrival (WS-02 acceptance / WS-13 semantics): a peer + /// declaring one chunk (`[0: u32 BE][64: u32 BE]`) and dribbling + /// its 64 payload bytes one byte per message, half a window apart, + /// never completes the chunk — the read pump must end with the + /// 1001 GoingAway close within the knob even though messages keep + /// arriving, so the demux channel machinery sees EOF. #[tokio::test] - async fn idle_read_timeout_resets_on_traffic() { + async fn idle_read_timeout_bounds_a_forever_dribble_with_no_chunk_progress() { let (client_io, server_io) = tokio::io::duplex(1 << 16); let ws = tokio_tungstenite::WebSocketStream::from_raw_socket( client_io, @@ -1172,7 +1276,7 @@ mod tests { ) .await; let knob = std::time::Duration::from_millis(200); - let (mut stream, _pumps) = split_tungstenite_to_bytes_idle(ws, Some(knob)); + let (_stream, pumps) = split_tungstenite_to_bytes_idle(ws, Some(knob)); let peer = tokio_tungstenite::WebSocketStream::from_raw_socket( server_io, @@ -1183,24 +1287,180 @@ mod tests { let (mut peer_sink, mut peer_stream) = peer.split(); use futures::{SinkExt, StreamExt}; - // Dribble: a message every 100 ms (half the knob) for 8 rounds — - // 1.6 s total, far past a single 200 ms window. The connection - // must stay open; every message surfaces on the adapter. - let mut buf = [0u8; 1]; - for round in 0u8..8 { - tokio::time::sleep(knob / 2).await; + let mut eof_rx = pumps.read_eof(); + let dribble = tokio::spawn(async move { + let mut header = vec![0u8; 8]; + header[4..8].copy_from_slice(&64u32.to_be_bytes()); peer_sink .send(tokio_tungstenite::tungstenite::Message::Binary( - vec![round].into(), + header.into(), + )) + .await + .expect("header write accepted"); + for byte in 0u8..64 { + tokio::time::sleep(knob / 4).await; + if peer_sink + .send(tokio_tungstenite::tungstenite::Message::Binary( + vec![byte].into(), + )) + .await + .is_err() + { + return; + } + } + loop { + tokio::time::sleep(knob).await; + if peer_sink + .send(tokio_tungstenite::tungstenite::Message::Binary( + vec![0u8].into(), + )) + .await + .is_err() + { + return; + } + } + }); + + let outcome = tokio::time::timeout(std::time::Duration::from_secs(10), async { + let close: Option = loop { + match peer_stream.next().await { + Some(Ok(tokio_tungstenite::tungstenite::Message::Close(cf))) => { + break cf.map(|f| u16::from(f.code)) + } + Some(Ok(_)) => continue, + Some(Err(_)) | None => break None, + } + }; + let _ = eof_rx.changed().await; + close + }) + .await + .expect("dribble must hit the progress deadline within 10 s"); + dribble.abort(); + assert_eq!( + outcome, + Some(WS_GOING_AWAY), + "the forever-dribble is evicted with 1001 despite arriving messages" + ); + } + + /// Progress semantics, survivor side (WS-13 accept 2): a session + /// whose inbound traffic keeps *making progress* (complete chunks + /// arriving, each within the window) is NOT disconnected even + /// though it never goes quiet in a way the WS-01 shape would + /// reward — every complete chunk re-arms the deadline from zero, + /// which is the "messages flowing that make progress" survival + /// the knob's spirit prescribes. + #[tokio::test] + async fn idle_read_timeout_survives_slow_productive_chunks() { + let (client_io, server_io) = tokio::io::duplex(1 << 16); + let ws = tokio_tungstenite::WebSocketStream::from_raw_socket( + client_io, + tokio_tungstenite::tungstenite::protocol::Role::Client, + None, + ) + .await; + let knob = std::time::Duration::from_millis(200); + let (mut stream, pumps) = split_tungstenite_to_bytes_idle(ws, Some(knob)); + let mut evict_rx = pumps.read_eof(); + + let peer = tokio_tungstenite::WebSocketStream::from_raw_socket( + server_io, + tokio_tungstenite::tungstenite::protocol::Role::Server, + None, + ) + .await; + let (mut peer_sink, _peer_stream) = peer.split(); + use futures::{SinkExt, StreamExt}; + + for round in 0u8..6u8 { + tokio::time::sleep(knob / 2).await; + let mut chunk = vec![0u8; 12]; + chunk[0..4].copy_from_slice(&0u32.to_be_bytes()); + chunk[4..8].copy_from_slice(&4u32.to_be_bytes()); + chunk[8..12].copy_from_slice(&[round; 4]); + peer_sink + .send(tokio_tungstenite::tungstenite::Message::Binary( + chunk.clone().into(), + )) + .await + .expect("chunk write accepted"); + let mut buf = vec![0u8; 12]; + tokio::time::timeout(knob, stream.read_exact(&mut buf)) + .await + .expect("connection alive between productive chunks") + .expect("read"); + assert_eq!(buf, chunk, "chunk bytes surfaced in order"); + } + let evicted_early = + tokio::time::timeout(std::time::Duration::from_millis(0), evict_rx.changed()).await; + assert!( + evicted_early.is_err(), + "no eviction while chunks keep landing inside the window" + ); + let mut header = vec![0u8; 8]; + header[4..8].copy_from_slice(&0u32.to_be_bytes()); + peer_sink + .send(tokio_tungstenite::tungstenite::Message::Binary( + header.into(), + )) + .await + .expect("EOF chunk write accepted"); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + match stream.read(&mut [0u8; 1]).await { + Ok(0) => return, + Ok(_) => continue, + Err(e) => panic!("read failed: {e}"), + } + } + }) + .await + .expect("demux EOF observed after the final chunk"); + } + + /// `idle_timeout = None` disables the deadline entirely — the + /// "no bound" arm of the WS-01 knob surface: no timer runs, so + /// messages arriving without chunk progress never evict. + #[tokio::test] + async fn idle_read_timeout_disabled_when_none() { + let (client_io, server_io) = tokio::io::duplex(1 << 16); + let ws = tokio_tungstenite::WebSocketStream::from_raw_socket( + client_io, + tokio_tungstenite::tungstenite::protocol::Role::Client, + None, + ) + .await; + let (_stream, pumps) = split_tungstenite_to_bytes_idle(ws, None); + + let peer = tokio_tungstenite::WebSocketStream::from_raw_socket( + server_io, + tokio_tungstenite::tungstenite::protocol::Role::Server, + None, + ) + .await; + let (mut peer_sink, _peer_stream) = peer.split(); + use futures::SinkExt; + + let eof_rx = pumps.read_eof(); + for _ in 0..4 { + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + peer_sink + .send(tokio_tungstenite::tungstenite::Message::Binary( + vec![0u8].into(), )) .await .expect("dribble write accepted"); - tokio::time::timeout(knob, stream.read_exact(&mut buf)) - .await - .expect("connection alive across the dribble") - .expect("read"); - assert_eq!(buf[0], round); } - let _ = peer_stream.next().await; + assert!( + !*eof_rx.borrow(), + "no idle timer ran with the knob disabled: arriving messages evict nothing" + ); + peer_sink + .send(tokio_tungstenite::tungstenite::Message::Close(None)) + .await + .expect("peer close accepted"); } } diff --git a/tests/ws_upgrade_session.rs b/tests/ws_upgrade_session.rs index 98a4dc9..dfd6afa 100644 --- a/tests/ws_upgrade_session.rs +++ b/tests/ws_upgrade_session.rs @@ -758,3 +758,163 @@ async fn session_cap_rejects_over_limit_with_503_and_frees_slots_on_end() { assert_eq!(env.r#type, EVENT_RESPONDED, "slot freed after session end"); third.close().await; } + +/// WS-13 acceptance on the axum path (the integration mirror; also the +/// COV-11b dark-knob gate for `with_ws_idle_timeout`): a client +/// dribbling a declared chunk's payload one byte per message, each gap +/// inside the idle window, is evicted with 1001 — message arrival does +/// not reset the deadline, only a completed chunk would. +#[tokio::test] +async fn idle_progress_knob_evicts_forever_dribble_over_axum_upgrade() { + use alkhttp::server::HttpAdapter; + + let registry_val = OperationRegistry::new(); + let registry = Arc::new(registry_val); + struct StaticTok; + impl IdentityProvider for StaticTok { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let s = String::from_utf8_lossy(&token.raw).to_string(); + (s == "tok-1").then(|| identity("alice", &[])) + } + } + let adapter = HttpAdapter::new(Arc::new(StaticTok), Arc::clone(®istry)) + .with_ws_idle_timeout(Some(std::time::Duration::from_millis(150))); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + let app: axum::Router = adapter.router().clone(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1") + .await + .unwrap(); + + let dribbler = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1") + .await + .unwrap(); + let dribble = tokio::spawn(async move { + let mut dribbler = dribbler; + let mut header = vec![0u8; 8]; + header[4..8].copy_from_slice(&64u32.to_be_bytes()); + dribbler.send_binary(header).await; + loop { + tokio::time::sleep(std::time::Duration::from_millis(40)).await; + dribbler.send_binary(vec![0u8]).await; + } + }); + + let close = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + match ws.next_close(std::time::Duration::from_millis(250)).await { + Some(Some(code)) => break code, + Some(None) => break 1006, + None => continue, + } + } + }) + .await + .expect("dribbling client must be evicted within 10 s"); + dribble.abort(); + assert_eq!( + close, + alkhttp::websocket::WS_GOING_AWAY, + "the forever-dribble hits the progress deadline despite arriving messages" + ); +} + +/// WS-13 survivor side on the axum path: a session whose traffic keeps +/// completing chunks (each chunk inside the window — the scaled-down +/// "messages flowing that make progress" shape) is NOT evicted, and +/// calls keep round-tripping across many windows. +#[tokio::test] +async fn idle_progress_knob_survives_productive_sessions_over_axum_upgrade() { + use alkhttp::server::HttpAdapter; + + let registry = echo_registry(); + struct StaticTok; + impl IdentityProvider for StaticTok { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let s = String::from_utf8_lossy(&token.raw).to_string(); + (s == "tok-1").then(|| identity("alice", &[])) + } + } + let adapter = HttpAdapter::new(Arc::new(StaticTok), Arc::clone(®istry)) + .with_ws_idle_timeout(Some(std::time::Duration::from_millis(150))); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + let app: axum::Router = adapter.router().clone(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1") + .await + .unwrap(); + + for i in 0..6 { + tokio::time::sleep(std::time::Duration::from_millis(75)).await; + let n = i * 7 + 1; + let env = call_and_await( + &mut ws, + &format!("req-{n}"), + "echo/run", + serde_json::json!({ "n": n }), + ) + .await; + assert_eq!(env.r#type, EVENT_RESPONDED); + assert_eq!(env.payload["output"]["n"], n); + } + ws.close().await; +} + +/// WS-13 `None` arm on the axum path: with the knob disabled a +/// dribbling client that completes no chunk is never evicted. +#[tokio::test] +async fn idle_progress_knob_none_disables_eviction_over_axum_upgrade() { + use alkhttp::server::HttpAdapter; + + let registry = echo_registry(); + struct StaticTok; + impl IdentityProvider for StaticTok { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let s = String::from_utf8_lossy(&token.raw).to_string(); + (s == "tok-1").then(|| identity("alice", &[])) + } + } + let adapter = + HttpAdapter::new(Arc::new(StaticTok), Arc::clone(®istry)).with_ws_idle_timeout(None); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + let app: axum::Router = adapter.router().clone(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1") + .await + .unwrap(); + + let mut header = vec![0u8; 8]; + header[4..8].copy_from_slice(&8u32.to_be_bytes()); + ws.send_binary(header).await; + for _ in 0..4 { + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + ws.send_binary_piece(&[0u8]).await; + } + let no_close = ws.next_close(std::time::Duration::from_millis(400)).await; + assert!( + !matches!(no_close, Some(Some(_))), + "knob disabled: no eviction arrives, got {no_close:?}" + ); + ws.close().await; +}