fix(websocket): idle deadline off demux chunk-progress (WS-13)

The WS-01 idle timer reset on WS message arrival, so the forever-dribble
stall (declare a chunk, deliver its payload one byte per message) reset
the deadline forever while the demux stayed parked on the partial chunk;
conversely the reset-on-message rule was the only thing slow-but-alive
sessions survived on.

Semantics: the deadline now resets on demux progress — bytes actually
forwarded into read_tx that complete inbound chunks (the frames the
demux routes), tracked byte-for-byte in line with alkcall's parse walk
(8-byte header -> payload skip). Message arrival without a completed
chunk resets nothing, so the dribble hits the deadline; every completed
chunk (even one per message, slowly) re-arms the window.

Tests (tungstenite path): the forever-dribble eviction with 1001 despite
arriving messages; a productive-progress session that survives across
many windows; the knob-disabled (None) arm; the stall and text/cap arms
unchanged.

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<M, S, F>(
     mut ws_stream: S,
     read_tx: mpsc::Sender<Vec<u8>>,
@@ -226,15 +286,18 @@ async fn run_read_pump<M, S, F>(
     S::Item: IntoWsResult<M>,
     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<M, S, F>(
         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<u16> = 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");
     }
 }