From f3ab5ef2a8deb6cd91a07d07c0c77d461be1332d Mon Sep 17 00:00:00 2001 From: brunota20 Date: Thu, 18 Jun 2026 11:36:01 -0300 Subject: [PATCH] feat(event-loop): WS reconnect with exponential backoff per stream (COW-1071) Replaces the previous "bail on WS drop" semantic (flagged as the "0.3 fix" in the source) with per-stream reconnect-aware tasks. Each chain's block subscription and each (module, chain) log subscription gets a dedicated task that: 1. Opens the subscription via `ProviderPool`. 2. Pumps items to an mpsc channel until the underlying stream yields `None` (WS drop) or `Err` (transport-level error). 3. Logs the drop + sleeps for `restart_policy::backoff_for(attempt)` (1s -> 2s -> 4s -> ... cap 5 min, reusing the COW-1033 policy). 4. Reopens. The first event after a reopen emits an `INFO ... reopened` line + increments `shepherd_stream_reconnects_total`. 5. Resets `attempt = 0` once the stream has been healthy for the `HEALTHY_WINDOW` (60 s of uninterrupted events) so a flaky-but- then-stable connection reverts to fast retries on the next drop. The event loop reads the channel as a regular `Stream` (wrapped with `futures::stream::unfold` to avoid pulling in `tokio-stream` just for `ReceiverStream`). A bare `None` from the merged stream now indicates the reconnect task itself exited (panic or channel closed); that path still bails the engine as before, but the log message updated to reflect the new semantic. ## Key behavioural change Public Sepolia (`wss://ethereum-sepolia-rpc.publicnode.com`) drops WS connections after ~20 min of sustained load. Pre-fix: engine bailed within seconds of the first drop, an operator restart re-opened the subscription but the engine had missed every event in between. Post-fix: the reconnect task waits 1s and reopens; only events that arrived during the 1s gap are missed. Multi-minute drops get progressively longer waits, capped at 5 min. ## New metric (consumed via COW-1034) `shepherd_stream_reconnects_total{kind, chain_id, module}` counter, incremented on every successful reopen. Operators write SLO alerts against this for "stream churn" (e.g. > 5 reconnects per 10 min on the same chain). ## Channel buffer + back-pressure Buffer is 64 events per task. Real-time dispatch usually drains in ~12 s (Sepolia block time) so the buffer is overkill for normal operation; it absorbs a brief dispatch-side stall (e.g. a stop-loss cow-api submit that takes 2 s) without dropping events at the WS boundary. ## Tests - `cargo test --workspace` -> 159 host tests + 6 doctests passing (unchanged shape - all existing tests still pass, including the `run_does_not_bail_when_both_stream_kinds_are_empty` regression guard which verifies the empty-stream path). - `cargo clippy --all-targets --workspace -- -D warnings` clean. - `cargo fmt --all --check` clean. - Live Sepolia happy path: `just run-m3` boots, all 3 modules dispatch normally, `subscription open` log line emitted, no reconnect activity in 60 s window (network was stable). Clean SIGTERM shutdown. ## Out of scope - WS endpoint failover (swap Alchemy <-> publicnode on failure). Operator concern; track via `[engine.chains.]` schema if demand arises. - Backfill of events missed during the drop window. Live-stream semantic only; backfill is an indexer concern outside the M4 engine scope. - Operator-tunable backoff / healthy-window via `engine.toml`. The current constants are workspace literals; configurable in 0.3. - Per-chain isolation across reconnects (COW-1073). The current patch already gives partial isolation: each chain's task drops + reconnects independently and one task's failure does not starve the others. COW-1073 covers the supervisor-side multi-chain coordination. Linear: COW-1071. Sixth M4 issue landed; stacks on #39 (COW-1033). --- crates/nexum-engine/src/runtime/event_loop.rs | 257 ++++++++++++++---- 1 file changed, 204 insertions(+), 53 deletions(-) diff --git a/crates/nexum-engine/src/runtime/event_loop.rs b/crates/nexum-engine/src/runtime/event_loop.rs index ff399fa2..8bf8f1d0 100644 --- a/crates/nexum-engine/src/runtime/event_loop.rs +++ b/crates/nexum-engine/src/runtime/event_loop.rs @@ -1,76 +1,228 @@ //! Open live `eth_subscribe` streams and dispatch their events to the //! supervisor until a shutdown signal arrives. +//! +//! ## COW-1071: per-stream reconnect with exponential backoff +//! +//! `open_block_streams` / `open_log_streams` no longer return a +//! `Vec` that ends on the first WebSocket drop. They each +//! spawn one reconnect-aware task per `(chain_id)` or `(module, +//! chain_id, filter)` tuple. The task: +//! +//! 1. Opens the subscription via the provider pool. +//! 2. Pumps items to an mpsc channel until the underlying stream +//! yields `None` (WS drop) or `Err` (transport-level error). +//! 3. Logs the drop + waits `restart_policy::backoff_for(attempt)` +//! (1s -> 2s -> ... cap 5min). +//! 4. Reopens. On the first event after a reopen, attempt resets +//! if the stream has been healthy for `HEALTHY_WINDOW`. +//! +//! The event loop reads the receiver as a regular `Stream`. The +//! reconnect tasks live for the lifetime of the engine; they exit +//! cleanly when their channel receiver is dropped (which happens +//! when `run` returns). + +use std::time::{Duration, Instant}; use futures::StreamExt; -use futures::stream::{BoxStream, FuturesUnordered, select_all}; +use futures::stream::{BoxStream, select_all}; +use tokio::sync::mpsc; use tracing::{info, warn}; use crate::bindings::nexum; use crate::host::provider_pool::ProviderPool; +use crate::runtime::restart_policy::backoff_for; use crate::supervisor::Supervisor; -/// Per-chain block subscriptions, one shared stream per chain id. +/// Time the wrapper stream must observe uninterrupted events before +/// the backoff counter resets to 0. Long enough that a brief but +/// real connection blip does not silently undo the doubling, short +/// enough that a healthy node reverts to fast retries on the next +/// drop. +const HEALTHY_WINDOW: Duration = Duration::from_secs(60); + +/// Channel buffer for the reconnect tasks. Each chain / module +/// subscription gets its own task -> channel pair; buffer is small +/// because the event loop drains in real time. +const RECONNECT_CHANNEL_BUF: usize = 64; + +/// Per-chain block subscriptions, one reconnect-aware task per chain id. pub async fn open_block_streams( pool: &ProviderPool, chains: &std::collections::BTreeSet, ) -> Vec { - let mut openings: FuturesUnordered<_> = chains - .iter() - .copied() - .map(|chain_id| async move { (chain_id, pool.subscribe_blocks(chain_id).await) }) - .collect(); - let mut streams = Vec::new(); - while let Some((chain_id, result)) = openings.next().await { - match result { - Ok(stream) => { - info!(chain_id, "block subscription open"); - let tagged: TaggedBlockStream = Box::pin(stream.map(move |item| { - item.map(|header| (chain_id, header)) - .map_err(anyhow::Error::from) - })); - streams.push(tagged); - } - Err(err) => { - warn!(chain_id, error = %err, "block subscription failed"); - } - } + for &chain_id in chains { + let (tx, rx) = mpsc::channel::>( + RECONNECT_CHANNEL_BUF, + ); + let pool = pool.clone(); + tokio::spawn(reconnecting_block_task(pool, chain_id, tx)); + let tagged: TaggedBlockStream = Box::pin(receiver_stream(rx)); + streams.push(tagged); } streams } -/// Per-module log subscriptions. Each entry is a stream tagged with -/// the owning module name + chain id. +/// Per-module log subscriptions. Each entry gets its own reconnect- +/// aware task tagged with the owning module name + chain id. pub async fn open_log_streams( pool: &ProviderPool, subs: Vec<(String, u64, alloy_rpc_types_eth::Filter)>, ) -> Vec { - let mut openings: FuturesUnordered<_> = subs - .into_iter() - .map(|(module, chain_id, filter)| async move { - let stream = pool.subscribe_logs(chain_id, filter).await; - (module, chain_id, stream) - }) - .collect(); - let mut streams = Vec::new(); - while let Some((module, chain_id, result)) = openings.next().await { - match result { - Ok(stream) => { - info!(module = %module, chain_id, "log subscription open"); - let module_name = module.clone(); - let tagged: TaggedLogStream = Box::pin(stream.map(move |item| { - item.map(|log| (module_name.clone(), chain_id, log)) - .map_err(anyhow::Error::from) - })); - streams.push(tagged); + for (module, chain_id, filter) in subs { + let (tx, rx) = mpsc::channel::< + Result<(String, u64, alloy_rpc_types_eth::Log), anyhow::Error>, + >(RECONNECT_CHANNEL_BUF); + let pool = pool.clone(); + tokio::spawn(reconnecting_log_task(pool, module, chain_id, filter, tx)); + let tagged: TaggedLogStream = Box::pin(receiver_stream(rx)); + streams.push(tagged); + } + streams +} + +/// Wrap an `mpsc::Receiver` as a `Stream` using +/// `futures::stream::unfold`. Avoids pulling in `tokio-stream` just +/// for `ReceiverStream`. +fn receiver_stream( + rx: mpsc::Receiver, +) -> impl futures::Stream + Send { + futures::stream::unfold(rx, |mut rx| async move { + rx.recv().await.map(|item| (item, rx)) + }) +} + +/// Reconnect-aware loop for a single chain's block subscription. +/// Holds `(pool, chain_id)` and re-opens the underlying alloy +/// `eth_subscribe` stream with exponential backoff after every drop +/// or transport error. +async fn reconnecting_block_task( + pool: ProviderPool, + chain_id: u64, + tx: mpsc::Sender>, +) { + let mut attempt: u32 = 0; + let mut last_event: Option = None; + loop { + match pool.subscribe_blocks(chain_id).await { + Ok(mut inner) => { + if attempt == 0 { + info!(chain_id, "block subscription open"); + } else { + info!(chain_id, attempt, "block subscription reopened"); + metrics::counter!( + "shepherd_stream_reconnects_total", + "kind" => "block", + "chain_id" => chain_id.to_string(), + ) + .increment(1); + } + while let Some(item) = inner.next().await { + let now = Instant::now(); + if attempt > 0 + && last_event.is_some_and(|t| now.duration_since(t) >= HEALTHY_WINDOW) + { + info!(chain_id, "block stream healthy - resetting backoff"); + attempt = 0; + } + last_event = Some(now); + let tagged = item + .map(|header| (chain_id, header)) + .map_err(anyhow::Error::from); + if tx.send(tagged).await.is_err() { + // Receiver dropped -> engine shutting down. + return; + } + } + warn!(chain_id, "block stream ended (WebSocket dropped?)"); + attempt = attempt.saturating_add(1); } Err(err) => { - warn!(module = %module, chain_id, error = %err, "log subscription failed"); + warn!(chain_id, error = %err, "block subscription failed"); + attempt = attempt.saturating_add(1); } } + let backoff = backoff_for(attempt); + warn!( + chain_id, + attempt, + backoff_ms = backoff.as_millis() as u64, + "reconnecting block subscription after backoff", + ); + tokio::time::sleep(backoff).await; + } +} + +/// Reconnect-aware loop for a single (module, chain) log subscription. +async fn reconnecting_log_task( + pool: ProviderPool, + module: String, + chain_id: u64, + filter: alloy_rpc_types_eth::Filter, + tx: mpsc::Sender>, +) { + let mut attempt: u32 = 0; + let mut last_event: Option = None; + loop { + match pool.subscribe_logs(chain_id, filter.clone()).await { + Ok(mut inner) => { + if attempt == 0 { + info!(module = %module, chain_id, "log subscription open"); + } else { + info!(module = %module, chain_id, attempt, "log subscription reopened"); + metrics::counter!( + "shepherd_stream_reconnects_total", + "kind" => "log", + "chain_id" => chain_id.to_string(), + "module" => module.clone(), + ) + .increment(1); + } + while let Some(item) = inner.next().await { + let now = Instant::now(); + if attempt > 0 + && last_event.is_some_and(|t| now.duration_since(t) >= HEALTHY_WINDOW) + { + info!( + module = %module, + chain_id, + "log stream healthy - resetting backoff" + ); + attempt = 0; + } + last_event = Some(now); + let module_name = module.clone(); + let tagged = item + .map(|log| (module_name, chain_id, log)) + .map_err(anyhow::Error::from); + if tx.send(tagged).await.is_err() { + return; + } + } + warn!(module = %module, chain_id, "log stream ended (WebSocket dropped?)"); + attempt = attempt.saturating_add(1); + } + Err(err) => { + warn!( + module = %module, + chain_id, + error = %err, + "log subscription failed" + ); + attempt = attempt.saturating_add(1); + } + } + let backoff = backoff_for(attempt); + warn!( + module = %module, + chain_id, + attempt, + backoff_ms = backoff.as_millis() as u64, + "reconnecting log subscription after backoff", + ); + tokio::time::sleep(backoff).await; } - streams } pub type TaggedBlockStream = std::pin::Pin< @@ -128,14 +280,13 @@ pub async fn run( } Some(Err(err)) => warn!(error = %err, "block stream error - continuing"), None => { - // alloy ends the stream with None when the - // WebSocket drops. Without this branch the loop - // keeps polling a dead stream and the operator - // sees no events with no indication anything is - // wrong. Bail out so the supervisor (or whatever - // wraps the engine) restarts us; a reconnect- - // with-backoff is the 0.3 fix. - warn!("block stream ended (WebSocket dropped?) - shutting down for restart"); + // COW-1071: WebSocket drops are now absorbed by + // the reconnect tasks behind `open_block_streams` + // / `open_log_streams`; the stream surfaced here + // only ends if the underlying task panicked or + // the channel was closed. Treat as an + // unrecoverable engine fault and bail. + warn!("block reconnect task ended unexpectedly - shutting down"); return; } }, @@ -145,7 +296,7 @@ pub async fn run( } Some(Err(err)) => warn!(error = %err, "log stream error - continuing"), None => { - warn!("log stream ended (WebSocket dropped?) - shutting down for restart"); + warn!("log reconnect task ended unexpectedly - shutting down"); return; } },