diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index 1b23123..2cafabb 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -113,12 +113,68 @@ pub const WS_INTERNAL_ERROR: u16 = 1011; /// 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): a peer whose +/// inbound byte stream stops producing **complete chunks** cannot +/// park the single demux loop longer than this (WS-13 progress +/// semantics — the deadline resets when a complete chunk's bytes are +/// forwarded into `read_tx`, not when WS messages arrive, so a +/// forever-dribble inside a declared chunk still hits it). Zero-wait +/// is spelled `None`; this duration is the deployment default +/// (`HttpAdapter::with_ws_idle_timeout`). 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 +260,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 +286,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 +313,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 +1226,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 +1243,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 +1254,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"); } }