diff --git a/docs/superpowers/plans/2026-06-29-backpressure-resilience.md b/docs/superpowers/plans/2026-06-29-backpressure-resilience.md new file mode 100644 index 0000000..80cf41b --- /dev/null +++ b/docs/superpowers/plans/2026-06-29-backpressure-resilience.md @@ -0,0 +1,534 @@ +# Backpressure without protocol starvation — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Stop a slow consumer from starving the socket loop's protocol handling — deliver inbound frames via `try_send` plus a backpressure sub-loop that holds one frame and keeps keepalive/outgoing alive without reading inbound, and raise the channel default so transient hiccups never trigger it. + +**Architecture:** In `socket_loop_split`, inbound frame handling is split into `handle_incoming` (protocol: ping/pong/close/errors) and `deliver_with_backpressure` (data delivery). Data is delivered with `try_send`; on a full channel the loop holds the single frame and waits on `Sender::reserve()` while still servicing outgoing sends and the keepalive timer, but deliberately stops reading the inbound stream (TCP backpressure, bounded memory, zero loss). The `CHANNEL_SIZE` const-generic default goes from 4 to 256. + +**Tech Stack:** Rust (edition 2024), `tokio` mpsc (`try_send`/`reserve`), `tokio-tungstenite`. + +## Global Constraints + +- Rust edition 2024, MSRV 1.85. +- `#![deny(missing_docs)]` — every public item needs a doc comment. +- Lint clean: `cargo clippy --all-features --tests --benches -- -Dclippy::all -Dclippy::pedantic`. +- Format clean: `cargo fmt --check`. +- Compile/lint clean across `--no-default-features`, `--features tracing`, `--all-features`. +- Conventional commits (feat:, fix:, refactor:, test:, docs:). +- Public API signatures unchanged except the `CHANNEL_SIZE` default value (4 → 256). +- End commit messages with: `Co-Authored-By: Claude Opus 4.8 (1M context) ` + +--- + +## File Structure + +- `src/mock_server.rs` (modify): add `backpressure_probe_server` — sends a fixed burst, then watches for a client keepalive ping and replies with an "alive" marker only if it arrives. This is the test discriminator for protocol-liveness-during-backpressure. +- `src/lib.rs` (modify): re-export `backpressure_probe_server` under the `mocking` feature; change the `CHANNEL_SIZE` default from `4` to `256` and update its doc. +- `src/socket_loop.rs` (modify): replace `socket_message_received` with `handle_incoming` (returns an `Incoming` enum) and add `deliver_with_backpressure`; rewire the `socket_loop_split` inbound arm. +- `src/config.rs` (modify): one doc sentence noting backpressure resilience depends on keepalive being enabled. +- `tests/integration.rs` (modify): probe-server validation test (Task 1) and the backpressure discriminator test (Task 2). + +--- + +## Task 1: Probe mock server + validation test + +Adds the `backpressure_probe_server` test harness and a test that validates it on a connection large enough that backpressure never engages (so it passes on the current, unmodified loop). + +**Files:** +- Modify: `src/mock_server.rs` +- Modify: `src/lib.rs` +- Test: `tests/integration.rs` + +**Interfaces:** +- Produces: `pub async fn backpressure_probe_server(ws: WebSocketStreamType) -> Result` (behind `mocking`). On connect it sends 5 `EchoControlMessage::Message("burst-0..4")` Text frames, waits up to 500ms for a client `Ping`, and — only if one arrives — sends a final `EchoControlMessage::Message("alive")` Text frame; then it pongs/drains until the client closes. + +- [ ] **Step 1: Add imports to `src/mock_server.rs`** + +Add these to the existing `use` block at the top of `src/mock_server.rs`: + +```rust +use std::time::Duration; +use tokio::time::timeout; +``` + +- [ ] **Step 2: Add the probe server** + +Append this function to `src/mock_server.rs` (after `auth_echo_server`, before `get_mock_address`): + +```rust +/// Probe server for the backpressure tests. +/// +/// On connect it immediately sends a burst of five `EchoControlMessage::Message` +/// frames (`"burst-0"`..`"burst-4"`), then waits up to 500ms for a client +/// keepalive [`Message::Ping`]. It sends a final `EchoControlMessage::Message` +/// `"alive"` frame ONLY if that ping arrives — so a client whose socket loop is +/// stalled by a full receive buffer (and therefore cannot send keepalives) never +/// receives the marker. After that it pongs and drains until the client closes. +/// +/// This is a test harness exposed under the `mocking` feature flag and is **not** +/// intended for production use. +/// # Errors +/// - If sending a burst frame, the marker, or a pong fails +pub async fn backpressure_probe_server( + ws: WebSocketStreamType, +) -> Result { + let (mut sink, mut stream) = ws.split(); + + // Immediately send a burst of five data frames. + for i in 0..5u32 { + let msg = EchoControlMessage::Message(format!("burst-{i}")); + let text = serde_json::to_string(&msg).unwrap(); + sink.send(Message::Text(text.into())).await?; + } + + // Wait for a client keepalive ping — proof the client kept the protocol + // alive while its consumer was behind. + let got_ping = timeout(Duration::from_millis(500), async { + while let Some(Ok(frame)) = stream.next().await { + if matches!(frame, Message::Ping(_)) { + return true; + } + } + false + }) + .await + .unwrap_or(false); + + if got_ping { + let alive = EchoControlMessage::Message("alive".to_string()); + let text = serde_json::to_string(&alive).unwrap(); + sink.send(Message::Text(text.into())).await?; + } + + // Drain until the client closes. + while let Some(Ok(frame)) = stream.next().await { + match frame { + Message::Ping(_) => sink.send(Message::Pong(Bytes::new())).await?, + Message::Close(_) => break, + _ => {} + } + } + Ok(true) +} +``` + +- [ ] **Step 3: Re-export it from `src/lib.rs`** + +In `src/lib.rs`, update the `mocking` re-export line to include `backpressure_probe_server`: + +```rust +#[cfg(feature = "mocking")] +pub use mock_server::{ + EchoControlMessage, auth_echo_server, backpressure_probe_server, echo_server, get_mock_address, +}; +``` + +- [ ] **Step 4: Write the validation test** + +In `tests/integration.rs`, ensure the top `use socketeer::{...}` import list includes `NoopHandler` and `backpressure_probe_server` (add them to the existing list). Then add: + +```rust +#[tokio::test] +async fn test_backpressure_probe_server_delivers_burst_and_marker() { + // Large buffer (16) > burst (5): no backpressure. Validates the probe + // server — the client buffers the burst, goes idle, and its keepalive ping + // prompts the "alive" marker. Passes on the unmodified loop. + let server_address = get_mock_address(backpressure_probe_server).await; + let options = ConnectOptions { + keepalive_interval: Some(Duration::from_millis(100)), + ..ConnectOptions::default() + }; + let mut socketeer: Socketeer = + Socketeer::connect_with(&format!("ws://{server_address}"), options) + .await + .unwrap(); + + let mut received = Vec::new(); + for _ in 0..5 { + match socketeer.next_message().await.unwrap() { + EchoControlMessage::Message(s) => received.push(s), + other => panic!("unexpected frame: {other:?}"), + } + } + assert_eq!( + received, + vec!["burst-0", "burst-1", "burst-2", "burst-3", "burst-4"] + ); + + let marker = socketeer.next_message().await.unwrap(); + assert_eq!(marker, EchoControlMessage::Message("alive".to_string())); + + socketeer.close_connection().await.unwrap(); +} +``` + +- [ ] **Step 5: Run the test** + +Run: `cargo test --all-features --test integration test_backpressure_probe_server_delivers_burst_and_marker` +Expected: PASS (the probe server works; no backpressure with a buffer of 16). + +- [ ] **Step 6: Lint, format, full suite** + +Run: `cargo clippy --all-features --tests --benches -- -Dclippy::all -Dclippy::pedantic` → clean. +Run: `cargo fmt --check` → clean. +Run: `cargo test --all-features` → PASS. + +- [ ] **Step 7: Commit** + +```bash +git add src/mock_server.rs src/lib.rs tests/integration.rs +git commit -m "test: add backpressure probe mock server + +Sends a burst then replies with an 'alive' marker only if it receives a client +keepalive ping — the discriminator for protocol-liveness during backpressure. + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 2: Backpressure-resilient delivery + +Restructures the inbound path so a full receive channel no longer blocks the `select!`. TDD: the discriminator test is RED on the current loop (a stalled loop sends no keepalive, so the probe server withholds the marker) and GREEN after the change. + +**Files:** +- Modify: `src/socket_loop.rs` +- Test: `tests/integration.rs` + +**Interfaces:** +- Consumes: `backpressure_probe_server` (Task 1); existing `send_socket_message`, `send_keepalive`, `LoopState`, `TxChannelPayload`. +- Produces (internal): `enum Incoming { State(LoopState), Data(Message) }`; `async fn handle_incoming(message, sink) -> Incoming`; `async fn deliver_with_backpressure(frame, sender, receiver, sink, keepalive_interval, keepalive_message) -> LoopState`. + +- [ ] **Step 1: Write the failing discriminator test** + +In `tests/integration.rs`, ensure `use tokio::time::timeout;` is present at the top (add it next to the existing `use tokio::time::sleep;`). Then add: + +```rust +#[tokio::test] +async fn test_backpressure_keeps_protocol_alive() { + // Small buffer (2) < burst (5) forces backpressure. The client pauses past + // the keepalive interval before draining; the "alive" marker proves the + // loop kept firing keepalives WHILE backpressured (the server withholds it + // if no ping arrives within 500ms). On the pre-backpressure loop the loop + // is blocked on a full-channel send and sends no ping until the drain at + // 800ms — past the server's window — so the marker never comes. + let server_address = get_mock_address(backpressure_probe_server).await; + let options = ConnectOptions { + keepalive_interval: Some(Duration::from_millis(100)), + ..ConnectOptions::default() + }; + let mut socketeer: Socketeer = + Socketeer::connect_with(&format!("ws://{server_address}"), options) + .await + .unwrap(); + + // Stay paused well past the server's 500ms ping window. + sleep(Duration::from_millis(800)).await; + + // All five burst frames must arrive, in order — zero loss. + let mut received = Vec::new(); + for _ in 0..5 { + match socketeer.next_message().await.unwrap() { + EchoControlMessage::Message(s) => received.push(s), + other => panic!("unexpected frame: {other:?}"), + } + } + assert_eq!( + received, + vec!["burst-0", "burst-1", "burst-2", "burst-3", "burst-4"] + ); + + // Marker present => the client kept the protocol alive while backpressured. + let marker = timeout(Duration::from_secs(2), socketeer.next_message()) + .await + .expect("alive marker should arrive, not hang") + .unwrap(); + assert_eq!(marker, EchoControlMessage::Message("alive".to_string())); + + socketeer.close_connection().await.unwrap(); +} +``` + +- [ ] **Step 2: Run it to verify it fails** + +Run: `cargo test --all-features --test integration test_backpressure_keeps_protocol_alive` +Expected: FAIL — the five burst frames still arrive (no loss even today), but the marker read hits the 2s timeout and `.expect("alive marker should arrive, not hang")` panics, because the stalled loop sent no keepalive within the server's 500ms window. + +- [ ] **Step 3: Replace `socket_message_received` with `handle_incoming`** + +In `src/socket_loop.rs`, delete the entire `socket_message_received` function and replace it with the `Incoming` enum and `handle_incoming` (same protocol handling, but data frames are returned instead of delivered): + +```rust +enum Incoming { + State(LoopState), + Data(Message), +} + +#[cfg_attr(feature = "tracing", instrument)] +async fn handle_incoming( + message: Option>, + sink: &mut SocketSink, +) -> Incoming { + const PONG_BYTES: Bytes = Bytes::from_static(b"pong"); + match message { + Some(Ok(message)) => match message { + Message::Ping(_) => { + let send_result = sink + .send(Message::Pong(PONG_BYTES)) + .await + .map_err(Error::from); + Incoming::State(match send_result { + Ok(()) => LoopState::Running, + Err(e) => { + #[cfg(feature = "tracing")] + error!("Error sending Pong: {:?}", e); + LoopState::Error(e) + } + }) + } + Message::Close(_) => { + let close_result = sink.close().await; + Incoming::State(match close_result { + Ok(()) => LoopState::Closed, + Err(e) => { + #[cfg(feature = "tracing")] + error!("Error sending Close: {:?}", e); + LoopState::Error(Error::from(e)) + } + }) + } + Message::Text(_) | Message::Binary(_) => Incoming::Data(message), + _ => Incoming::State(LoopState::Running), + }, + Some(Err(e)) => { + #[cfg(feature = "tracing")] + error!("Error receiving message: {:?}", e); + Incoming::State(LoopState::Error(Error::WebsocketError(e))) + } + None => { + #[cfg(feature = "tracing")] + info!("Websocket Closed, closing rx channel"); + Incoming::State(LoopState::Error(Error::WebsocketClosed)) + } + } +} +``` + +- [ ] **Step 4: Add `deliver_with_backpressure`** + +In `src/socket_loop.rs`, add this function immediately after `handle_incoming`: + +```rust +/// Deliver one inbound data frame to the consumer. +/// +/// Fast path: `try_send`. If the receive channel is full, hold the single frame +/// and wait for capacity via `reserve()` while still servicing outgoing sends +/// and firing keepalives — but do NOT read the inbound stream. Unread bytes stay +/// in the TCP buffer (end-to-end backpressure), memory is bounded to this one +/// frame, and ordering is preserved (this frame is delivered before the loop +/// reads the next). +async fn deliver_with_backpressure( + frame: Message, + sender: &mut mpsc::Sender, + receiver: &mut mpsc::Receiver, + sink: &mut SocketSink, + keepalive_interval: Option, + keepalive_message: Option<&Message>, +) -> LoopState { + let frame = match sender.try_send(frame) { + Ok(()) => return LoopState::Running, + Err(mpsc::error::TrySendError::Closed(_)) => { + return LoopState::Error(Error::SocketeerDroppedWithoutClosing); + } + Err(mpsc::error::TrySendError::Full(frame)) => frame, + }; + #[cfg(feature = "tracing")] + trace!("Receive buffer full; holding frame and applying backpressure"); + loop { + if let Some(interval) = keepalive_interval { + select! { + permit = sender.reserve() => match permit { + Ok(permit) => { + permit.send(frame); + return LoopState::Running; + } + Err(_) => return LoopState::Error(Error::SocketeerDroppedWithoutClosing), + }, + outgoing = receiver.recv() => match send_socket_message(outgoing, sink).await { + LoopState::Running => {} + other => return other, + }, + () = sleep(interval) => match send_keepalive(sink, keepalive_message).await { + LoopState::Running => {} + other => return other, + }, + } + } else { + select! { + permit = sender.reserve() => match permit { + Ok(permit) => { + permit.send(frame); + return LoopState::Running; + } + Err(_) => return LoopState::Error(Error::SocketeerDroppedWithoutClosing), + }, + outgoing = receiver.recv() => match send_socket_message(outgoing, sink).await { + LoopState::Running => {} + other => return other, + }, + } + } + } +} +``` + +(Note: `frame` is moved only in the `permit` arm, which returns; the other arms don't touch it, so it stays valid across loop iterations. This compiles under NLL — no `Option`/`take` wrapper is needed.) + +- [ ] **Step 5: Rewire the `socket_loop_split` inbound arm** + +In `src/socket_loop.rs`, in `socket_loop_split`, replace BOTH `incoming_message = stream.next() => ...` arms (the one in the `if let Some(interval)` branch and the one in the `else` branch) so they route data through `deliver_with_backpressure`. The two `select!` blocks become: + +```rust + state = if let Some(interval) = keepalive_interval { + select! { + outgoing_message = receiver.recv() => send_socket_message(outgoing_message, &mut sink).await, + incoming_message = stream.next() => match handle_incoming(incoming_message, &mut sink).await { + Incoming::State(s) => s, + Incoming::Data(frame) => deliver_with_backpressure( + frame, + &mut sender, + &mut receiver, + &mut sink, + keepalive_interval, + keepalive_message.as_ref(), + ).await, + }, + () = sleep(interval) => send_keepalive(&mut sink, keepalive_message.as_ref()).await, + } + } else { + select! { + outgoing_message = receiver.recv() => send_socket_message(outgoing_message, &mut sink).await, + incoming_message = stream.next() => match handle_incoming(incoming_message, &mut sink).await { + Incoming::State(s) => s, + Incoming::Data(frame) => deliver_with_backpressure( + frame, + &mut sender, + &mut receiver, + &mut sink, + keepalive_interval, + keepalive_message.as_ref(), + ).await, + }, + } + }; +``` + +- [ ] **Step 6: Run the discriminator test to verify it passes** + +Run: `cargo test --all-features --test integration test_backpressure_keeps_protocol_alive` +Expected: PASS — the five burst frames arrive in order and the "alive" marker is received (the loop fired a keepalive while backpressured). + +- [ ] **Step 7: Lint, format, full suite** + +Run: `cargo clippy --all-features --tests --benches -- -Dclippy::all -Dclippy::pedantic` → clean. +Run: `cargo fmt --check` → clean. +Run: `cargo test --all-features` → PASS (all existing tests still green). + +- [ ] **Step 8: Commit** + +```bash +git add src/socket_loop.rs tests/integration.rs +git commit -m "feat: keep protocol alive under receive backpressure + +Inbound delivery now uses try_send + a backpressure sub-loop that holds one +frame and keeps keepalive/outgoing alive (without reading inbound) until the +consumer frees capacity. A slow consumer no longer stalls ping/pong and +keepalive, so the server no longer drops the connection. Zero data loss, +bounded memory, ordering preserved. + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 3: Raise the channel default + docs + +Raises the `CHANNEL_SIZE` default so transient consumer hiccups fit in the channel and never trigger the backpressure sub-loop, and documents the keepalive dependency. + +**Files:** +- Modify: `src/lib.rs` +- Modify: `src/config.rs` + +**Interfaces:** +- Produces: `Socketeer` (default changed from 4). + +- [ ] **Step 1: Change the default and its doc in `src/lib.rs`** + +In `src/lib.rs`, update the `CHANNEL_SIZE` doc bullet in the `Socketeer` type-parameter docs: + +```rust +/// - `CHANNEL_SIZE`: Capacity of the internal mpsc channels between the +/// connection task and the handle. Defaults to 256. A larger buffer absorbs +/// transient consumer slowdowns so backpressure (which holds one frame and +/// relies on keepalives to stay alive) engages only under sustained overload; +/// tune it for your feed's burstiness. +``` + +Then change the struct declaration default: + +```rust +pub struct Socketeer +``` + +- [ ] **Step 2: Document the keepalive dependency in `src/config.rs`** + +In `src/config.rs`, extend the doc comment on the `keepalive_interval` field to add a final sentence: + +```rust + /// Idle timeout before sending a keepalive. `None` disables keepalives entirely. + /// Defaults to 2 seconds. + /// + /// Keepalives also keep the connection alive while the receive buffer is + /// full and the reader is applying backpressure; with keepalives disabled, a + /// persistently slow consumer may eventually be disconnected by the server. + pub keepalive_interval: Option, +``` + +- [ ] **Step 3: Build and run the full suite** + +Run: `cargo test --all-features` +Expected: PASS — all existing tests still green with the larger default (none depend on the buffer being 4). + +- [ ] **Step 4: Lint and format across feature sets** + +Run: `cargo clippy --all-features --tests --benches -- -Dclippy::all -Dclippy::pedantic` → clean. +Run: `cargo clippy --no-default-features -- -Dclippy::all -Dclippy::pedantic` → clean. +Run: `cargo clippy --features tracing -- -Dclippy::all -Dclippy::pedantic` → clean. +Run: `cargo fmt --check` → clean. + +- [ ] **Step 5: Commit** + +```bash +git add src/lib.rs src/config.rs +git commit -m "feat: raise default channel size to 256 and document backpressure + +A larger default buffer absorbs transient consumer slowdowns so backpressure +engages only under sustained overload. Documents that backpressure resilience +relies on keepalive being enabled. + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Final verification (after Task 3) + +- [ ] `cargo test --all-features` — all pass (existing + 2 new backpressure tests). +- [ ] `cargo clippy --all-features --tests --benches -- -Dclippy::all -Dclippy::pedantic` — clean. +- [ ] `cargo clippy --no-default-features -- -Dclippy::all -Dclippy::pedantic` — clean. +- [ ] `cargo clippy --features tracing -- -Dclippy::all -Dclippy::pedantic` — clean. +- [ ] `cargo fmt --check` — clean. +- [ ] Skim the spec (`docs/superpowers/specs/2026-06-29-backpressure-resilience-design.md`) — every acceptance criterion met. +- [ ] Note: spec Testing item #3 (dead peer detected via keepalive during backpressure) is intentionally NOT a separate test. Its distinguishing case — a *frozen* consumer that never reads, where the new loop self-terminates via a failing keepalive but the old loop would block forever — is not observable through the public API (no exposed handle on the background task), and the variant where the consumer does read resolves identically on old and new code. The keepalive-failure-terminates-the-loop path is exercised by the existing `test_abrupt_close_surfaces_websocket_error` and the keepalive arm in `deliver_with_backpressure`. Documented here so the coverage decision is explicit, not silent. +- [ ] Note for the controller: `CLAUDE.md` is gitignored; if updating the local copy, mention the loop's new backpressure behavior there, but it won't be part of the branch. diff --git a/docs/superpowers/specs/2026-06-29-backpressure-resilience-design.md b/docs/superpowers/specs/2026-06-29-backpressure-resilience-design.md new file mode 100644 index 0000000..2273950 --- /dev/null +++ b/docs/superpowers/specs/2026-06-29-backpressure-resilience-design.md @@ -0,0 +1,168 @@ +# Backpressure without protocol starvation + +**Status:** Approved design — ready for implementation planning +**Date:** 2026-06-29 +**Scope:** PR 2 of the post-architecture-review series (issues #3 + #9). Stacks on +PR 3 (`feature/split-stream-api`), since it touches the socket loop's delivery +path that PR 3's `Stream`/`split` consumers read from. + +## Problem + +`socket_loop_split` runs a single `select!` that races three arms: outgoing +sends (`receiver.recv()`), inbound frames (`stream.next()`), and the keepalive +timer (`sleep(interval)`). When an inbound data frame arrives, +`socket_message_received` delivers it with `rx_sender.send(message).await`. If +the consumer has fallen behind and the rx channel (default `CHANNEL_SIZE = 4`) +is full, that `.await` blocks **inside the `select!` arm** — so the loop stops +servicing the other arms: it no longer pongs incoming pings and no longer fires +keepalives. The server sees an unresponsive client and drops the connection. +For a bursty data feed with an occasionally-slow consumer this is a latent +disconnect whose cause (protocol timeout) is opaque and misleading. + +Issue #9 is the adjacent observation that the default buffer of `4` is far too +small for a feed. + +## Goals + +- A slow or briefly-paused consumer must never cause a protocol-timeout + disconnect. +- Zero data loss and bounded memory: all delivered frames reach the consumer, + in order; the loop never buffers unboundedly. +- Keep the change minimal and non-breaking (no public API signature changes). + +## Non-goals + +- Dropping/overflow policies (latest-only feeds). The chosen philosophy is + preserve-all-data backpressure, not drop. +- Removing or changing the `CHANNEL_SIZE` const generic (only its default + value changes). +- Guaranteeing liveness against a server that demands prompt pongs to its own + pings during *sustained* overload (see Accepted caveat). + +## Design + +### 1. Deliver via a held permit, never a blind blocking send + +When an inbound data frame (`Text`/`Binary`) arrives, attempt non-blocking +delivery first: `rx_sender.try_send(frame)`. + +- `Ok(())` — fast path, continue the main loop. +- `Err(TrySendError::Closed(_))` — the consumer half was dropped; terminate + with `Error::SocketeerDroppedWithoutClosing` (same as today's closed-channel + handling). +- `Err(TrySendError::Full(frame))` — enter the **backpressure sub-loop** holding + exactly that one `frame`. + +The backpressure sub-loop `select!`s on: + +- `rx_sender.reserve()` → a permit becomes available (consumer freed a slot) → + `permit.send(frame)`; return `LoopState::Running` to the main loop, which + resumes reading the socket and pongs any piled-up pings in order. +- `receiver.recv()` (outgoing) → keep sending the user's outbound frames via the + existing send path; propagate any non-`Running` `LoopState` it produces + (e.g. a socket error or `SocketeerDroppedWithoutClosing`). +- `sleep(keepalive_interval)` → fire a keepalive (the existing + `send_keepalive`); propagate a non-`Running` result if the keepalive send + fails (peer gone). Only present when `keepalive_interval` is `Some`. + +Crucially, the sub-loop does **not** poll `stream.next()`. Holding the one +undelivered frame and not reading further means: memory is bounded to a single +in-flight frame, and unread socket bytes stay in the OS/TCP receive buffer, +applying genuine end-to-end backpressure to the server (its writes block via TCP +flow control). No data is lost; ordering is preserved (the held frame is +delivered before the loop reads the next one). + +### 2. Liveness during backpressure comes from our keepalive pings + +Because the sub-loop does not read inbound frames, it does not pong the server's +own pings until the consumer drains and the main loop resumes. The connection is +kept alive by our outgoing keepalive pings (default interval 2s). If a keepalive +`sink.send` fails, that surfaces as the loop's terminal error (the peer is gone) +rather than a hang. + +### 3. Raise the channel default (issue #9) + +Change the `CHANNEL_SIZE` const-generic **default** from `4` to `256`. The const +generic is retained (non-breaking) as the tuning knob. A larger default means +transient consumer hiccups fit entirely within the channel and never trigger the +backpressure sub-loop (and therefore never open a no-pong window); backpressure +engages only under sustained overload, which is exactly when riding it out +gracefully matters. This is why no internal "keep ponging while backpressured" +buffer is added — the larger channel already absorbs the transient case where +ponging-through would matter, at a fraction of the complexity. + +## Accepted caveat + +During *sustained* backpressure the loop relies solely on our keepalive pings +for liveness and does not pong the server's pings until it resumes reading. This +is safe for servers that treat client keepalive pings as liveness (the common +case) but a server that strictly requires prompt pongs to its own pings within a +tight window could still drop the connection under sustained overload. This is an +accepted trade-off versus the complexity of bounded ping-through buffering. + +Additionally, backpressure resilience depends on keepalive being enabled (the +default). With `keepalive_interval = None`, a stalled consumer sends nothing +proactively during backpressure and the connection may still eventually drop; +this will be documented. + +## Components touched + +- `src/socket_loop.rs`: the delivery path (currently `sender.send(message).await` + in `socket_message_received`) is restructured to `try_send` + a backpressure + sub-loop. The sub-loop needs access to the rx sender, the outgoing receiver, + the sink, and the keepalive config — so the delivery decision moves up into / + alongside `socket_loop_split` rather than staying buried in the + `socket_message_received` helper. Keep the helper focused (ping/pong/close + + the data-frame hand-off); the backpressure wait belongs in the loop. +- `src/lib.rs`: change the `CHANNEL_SIZE` default in the `Socketeer` struct + declaration from `4` to `256`. Update the type-parameter doc comment. +- Possibly `src/config.rs` doc: note the keepalive dependency for backpressure. + +## Testing + +Integration tests (mocking feature): + +1. **No-loss-under-backpressure / no-disconnect.** Connect with a small + `CHANNEL_SIZE` (e.g. `2`) and a short keepalive interval. Have the server + emit a burst of N messages (N > buffer). On the client, deliberately do not + read for longer than the keepalive interval (so a keepalive must fire during + backpressure), then drain all N. Assert: all N received **in order**, none + lost, and the connection is still alive afterward (round-trip one more + message, then close cleanly). +2. **Outgoing sends still flow during backpressure** (optional, if cleanly + testable): while the consumer is paused and the channel full, a `send` from + another task still reaches the server. +3. **Dead peer during backpressure is detected via keepalive** (keepalive + enabled). While the consumer is paused and the channel is full, the server + drops the connection abruptly. The next keepalive `sink.send` fails, so the + loop terminates with a `WebsocketError` (surfaced via the terminal-error + path) rather than hanging. Document the inherent limitation that with + `keepalive_interval = None`, a dead peer is not detected until the consumer + drains and the loop resumes reading — do not test that as a guarantee. + +A new mock-server capability may be needed: a server mode that emits a burst of +M messages on request (so the client can overflow its buffer deterministically). +Reuse `EchoControlMessage` with a new variant (e.g. `Burst(u32)`) or a dedicated +burst server. + +## Acceptance criteria + +- A consumer that pauses past the keepalive interval and then resumes loses no + messages and keeps the connection alive (covered by Test 1). +- `cargo test --all-features` green (existing + new tests). +- `cargo clippy --all-features --tests --benches -- -Dclippy::all + -Dclippy::pedantic` clean; clean across `--no-default-features` and + `--features tracing`. +- `cargo fmt --check` clean. +- Public API signatures unchanged except the `CHANNEL_SIZE` default value. +- `#![deny(missing_docs)]` satisfied. + +## Open/Deferred + +- Drop/latest-only overflow policy: deferred (different philosophy; revisit only + if a concrete feed needs it). +- Ping-through bounded buffering during sustained backpressure: deferred in + favor of the larger default channel. +- Migrating buffer sizing from the const generic to a runtime `ConnectOptions` + field: deferred (would be a larger API change; the const generic plus a saner + default suffices). diff --git a/src/config.rs b/src/config.rs index 1e10aca..1f4f2a0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -14,6 +14,13 @@ pub struct ConnectOptions { pub extra_headers: http::HeaderMap, /// Idle timeout before sending a keepalive. `None` disables keepalives entirely. /// Defaults to 2 seconds. + /// + /// Keepalives also keep the connection alive while the receive buffer is + /// full and the reader is applying backpressure; with keepalives disabled, + /// while the receive buffer is full the loop is not reading inbound frames + /// (so it will not observe a server close), and a consumer that stops + /// reading entirely can leave the connection parked until it resumes — in + /// addition to the server potentially dropping the idle connection. pub keepalive_interval: Option, /// If set, send this message as the keepalive instead of a WebSocket Ping frame. /// Useful for APIs that expect a custom keepalive payload — e.g. Interactive diff --git a/src/lib.rs b/src/lib.rs index dc01125..b64112a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,11 +18,16 @@ pub use handler::{ConnectionHandler, HandshakeContext, NoopHandler}; #[cfg(all(feature = "mocking", feature = "msgpack"))] pub use mock_server::msgpack_echo_server; #[cfg(feature = "mocking")] -pub use mock_server::{EchoControlMessage, auth_echo_server, echo_server, get_mock_address}; +pub use mock_server::{ + EchoControlMessage, auth_echo_server, backpressure_probe_server, echo_server, get_mock_address, +}; pub use split::{ReuniteError, SocketeerRx, SocketeerTx}; pub(crate) use socket_loop::WebSocketStreamType; -use socket_loop::{TerminalError, TxChannelPayload, send_close, send_confirmed, socket_loop_split}; +use socket_loop::{ + TerminalError, TxChannelPayload, poll_recv_raw, recv_raw, send_close, send_confirmed, + socket_loop_split, +}; use std::pin::Pin; use std::sync::{Arc, Mutex, PoisonError}; @@ -33,7 +38,7 @@ use tokio::sync::mpsc; use tokio_tungstenite::{connect_async, tungstenite::Message}; #[cfg(feature = "tracing")] -use tracing::{debug, info, instrument, trace}; +use tracing::{debug, info, instrument}; use url::Url; /// A WebSocket client that manages the connection to a WebSocket server. @@ -47,9 +52,12 @@ use url::Url; /// [`RawCodec`] for direct [`Message`] access. /// - `Handler`: A [`ConnectionHandler`] for lifecycle hooks (auth, subscriptions). /// Defaults to [`NoopHandler`] for the simple case. -/// - `CHANNEL_SIZE`: The size of the internal channels used to communicate between -/// the task managing the WebSocket connection and the client. -pub struct Socketeer +/// - `CHANNEL_SIZE`: Capacity of the internal mpsc channels between the +/// connection task and the handle. Defaults to 256. A larger buffer absorbs +/// transient consumer slowdowns so backpressure (which holds one frame and +/// relies on keepalives to stay alive) engages only under sustained overload; +/// tune it for your feed's burstiness. +pub struct Socketeer where Handler: ConnectionHandler, { @@ -84,14 +92,8 @@ where fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let this = self.get_mut(); - match this.receiver.poll_recv(cx) { - Poll::Ready(Some(message)) => Poll::Ready(Some(this.codec.decode(&message))), - Poll::Ready(None) => match socket_loop::take_terminal_error_opt(&this.terminal_error) { - Some(error) => Poll::Ready(Some(Err(error))), - None => Poll::Ready(None), - }, - Poll::Pending => Poll::Pending, - } + poll_recv_raw(&mut this.receiver, &this.terminal_error, cx) + .map(|frame| frame.map(|result| result.and_then(|message| this.codec.decode(&message)))) } } @@ -223,24 +225,10 @@ where /// - If the codec fails to decode the frame #[cfg_attr(feature = "tracing", instrument(skip(self)))] pub async fn next_message(&mut self) -> Result { - let Some(message) = self.receiver.recv().await else { - return Err(self.take_terminal_error()); - }; - #[cfg(feature = "tracing")] - trace!("Received message: {:?}", message); + let message = recv_raw(&mut self.receiver, &self.terminal_error).await?; self.codec.decode(&message) } - /// Report the error the socket loop recorded when it terminated, falling - /// back to the generic [`Error::WebsocketClosed`] for a graceful close (or - /// once the cause has already been taken). - /// - /// The cause is surfaced once: the first caller to observe the closed - /// connection consumes it, and subsequent observers see `WebsocketClosed`. - fn take_terminal_error(&self) -> Error { - socket_loop::take_terminal_error(&self.terminal_error) - } - /// Encode and send a message via the connection's [`Codec`]. /// This function will wait for the message to be sent before returning. /// @@ -264,10 +252,7 @@ where /// /// - If the WebSocket connection is closed or otherwise errored pub async fn next_raw_message(&mut self) -> Result { - match self.receiver.recv().await { - Some(message) => Ok(message), - None => Err(self.take_terminal_error()), - } + recv_raw(&mut self.receiver, &self.terminal_error).await } /// Send a raw [`Message`] to the WebSocket connection without running the codec. diff --git a/src/mock_server.rs b/src/mock_server.rs index 7f21ac4..a8d73ff 100644 --- a/src/mock_server.rs +++ b/src/mock_server.rs @@ -3,8 +3,10 @@ use bytes::Bytes; use futures::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use std::net::SocketAddr; +use std::time::Duration; use std::{fmt::Debug, future::Future}; use tokio::net::TcpListener; +use tokio::time::timeout; use tokio_tungstenite::{ MaybeTlsStream, tungstenite::{ @@ -202,6 +204,66 @@ pub async fn auth_echo_server(ws: WebSocketStreamType) -> Result Result { + let (mut sink, mut stream) = ws.split(); + + // Immediately send a burst of five data frames. + for i in 0..5u32 { + let msg = EchoControlMessage::Message(format!("burst-{i}")); + let text = serde_json::to_string(&msg).unwrap(); + sink.send(Message::Text(text.into())).await?; + } + + // Wait for a client keepalive ping — proof the client kept the protocol + // alive while its consumer was behind. + let got_ping = timeout(Duration::from_millis(500), async { + while let Some(Ok(frame)) = stream.next().await { + if matches!(frame, Message::Ping(_)) { + return true; + } + } + false + }) + .await + .unwrap_or(false); + + if got_ping { + let alive = EchoControlMessage::Message("alive".to_string()); + let text = serde_json::to_string(&alive).unwrap(); + sink.send(Message::Text(text.into())).await?; + } + + // Drain until the client closes. + while let Some(Ok(frame)) = stream.next().await { + match frame { + Message::Ping(_) => sink.send(Message::Pong(Bytes::new())).await?, + Message::Close(_) => { + sink.close().await?; + break; + } + _ => {} + } + } + Ok(true) +} + /// Create a WebSocket server that handles a customizable set of /// requests and exits. /// If the spawned socket handler returns true, the server will exit. diff --git a/src/socket_loop.rs b/src/socket_loop.rs index 1d7c932..3f73df6 100644 --- a/src/socket_loop.rs +++ b/src/socket_loop.rs @@ -11,6 +11,7 @@ use bytes::Bytes; use futures::{SinkExt, StreamExt, stream::SplitSink, stream::SplitStream}; use std::sync::{Arc, Mutex, PoisonError}; +use std::task::{Context, Poll}; use std::time::Duration; use tokio::{ net::TcpStream, @@ -120,6 +121,42 @@ pub(crate) async fn send_close(sender: &mpsc::Sender) -> Resul } } +/// Await the next inbound data frame, mapping a closed channel to the loop's +/// recorded terminal cause. The async counterpart to [`send_confirmed`]: both +/// honor the same terminal-error contract so the handle and its split half share +/// one receive path. +pub(crate) async fn recv_raw( + receiver: &mut mpsc::Receiver, + terminal_error: &TerminalError, +) -> Result { + match receiver.recv().await { + Some(message) => { + #[cfg(feature = "tracing")] + trace!("Received message: {:?}", message); + Ok(message) + } + None => Err(take_terminal_error(terminal_error)), + } +} + +/// Poll for the next inbound data frame, for `Stream` implementations. A closed +/// channel surfaces the recorded terminal cause once (subsequent polls see the +/// stream end). The polling counterpart to [`recv_raw`]. +pub(crate) fn poll_recv_raw( + receiver: &mut mpsc::Receiver, + terminal_error: &TerminalError, + cx: &mut Context<'_>, +) -> Poll>> { + match receiver.poll_recv(cx) { + Poll::Ready(Some(message)) => Poll::Ready(Some(Ok(message))), + Poll::Ready(None) => match take_terminal_error_opt(terminal_error) { + Some(error) => Poll::Ready(Some(Err(error))), + None => Poll::Ready(None), + }, + Poll::Pending => Poll::Pending, + } +} + #[cfg_attr( feature = "tracing", instrument(skip(keepalive_interval, keepalive_message, terminal_error)) @@ -138,13 +175,33 @@ pub(crate) async fn socket_loop_split( state = if let Some(interval) = keepalive_interval { select! { outgoing_message = receiver.recv() => send_socket_message(outgoing_message, &mut sink).await, - incoming_message = stream.next() => socket_message_received(incoming_message, &mut sender, &mut sink).await, + incoming_message = stream.next() => match handle_incoming(incoming_message, &mut sink).await { + Incoming::State(s) => s, + Incoming::Data(frame) => deliver_with_backpressure( + frame, + &mut sender, + &mut receiver, + &mut sink, + keepalive_interval, + keepalive_message.as_ref(), + ).await, + }, () = sleep(interval) => send_keepalive(&mut sink, keepalive_message.as_ref()).await, } } else { select! { outgoing_message = receiver.recv() => send_socket_message(outgoing_message, &mut sink).await, - incoming_message = stream.next() => socket_message_received(incoming_message, &mut sender, &mut sink).await, + incoming_message = stream.next() => match handle_incoming(incoming_message, &mut sink).await { + Incoming::State(s) => s, + Incoming::Data(frame) => deliver_with_backpressure( + frame, + &mut sender, + &mut receiver, + &mut sink, + keepalive_interval, + keepalive_message.as_ref(), + ).await, + }, } }; } @@ -190,12 +247,16 @@ async fn send_socket_message( } } +enum Incoming { + State(LoopState), + Data(Message), +} + #[cfg_attr(feature = "tracing", instrument)] -async fn socket_message_received( +async fn handle_incoming( message: Option>, - sender: &mut mpsc::Sender, sink: &mut SocketSink, -) -> LoopState { +) -> Incoming { const PONG_BYTES: Bytes = Bytes::from_static(b"pong"); match message { Some(Ok(message)) => match message { @@ -204,41 +265,100 @@ async fn socket_message_received( .send(Message::Pong(PONG_BYTES)) .await .map_err(Error::from); - match send_result { + Incoming::State(match send_result { Ok(()) => LoopState::Running, Err(e) => { #[cfg(feature = "tracing")] error!("Error sending Pong: {:?}", e); LoopState::Error(e) } - } + }) } Message::Close(_) => { let close_result = sink.close().await; - match close_result { + Incoming::State(match close_result { Ok(()) => LoopState::Closed, Err(e) => { #[cfg(feature = "tracing")] error!("Error sending Close: {:?}", e); LoopState::Error(Error::from(e)) } - } + }) } - Message::Text(_) | Message::Binary(_) => match sender.send(message).await { - Ok(()) => LoopState::Running, - Err(_) => LoopState::Error(Error::SocketeerDroppedWithoutClosing), - }, - _ => LoopState::Running, + Message::Text(_) | Message::Binary(_) => Incoming::Data(message), + _ => Incoming::State(LoopState::Running), }, Some(Err(e)) => { #[cfg(feature = "tracing")] error!("Error receiving message: {:?}", e); - LoopState::Error(Error::WebsocketError(e)) + Incoming::State(LoopState::Error(Error::WebsocketError(e))) } None => { #[cfg(feature = "tracing")] info!("Websocket Closed, closing rx channel"); - LoopState::Error(Error::WebsocketClosed) + Incoming::State(LoopState::Error(Error::WebsocketClosed)) + } + } +} + +/// Deliver one inbound data frame to the consumer. +/// +/// Fast path: `try_send`. If the receive channel is full, hold the single frame +/// and wait for capacity via `reserve()` while still servicing outgoing sends +/// and firing keepalives — but do NOT read the inbound stream. Unread bytes stay +/// in the TCP buffer (end-to-end backpressure), memory is bounded to this one +/// frame, and ordering is preserved (this frame is delivered before the loop +/// reads the next). +async fn deliver_with_backpressure( + frame: Message, + sender: &mut mpsc::Sender, + receiver: &mut mpsc::Receiver, + sink: &mut SocketSink, + keepalive_interval: Option, + keepalive_message: Option<&Message>, +) -> LoopState { + let frame = match sender.try_send(frame) { + Ok(()) => return LoopState::Running, + Err(mpsc::error::TrySendError::Closed(_)) => { + return LoopState::Error(Error::SocketeerDroppedWithoutClosing); + } + Err(mpsc::error::TrySendError::Full(frame)) => frame, + }; + #[cfg(feature = "tracing")] + trace!("Receive buffer full; holding frame and applying backpressure"); + loop { + if let Some(interval) = keepalive_interval { + select! { + permit = sender.reserve() => match permit { + Ok(permit) => { + permit.send(frame); + return LoopState::Running; + } + Err(_) => return LoopState::Error(Error::SocketeerDroppedWithoutClosing), + }, + outgoing = receiver.recv() => match send_socket_message(outgoing, sink).await { + LoopState::Running => {} + other => return other, + }, + () = sleep(interval) => match send_keepalive(sink, keepalive_message).await { + LoopState::Running => {} + other => return other, + }, + } + } else { + select! { + permit = sender.reserve() => match permit { + Ok(permit) => { + permit.send(frame); + return LoopState::Running; + } + Err(_) => return LoopState::Error(Error::SocketeerDroppedWithoutClosing), + }, + outgoing = receiver.recv() => match send_socket_message(outgoing, sink).await { + LoopState::Running => {} + other => return other, + }, + } } } } diff --git a/src/split.rs b/src/split.rs index 7adcac2..becaef7 100644 --- a/src/split.rs +++ b/src/split.rs @@ -16,8 +16,8 @@ use tokio_tungstenite::tungstenite::Message; use url::Url; use crate::socket_loop::{ - TerminalError, TxChannelPayload, send_close, send_confirmed, send_unconfirmed, - take_terminal_error, take_terminal_error_opt, + TerminalError, TxChannelPayload, poll_recv_raw, recv_raw, send_close, send_confirmed, + send_unconfirmed, }; use crate::{Codec, ConnectOptions, ConnectionHandler, Error, NoopHandler, Socketeer}; @@ -96,7 +96,7 @@ impl SocketeerTx { /// [`Socketeer::split`](crate::Socketeer::split). Implements /// [`Stream`](futures::Stream) and can be recombined with a [`SocketeerTx`] via /// [`reunite`](Self::reunite). -pub struct SocketeerRx +pub struct SocketeerRx where Handler: ConnectionHandler, { @@ -131,20 +131,15 @@ where /// - If the connection is closed or errored /// - If the codec fails to decode the frame pub async fn next_message(&mut self) -> Result { - match self.receiver.recv().await { - Some(message) => self.codec.decode(&message), - None => Err(take_terminal_error(&self.terminal_error)), - } + let message = recv_raw(&mut self.receiver, &self.terminal_error).await?; + self.codec.decode(&message) } /// Wait for the next raw [`Message`] without decoding. /// # Errors /// - If the connection is closed or errored pub async fn next_raw_message(&mut self) -> Result { - match self.receiver.recv().await { - Some(message) => Ok(message), - None => Err(take_terminal_error(&self.terminal_error)), - } + recv_raw(&mut self.receiver, &self.terminal_error).await } /// Recombine with a [`SocketeerTx`] to restore the full @@ -181,7 +176,7 @@ where /// Error returned by [`SocketeerRx::reunite`] when the two halves did not come /// from the same [`Socketeer::split`](crate::Socketeer::split). Carries both /// halves back so the caller can recover them. -pub struct ReuniteError +pub struct ReuniteError where Handler: ConnectionHandler, { @@ -227,16 +222,7 @@ where fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let this = self.get_mut(); - match this.receiver.poll_recv(cx) { - Poll::Ready(Some(message)) => Poll::Ready(Some(this.codec.decode(&message))), - Poll::Ready(None) => { - // Channel closed: surface the recorded cause once, then end. - match take_terminal_error_opt(&this.terminal_error) { - Some(error) => Poll::Ready(Some(Err(error))), - None => Poll::Ready(None), - } - } - Poll::Pending => Poll::Pending, - } + poll_recv_raw(&mut this.receiver, &this.terminal_error, cx) + .map(|frame| frame.map(|result| result.and_then(|message| this.codec.decode(&message)))) } } diff --git a/tests/integration.rs b/tests/integration.rs index 364e56f..0002027 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -6,13 +6,14 @@ use std::time::Duration; use bytes::Bytes; -use tokio::time::sleep; +use tokio::time::{sleep, timeout}; use tokio_tungstenite::tungstenite::Message; use futures::StreamExt; use socketeer::{ Codec, ConnectOptions, ConnectionHandler, EchoControlMessage, Error, HandshakeContext, - JsonCodec, RawCodec, Socketeer, auth_echo_server, echo_server, get_mock_address, + JsonCodec, NoopHandler, RawCodec, Socketeer, auth_echo_server, backpressure_probe_server, + echo_server, get_mock_address, }; #[cfg(feature = "msgpack")] @@ -755,6 +756,39 @@ async fn test_socketeer_as_stream() { assert_eq!(item, message); } +#[tokio::test] +async fn test_backpressure_probe_server_delivers_burst_and_marker() { + // Large buffer (16) > burst (5): no backpressure. Validates the probe + // server — the client buffers the burst, goes idle, and its keepalive ping + // prompts the "alive" marker. Passes on the unmodified loop. + let server_address = get_mock_address(backpressure_probe_server).await; + let options = ConnectOptions { + keepalive_interval: Some(Duration::from_millis(100)), + ..ConnectOptions::default() + }; + let mut socketeer: Socketeer = + Socketeer::connect_with(&format!("ws://{server_address}"), options) + .await + .unwrap(); + + let mut received = Vec::new(); + for _ in 0..5 { + match socketeer.next_message().await.unwrap() { + EchoControlMessage::Message(s) => received.push(s), + other => panic!("unexpected frame: {other:?}"), + } + } + assert_eq!( + received, + vec!["burst-0", "burst-1", "burst-2", "burst-3", "burst-4"] + ); + + let marker = socketeer.next_message().await.unwrap(); + assert_eq!(marker, EchoControlMessage::Message("alive".to_string())); + + socketeer.close_connection().await.unwrap(); +} + #[tokio::test] async fn test_binary_custom_keepalive() { // The widening of custom_keepalive_message from Option to @@ -780,3 +814,134 @@ async fn test_binary_custom_keepalive() { assert_eq!(socketeer.next_message().await.unwrap(), message); socketeer.close_connection().await.unwrap(); } + +#[tokio::test] +async fn test_backpressure_keeps_protocol_alive() { + // Small buffer (2) < burst (5) forces backpressure. The client pauses past + // the keepalive interval before draining; the "alive" marker proves the + // loop kept firing keepalives WHILE backpressured (the server withholds it + // if no ping arrives within 500ms). On the pre-backpressure loop the loop + // is blocked on a full-channel send and sends no ping until the drain at + // 800ms — past the server's window — so the marker never comes. + let server_address = get_mock_address(backpressure_probe_server).await; + let options = ConnectOptions { + keepalive_interval: Some(Duration::from_millis(100)), + ..ConnectOptions::default() + }; + let mut socketeer: Socketeer = + Socketeer::connect_with(&format!("ws://{server_address}"), options) + .await + .unwrap(); + + // Stay paused well past the server's 500ms ping window. + sleep(Duration::from_millis(800)).await; + + // All five burst frames must arrive, in order — zero loss. + let mut received = Vec::new(); + for _ in 0..5 { + match socketeer.next_message().await.unwrap() { + EchoControlMessage::Message(s) => received.push(s), + other => panic!("unexpected frame: {other:?}"), + } + } + assert_eq!( + received, + vec!["burst-0", "burst-1", "burst-2", "burst-3", "burst-4"] + ); + + // Marker present => the client kept the protocol alive while backpressured. + let marker = timeout(Duration::from_secs(2), socketeer.next_message()) + .await + .expect("alive marker should arrive, not hang") + .unwrap(); + assert_eq!(marker, EchoControlMessage::Message("alive".to_string())); + + socketeer.close_connection().await.unwrap(); +} + +#[tokio::test] +async fn test_backpressure_without_keepalive_preserves_frames() { + // Small buffer (2) < burst (5) forces backpressure, and keepalives are + // disabled — exercising the no-keepalive arm of `deliver_with_backpressure`. + // The consumer pauses, then drains: every burst frame must still arrive in + // order with zero loss. (No keepalive => no ping => the server never emits + // the "alive" marker, so we only assert the burst.) + let server_address = get_mock_address(backpressure_probe_server).await; + let options = ConnectOptions { + keepalive_interval: None, + ..ConnectOptions::default() + }; + let mut socketeer: Socketeer = + Socketeer::connect_with(&format!("ws://{server_address}"), options) + .await + .unwrap(); + + // Let the burst fill the buffer and park the loop in backpressure. + sleep(Duration::from_millis(200)).await; + + let mut received = Vec::new(); + for _ in 0..5 { + match timeout(Duration::from_secs(2), socketeer.next_message()) + .await + .expect("burst frame should arrive, not hang") + .unwrap() + { + EchoControlMessage::Message(s) => received.push(s), + other => panic!("unexpected frame: {other:?}"), + } + } + assert_eq!( + received, + vec!["burst-0", "burst-1", "burst-2", "burst-3", "burst-4"] + ); + + socketeer.close_connection().await.unwrap(); +} + +#[tokio::test] +async fn test_backpressure_allows_outgoing_sends() { + // Small buffer (2) < burst (5) parks the loop in backpressure while the + // consumer is not reading. A confirmed send issued in that window must still + // complete: it exercises the `receiver.recv()` arm of the backpressure loop, + // proving a slow receiver does not block the outgoing path. After that the + // held burst still drains in full and in order. + let server_address = get_mock_address(backpressure_probe_server).await; + let options = ConnectOptions { + keepalive_interval: Some(Duration::from_millis(100)), + ..ConnectOptions::default() + }; + let mut socketeer: Socketeer = + Socketeer::connect_with(&format!("ws://{server_address}"), options) + .await + .unwrap(); + + // Let the burst fill the buffer and park the loop in backpressure before we + // send, so the send is serviced from inside the backpressure loop rather + // than the outer select. + sleep(Duration::from_millis(200)).await; + + // Confirmed send completes even though the receive buffer is full and the + // consumer has read nothing. + socketeer + .send(EchoControlMessage::Message("while-backpressured".into())) + .await + .unwrap(); + + let mut received = Vec::new(); + for _ in 0..5 { + match timeout(Duration::from_secs(2), socketeer.next_message()) + .await + .expect("burst frame should arrive, not hang") + .unwrap() + { + EchoControlMessage::Message(s) => received.push(s), + other => panic!("unexpected frame: {other:?}"), + } + } + assert_eq!( + received, + vec!["burst-0", "burst-1", "burst-2", "burst-3", "burst-4"] + ); + + socketeer.close_connection().await.unwrap(); +}