diff --git a/src/metrics.rs b/src/metrics.rs index ba39461..c5f739b 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -68,9 +68,14 @@ pub struct Metrics { /// eth_getLogs for a block exhausted retries — that block's logs are /// permanently lost (dispatcher still bumps `latest_processed_block` to /// keep lag health honest). Sum across erc20 and weth9 paths; one failed - /// source on one block bumps by 1. **Data-loss counter — page on any - /// non-zero rate.** Snapshot loop (60s cadence) is the recovery path. + /// source bumps by the number of blocks in the failed range (1 on the + /// per-head path). **Data-loss counter — page on any non-zero rate.** + /// Snapshot loop (60s cadence) is the recovery path. pub event_dispatcher_missed_block_logs_total: Counter, + /// blocks fetched via gap backfill (range eth_getLogs) instead of the + /// per-head path — i.e. heads the WS subscription never delivered + /// (reconnect gaps, skipped heads). Steady non-zero rate = flaky WS. + pub event_dispatcher_backfilled_blocks_total: Counter, /// token list fetched ok pub token_list_loaded_total: Counter, @@ -133,6 +138,9 @@ impl Metrics { event_dispatcher_missed_block_logs_total: counter!( "event_dispatcher_missed_block_logs_total" ), + event_dispatcher_backfilled_blocks_total: counter!( + "event_dispatcher_backfilled_blocks_total" + ), token_list_loaded_total: counter!("token_list_loaded_total"), token_list_load_failed_total: counter!("token_list_load_failed_total"), diff --git a/src/services/block_watcher.rs b/src/services/block_watcher.rs index 99b8124..30e770f 100644 --- a/src/services/block_watcher.rs +++ b/src/services/block_watcher.rs @@ -23,11 +23,21 @@ //! (`block_time × STALL_TIMEOUT_BLOCKS`, floored at `MIN_STALL_DURATION`) //! that forces a fresh subscription when headers stop arriving, plus the //! state atomics. +//! +//! **Stall cross-check.** "No header for N seconds" is not proof the +//! stream is dead — chains with dynamic block production (Linea batches, +//! bursty low-activity periods) legitimately go quiet for longer than any +//! fixed timeout. When the watchdog fires, one HTTP `eth_blockNumber` +//! settles it: head still at our last observed block → the chain is idle, +//! keep the stream and stay healthy; head moved past us → the stream +//! really stalled, resubscribe (the dispatcher's gap backfill recovers +//! the missed range). Check unavailable → assume the worst, resubscribe. use crate::domain::EvmNetwork; use crate::graceful_shutdown::LifeCycle; use crate::metrics::Metrics; use crate::services::health::SubsystemHealth; +use crate::services::rpc_client::RpcClient; use crate::ws_connection::{ManagedWsSubscription, WsConnection}; use alloy::primitives::BlockNumber; use alloy::rpc::types::Header; @@ -44,6 +54,21 @@ const POST_DISCONNECT_DELAY: Duration = Duration::from_millis(200); const BLOCK_CHANNEL_CAP: usize = 256; +/// Outcome of cross-checking a stalled `newHeads` stream against the HTTP +/// head. See the module docs for the decision table. +#[derive(Debug, PartialEq, Eq)] +enum StallVerdict { + /// HTTP head has not moved past our last observed block — the chain is + /// simply not producing; the stream is presumed fine. + ChainIdle, + /// HTTP head is ahead of the stream — the subscription genuinely + /// stalled and must be re-established. + StreamStalled { http_head: BlockNumber }, + /// No baseline yet (no header this process) or the HTTP check itself + /// failed — cannot tell, so assume the worst. + Unconfirmed, +} + /// Process-wide WS subscription to `newHeads`. See module docs for the reconnect /// strategy and the rationale for a dedicated provider. pub struct BlockWatcher { @@ -52,6 +77,8 @@ pub struct BlockWatcher { connected: AtomicBool, latest_block: AtomicU64, ws_connection: WsConnection, + /// HTTP client for the stall check (`eth_blockNumber` cross-check). + rpc_client: Arc, on_connect_tx: watch::Sender, latest_block_tx: mpsc::Sender, } @@ -64,6 +91,7 @@ impl BlockWatcher { metrics: Arc, lifecycle: LifeCycle, ws_connection: WsConnection, + rpc_client: Arc, ) -> (Arc, mpsc::Receiver) { let (on_connect_tx, _) = watch::channel(false); let (latest_block_tx, latest_block_rx) = mpsc::channel::(BLOCK_CHANNEL_CAP); @@ -74,6 +102,7 @@ impl BlockWatcher { connected: AtomicBool::new(false), latest_block: AtomicU64::new(0), ws_connection, + rpc_client, on_connect_tx, latest_block_tx, }); @@ -87,9 +116,10 @@ impl BlockWatcher { } /// Structured health for `/health`. Healthy iff the watcher has received - /// at least one header since the most recent reconnect and the stream has - /// not since closed or stalled. The `Unhealthy` variant carries a concrete - /// reason string. + /// at least one header since the most recent reconnect — or the stall + /// check has since confirmed the chain is idle at our last observed + /// head — and the stream has not since closed or stalled for real. The + /// `Unhealthy` variant carries a concrete reason string. pub fn health_status(&self) -> SubsystemHealth { if self.connected.load(Ordering::Relaxed) { SubsystemHealth::Healthy @@ -159,8 +189,38 @@ impl BlockWatcher { return; }, Err(_) => { - tracing::warn!(stall_timeout_s = stall_timeout.as_secs(), "stream stalled"); - return; + match self.confirm_stall().await { + StallVerdict::ChainIdle => { + tracing::debug!( + stall_timeout_s = stall_timeout.as_secs(), + last_observed = self.latest_block.load(Ordering::Relaxed), + "no header within stall timeout but http head has not \ + advanced — chain idle, keeping stream" + ); + // Confirmed in sync with the chain head, so the + // canary may report healthy even if no header + // arrived since the last resubscribe. + self.connected.store(true, Ordering::Relaxed); + } + StallVerdict::StreamStalled { http_head } => { + tracing::warn!( + stall_timeout_s = stall_timeout.as_secs(), + http_head, + last_observed = self.latest_block.load(Ordering::Relaxed), + "stream stalled: chain advanced past last observed \ + header, resubscribing" + ); + return; + } + StallVerdict::Unconfirmed => { + tracing::warn!( + stall_timeout_s = stall_timeout.as_secs(), + "stream stalled (head check unavailable or no baseline), \ + resubscribing" + ); + return; + } + } } } } @@ -191,4 +251,63 @@ impl BlockWatcher { fn stall_timeout(block_time: Duration) -> Duration { (block_time * STALL_TIMEOUT_BLOCKS).max(MIN_STALL_DURATION) } + + fn stall_verdict(last_observed: BlockNumber, http_head: BlockNumber) -> StallVerdict { + if http_head <= last_observed { + StallVerdict::ChainIdle + } else { + StallVerdict::StreamStalled { http_head } + } + } + + /// Cross-check a fired stall watchdog against the HTTP head. Requires a + /// baseline (`latest_block != 0` — at least one header seen this + /// process); without one, or when the HTTP check fails or times out, + /// the verdict is [`StallVerdict::Unconfirmed`] and the caller falls + /// back to the plain reconnect path. + async fn confirm_stall(&self) -> StallVerdict { + let last_observed = self.latest_block.load(Ordering::Relaxed); + if last_observed == 0 { + return StallVerdict::Unconfirmed; + } + + match self.rpc_client.latest_block_number().await { + Ok(http_head) => Self::stall_verdict(last_observed, http_head), + Err(err) => { + tracing::warn!(error = %err, "stall check: eth_blockNumber failed"); + StallVerdict::Unconfirmed + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn head_at_last_observed_means_idle() { + assert_eq!( + BlockWatcher::stall_verdict(100, 100), + StallVerdict::ChainIdle + ); + } + + #[test] + fn head_behind_last_observed_means_idle() { + // http node briefly behind the ws view, or a reorg to a lower head — + // neither is evidence the stream is broken + assert_eq!( + BlockWatcher::stall_verdict(100, 97), + StallVerdict::ChainIdle + ); + } + + #[test] + fn head_past_last_observed_means_stalled() { + assert_eq!( + BlockWatcher::stall_verdict(100, 106), + StallVerdict::StreamStalled { http_head: 106 } + ); + } } diff --git a/src/services/event_dispatcher.rs b/src/services/event_dispatcher.rs index 08ed85c..da1058d 100644 --- a/src/services/event_dispatcher.rs +++ b/src/services/event_dispatcher.rs @@ -10,12 +10,24 @@ //! //! Flow: //! BlockWatcher --(watch::Receiver>)--> Dispatcher -//! Dispatcher --(per-block eth_getLogs × 2)--> RpcClient +//! Dispatcher --(per-range eth_getLogs × 2)--> RpcClient //! Dispatcher --(Erc20TransferEvent)--> SessionManager router //! -//! Each new block triggers exactly two `eth_getLogs` calls with a -//! single-block range, so the cost is fixed per block (~two HTTP RPCs -//! per ~12s on mainnet) regardless of active session count. +//! On the happy path each new head triggers exactly two `eth_getLogs` +//! calls with a single-block range, so the cost is fixed per block (~two +//! HTTP RPCs per ~12s on mainnet) regardless of active session count. +//! +//! **Gap backfill.** WS `newHeads` skips heads across reconnects (and can +//! skip under load). When an incoming head is more than one block past +//! `latest_processed_block`, the dispatcher fetches the missing range too: +//! sequential `eth_getLogs` chunks sized to [`BACKFILL_CHUNK_SECONDS`] of +//! chain activity (see [`Erc20TransferEventDispatcher::backfill_chunk_blocks`]), advancing the cursor +//! after each chunk — so a chunk that exhausts retries leaves a smaller +//! residual gap that the next head re-triggers. The backfill is capped at +//! the chain's `max_block_lag()` blocks; anything older is deliberately +//! skipped (the 60s snapshot loop is the recovery path, and processing a +//! log only schedules a balance refresh, so replays and reorg leftovers +//! are harmless — refreshes read current chain state). use crate::domain::EvmNetwork; use crate::evm::wrapped::WrappedToken; @@ -29,11 +41,33 @@ use alloy::rpc::types::Log; use alloy::sol_types::SolEvent; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::Arc; +use std::time::Duration; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; const ERC20_TRANSFER_TOPICS_LEN: usize = 3; +/// Budget per backfill `eth_getLogs` range, expressed in chain time: log +/// volume scales with seconds of chain activity, not block count, so a +/// fixed block-count chunk would be huge on a 12s-block chain and trivial +/// on a 250ms one. Providers cap response size; sequential chunks +/// self-throttle. +const BACKFILL_CHUNK_SECONDS: Duration = Duration::from_secs(10); + +/// Fetch work derived from one incoming head vs the dispatcher's cursor. +/// Produced by [`Erc20TransferEventDispatcher::plan_fetch`], executed by +/// the dispatcher loop. +#[derive(Debug, PartialEq, Eq)] +struct FetchPlan { + /// Blocks deliberately dropped because the gap exceeded the backfill + /// cap. Healed by the snapshot loop; logged, not replayed. + skipped_blocks: u64, + /// Inclusive `(from, to)` ranges to fetch, oldest first, each at most + /// one backfill chunk wide (see [`Erc20TransferEventDispatcher::backfill_chunk_blocks`]). Empty = + /// stale or duplicate head. + ranges: Vec<(BlockNumber, BlockNumber)>, +} + pub struct Erc20TransferEvent { pub owner: Address, pub token: Address, @@ -53,6 +87,9 @@ pub struct Erc20TransferEventDispatcher { /// dispatcher was spawned for. Kept as a plain field so the health /// check stays lock-free. max_block_lag: u64, + /// Blocks per backfill `eth_getLogs` chunk, derived once from the + /// chain's block time (see [`Erc20TransferEventDispatcher::backfill_chunk_blocks`]). + backfill_chunk_blocks: u64, } impl Erc20TransferEventDispatcher { @@ -75,6 +112,7 @@ impl Erc20TransferEventDispatcher { weth9_address: network.weth9_address(), latest_processed_block: AtomicU64::new(0), max_block_lag: network.max_block_lag(), + backfill_chunk_blocks: Self::backfill_chunk_blocks(network.block_time()), }); let dispatcher_for_spawn = Arc::clone(&dispatcher); @@ -85,6 +123,60 @@ impl Erc20TransferEventDispatcher { dispatcher } + /// Blocks per backfill chunk for a chain with the given block time — + /// [`BACKFILL_CHUNK_SECONDS`] worth of blocks, at least 1 (chains whose + /// block time exceeds the budget, e.g. mainnet's 12s, go block-by-block). + fn backfill_chunk_blocks(block_time: Duration) -> u64 { + ((BACKFILL_CHUNK_SECONDS.as_millis() / block_time.as_millis().max(1)) as u64).max(1) + } + + /// Plan the fetch for one incoming head: skip stale heads, fetch the head + /// itself, and chunk any gap since `latest_processed` — capped at + /// `max_backfill` blocks counted back from `incoming` (the newest blocks + /// win; stale history is the snapshot loop's job). + /// + /// `latest_processed == 0` means nothing was processed yet: fetch just the + /// incoming head, there is no baseline to backfill from. + fn plan_fetch( + latest_processed: BlockNumber, + incoming: BlockNumber, + max_backfill: u64, + chunk_blocks: u64, + ) -> FetchPlan { + if latest_processed != 0 && incoming <= latest_processed { + return FetchPlan { + skipped_blocks: 0, + ranges: Vec::new(), + }; + } + + let fetch_from = if latest_processed == 0 { + incoming + } else { + let cap_floor = incoming.saturating_sub(max_backfill) + 1; + (latest_processed + 1).max(cap_floor) + }; + + let mut ranges = Vec::new(); + let mut from = fetch_from; + while from <= incoming { + let to = incoming.min(from + chunk_blocks - 1); + ranges.push((from, to)); + from = to + 1; + } + + let skipped_blocks = if latest_processed == 0 { + 0 + } else { + fetch_from - (latest_processed + 1) + }; + + FetchPlan { + skipped_blocks, + ranges, + } + } + /// Structured health for `/health`. Healthy from the moment the dispatcher /// has seen its first block notification, unless block lag has climbed /// past `self.max_block_lag`. Per-block fetch failures do NOT toggle this — @@ -138,13 +230,56 @@ impl Erc20TransferEventDispatcher { return; }; - let _ = tokio::join!( - self.fetch_erc20_transfer_logs_and_process_block(block_number), - self.fetch_and_process_weth9_logs(block_number), + let latest_processed = self.latest_processed_block.load(Ordering::Relaxed); + let plan = Self::plan_fetch( + latest_processed, + block_number, + self.max_block_lag, + self.backfill_chunk_blocks, ); - self.latest_processed_block.store(block_number, Ordering::Relaxed); - self.metrics.event_dispatcher_blocks_processed_total.increment(1); + if plan.ranges.is_empty() { + tracing::debug!( + block = block_number, + latest_processed, + "event dispatcher: stale or duplicate head, skipping" + ); + continue; + } + + if plan.skipped_blocks > 0 { + tracing::warn!( + block = block_number, + latest_processed, + skipped = plan.skipped_blocks, + max_backfill = self.max_block_lag, + "event dispatcher: gap exceeds backfill cap, skipping oldest blocks" + ); + } + + let total_blocks: u64 = plan.ranges.iter().map(|(from, to)| to - from + 1).sum(); + let backfilled = total_blocks - 1; // the head itself is the ordinary path + if backfilled > 0 { + self.metrics.event_dispatcher_backfilled_blocks_total.increment(backfilled); + tracing::info!( + block = block_number, + latest_processed, + backfilled, + "event dispatcher: backfilling gap left by ws heads" + ); + } + + for (from, to) in plan.ranges { + let _ = tokio::join!( + self.fetch_erc20_transfer_logs_and_process(from, to), + self.fetch_and_process_weth9_logs(from, to), + ); + + // Advance after every chunk: a later chunk failing + // leaves only the residual gap for the next head. + self.latest_processed_block.store(to, Ordering::Relaxed); + self.metrics.event_dispatcher_blocks_processed_total.increment(to - from + 1); + } // update dispatcher lag gauge let latest_head = self.block_watcher.latest_block(); @@ -157,26 +292,24 @@ impl Erc20TransferEventDispatcher { self.is_active.store(false, Ordering::Relaxed); } - async fn fetch_and_process_weth9_logs(&self, block_number: BlockNumber) { - tracing::debug!( - block = block_number, - "event dispatcher: fetching weth9 logs for block" - ); + async fn fetch_and_process_weth9_logs(&self, from: BlockNumber, to: BlockNumber) { + tracing::debug!(from, to, "event dispatcher: fetching weth9 logs for range"); let t0 = std::time::Instant::now(); match self .rpc_client - .fetch_weth9_logs_for_block(self.weth9_address, block_number) + .fetch_weth9_logs_in_range(self.weth9_address, from, to) .await { Ok(logs) => { let elapsed_ms = t0.elapsed().as_millis() as f64; self.metrics.eth_get_logs_duration_ms.record(elapsed_ms); tracing::debug!( - block = block_number, + from, + to, count = logs.len(), duration_ms = elapsed_ms as u64, - "event dispatcher: got weth9 logs for block", + "event dispatcher: got weth9 logs for range", ); for log in logs { @@ -186,36 +319,35 @@ impl Erc20TransferEventDispatcher { Err(err) => { self.metrics .event_dispatcher_missed_block_logs_total - .increment(1); + .increment(to - from + 1); tracing::warn!( - block = block_number, + from, + to, error = %err, - "event dispatcher: eth_getLogs for weth9 exhausted retries, block logs lost" + "event dispatcher: eth_getLogs for weth9 exhausted retries, range logs lost" ); } } } - async fn fetch_erc20_transfer_logs_and_process_block(&self, block_number: BlockNumber) { + async fn fetch_erc20_transfer_logs_and_process(&self, from: BlockNumber, to: BlockNumber) { tracing::debug!( - block = block_number, - "event dispatcher: fetching erc20 transfer logs for block" + from, + to, + "event dispatcher: fetching erc20 transfer logs for range" ); let t0 = std::time::Instant::now(); - match self - .rpc_client - .fetch_transfer_logs_for_block(block_number) - .await - { + match self.rpc_client.fetch_transfer_logs_in_range(from, to).await { Ok(logs) => { let elapsed_ms = t0.elapsed().as_millis() as f64; self.metrics.eth_get_logs_duration_ms.record(elapsed_ms); tracing::debug!( - block = block_number, + from, + to, count = logs.len(), duration_ms = elapsed_ms as u64, - "event dispatcher: got erc20 transfer logs for block" + "event dispatcher: got erc20 transfer logs for range" ); for log in logs { self.on_erc20_log(log).await; @@ -224,11 +356,12 @@ impl Erc20TransferEventDispatcher { Err(err) => { self.metrics .event_dispatcher_missed_block_logs_total - .increment(1); + .increment(to - from + 1); tracing::warn!( - block = block_number, + from, + to, error = %err, - "event dispatcher: eth_getLogs for erc20 exhausted retries, block logs lost" + "event dispatcher: eth_getLogs for erc20 exhausted retries, range logs lost" ); } } @@ -303,3 +436,104 @@ impl Erc20TransferEventDispatcher { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn first_head_fetches_only_that_block() { + let plan = Erc20TransferEventDispatcher::plan_fetch(0, 100, 30, 10); + assert_eq!(plan.skipped_blocks, 0); + assert_eq!(plan.ranges, vec![(100, 100)]); + } + + #[test] + fn stale_and_duplicate_heads_are_skipped() { + assert!(Erc20TransferEventDispatcher::plan_fetch(100, 100, 30, 10) + .ranges + .is_empty()); + assert!(Erc20TransferEventDispatcher::plan_fetch(100, 97, 30, 10) + .ranges + .is_empty()); + } + + #[test] + fn next_head_is_a_single_block_range() { + let plan = Erc20TransferEventDispatcher::plan_fetch(100, 101, 30, 10); + assert_eq!(plan.skipped_blocks, 0); + assert_eq!(plan.ranges, vec![(101, 101)]); + } + + #[test] + fn small_gap_backfills_from_cursor_in_one_range() { + let plan = Erc20TransferEventDispatcher::plan_fetch(100, 105, 30, 10); + assert_eq!(plan.skipped_blocks, 0); + assert_eq!(plan.ranges, vec![(101, 105)]); + } + + #[test] + fn wide_gap_is_chunked_with_remainder() { + let plan = Erc20TransferEventDispatcher::plan_fetch(100, 125, 30, 10); + assert_eq!(plan.skipped_blocks, 0); + assert_eq!(plan.ranges, vec![(101, 110), (111, 120), (121, 125)]); + } + + #[test] + fn gap_at_exact_chunk_boundary() { + let plan = Erc20TransferEventDispatcher::plan_fetch(100, 120, 30, 10); + assert_eq!(plan.skipped_blocks, 0); + assert_eq!(plan.ranges, vec![(101, 110), (111, 120)]); + } + + #[test] + fn gap_equal_to_cap_backfills_everything() { + // gap == max_backfill: nothing skipped + let plan = Erc20TransferEventDispatcher::plan_fetch(100, 130, 30, 10); + assert_eq!(plan.skipped_blocks, 0); + assert_eq!(plan.ranges, vec![(101, 110), (111, 120), (121, 130)]); + } + + #[test] + fn gap_over_cap_skips_oldest_blocks() { + // gap = 100, cap = 30: fetch the newest 30 blocks, skip the 70 oldest + let plan = Erc20TransferEventDispatcher::plan_fetch(100, 200, 30, 10); + assert_eq!(plan.skipped_blocks, 70); + assert_eq!(plan.ranges, vec![(171, 180), (181, 190), (191, 200)]); + let fetched: u64 = plan.ranges.iter().map(|(f, t)| t - f + 1).sum(); + assert_eq!(fetched, 30); + } + + #[test] + fn tight_cap_still_fetches_the_head() { + // mainnet-style cap of 3: only the newest 3 blocks survive the cap + let plan = Erc20TransferEventDispatcher::plan_fetch(10, 20, 3, 10); + assert_eq!(plan.skipped_blocks, 7); + assert_eq!(plan.ranges, vec![(18, 20)]); + } + + #[test] + fn chunk_is_time_normalized_per_network() { + // ~BACKFILL_CHUNK_SECONDS of chain activity per request, floor 1 + assert_eq!( + Erc20TransferEventDispatcher::backfill_chunk_blocks(EvmNetwork::Eth.block_time()), + 1 + ); + assert_eq!( + Erc20TransferEventDispatcher::backfill_chunk_blocks(EvmNetwork::Arbitrum.block_time()), + 40 + ); + assert_eq!( + Erc20TransferEventDispatcher::backfill_chunk_blocks(EvmNetwork::Bnb.block_time()), + 13 + ); + assert_eq!( + Erc20TransferEventDispatcher::backfill_chunk_blocks(EvmNetwork::Plasma.block_time()), + 10 + ); + assert_eq!( + Erc20TransferEventDispatcher::backfill_chunk_blocks(EvmNetwork::Linea.block_time()), + 2 + ); + } +} diff --git a/src/services/rpc_client.rs b/src/services/rpc_client.rs index 128405d..cd13b3a 100644 --- a/src/services/rpc_client.rs +++ b/src/services/rpc_client.rs @@ -19,9 +19,10 @@ //! ``` //! //! 2. **Log fetches for the process-wide event dispatcher** — -//! [`RpcClient::fetch_transfer_logs_for_block`] (ERC20 Transfer, global topic -//! filter) and [`RpcClient::fetch_weth9_logs_for_block`] (WETH9 Deposit / -//! Withdrawal, address-filtered). Called once per block by +//! [`RpcClient::fetch_transfer_logs_in_range`] (ERC20 Transfer, global topic +//! filter) and [`RpcClient::fetch_weth9_logs_in_range`] (WETH9 Deposit / +//! Withdrawal, address-filtered). Called per head block (single-block range) +//! or per backfill chunk after a gap by //! [`crate::services::event_dispatcher::Erc20TransferEventDispatcher`]. use crate::config::constants::MULTICALL_PERMITS_COUNT; @@ -43,6 +44,12 @@ use tokio_stream::Stream; const MULTICALL_CHUNK_SIZE: usize = 500; +/// Upper bound on the `eth_blockNumber` round-trip in +/// [`RpcClient::latest_block_number`]. The block-watcher stall check must +/// not be held hostage by a slow node — on timeout the caller treats the +/// head as unknown and falls back to the plain reconnect path. +const HEADER_TIMEOUT: Duration = Duration::from_secs(3); + /// `(token → balance, block_number_the_batch_was_read_at)`. pub type BalancesWithBlock = (HashMap, BlockNumber); @@ -74,35 +81,57 @@ impl RpcClient { } } - /// Fetch every ERC20 `Transfer` log emitted in `block_number`. Uses HTTP - /// `eth_getLogs` (server-side topic filter); returns 100% of matching - /// logs in the block, in contrast to WS `eth_subscribe` whose internal + /// Current chain head via HTTP `eth_blockNumber`. Single attempt, no + /// retry/backoff, bounded by [`HEADER_TIMEOUT`] — the only caller is + /// the block-watcher stall check, which needs a fast verdict and treats + /// any error as "unconfirmed" (falls back to the plain reconnect path). + pub async fn latest_block_number(&self) -> Result { + match tokio::time::timeout(HEADER_TIMEOUT, self.provider.get_block_number()).await { + Ok(result) => result.map_err(|err| RpcError::Exhausted(err.to_string())), + Err(_) => Err(RpcError::Exhausted(format!( + "eth_blockNumber timed out after {}s", + HEADER_TIMEOUT.as_secs() + ))), + } + } + + /// Fetch every ERC20 `Transfer` log emitted in blocks `from..=to`. Uses + /// HTTP `eth_getLogs` (server-side topic filter); returns 100% of matching + /// logs in the range, in contrast to WS `eth_subscribe` whose internal /// buffer drops the tail of burst blocks. /// + /// The happy path is a single-block range (`from == to`, one call per + /// head); wider ranges are used by the dispatcher's gap backfill and must + /// stay small enough for the provider's per-response log limit — the + /// caller owns chunking. + /// /// Transient RPC failures are retried per [`Self::backoff`]; the final /// `Err` surfaces as [`RpcError::Exhausted`]. - pub async fn fetch_transfer_logs_for_block( + pub async fn fetch_transfer_logs_in_range( &self, - block_number: BlockNumber, + from: BlockNumber, + to: BlockNumber, ) -> Result, RpcError> { let filter = Filter::new() - .from_block(block_number) - .to_block(block_number) + .from_block(from) + .to_block(to) .event_signature(ERC20::Transfer::SIGNATURE_HASH); self.get_logs_with_retries(filter).await } - /// Fetch WETH9 `Deposit` and `Withdrawal` logs emitted in `block_number`, - /// address-filtered to the canonical WETH9 contract for this chain. - /// Needed alongside [`Self::fetch_transfer_logs_for_block`] because the - /// canonical WETH9 impl does not emit `Transfer` on wrap / unwrap. + /// Fetch WETH9 `Deposit` and `Withdrawal` logs emitted in blocks + /// `from..=to`, address-filtered to the canonical WETH9 contract for this + /// chain. Needed alongside [`Self::fetch_transfer_logs_in_range`] because + /// the canonical WETH9 impl does not emit `Transfer` on wrap / unwrap. /// - /// Same retry policy as [`Self::fetch_transfer_logs_for_block`]. - pub async fn fetch_weth9_logs_for_block( + /// Same retry policy and chunking contract as + /// [`Self::fetch_transfer_logs_in_range`]. + pub async fn fetch_weth9_logs_in_range( &self, weth9_address: Address, - block_number: BlockNumber, + from: BlockNumber, + to: BlockNumber, ) -> Result, RpcError> { let event_signatures = vec![ WrappedToken::Deposit::SIGNATURE_HASH, @@ -110,8 +139,8 @@ impl RpcClient { ]; let filter = Filter::new() - .from_block(block_number) - .to_block(block_number) + .from_block(from) + .to_block(to) .address(weth9_address) .event_signature(event_signatures); diff --git a/src/services/session_manager.rs b/src/services/session_manager.rs index 36ebb0d..86778a6 100644 --- a/src/services/session_manager.rs +++ b/src/services/session_manager.rs @@ -152,6 +152,7 @@ impl SessionManager { Arc::clone(&metrics), lifecycle.clone(), block_ws, + Arc::clone(&rpc_client), ); let (tx, rx) = mpsc::channel::(256); diff --git a/tests/common/env.rs b/tests/common/env.rs index bba6c6e..5e6b662 100644 --- a/tests/common/env.rs +++ b/tests/common/env.rs @@ -59,6 +59,15 @@ impl Env { /// balances-watcher service. Anything failing aborts the test with a /// pointed panic message. pub async fn spawn() -> Self { + Self::spawn_with_network(EvmNetwork::Eth).await + } + + /// Same as [`Self::spawn`] but with an explicit network. The network + /// picks per-chain timings (`block_time`-derived stall timeout, + /// `max_block_lag`) — timing-sensitive tests spawn a fast chain (e.g. + /// Ink: 3s stall timeout) so they run in seconds instead of mainnet's + /// 36s windows. Anvil doesn't care what chain we claim to be. + pub async fn spawn_with_network(network: EvmNetwork) -> Self { let (anvil, provider, deployer, deployer_wallet) = spawn_anvil().await; install_infrastructure(&provider).await; @@ -70,7 +79,7 @@ impl Env { let metrics = Arc::new(Metrics::install()); let network_cfg = NetworkConfig { - network: EvmNetwork::Eth, + network, rpc_http_url: anvil.endpoint(), snapshot_interval: 5, max_watched_tokens_limit: 1500, diff --git a/tests/stall_health.rs b/tests/stall_health.rs new file mode 100644 index 0000000..c48d5e2 --- /dev/null +++ b/tests/stall_health.rs @@ -0,0 +1,101 @@ +//! Stall cross-check behaviour on an idle chain, against a real anvil node. +//! +//! Anvil (automine) mines a block only when a transaction lands, so an +//! empty chain produces no `newHeads` at all — we don't delay header +//! delivery, we withhold the headers by not submitting transactions. From +//! the watcher's side this is identical to a quiet chain: `stream.next()` +//! yields nothing within the stall timeout. That reproduces the idle +//! condition that bites Linea in production (its sequencer batches +//! transactions and goes quiet for longer than the timeout). +//! +//! The behaviour under test is chain-agnostic, so we spawn as Arbitrum — +//! the fastest timings in the table (250ms block time → stall timeout +//! floored at `MIN_STALL_DURATION` = 2s) — purely to keep the test short. +//! +//! This exercises only the `ChainIdle` branch (HTTP head has not advanced +//! either, since automine stops both the WS and HTTP views together). The +//! `StreamStalled` branch — socket silent while the chain keeps moving — +//! needs a TCP proxy between the service and anvil to desync the WS stream +//! from the HTTP head, and is not covered here. +//! +//! `#[ignore]` per the suite convention; run with +//! `cargo test --ignored -- --test-threads=1`. + +use alloy::primitives::U256; +use balances_watcher::domain::EvmNetwork; +use std::time::Duration; + +mod common; +use common::{fetch_metric, Env}; + +async fn health_status(service_url: &str) -> reqwest::StatusCode { + reqwest::get(format!("{service_url}/health")) + .await + .expect("GET /health") + .status() +} + +/// An idle chain must not flip `/health`: the stall watchdog fires (2s on +/// Arbitrum timings), the head check sees the HTTP head unchanged, the +/// stream is kept and the canary stays green. Without the check this test +/// fails — the watchdog tears the subscription down, no new header ever +/// arrives on the idle chain, and `/health` sits at 503 for the whole +/// quiet window. +#[tokio::test] +#[ignore] +async fn idle_chain_stays_healthy_past_stall_timeout() { + let env = Env::spawn_with_network(EvmNetwork::Arbitrum).await; + + // A single transaction makes anvil mine exactly one block, so the + // watcher observes one header (its health baseline) and the stall check + // gets a non-zero last_observed to compare against. + env.weth_deposit(U256::from(1)).await; + + // Wait out startup races: subscription + first header. + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + if health_status(&env.service_url).await == reqwest::StatusCode::OK { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "service never became healthy after the first mined block" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + + // Idle: no blocks well past the 2s stall timeout, so the watchdog fires + // at least twice; every firing must resolve to ChainIdle and keep + // health green. + tokio::time::sleep(Duration::from_secs(5)).await; + assert_eq!( + health_status(&env.service_url).await, + reqwest::StatusCode::OK, + "idle chain flipped /health — stall check did not hold the stream" + ); + + // The kept stream must still be live: one more transaction mines one + // more block, and that header has to arrive through the very same + // subscription (no resubscribe happened). + let accepted_before = fetch_metric(&env.service_url, "block_accepted_total").await; + env.weth_deposit(U256::from(1)).await; + + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let accepted_now = fetch_metric(&env.service_url, "block_accepted_total").await; + if accepted_now > accepted_before { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "no header consumed after idle window — stream was not kept alive" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + + assert_eq!( + health_status(&env.service_url).await, + reqwest::StatusCode::OK, + "health must stay green after the chain resumes" + ); +}