alkdev/alkhttp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'wt/review-002-ws13-idle-progress'

Browse Source
This commit is contained in:
glm-5.3-flash
2026-08-30 19:57:04 +00:00
parent 18c78acc79 b73804d9bc
commit 84bf538c1d
4 changed files with 516 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/adapter.rs
+14 -5
View File
@@ -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<Duration>) -> Self {
src/websocket/byte_adapter.rs
+298 -38
View File
@@ -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<M, S, F>(
mut ws_stream: S,
read_tx: mpsc::Sender<Vec<u8>>,
@@ -226,15 +319,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 +346,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 +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<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");
}
}