From bbb0fc4e2f5c3bd61f660249ac6d1b14c0bb11be Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 27 May 2026 00:19:16 +0200 Subject: [PATCH 1/4] feat(scanner_ws): add client-side Ping/Pong keepalive to stop spurious watchdog reconnects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On Mainnet (10 min mean block time), the existing 90 s liveness watchdog fires every ~2 min on quiet stretches between blocks because nothing flows on the WebSocket between block events. The scanner reconnects ~27 times per hour — operationally harmless (HTTP fallback handles block fetches, scanner recovers) but noisy and trips log alerts. RFC 6455 §5.5 mandates the peer respond to every Ping with a Pong; the Pong arrives on the same reader the watchdog observes. Adding a periodic client-side Ping makes the watchdog reset every ~30 s regardless of block-event cadence, restoring its original meaning ("no pong + no event in 90 s = connection genuinely dead, reconnect"). - New `DEFAULT_PING_INTERVAL: Duration = 30s` and `ScannerWsConfig::ping_interval`. - Compile-time assertion that `ping_interval * 2 < liveness_timeout` so any future tweak to either constant trips the build instead of silently drifting back into the spurious-reconnect regime. - `connect_and_drain` now splits the WebSocketStream (`StreamExt::split`) and runs reader + ping ticker inside one `tokio::select!`. The liveness deadline is tracked manually via `tokio::time::sleep_until(deadline)` rather than wrapping each `next()` in `timeout(...)`, so select-arm cancellation cannot silently reset the watchdog (only inbound frames do). - Marker `scanner-polling-ok:` added on the `tokio::time::interval` / `sleep_until` lines per the CI lint enforcing CONTRIBUTING.md § "No polling — events only". Tests: - `ping_interval_is_strictly_below_half_liveness_timeout` — const sanity. - `run_scanner_ws_pongs_keep_connection_alive_past_liveness_timeout` — quiet server (no blocks, draining reader auto-pongs); scanner stays on a single connection through multiple watchdog windows. - `run_scanner_ws_watchdog_fires_when_pongs_are_dropped` — server stops reading after subscribe; no pongs come back; watchdog still fires and the scanner reconnects (count ≥ 2 in a 1.5 s window). Existing tests get an explicit `ping_interval` field; values are set well above the test budget so the keepalive does not interfere with the behaviour each test is targeting. --- node/src/scanner_ws.rs | 174 +++++++++++++++++++++++------- node/src/scanner_ws_tests.rs | 198 +++++++++++++++++++++++++++++++++++ 2 files changed, 335 insertions(+), 37 deletions(-) diff --git a/node/src/scanner_ws.rs b/node/src/scanner_ws.rs index e21df9bf..4277a61d 100644 --- a/node/src/scanner_ws.rs +++ b/node/src/scanner_ws.rs @@ -29,6 +29,15 @@ //! the `tokio::time::` reference in event-driven code (documented //! in CONTRIBUTING.md, enforced by the CI lint added in the same //! PR). +//! - 30 s client-side Ping keepalive (`ping_interval`). A tokio +//! `interval` ticker running alongside the reader sends a +//! `WsMessage::Ping` to the peer every `ping_interval`. RFC 6455 +//! §5.5 mandates a Pong response, which arrives on the same +//! reader and resets the liveness watchdog. Without this, a quiet +//! Mainnet-tier upstream (10-min mean block time) had nothing +//! flowing in the watchdog window and reconnected every ~2 min; +//! the keepalive turns the watchdog into the half-open detector +//! it was always meant to be (no pong + no event = dead). //! - On reconnect, fetch the current tip via the existing //! `EsploraClient::get_tip_hash` and push that hash into the //! channel too. This plugs the gap that opened while we were @@ -76,6 +85,26 @@ pub const DEFAULT_ESPLORA_WS_URL: &str = "wss://mutinynet.com/api/v1/ws"; /// socket is half-open" signal. pub const DEFAULT_LIVENESS_TIMEOUT: Duration = Duration::from_secs(90); +/// Default cadence of the client-side Ping keepalive. The scanner +/// sends `WsMessage::Ping` to the peer every `DEFAULT_PING_INTERVAL`; +/// the peer's mandatory Pong reply (RFC 6455 §5.5) arrives on the +/// same reader and resets the liveness watchdog. Must stay strictly +/// less than `DEFAULT_LIVENESS_TIMEOUT / 2` so that at least one +/// ping + pong round-trip fits inside every watchdog window even on +/// a marginal link (a single dropped pong should not be enough to +/// trip the watchdog). +pub const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(30); + +/// Compile-time assertion that the ping cadence leaves enough margin +/// inside the watchdog window. Encoded as a `const` evaluation so +/// any future tweak to either constant trips the build instead of +/// quietly drifting into a configuration where the watchdog could +/// fire between pings. +const _PING_INTERVAL_FITS_LIVENESS: () = assert!( + DEFAULT_PING_INTERVAL.as_millis() * 2 < DEFAULT_LIVENESS_TIMEOUT.as_millis(), + "DEFAULT_PING_INTERVAL must be < DEFAULT_LIVENESS_TIMEOUT / 2" +); + /// Default initial reconnect delay. Doubled on each consecutive /// failure up to `DEFAULT_RECONNECT_MAX`. pub const DEFAULT_RECONNECT_MIN: Duration = Duration::from_millis(500); @@ -173,6 +202,13 @@ pub struct ScannerWsConfig { /// Force-reconnect deadline for `ws.next()`. A silent half-open /// socket would otherwise wedge the scanner indefinitely. pub liveness_timeout: Duration, + /// Cadence of the client-side Ping keepalive. Each tick sends a + /// `WsMessage::Ping` frame; the peer's Pong reply (RFC 6455 + /// §5.5) flows back through `ws.next()` and resets the liveness + /// watchdog. Without keepalive a quiet Mainnet upstream produced + /// nothing on the reader for minutes at a time and the watchdog + /// reconnected every ~2 min unnecessarily. + pub ping_interval: Duration, } impl ScannerWsConfig { @@ -190,6 +226,7 @@ impl ScannerWsConfig { reconnect_min: DEFAULT_RECONNECT_MIN, reconnect_max: DEFAULT_RECONNECT_MAX, liveness_timeout: DEFAULT_LIVENESS_TIMEOUT, + ping_interval: DEFAULT_PING_INTERVAL, } } } @@ -271,65 +308,128 @@ pub async fn run_scanner_ws(config: ScannerWsConfig, tip_tx: mpsc::Sender, ) -> Result<(), WsError> { - let mut ws = connect_with_timeout(&config.url).await?; + let ws = connect_with_timeout(&config.url).await?; println!("scanner_ws: connected to {}", config.url); + let (mut sink, mut stream) = ws.split(); + let subscribe = serde_json::json!({ "action": "want", "data": ["blocks"] }).to_string(); - ws.send(WsMessage::Text(subscribe)) + sink.send(WsMessage::Text(subscribe)) .await .map_err(|e| WsError::Subscribe(e.to_string()))?; + // Client-side Ping keepalive. The ticker's first tick fires + // immediately (default tokio behaviour); that's fine — sending an + // initial ping right after subscribe gives us the fastest possible + // confirmation that the peer is live. `Burst` is the default + // missed-tick behaviour; if a tick is missed (e.g. busy reader) + // we explicitly opt into `Delay` below so we never send a flurry + // of pings to "catch up". The line below carries the required + // `scanner-polling-ok:` marker for the CI lint enforcing + // CONTRIBUTING.md § "No polling — events only". + let mut ping_ticker = tokio::time::interval(config.ping_interval); // scanner-polling-ok: client-side WS Ping keepalive cadence (RFC 6455 §5.5), not a chain-tip poll + ping_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + // Track the liveness deadline manually rather than wrapping each + // `stream.next()` in `tokio::time::timeout`, because `select!` + // drops the losing branch's future on every iteration. With a + // wrapper-based watchdog the timer would silently reset every + // time the ping arm fires, defeating the watchdog. The manual + // deadline is reset ONLY when an inbound frame arrives — exactly + // the invariant we want. + let mut deadline = tokio::time::Instant::now() + config.liveness_timeout; + loop { - let next = tokio::time::timeout(config.liveness_timeout, ws.next()).await; - let frame = match next { - Ok(Some(Ok(m))) => m, - Ok(Some(Err(e))) => return Err(WsError::Stream(e.to_string())), - Ok(None) => return Ok(()), // clean close - Err(_) => { + tokio::select! { + biased; + + // Liveness watchdog. Fires only if no inbound frame has + // arrived for `liveness_timeout`. A live peer answers our + // pings, so this should only fire on a genuinely dead + // socket. + _ = tokio::time::sleep_until(deadline) => { // scanner-polling-ok: liveness watchdog deadline, not a chain-tip poll return Err(WsError::Stream(format!( "no frame in {:?} (liveness watchdog)", config.liveness_timeout ))); } - }; - match frame { - WsMessage::Text(text) => { - for hash in parse_ws_frame(&text) { - if tip_tx.send(hash).await.is_err() { - // Receiver dropped → scanner_runtime is - // shutting down; drop any remaining hashes in - // this frame (anchor_on_current_tip on the - // next session would replay the latest tip - // anyway). Issue #84 review (round 4) MAJOR 3. - return Err(WsError::Stream("receiver dropped".into())); - } + // Outbound ping. RFC 6455 §5.5 requires the peer to reply + // with a Pong carrying the same payload; that Pong arrives + // on `stream.next()` and resets the deadline. + _ = ping_ticker.tick() => { + if let Err(e) = sink.send(WsMessage::Ping(PING_PAYLOAD.to_vec())).await { + return Err(WsError::Stream(format!("ping send failed: {}", e))); } } - WsMessage::Binary(_) => { - // Esplora WS does not send binary frames for the - // `blocks` subscription, but tungstenite delivers - // protocol frames here too. Ignore quietly. - } - WsMessage::Ping(_) | WsMessage::Pong(_) => { - // tungstenite handles ping/pong internally; nothing - // to do. + + // Inbound frame. Any frame — Text, Binary, Ping, Pong, + // Close — counts as evidence the socket is alive and + // resets the deadline. The frame variant then drives the + // per-shape handling below. + next = stream.next() => { + let frame = match next { + Some(Ok(m)) => m, + Some(Err(e)) => return Err(WsError::Stream(e.to_string())), + None => return Ok(()), // clean close + }; + deadline = tokio::time::Instant::now() + config.liveness_timeout; + + match frame { + WsMessage::Text(text) => { + for hash in parse_ws_frame(&text) { + if tip_tx.send(hash).await.is_err() { + // Receiver dropped → scanner_runtime is + // shutting down; drop any remaining hashes in + // this frame (anchor_on_current_tip on the + // next session would replay the latest tip + // anyway). Issue #84 review (round 4) MAJOR 3. + return Err(WsError::Stream("receiver dropped".into())); + } + } + } + WsMessage::Binary(_) => { + // Esplora WS does not send binary frames for the + // `blocks` subscription, but tungstenite delivers + // protocol frames here too. Ignore quietly. + } + WsMessage::Ping(_) | WsMessage::Pong(_) => { + // tungstenite auto-responds to inbound Pings; + // inbound Pongs are the response to OUR outbound + // keepalive pings. Either way, the deadline + // reset above is the whole job — nothing to do. + } + WsMessage::Close(_) => return Ok(()), + // The `Frame` variant of `tungstenite::Message` only + // surfaces under the `frame` cargo feature, which we do + // not enable. Keep the arm here as a defensive catch-all + // so a future tungstenite upgrade that flips the feature + // default does not break the build via a non-exhaustive + // match warning. + #[allow(unreachable_patterns)] + WsMessage::Frame(_) => {} + } } - WsMessage::Close(_) => return Ok(()), - // The `Frame` variant of `tungstenite::Message` only - // surfaces under the `frame` cargo feature, which we do - // not enable. Keep the arm here as a defensive catch-all - // so a future tungstenite upgrade that flips the feature - // default does not break the build via a non-exhaustive - // match warning. - #[allow(unreachable_patterns)] - WsMessage::Frame(_) => {} } } } diff --git a/node/src/scanner_ws_tests.rs b/node/src/scanner_ws_tests.rs index a363708e..662bc089 100644 --- a/node/src/scanner_ws_tests.rs +++ b/node/src/scanner_ws_tests.rs @@ -113,6 +113,10 @@ async fn run_scanner_ws_publishes_blocks_from_server() { reconnect_min: Duration::from_millis(10), reconnect_max: Duration::from_millis(50), liveness_timeout: Duration::from_secs(5), + // Pin the ping cadence well above the test budget — the + // happy-path coverage here is about block delivery, not the + // keepalive (separate test below). + ping_interval: Duration::from_secs(60), }; let handle = tokio::spawn(run_scanner_ws(config, tx)); @@ -170,6 +174,9 @@ async fn run_scanner_ws_reconnects_after_server_close() { reconnect_min: Duration::from_millis(10), reconnect_max: Duration::from_millis(50), liveness_timeout: Duration::from_secs(5), + // Same rationale as the previous test — keepalive is not + // under examination here. + ping_interval: Duration::from_secs(60), }; let handle = tokio::spawn(run_scanner_ws(config, tx)); @@ -246,6 +253,14 @@ async fn run_scanner_ws_force_reconnects_on_liveness_timeout() { reconnect_max: Duration::from_millis(50), // Aggressive watchdog so the test stays fast. liveness_timeout: Duration::from_millis(300), + // For this test the handler explicitly STOPS reading on + // server side after the first block — so an outbound ping + // gets no auto-Pong reply. Pin ping_interval well above the + // 300 ms watchdog so the watchdog fires for the documented + // "no inbound frame in window" reason rather than racing the + // ping-pong round-trip. The keepalive-specific behaviour is + // covered by the dedicated tests further down. + ping_interval: Duration::from_secs(60), }; let handle = tokio::spawn(run_scanner_ws(config, tx)); @@ -375,3 +390,186 @@ fn scanner_ws_config_from_env_uses_defaults_when_unset() { assert_eq!(DEFAULT_LIVENESS_TIMEOUT, Duration::from_secs(90)); assert!(DEFAULT_RECONNECT_MIN < DEFAULT_RECONNECT_MAX); } + +// ----------------------------------------------------------------------------- +// Ping keepalive — RFC 6455 §5.5 Pong-driven liveness +// ----------------------------------------------------------------------------- + +/// Sanity: the default ping cadence leaves room for at least one +/// full ping + pong round-trip inside the watchdog window with +/// margin. A drifted constant (e.g. someone bumping +/// `DEFAULT_PING_INTERVAL` to 60s without raising the watchdog) +/// would silently reintroduce the spurious-reconnect class this +/// keepalive is here to fix; the assertion turns that into a +/// build-time test failure. +#[test] +fn ping_interval_is_strictly_below_half_liveness_timeout() { + assert!( + DEFAULT_PING_INTERVAL * 2 < DEFAULT_LIVENESS_TIMEOUT, + "DEFAULT_PING_INTERVAL ({:?}) must be < DEFAULT_LIVENESS_TIMEOUT/2 ({:?})", + DEFAULT_PING_INTERVAL, + DEFAULT_LIVENESS_TIMEOUT, + ); + // And it should be non-trivially smaller than the watchdog + // itself; a value within one tick of the watchdog would race + // the watchdog under any timer jitter. + assert!(DEFAULT_PING_INTERVAL < DEFAULT_LIVENESS_TIMEOUT); +} + +/// Quiet-server test: the server completes the subscribe handshake, +/// sends no further block frames, but DOES keep draining its read +/// half. Tungstenite auto-pongs every inbound Ping, so each of our +/// keepalive pings produces an inbound Pong on the scanner's reader +/// and resets the liveness deadline. With keepalive working the +/// scanner stays connected through several watchdog windows back-to- +/// back; without keepalive (the pre-fix shape) the watchdog would +/// fire after one window and the test handler would see a second +/// `accept()`. +#[tokio::test] +async fn run_scanner_ws_pongs_keep_connection_alive_past_liveness_timeout() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let url = format!("ws://{}", addr); + + // Track how many connections the scanner opens. If the keepalive + // works, this stays at 1 for the entire test window. If the + // keepalive is broken, the watchdog fires and the scanner + // reconnects (count goes ≥ 2 well inside our budget). + let connection_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let cc_for_server = std::sync::Arc::clone(&connection_count); + + tokio::spawn(async move { + loop { + let (stream, _) = listener.accept().await.unwrap(); + cc_for_server.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + // Consume the subscribe frame and then keep draining the + // socket forever. tungstenite auto-queues a Pong reply + // for every inbound Ping while the stream is being polled, + // so the scanner sees a Pong on every keepalive tick. + // Crucially we send NO block frames — the only thing + // reaching the scanner's reader is the pong stream. + while let Some(msg) = ws.next().await { + if msg.is_err() { + break; + } + // Drop the message and continue. We never send any + // application-level frame. + } + } + }); + + let (tx, mut rx) = mpsc::channel::(8); + // Watchdog short enough that the test stays fast; ping interval + // strictly < watchdog/2 (matching the production invariant) so a + // pong fits comfortably inside every watchdog window. + let config = ScannerWsConfig { + url, + http_url: "http://127.0.0.1:1/api".to_string(), + reconnect_min: Duration::from_millis(10), + reconnect_max: Duration::from_millis(50), + liveness_timeout: Duration::from_millis(400), + ping_interval: Duration::from_millis(100), + }; + let handle = tokio::spawn(run_scanner_ws(config, tx)); + + // Wait for >> liveness_timeout. Without keepalive the scanner + // would fire the watchdog after ~400 ms and reconnect; with + // keepalive the connection_count stays at 1. + tokio::time::sleep(Duration::from_millis(1500)).await; + + // Block channel must be empty (server sent no block frames at + // all) — keepalive must not introduce phantom tip events. + assert!( + rx.try_recv().is_err(), + "scanner must not publish any BlockHash when the server only echoes pings" + ); + + let observed = connection_count.load(std::sync::atomic::Ordering::SeqCst); + assert_eq!( + observed, 1, + "expected exactly 1 connection (keepalive should prevent the watchdog reconnect); \ + saw {} connections, which means the watchdog fired", + observed, + ); + + handle.abort(); +} + +/// Failure-mode test: the server completes the subscribe handshake +/// and then stops reading entirely. Outbound pings pile up in the +/// server's TCP receive buffer; no Pong ever comes back; nothing +/// resets the deadline. The watchdog MUST fire after +/// `liveness_timeout` and the scanner MUST reconnect. Asserts the +/// brief's "no pong + no event in 90 s = connection genuinely dead" +/// semantic. +#[tokio::test] +async fn run_scanner_ws_watchdog_fires_when_pongs_are_dropped() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let url = format!("ws://{}", addr); + + let connection_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let cc_for_server = std::sync::Arc::clone(&connection_count); + + tokio::spawn(async move { + // Accept connections in a loop and hand each off to a + // dedicated task that parks forever — that way subsequent + // accepts can run while earlier connections are still being + // held open. The scanner reconnects after the watchdog, so + // the listener must keep accepting beyond the first + // connection for the test to observe count ≥ 2. + loop { + let (stream, _) = listener.accept().await.unwrap(); + cc_for_server.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + tokio::spawn(async move { + let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + // Read exactly the subscribe frame so the handshake + // completes, then stop touching the socket. The pings + // the scanner sends from now on are never observed + // and never auto-ponged; the scanner's deadline must + // elapse. + let _ = ws.next().await; + // Hold the socket open so the scanner's only path + // out is the watchdog. + std::future::pending::<()>().await; + }); + } + }); + + let (tx, mut rx) = mpsc::channel::(8); + let config = ScannerWsConfig { + url, + http_url: "http://127.0.0.1:1/api".to_string(), + reconnect_min: Duration::from_millis(10), + reconnect_max: Duration::from_millis(50), + liveness_timeout: Duration::from_millis(300), + // Ping cadence is well inside the watchdog window — but + // since the server never auto-pongs, the watchdog still + // fires. + ping_interval: Duration::from_millis(100), + }; + let handle = tokio::spawn(run_scanner_ws(config, tx)); + + // Allow the watchdog to fire at least once and the scanner to + // open a fresh connection. 1.5 s is enough for several watchdog + // windows back to back. + tokio::time::sleep(Duration::from_millis(1500)).await; + + let observed = connection_count.load(std::sync::atomic::Ordering::SeqCst); + assert!( + observed >= 2, + "expected ≥ 2 connections (watchdog must fire when pongs are dropped); \ + saw {} connections", + observed, + ); + + // No block frames ever flowed, so the scanner channel must be + // empty — keepalive doesn't conjure tips out of dropped pongs. + assert!( + rx.try_recv().is_err(), + "scanner must not publish any BlockHash when no block frames are sent", + ); + + handle.abort(); +} From d319a0fe770cf79e67ca297946c6bf5d54df4a6c Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 27 May 2026 00:42:05 +0200 Subject: [PATCH 2/4] fix(scanner_ws): make ping send cancel-safe + cover send-error reconnect path Three findings from independent review of the keepalive PR: 1. MAJOR - sink.send(Ping).await inside tokio::select! was not cancel-safe. SinkExt::send is not cancel-safe by contract: if the read arm wins a race against a half-completed send, the future is dropped and the sink can be left in a torn state mid-frame. Risk was low in practice (tiny payload, biased branch order) but a real correctness footgun. Switched to a dedicated writer task that owns the SplitSink and drains a 1-slot tokio::sync::mpsc::Receiver via feed + flush. The main loop's ping arm now does out_tx.send(msg) which IS cancel-safe (documented), and the actual wire-level send runs outside any select! boundary in the writer task. The 1-slot channel provides natural back-pressure against a stalled writer rather than letting an unbounded queue of pings grow against a slow peer. A scoped AbortOnDrop guard guarantees the writer task is torn down on every return path, so the underlying TCP socket is freed deterministically. 2. MINOR - no test for the send-error reconnect branch. Added run_scanner_ws_reconnects_when_ping_send_errors: the fake server accepts the WS, completes the subscribe handshake, then drops the stream. The scanner detects the closed socket via the close path (ping-send error or read error - both go through the cancel-safe plumbing) and reconnects well within liveness_timeout. The test pins liveness_timeout at 30 s and observes >= 2 connections inside a 2 s budget so the reconnect cannot be attributed to the watchdog. 3. NIT - _PING_INTERVAL_FITS_LIVENESS const assertion read as "interval * 2 < timeout" which is algebraically equivalent to but directionally inverse of the docstring. Flipped to "interval < timeout / 2" so the expression matches the prose. --- node/src/scanner_ws.rs | 96 ++++++++++++++++++++++++++++++---- node/src/scanner_ws_tests.rs | 99 ++++++++++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+), 10 deletions(-) diff --git a/node/src/scanner_ws.rs b/node/src/scanner_ws.rs index 4277a61d..21303af0 100644 --- a/node/src/scanner_ws.rs +++ b/node/src/scanner_ws.rs @@ -101,7 +101,7 @@ pub const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(30); /// quietly drifting into a configuration where the watchdog could /// fire between pings. const _PING_INTERVAL_FITS_LIVENESS: () = assert!( - DEFAULT_PING_INTERVAL.as_millis() * 2 < DEFAULT_LIVENESS_TIMEOUT.as_millis(), + DEFAULT_PING_INTERVAL.as_millis() < DEFAULT_LIVENESS_TIMEOUT.as_millis() / 2, "DEFAULT_PING_INTERVAL must be < DEFAULT_LIVENESS_TIMEOUT / 2" ); @@ -316,14 +316,24 @@ const PING_PAYLOAD: &[u8] = b"zkcoins-scanner-keepalive"; /// Single connect → subscribe → drain cycle. Returns Ok on a clean /// close, Err on any failure. Caller schedules the reconnect. /// -/// The reader and the ping ticker run inside a single `tokio::select!` -/// on a split stream: the reader half (`SplitStream`) feeds the frame -/// loop, the writer half (`SplitSink`) carries the periodic -/// `WsMessage::Ping`. Splitting (vs. two tasks) keeps error -/// propagation linear and avoids a shutdown handshake between halves; -/// `select!` (vs. polling the ticker between reads) preserves the -/// invariant that the liveness deadline is reset ONLY by inbound -/// frames, not by our own send activity. +/// Architecture: the WS stream is split into a reader (`SplitStream`) +/// and a writer (`SplitSink`). The writer half is moved into a +/// dedicated `tokio::spawn`ed writer task that drains a 1-slot +/// `tokio::sync::mpsc::Receiver` and runs `feed` + `flush` +/// against the sink. The main loop's `tokio::select!` polls only the +/// reader, the liveness watchdog, and an mpsc `out_tx.send().await` +/// driven by the ping ticker. +/// +/// Why the writer-task split (and not `sink.send(...).await` inline +/// in the select): `SinkExt::send` is NOT cancel-safe — if the read +/// arm wins a race against a half-completed send, the send-future is +/// dropped and the sink can be left in a torn state mid-frame. By +/// contrast `tokio::sync::mpsc::Sender::send().await` IS cancel-safe, +/// and the writer task awaits the actual wire-level send to +/// completion outside any `select!` boundary, so the sink is never +/// cancelled mid-poll. The `select!`-on-ticker invariant that the +/// liveness deadline is reset ONLY by inbound frames (never by our +/// own send activity) is preserved exactly as before. async fn connect_and_drain( config: &ScannerWsConfig, tip_tx: &mpsc::Sender, @@ -338,6 +348,57 @@ async fn connect_and_drain( .await .map_err(|e| WsError::Subscribe(e.to_string()))?; + // Outbound writer task. Owns `sink` outright and drives every + // outbound frame to completion via `feed` + `flush` — the + // `feed`/`flush` split keeps the partial-write window the + // narrowest the API allows. The writer's body is plain + // `loop { rx.recv().await ... }`, with no `select!` around the + // send, so the send-future is never cancelled mid-poll and the + // sink can never be left in a torn state. + // + // The main loop talks to this task via `tokio::sync::mpsc::Sender`, + // whose `send().await` IS cancel-safe (documented: dropping the + // future before completion is sound — the message is never + // delivered, but the channel and sender remain consistent). This + // is the cancel-safety argument for the ping arm in the `select!` + // below: instead of `sink.send(Ping).await` (NOT cancel-safe) we + // do `out_tx.send(Ping).await`, and the writer task takes care of + // the actual wire-level send outside any `select!` boundary. + // + // The channel is bounded at 1 so a stalled writer applies + // immediate back-pressure to the main loop (the second ping tick + // would block) — far preferable to growing an unbounded queue of + // pings against a peer that cannot drain them. + let (out_tx, mut out_rx) = mpsc::channel::(1); + let writer = tokio::spawn(async move { + while let Some(msg) = out_rx.recv().await { + // `feed` queues the frame into the sink's internal + // buffer; `flush` drives it onto the wire. Splitting (vs. + // `send`) bounds the partial-write window and makes the + // two halves explicit. On error we surface it to the main + // loop by dropping `out_tx` from the writer side (closing + // the channel from the producer's perspective is achieved + // by the writer task exiting); the main loop's next + // `out_tx.send` will then fail and trigger reconnect. + sink.feed(msg).await?; + sink.flush().await?; + } + Ok::<(), tokio_tungstenite::tungstenite::Error>(()) + }); + // Always abort the writer when this function returns, regardless + // of how we exit. Without this, a returning main-loop iteration + // could leave the writer task parked in `out_rx.recv().await` and + // leak the `sink` (and thus the underlying TCP socket) until the + // tokio runtime tears down. `AbortOnDrop` makes that cleanup + // deterministic and exception-safe. + struct AbortOnDrop(tokio::task::JoinHandle>); + impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } + } + let _writer_guard = AbortOnDrop(writer); + // Client-side Ping keepalive. The ticker's first tick fires // immediately (default tokio behaviour); that's fine — sending an // initial ping right after subscribe gives us the fastest possible @@ -377,8 +438,18 @@ async fn connect_and_drain( // Outbound ping. RFC 6455 §5.5 requires the peer to reply // with a Pong carrying the same payload; that Pong arrives // on `stream.next()` and resets the deadline. + // + // Cancel-safety: `tokio::sync::mpsc::Sender::send().await` + // is documented as cancel-safe, so if the read arm wins + // this race the half-completed send-future can be dropped + // without corrupting the channel or the underlying sink. + // The actual wire-level write happens inside the dedicated + // writer task above, never inside this `select!`. A send + // error here means the writer task has exited (e.g. the + // peer closed mid-write) — surface as a stream error so + // the reconnect loop kicks in. _ = ping_ticker.tick() => { - if let Err(e) = sink.send(WsMessage::Ping(PING_PAYLOAD.to_vec())).await { + if let Err(e) = out_tx.send(WsMessage::Ping(PING_PAYLOAD.to_vec())).await { return Err(WsError::Stream(format!("ping send failed: {}", e))); } } @@ -387,6 +458,11 @@ async fn connect_and_drain( // Close — counts as evidence the socket is alive and // resets the deadline. The frame variant then drives the // per-shape handling below. + // + // Cancel-safety: `StreamExt::next` is documented as + // cancel-safe (futures-util 0.3), so dropping this arm's + // future when another arm wins is sound — no frame is + // lost. next = stream.next() => { let frame = match next { Some(Ok(m)) => m, diff --git a/node/src/scanner_ws_tests.rs b/node/src/scanner_ws_tests.rs index 662bc089..dcea1915 100644 --- a/node/src/scanner_ws_tests.rs +++ b/node/src/scanner_ws_tests.rs @@ -573,3 +573,102 @@ async fn run_scanner_ws_watchdog_fires_when_pongs_are_dropped() { handle.abort(); } + +/// Send-error reconnect: the server completes the subscribe handshake +/// and then drops the TCP socket abruptly (no clean WS close frame, +/// no graceful FIN handshake — just `drop(ws)` which closes the +/// underlying TcpStream). The scanner's next ping-ticker tick attempts +/// to write a Ping frame to the now-closed socket; the writer task's +/// `sink.feed`/`flush` returns `Err` (broken pipe / connection reset) +/// and the main loop surfaces that as `WsError::Stream("ping send +/// failed: ...")`, driving a reconnect via the normal backoff loop. +/// +/// Race-note: in practice the reader arm may also observe the close +/// (as `Some(Err(_))` or `None`) on roughly the same scheduling tick +/// as the ping arm. Both paths produce the SAME observable behaviour +/// — fast reconnect well inside `liveness_timeout` — and both go +/// through the cancel-safe writer-task plumbing introduced for the +/// ping-send branch, so either winning the race exercises the +/// cancel-safety guarantee. The assertion below pins the observable +/// invariant: reconnect happens MUCH faster than the watchdog window, +/// which is only achievable if a non-watchdog reconnect path fired. +#[tokio::test] +async fn run_scanner_ws_reconnects_when_ping_send_errors() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let url = format!("ws://{}", addr); + + let connection_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let cc_for_server = std::sync::Arc::clone(&connection_count); + + tokio::spawn(async move { + // Accept connections in a loop; per-connection handler drops + // the WS as soon as the subscribe frame arrives. Subsequent + // accepts continue to fire so the scanner's reconnect attempt + // can land cleanly. + loop { + let (stream, _) = listener.accept().await.unwrap(); + cc_for_server.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + tokio::spawn(async move { + let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + // Wait for the subscribe frame so the handshake is + // observably complete (the scanner has transitioned + // out of connect/subscribe and into the steady-state + // select loop) before we tear the socket down. + let _ = ws.next().await; + // Drop the WS — this drops the underlying TcpStream, + // which closes the connection from the server side. + // The scanner's next outbound write (ping-tick) sees + // a broken pipe; the reader sees an EOF/error around + // the same time. Either way the scanner exits the + // current session via a non-watchdog path and the + // outer reconnect loop opens a fresh TCP connection + // (which lands here, incrementing the counter). + drop(ws); + }); + } + }); + + let (tx, mut rx) = mpsc::channel::(8); + // Liveness watchdog is set to a value LARGER than the test budget + // below so that a count ≥ 2 within the budget cannot possibly be + // attributed to a watchdog firing — the reconnect MUST have come + // from the close-detection path (ping-send error or read error). + // Ping cadence is tight so the first ping tick fires within a few + // ms of the subscribe completing, giving the send-error path the + // best chance to be the path that actually drives the reconnect. + let config = ScannerWsConfig { + url, + http_url: "http://127.0.0.1:1/api".to_string(), + reconnect_min: Duration::from_millis(10), + reconnect_max: Duration::from_millis(50), + liveness_timeout: Duration::from_secs(30), + ping_interval: Duration::from_millis(50), + }; + let handle = tokio::spawn(run_scanner_ws(config, tx)); + + // Budget for observing the reconnect. Must be ≫ ping_interval + + // reconnect_max but ≪ liveness_timeout, so any observed + // reconnect MUST be driven by the close-detection path, not the + // watchdog. 2 s comfortably satisfies both. + tokio::time::sleep(Duration::from_secs(2)).await; + + let observed = connection_count.load(std::sync::atomic::Ordering::SeqCst); + assert!( + observed >= 2, + "expected ≥ 2 connections (server dropped socket — scanner must \ + reconnect via the ping-send-error path, well inside the 30 s \ + liveness watchdog window); saw {} connections", + observed, + ); + + // The server never sent any block frames, only the implicit + // subscribe-then-drop. The channel must therefore be empty — + // failure-path reconnects must not inject phantom tips. + assert!( + rx.try_recv().is_err(), + "scanner must not publish any BlockHash when no block frames are sent", + ); + + handle.abort(); +} From df6dd5b3b7af8f1bbc8b7d8dfbc3e8e24c0ab4b9 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 27 May 2026 01:34:56 +0200 Subject: [PATCH 3/4] fix(scanner_ws): timeout writer send + reword race-note + extract queue-capacity const --- Cargo.lock | 1 + node/Cargo.toml | 7 ++ node/src/scanner_ws.rs | 56 +++++++++++-- node/src/scanner_ws_tests.rs | 150 ++++++++++++++++++++++++++++++++++- 4 files changed, 204 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cda8f445..34ac69c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1962,6 +1962,7 @@ dependencies = [ "hex", "http-body-util", "lazy_static", + "libc", "rand 0.8.6", "reqwest 0.12.28", "serde", diff --git a/node/Cargo.toml b/node/Cargo.toml index 63edf322..d932271c 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -90,6 +90,13 @@ rand = "0.8" # the directory even when the test panics, so no test leaves a # leaked /tmp/zkcoins-* tree behind. tempfile = "3" +# Used by `scanner_ws_tests` to shrink the OS TCP receive buffer on +# the test server socket so a kernel-level backpressure wedge fires +# inside a millisecond-scale test budget instead of needing minutes +# of pings to fill a default-size buffer. Pure-Unix syscall surface +# (`setsockopt`), no Windows path because tests already gate on +# Unix-only fixtures. +libc = "0.2" [features] # All non-MVP features are off by default. When a feature is not enabled, the diff --git a/node/src/scanner_ws.rs b/node/src/scanner_ws.rs index 21303af0..0d3b10dc 100644 --- a/node/src/scanner_ws.rs +++ b/node/src/scanner_ws.rs @@ -95,6 +95,24 @@ pub const DEFAULT_LIVENESS_TIMEOUT: Duration = Duration::from_secs(90); /// trip the watchdog). pub const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(30); +/// Capacity of the bounded `mpsc` channel feeding the dedicated +/// writer task in `connect_and_drain`. Fixed at `1` on purpose: +/// +/// - Strict back-pressure. A second ping cannot queue until the +/// first one has fully flushed onto the wire, so the producer +/// side (the `select!` loop) observes a stalled writer +/// immediately rather than absorbing it into a growing queue. +/// - Latest-ping-wins is acceptable because we never have anything +/// useful to "catch up" on — a stale ping in the queue would buy +/// us nothing the next live ping wouldn't. +/// - No unbounded queue. If the peer accepts TCP but never reads +/// (a stalled writer), the producer's `out_tx.send(...).await` +/// is the natural choke-point; combined with the +/// `liveness_timeout`-bounded `tokio::time::timeout` wrapper +/// around that send, a wedged writer becomes a reconnect rather +/// than a deadlocked task. +pub const WRITER_QUEUE_CAPACITY: usize = 1; + /// Compile-time assertion that the ping cadence leaves enough margin /// inside the watchdog window. Encoded as a `const` evaluation so /// any future tweak to either constant trips the build instead of @@ -365,11 +383,13 @@ async fn connect_and_drain( // do `out_tx.send(Ping).await`, and the writer task takes care of // the actual wire-level send outside any `select!` boundary. // - // The channel is bounded at 1 so a stalled writer applies - // immediate back-pressure to the main loop (the second ping tick - // would block) — far preferable to growing an unbounded queue of - // pings against a peer that cannot drain them. - let (out_tx, mut out_rx) = mpsc::channel::(1); + // The channel is bounded at `WRITER_QUEUE_CAPACITY` (= 1) so a + // stalled writer applies immediate back-pressure to the main loop + // (the second ping tick would block) — far preferable to growing + // an unbounded queue of pings against a peer that cannot drain + // them. The constant lives at the top of the file alongside the + // other tunables and carries the full rationale. + let (out_tx, mut out_rx) = mpsc::channel::(WRITER_QUEUE_CAPACITY); let writer = tokio::spawn(async move { while let Some(msg) = out_rx.recv().await { // `feed` queues the frame into the sink's internal @@ -448,9 +468,31 @@ async fn connect_and_drain( // error here means the writer task has exited (e.g. the // peer closed mid-write) — surface as a stream error so // the reconnect loop kicks in. + // + // Backpressure-deadlock guard: if the peer accepts TCP but + // never reads, the writer task wedges in `sink.flush()` + // forever. The 1-slot `out_tx` then fills with the first + // unflushed ping, and a subsequent `out_tx.send(...).await` + // would block this arm indefinitely — preventing the + // `select!` from advancing to the watchdog arm too. We wrap + // the send in `tokio::time::timeout(liveness_timeout, ...)` + // so a wedged writer surfaces as a reconnect-triggering + // error within the same upper bound the watchdog uses for + // "this connection is dead", keeping the two failure modes + // semantically aligned. _ = ping_ticker.tick() => { - if let Err(e) = out_tx.send(WsMessage::Ping(PING_PAYLOAD.to_vec())).await { - return Err(WsError::Stream(format!("ping send failed: {}", e))); + let send_fut = out_tx.send(WsMessage::Ping(PING_PAYLOAD.to_vec())); + match tokio::time::timeout(config.liveness_timeout, send_fut).await { + Ok(Ok(())) => {} + Ok(Err(e)) => { + return Err(WsError::Stream(format!("ping send failed: {}", e))); + } + Err(_) => { + return Err(WsError::Stream(format!( + "ping send stalled for {:?} (writer wedged, peer not reading)", + config.liveness_timeout + ))); + } } } diff --git a/node/src/scanner_ws_tests.rs b/node/src/scanner_ws_tests.rs index dcea1915..9d6793a8 100644 --- a/node/src/scanner_ws_tests.rs +++ b/node/src/scanner_ws_tests.rs @@ -656,9 +656,9 @@ async fn run_scanner_ws_reconnects_when_ping_send_errors() { let observed = connection_count.load(std::sync::atomic::Ordering::SeqCst); assert!( observed >= 2, - "expected ≥ 2 connections (server dropped socket — scanner must \ - reconnect via the ping-send-error path, well inside the 30 s \ - liveness watchdog window); saw {} connections", + "scanner must reconnect via the close-detection path \ + (ping-send-error OR read-error) — observed only {} connections \ + in the budget window, well inside the 30 s liveness watchdog", observed, ); @@ -672,3 +672,147 @@ async fn run_scanner_ws_reconnects_when_ping_send_errors() { handle.abort(); } + +/// Writer-send-timeout reconnect: exercises the +/// `tokio::time::timeout(liveness_timeout, out_tx.send(...))` guard +/// added around the ping arm's `out_tx.send` to defuse the +/// backpressure-deadlock window. The deadlock shape it defuses: +/// +/// 1. The peer accepts the TCP socket and completes the WS upgrade, +/// but then never reads from the socket again. The OS-level TCP +/// send window on the scanner side fills up. +/// 2. The dedicated writer task wedges inside `sink.flush().await` +/// waiting for the kernel to drain that buffer. +/// 3. The 1-slot `out_tx` channel fills with the first un-flushed +/// ping (writer holds it, can't progress). +/// 4. The next `ping_ticker.tick()` fires; its body calls +/// `out_tx.send(...).await`, which now blocks because the queue +/// is full and the writer can't drain it. +/// 5. WITHOUT the timeout wrap, this `.await` sits forever — the +/// enclosing `select!` has already exited (the ping arm won), +/// so the watchdog arm can't fire to break the deadlock. +/// 6. WITH the timeout wrap, the wedge surfaces as a stream error +/// inside `liveness_timeout`, the outer reconnect loop kicks in, +/// and a fresh TCP connection lands at the server. +/// +/// Setup: we shrink the server-side `SO_RCVBUF` on the listener to +/// the OS-minimum BEFORE `accept()`, so each accepted socket inherits +/// a tiny receive buffer (a few KB). Combined with a server that +/// reads exactly one frame (the subscribe) and then parks on +/// `pending()`, the client's kernel send buffer + the server's +/// receive buffer fill up after a handful of pings, and `flush()` +/// wedges inside the test budget. +/// +/// `liveness_timeout = 300 ms` + `ping_interval = 50 ms` match the +/// reviewer's spec: the wedge bites well before either the watchdog +/// or the test-budget timeout, so observing ≥ 2 server-side accepts +/// inside ~3 s is positive evidence that a non-deadlock reconnect +/// path drove the reconnect. +#[cfg(unix)] +#[tokio::test] +async fn run_scanner_ws_reconnects_when_writer_send_times_out() { + use std::os::unix::io::AsRawFd; + + // Bind a std listener first so we can `setsockopt(SO_RCVBUF)` + // BEFORE handing it to tokio. Accepted sockets inherit the small + // receive buffer, which is what fills the client's send window + // and wedges `sink.flush()`. The exact size is platform-clamped: + // Linux rounds up to its minimum (typically ~2 KB); macOS honors + // it closer to the literal value. Either way the result is small + // enough that a handful of WS Ping frames + WS framing overhead + // saturate it inside the test budget. + let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + std_listener.set_nonblocking(true).unwrap(); + { + let fd = std_listener.as_raw_fd(); + let bufsize: libc::c_int = 1024; + // SAFETY: `fd` is a live socket file descriptor owned by + // `std_listener` for the duration of this call; the option + // name and value pointer are well-formed per setsockopt(2); + // the return value is checked below. + let rc = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_RCVBUF, + &bufsize as *const _ as *const libc::c_void, + std::mem::size_of_val(&bufsize) as libc::socklen_t, + ) + }; + assert_eq!(rc, 0, "setsockopt SO_RCVBUF must succeed"); + } + let listener = TcpListener::from_std(std_listener).unwrap(); + let addr = listener.local_addr().unwrap(); + let url = format!("ws://{}", addr); + + let connection_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let cc_for_server = std::sync::Arc::clone(&connection_count); + + tokio::spawn(async move { + // Per-connection handler: complete the WS handshake, read the + // subscribe frame so the scanner observably transitions into + // its steady-state select loop, then PARK without reading + // anything else. The server-side socket's receive buffer + // (shrunk via SO_RCVBUF on the listener above) fills up after + // a handful of pings; the client's `sink.flush()` then wedges + // on TCP backpressure, the 1-slot `out_tx` fills, and the + // next `out_tx.send(...).await` in the ping arm hits the new + // `liveness_timeout` wrap. + loop { + let (stream, _) = listener.accept().await.unwrap(); + cc_for_server.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + tokio::spawn(async move { + let mut ws = match tokio_tungstenite::accept_async(stream).await { + Ok(ws) => ws, + Err(_) => return, + }; + // Read the subscribe frame so the handshake is fully + // observable as complete; subsequent reads are + // intentionally omitted so the receive buffer fills. + let _ = ws.next().await; + // Hold the socket open forever — do NOT read anything + // else. The scanner's pings will pile up in the + // (small) kernel buffer and wedge the writer. + std::future::pending::<()>().await; + }); + } + }); + + let (tx, mut rx) = mpsc::channel::(8); + // `liveness_timeout = 300 ms`: short enough that the timeout-wrap + // around `out_tx.send` fires inside the test budget, long enough + // that we don't race a slow CI scheduler. + // `ping_interval = 50 ms`: fast enough to fill the 1-slot writer + // queue + saturate the small receive buffer well before the + // first watchdog window elapses. + let config = ScannerWsConfig { + url, + http_url: "http://127.0.0.1:1/api".to_string(), + reconnect_min: Duration::from_millis(10), + reconnect_max: Duration::from_millis(50), + liveness_timeout: Duration::from_millis(300), + ping_interval: Duration::from_millis(50), + }; + let handle = tokio::spawn(run_scanner_ws(config, tx)); + + // Budget: ≫ liveness_timeout + reconnect_max so at least one + // reconnect cycle is observable, ≪ any realistic CI flake budget. + tokio::time::sleep(Duration::from_secs(3)).await; + + let observed = connection_count.load(std::sync::atomic::Ordering::SeqCst); + assert!( + observed >= 2, + "scanner must reconnect when the writer wedges — observed only \ + {} connections in the budget window (timeout-wrap path OR \ + watchdog path, both valid reconnect drivers)", + observed, + ); + + // No block frames were ever sent, so the channel must be empty. + assert!( + rx.try_recv().is_err(), + "scanner must not publish any BlockHash when no block frames are sent", + ); + + handle.abort(); +} From 627fc0a0e62535e07d4f4834a03646afbdf9413e Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 27 May 2026 02:13:40 +0200 Subject: [PATCH 4/4] fix(scanner_ws): drop libc dev-dep for socket2 + tighten timeout-wrap regression test --- Cargo.lock | 2 +- node/Cargo.toml | 10 ++- node/src/scanner_ws.rs | 2 +- node/src/scanner_ws_tests.rs | 139 +++++++++++++++++------------------ 4 files changed, 74 insertions(+), 79 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 34ac69c3..79c11984 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1962,13 +1962,13 @@ dependencies = [ "hex", "http-body-util", "lazy_static", - "libc", "rand 0.8.6", "reqwest 0.12.28", "serde", "serde_json", "sha2", "shared", + "socket2 0.5.10", "sqlx", "tempfile", "testcontainers", diff --git a/node/Cargo.toml b/node/Cargo.toml index d932271c..f934c0b0 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -93,10 +93,12 @@ tempfile = "3" # Used by `scanner_ws_tests` to shrink the OS TCP receive buffer on # the test server socket so a kernel-level backpressure wedge fires # inside a millisecond-scale test budget instead of needing minutes -# of pings to fill a default-size buffer. Pure-Unix syscall surface -# (`setsockopt`), no Windows path because tests already gate on -# Unix-only fixtures. -libc = "0.2" +# of pings to fill a default-size buffer. `socket2::SockRef` +# provides a safe, cross-platform wrapper around the underlying +# `setsockopt(SO_RCVBUF)` syscall — no unsafe block needed in the +# test fixture. Already a transitive dep via tokio/h2/reqwest, but +# must be declared here to be reachable from test code. +socket2 = "0.5" [features] # All non-MVP features are off by default. When a feature is not enabled, the diff --git a/node/src/scanner_ws.rs b/node/src/scanner_ws.rs index 0d3b10dc..6e86083f 100644 --- a/node/src/scanner_ws.rs +++ b/node/src/scanner_ws.rs @@ -111,7 +111,7 @@ pub const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(30); /// `liveness_timeout`-bounded `tokio::time::timeout` wrapper /// around that send, a wedged writer becomes a reconnect rather /// than a deadlocked task. -pub const WRITER_QUEUE_CAPACITY: usize = 1; +const WRITER_QUEUE_CAPACITY: usize = 1; /// Compile-time assertion that the ping cadence leaves enough margin /// inside the watchdog window. Encoded as a `const` evaluation so diff --git a/node/src/scanner_ws_tests.rs b/node/src/scanner_ws_tests.rs index 9d6793a8..d766a378 100644 --- a/node/src/scanner_ws_tests.rs +++ b/node/src/scanner_ws_tests.rs @@ -673,75 +673,61 @@ async fn run_scanner_ws_reconnects_when_ping_send_errors() { handle.abort(); } -/// Writer-send-timeout reconnect: exercises the -/// `tokio::time::timeout(liveness_timeout, out_tx.send(...))` guard -/// added around the ping arm's `out_tx.send` to defuse the -/// backpressure-deadlock window. The deadlock shape it defuses: +/// Peer-stops-reading reconnect: covers the failure mode where the +/// peer accepts the TCP socket and completes the WS upgrade but +/// then stops reading entirely. The +/// `tokio::time::timeout(liveness_timeout, out_tx.send(...))` wrap +/// in `scanner_ws.rs` is in place to defuse the deadlock shape: /// -/// 1. The peer accepts the TCP socket and completes the WS upgrade, -/// but then never reads from the socket again. The OS-level TCP -/// send window on the scanner side fills up. +/// 1. Peer accepts and never reads again; OS-level TCP send window +/// on the scanner side fills up. /// 2. The dedicated writer task wedges inside `sink.flush().await` /// waiting for the kernel to drain that buffer. /// 3. The 1-slot `out_tx` channel fills with the first un-flushed /// ping (writer holds it, can't progress). -/// 4. The next `ping_ticker.tick()` fires; its body calls -/// `out_tx.send(...).await`, which now blocks because the queue -/// is full and the writer can't drain it. +/// 4. The next `ping_ticker.tick()` body calls +/// `out_tx.send(...).await`, which now blocks (queue full). /// 5. WITHOUT the timeout wrap, this `.await` sits forever — the /// enclosing `select!` has already exited (the ping arm won), /// so the watchdog arm can't fire to break the deadlock. /// 6. WITH the timeout wrap, the wedge surfaces as a stream error -/// inside `liveness_timeout`, the outer reconnect loop kicks in, -/// and a fresh TCP connection lands at the server. +/// inside `liveness_timeout`, the outer reconnect loop kicks +/// in, and a fresh TCP connection lands at the server. /// -/// Setup: we shrink the server-side `SO_RCVBUF` on the listener to -/// the OS-minimum BEFORE `accept()`, so each accepted socket inherits -/// a tiny receive buffer (a few KB). Combined with a server that -/// reads exactly one frame (the subscribe) and then parks on -/// `pending()`, the client's kernel send buffer + the server's -/// receive buffer fill up after a handful of pings, and `flush()` -/// wedges inside the test budget. +/// Setup: `SO_RCVBUF = 1 KiB` on each ACCEPTED socket (NOT on the +/// listener — macOS does not propagate the listener-level recv +/// buffer to accepted children). `socket2::SockRef` provides a +/// safe wrapper around `setsockopt`, no unsafe block needed. /// -/// `liveness_timeout = 300 ms` + `ping_interval = 50 ms` match the -/// reviewer's spec: the wedge bites well before either the watchdog -/// or the test-budget timeout, so observing ≥ 2 server-side accepts -/// inside ~3 s is positive evidence that a non-deadlock reconnect -/// path drove the reconnect. +/// Honesty note (reviewer round 3): isolating the wrap path from +/// the watchdog path in this fixture is not achievable in a few- +/// second budget on the m3-ultra CI runner pool (macOS). The macOS +/// TCP loopback implementation buffers up to ~150 KiB on the +/// sender side and dynamically drains/grows in ways that prevent +/// the scanner's writer-task `flush()` from blocking reliably +/// inside a 30 s window at any practical ping cadence. As a +/// result, the path that drives the reconnect observed below is +/// the liveness watchdog (`liveness_timeout = 300 ms` here), not +/// the wrap. The wrap remains production-correct code — on links +/// where the kernel actually wedges the writer (smaller buffers, +/// non-loopback peer, paths with real RTT) the wrap is the path +/// that fires — but a "wrap-only" isolation test would need a +/// custom Sink fixture that sidesteps TCP, which is out of scope +/// for this PR. The assertion pins the OBSERVABLE invariant +/// (reconnect within the budget) rather than the specific path, +/// matching the production guarantee. #[cfg(unix)] #[tokio::test] async fn run_scanner_ws_reconnects_when_writer_send_times_out() { - use std::os::unix::io::AsRawFd; - - // Bind a std listener first so we can `setsockopt(SO_RCVBUF)` - // BEFORE handing it to tokio. Accepted sockets inherit the small - // receive buffer, which is what fills the client's send window - // and wedges `sink.flush()`. The exact size is platform-clamped: - // Linux rounds up to its minimum (typically ~2 KB); macOS honors - // it closer to the literal value. Either way the result is small - // enough that a handful of WS Ping frames + WS framing overhead - // saturate it inside the test budget. - let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - std_listener.set_nonblocking(true).unwrap(); - { - let fd = std_listener.as_raw_fd(); - let bufsize: libc::c_int = 1024; - // SAFETY: `fd` is a live socket file descriptor owned by - // `std_listener` for the duration of this call; the option - // name and value pointer are well-formed per setsockopt(2); - // the return value is checked below. - let rc = unsafe { - libc::setsockopt( - fd, - libc::SOL_SOCKET, - libc::SO_RCVBUF, - &bufsize as *const _ as *const libc::c_void, - std::mem::size_of_val(&bufsize) as libc::socklen_t, - ) - }; - assert_eq!(rc, 0, "setsockopt SO_RCVBUF must succeed"); - } - let listener = TcpListener::from_std(std_listener).unwrap(); + // Shrink `SO_RCVBUF` on each ACCEPTED socket so the kernel + // advertises a tiny TCP receive window. `socket2::SockRef` + // borrows the tokio TcpStream's socket and exposes + // `set_recv_buffer_size`, a safe cross-platform wrapper around + // the underlying `setsockopt(SO_RCVBUF)` syscall — no unsafe + // block needed in the fixture. The exact size is platform- + // clamped: Linux rounds up to its minimum (typically ~2 KiB); + // macOS honors values down to a few hundred bytes. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let url = format!("ws://{}", addr); @@ -752,15 +738,16 @@ async fn run_scanner_ws_reconnects_when_writer_send_times_out() { // Per-connection handler: complete the WS handshake, read the // subscribe frame so the scanner observably transitions into // its steady-state select loop, then PARK without reading - // anything else. The server-side socket's receive buffer - // (shrunk via SO_RCVBUF on the listener above) fills up after - // a handful of pings; the client's `sink.flush()` then wedges - // on TCP backpressure, the 1-slot `out_tx` fills, and the - // next `out_tx.send(...).await` in the ping arm hits the new - // `liveness_timeout` wrap. + // anything else. loop { let (stream, _) = listener.accept().await.unwrap(); cc_for_server.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + // Shrink the accepted socket's recv buffer BEFORE the WS + // upgrade handshake completes, so the tiny window is in + // effect for every byte the client sends after subscribe. + socket2::SockRef::from(&stream) + .set_recv_buffer_size(1024) + .expect("set_recv_buffer_size must succeed on a freshly accepted socket"); tokio::spawn(async move { let mut ws = match tokio_tungstenite::accept_async(stream).await { Ok(ws) => ws, @@ -768,23 +755,24 @@ async fn run_scanner_ws_reconnects_when_writer_send_times_out() { }; // Read the subscribe frame so the handshake is fully // observable as complete; subsequent reads are - // intentionally omitted so the receive buffer fills. + // intentionally omitted. let _ = ws.next().await; // Hold the socket open forever — do NOT read anything - // else. The scanner's pings will pile up in the - // (small) kernel buffer and wedge the writer. + // else. The scanner's pings pile up in the (tiny) + // advertised receive window; the inbound side stays + // silent so the watchdog deadline is never reset. std::future::pending::<()>().await; }); } }); let (tx, mut rx) = mpsc::channel::(8); - // `liveness_timeout = 300 ms`: short enough that the timeout-wrap - // around `out_tx.send` fires inside the test budget, long enough - // that we don't race a slow CI scheduler. - // `ping_interval = 50 ms`: fast enough to fill the 1-slot writer - // queue + saturate the small receive buffer well before the - // first watchdog window elapses. + // `liveness_timeout = 300 ms`: the test's reconnect path (see the + // honesty note in the doc-comment above). `ping_interval = 50 ms`: + // fast enough that several ping ticks land inside one watchdog + // window so any "ping itself accidentally resets the deadline" + // regression would surface as a hung connection rather than a + // false positive. let config = ScannerWsConfig { url, http_url: "http://127.0.0.1:1/api".to_string(), @@ -802,9 +790,14 @@ async fn run_scanner_ws_reconnects_when_writer_send_times_out() { let observed = connection_count.load(std::sync::atomic::Ordering::SeqCst); assert!( observed >= 2, - "scanner must reconnect when the writer wedges — observed only \ - {} connections in the budget window (timeout-wrap path OR \ - watchdog path, both valid reconnect drivers)", + "scanner must reconnect when the peer accepts the WS upgrade \ + then stops reading entirely; observed only {} connections in \ + the 3 s budget. On the m3-ultra CI runner pool (macOS) the \ + path that drives this reconnect is the watchdog at \ + liveness_timeout (300 ms); on links where the writer wedge \ + actually develops, the timeout-wrap fires first. Both are \ + production-correct reconnect drivers — the assertion only \ + pins the observable invariant", observed, );