From 22f9780ff281b241048b0cef3d9f21da38ec96af Mon Sep 17 00:00:00 2001 From: Denis Makarov Date: Wed, 29 Jul 2026 21:05:22 +0400 Subject: [PATCH] fix: request blocks by range filter if the node issues with traffic --- Cargo.lock | 1 + Cargo.toml | 2 +- src/domain/evm_network.rs | 31 +++ src/services/event_dispatcher.rs | 443 ++++++++++++++++++++++++------- src/services/rpc_client.rs | 124 ++++++++- tests/common/env.rs | 38 ++- tests/common/mod.rs | 2 + tests/common/rpc_proxy.rs | 199 ++++++++++++++ tests/degraded_rpc.rs | 248 +++++++++++++++++ 9 files changed, 975 insertions(+), 113 deletions(-) create mode 100644 tests/common/rpc_proxy.rs create mode 100644 tests/degraded_rpc.rs diff --git a/Cargo.lock b/Cargo.lock index 0dac2ba..8b5879d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -34,6 +34,7 @@ dependencies = [ "alloy-core", "alloy-eips", "alloy-genesis", + "alloy-json-rpc", "alloy-network", "alloy-node-bindings", "alloy-provider", diff --git a/Cargo.toml b/Cargo.toml index ba4b494..3cbfa66 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ serde_json = "1.0.149" futures = "0.3.32" tokio-stream = "0.1.18" clap = { version = "4.6.1", features = ["env", "derive"] } -alloy = { version = "2.0.4", features = ["provider-ws", "getrandom", "node-bindings", "sol-types", "contract", "signers", "signer-local"] } +alloy = { version = "2.0.4", features = ["provider-ws", "getrandom", "node-bindings", "sol-types", "contract", "signers", "signer-local", "json-rpc"] } thiserror = "2.0.18" anyhow = "1.0.102" tracing = "0.1.44" diff --git a/src/domain/evm_network.rs b/src/domain/evm_network.rs index 03509cd..c274d88 100644 --- a/src/domain/evm_network.rs +++ b/src/domain/evm_network.rs @@ -96,6 +96,37 @@ impl EvmNetwork { EvmNetwork::Sepolia => 3, // * 12s = 36s } } + + /// Blocks one backfill `eth_getLogs` chunk may span for this chain. The + /// binding constraint is the provider's per-response log cap (10k on + /// Alchemy-class nodes, 20k on reth), and Transfer-log volume scales + /// with seconds of chain activity — so the budget is per-chain, not + /// one-size-fits-all. Each value targets roughly half of the 10k budget + /// at the chain's typical log volume. + /// + /// Ethereum and Polygon are grounded in the 2026-07-29 range bench + /// (~550 Transfer logs/block on mainnet via OVH reth, ~400 on Polygon + /// via Alchemy); chains without bench data keep their previous + /// heuristic of ~10s of chain activity. The dispatcher bisects a + /// rejected range down to single blocks, so an overestimate degrades + /// gracefully on restrictive fallbacks. Values never exceed + /// `max_block_lag()`: a chunk wider than the backfill cap could never + /// be filled. + pub fn backfill_chunk_blocks(self) -> u64 { + match self { + EvmNetwork::Eth => 2, // ~1.1k logs; covers the 3-block cap in 2 requests + EvmNetwork::Bnb => 13, // ~10s heuristic + EvmNetwork::Gnosis => 2, // ~10s heuristic + EvmNetwork::Polygon => 15, // ~6k logs, p50 ~1.7s — the whole cap in one request + EvmNetwork::Base => 5, // ~10s heuristic + EvmNetwork::Plasma => 10, // ~10s heuristic + EvmNetwork::Arbitrum => 40, // ~10s heuristic + EvmNetwork::Avalanche => 5, // ~10s heuristic + EvmNetwork::Ink => 10, // ~10s heuristic + EvmNetwork::Linea => 2, // ~10s heuristic + EvmNetwork::Sepolia => 1, // ~10s heuristic (12s blocks: block-by-block) + } + } } impl TryFrom for EvmNetwork { diff --git a/src/services/event_dispatcher.rs b/src/services/event_dispatcher.rs index da1058d..32ed9be 100644 --- a/src/services/event_dispatcher.rs +++ b/src/services/event_dispatcher.rs @@ -20,14 +20,35 @@ //! **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 +//! sequential `eth_getLogs` chunks of at most +//! [`EvmNetwork::backfill_chunk_blocks`] blocks (a per-chain budget sized +//! against provider log caps), 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). +//! +//! **Queue drain.** Heads that arrive while a fetch cycle is in flight +//! queue up in the block channel. Processing them one recv at a time +//! would fire one single-block fetch cycle per queued head — on a provider +//! that is even slightly slower than the chain's block time the dispatcher +//! then falls behind monotonically with no way to amortize (this is +//! exactly how the 2026-07-29 Polygon crashloop unfolded: a fallback node +//! ~15% over block time grew lag past `max_block_lag` and `/health` killed +//! the pod). Instead, every cycle drains the channel to the newest queued +//! head and hands `plan_fetch` the whole span at once, so a backlog +//! collapses into one ranged plan (ordinary ranged `eth_getLogs` amortizes +//! its per-request overhead: ~2× the latency buys ~10× the blocks). +//! +//! **Range bisect.** Some providers reject ranged topic-only queries that +//! a healthy node accepts — log-count / response-size caps, or public +//! fallbacks that require an address filter beyond a single block. A +//! rejected range is split in half and each half refetched, bottoming out +//! at single blocks — whose failure loses only that block's logs (the +//! pre-range behavior). Worst case therefore degrades to block-by-block +//! fetching, never below it. use crate::domain::EvmNetwork; use crate::evm::wrapped::WrappedToken; @@ -35,25 +56,18 @@ use crate::graceful_shutdown::LifeCycle; use crate::metrics::Metrics; use crate::services::block_watcher::BlockWatcher; use crate::services::health::SubsystemHealth; -use crate::services::rpc_client::RpcClient; +use crate::services::rpc_client::{RpcClient, RpcError}; use alloy::primitives::{Address, BlockNumber}; use alloy::rpc::types::Log; use alloy::sol_types::SolEvent; +use std::future::Future; 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. @@ -63,8 +77,8 @@ struct FetchPlan { /// 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. + /// one backfill chunk wide (see [`EvmNetwork::backfill_chunk_blocks`]). + /// Empty = stale or duplicate head. ranges: Vec<(BlockNumber, BlockNumber)>, } @@ -87,8 +101,8 @@ 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`]). + /// Snapshot of [`EvmNetwork::backfill_chunk_blocks`] — blocks one + /// backfill `eth_getLogs` chunk may span for this chain. backfill_chunk_blocks: u64, } @@ -112,7 +126,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()), + backfill_chunk_blocks: network.backfill_chunk_blocks(), }); let dispatcher_for_spawn = Arc::clone(&dispatcher); @@ -123,13 +137,6 @@ 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 @@ -177,6 +184,89 @@ impl Erc20TransferEventDispatcher { } } + /// Empty the block channel and fold everything queued into the newest + /// head (max, not last — defensive against a reordered head). One + /// ranged plan against the newest head replaces one fetch cycle per + /// queued head; see the module docs ("Queue drain") for why processing + /// the queue one head at a time cannot keep up with a slow provider. + fn drain_queued_heads( + rx: &mut mpsc::Receiver, + first: BlockNumber, + ) -> (BlockNumber, u64) { + let mut newest = first; + let mut drained = 0; + while let Ok(queued) = rx.try_recv() { + drained += 1; + newest = newest.max(queued); + } + (newest, drained) + } + + /// Fetch `from..=to` via `fetch`, bisecting on failure: a failed + /// multi-block range is split in half and each half refetched, + /// bottoming out at single blocks. Only blocks whose single-block + /// fetch also fails lose their logs. + /// + /// Each subrange's logs are handed to `handle_log` (oldest subrange + /// first) as soon as that subrange lands — downstream sessions see + /// events from the resolved half while the other half is still being + /// bisected, instead of waiting for the whole traversal. Retry/backoff + /// lives inside `fetch` ([`RpcClient`] short-circuits deterministic + /// "query too big" rejections so the bisect reacts immediately instead + /// of after a full backoff round). + /// + /// Returns `(handled_logs, lost_blocks)`. + async fn fetch_range_with_bisect( + from: BlockNumber, + to: BlockNumber, + fetch: F, + mut handle_log: H, + ) -> (u64, u64) + where + F: Fn(BlockNumber, BlockNumber) -> Fut, + Fut: Future, RpcError>>, + H: FnMut(Log) -> HFut, + HFut: Future, + { + let mut handled_logs = 0; + let mut lost_blocks = 0; + // LIFO with the older half pushed last → subranges resolve oldest + // first, preserving the pre-bisect log order. + let mut pending = vec![(from, to)]; + + while let Some((from, to)) = pending.pop() { + match fetch(from, to).await { + Ok(batch) => { + for log in batch { + handled_logs += 1; + handle_log(log).await; + } + } + Err(err) if from < to => { + let mid = from + (to - from) / 2; + tracing::warn!( + from, + to, + error = %err, + "event dispatcher: ranged eth_getLogs failed, bisecting" + ); + pending.push((mid + 1, to)); + pending.push((from, mid)); + } + Err(err) => { + lost_blocks += 1; + tracing::warn!( + block = from, + error = %err, + "event dispatcher: single-block eth_getLogs failed, block logs lost" + ); + } + } + } + + (handled_logs, lost_blocks) + } + /// 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 — @@ -230,6 +320,16 @@ impl Erc20TransferEventDispatcher { return; }; + let (block_number, drained_heads) = + Self::drain_queued_heads(&mut block_number_rx, block_number); + if drained_heads > 0 { + tracing::info!( + drained_heads, + head = block_number, + "event dispatcher: drained queued heads, catching up in one ranged plan" + ); + } + let latest_processed = self.latest_processed_block.load(Ordering::Relaxed); let plan = Self::plan_fetch( latest_processed, @@ -296,38 +396,32 @@ impl Erc20TransferEventDispatcher { tracing::debug!(from, to, "event dispatcher: fetching weth9 logs for range"); let t0 = std::time::Instant::now(); - match self - .rpc_client - .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!( - from, - to, - count = logs.len(), - duration_ms = elapsed_ms as u64, - "event dispatcher: got weth9 logs for range", - ); - - for log in logs { - self.on_weth9_log(log).await; - } - } - Err(err) => { - self.metrics - .event_dispatcher_missed_block_logs_total - .increment(to - from + 1); - tracing::warn!( - from, - to, - error = %err, - "event dispatcher: eth_getLogs for weth9 exhausted retries, range logs lost" - ); - } + let (count, lost_blocks) = Self::fetch_range_with_bisect( + from, + to, + |from, to| { + self.rpc_client + .fetch_weth9_logs_in_range(self.weth9_address, from, to) + }, + |log| self.on_weth9_log(log), + ) + .await; + + let elapsed_ms = t0.elapsed().as_millis() as f64; + self.metrics.eth_get_logs_duration_ms.record(elapsed_ms); + if lost_blocks > 0 { + self.metrics + .event_dispatcher_missed_block_logs_total + .increment(lost_blocks); } + tracing::debug!( + from, + to, + count, + lost_blocks, + duration_ms = elapsed_ms as u64, + "event dispatcher: got weth9 logs for range", + ); } async fn fetch_erc20_transfer_logs_and_process(&self, from: BlockNumber, to: BlockNumber) { @@ -338,33 +432,29 @@ impl Erc20TransferEventDispatcher { ); let t0 = std::time::Instant::now(); - 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!( - from, - to, - count = logs.len(), - duration_ms = elapsed_ms as u64, - "event dispatcher: got erc20 transfer logs for range" - ); - for log in logs { - self.on_erc20_log(log).await; - } - } - Err(err) => { - self.metrics - .event_dispatcher_missed_block_logs_total - .increment(to - from + 1); - tracing::warn!( - from, - to, - error = %err, - "event dispatcher: eth_getLogs for erc20 exhausted retries, range logs lost" - ); - } + let (count, lost_blocks) = Self::fetch_range_with_bisect( + from, + to, + |from, to| self.rpc_client.fetch_transfer_logs_in_range(from, to), + |log| self.on_erc20_log(log), + ) + .await; + + let elapsed_ms = t0.elapsed().as_millis() as f64; + self.metrics.eth_get_logs_duration_ms.record(elapsed_ms); + if lost_blocks > 0 { + self.metrics + .event_dispatcher_missed_block_logs_total + .increment(lost_blocks); } + tracing::debug!( + from, + to, + count, + lost_blocks, + duration_ms = elapsed_ms as u64, + "event dispatcher: got erc20 transfer logs for range" + ); } async fn on_weth9_log(&self, log: Log) { @@ -513,27 +603,180 @@ mod tests { } #[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 - ); + fn drain_folds_queue_into_newest_head_and_counts() { + let (tx, mut rx) = mpsc::channel(8); + tx.try_send(101).unwrap(); + tx.try_send(103).unwrap(); + tx.try_send(102).unwrap(); + + let (newest, drained) = Erc20TransferEventDispatcher::drain_queued_heads(&mut rx, 100); + assert_eq!(newest, 103); + assert_eq!(drained, 3); + } + + #[test] + fn drain_on_empty_queue_keeps_first_head() { + let (_tx, mut rx) = mpsc::channel::(8); + let (newest, drained) = Erc20TransferEventDispatcher::drain_queued_heads(&mut rx, 100); + assert_eq!(newest, 100); + assert_eq!(drained, 0); + } + + /// One synthetic log per block in `from..=to`, tagged via `block_number`. + fn logs_for_range(from: BlockNumber, to: BlockNumber) -> Vec { + (from..=to) + .map(|block| Log { + block_number: Some(block), + ..Log::default() + }) + .collect() + } + + /// Run the bisect over a fake provider that rejects ranges for which + /// `reject` returns true, collecting handled logs' block numbers in + /// arrival order. + async fn run_bisect( + from: BlockNumber, + to: BlockNumber, + reject: impl Fn(BlockNumber, BlockNumber) -> bool, + ) -> (Vec, u64, u64) { + let handled = std::cell::RefCell::new(Vec::new()); + let (count, lost) = Erc20TransferEventDispatcher::fetch_range_with_bisect( + from, + to, + |from, to| { + let rejected = reject(from, to); + async move { + if rejected { + Err(RpcError::Exhausted("rejected".into())) + } else { + Ok(logs_for_range(from, to)) + } + } + }, + |log| { + handled.borrow_mut().extend(log.block_number); + async {} + }, + ) + .await; + (handled.into_inner(), count, lost) + } + + #[tokio::test] + async fn bisect_splits_until_provider_accepts() { + // provider rejects any range wider than 2 blocks + let (handled, count, lost) = run_bisect(1, 5, |from, to| to - from + 1 > 2).await; + + assert_eq!(lost, 0); + assert_eq!(count, 5); + assert_eq!(handled, vec![1, 2, 3, 4, 5]); + } + + #[tokio::test] + async fn bisect_loses_only_the_block_that_fails_alone() { + // any range touching block 3 fails; the rest succeeds + let (handled, count, lost) = run_bisect(1, 5, |from, to| (from..=to).contains(&3)).await; + + assert_eq!(lost, 1); + assert_eq!(count, 4); + assert_eq!(handled, vec![1, 2, 4, 5]); + } + + #[tokio::test] + async fn bisect_total_failure_loses_every_block() { + let (handled, count, lost) = run_bisect(1, 5, |_, _| true).await; + + assert!(handled.is_empty()); + assert_eq!(count, 0); + assert_eq!(lost, 5); + } + + #[tokio::test] + async fn bisect_single_block_range_does_not_split() { + let (handled, count, lost) = run_bisect(7, 7, |from, to| { + assert_eq!((from, to), (7, 7)); + false + }) + .await; + + assert_eq!(lost, 0); + assert_eq!(count, 1); + assert_eq!(handled, vec![7]); + } + + #[tokio::test] + async fn bisect_streams_logs_before_the_traversal_completes() { + // Interleaving proof: with a provider that rejects ranges wider + // than 2 blocks, logs of an accepted older subrange must reach the + // handler before the younger subranges are even fetched. + let events = std::cell::RefCell::new(Vec::new()); + let _ = Erc20TransferEventDispatcher::fetch_range_with_bisect( + 1, + 5, + |from, to| { + events.borrow_mut().push(format!("fetch {from}-{to}")); + let rejected = to - from + 1 > 2; + async move { + if rejected { + Err(RpcError::Exhausted("rejected".into())) + } else { + Ok(logs_for_range(from, to)) + } + } + }, + |log| { + events + .borrow_mut() + .push(format!("handle {}", log.block_number.unwrap())); + async {} + }, + ) + .await; + assert_eq!( - Erc20TransferEventDispatcher::backfill_chunk_blocks(EvmNetwork::Linea.block_time()), - 2 + events.into_inner(), + vec![ + "fetch 1-5", // rejected, bisect into 1-3 / 4-5 + "fetch 1-3", // rejected, bisect into 1-2 / 3-3 + "fetch 1-2", + "handle 1", // streamed before the rest of the traversal + "handle 2", + "fetch 3-3", + "handle 3", + "fetch 4-5", + "handle 4", + "handle 5", + ] ); } + + #[test] + fn chunk_never_exceeds_backfill_cap_and_is_at_least_one_block() { + // A chunk wider than max_block_lag could never be filled: plan_fetch + // caps the whole backfill below it. Zero would stall the plan loop. + const ALL_NETWORKS: [EvmNetwork; 11] = [ + EvmNetwork::Eth, + EvmNetwork::Bnb, + EvmNetwork::Gnosis, + EvmNetwork::Polygon, + EvmNetwork::Base, + EvmNetwork::Plasma, + EvmNetwork::Arbitrum, + EvmNetwork::Avalanche, + EvmNetwork::Ink, + EvmNetwork::Linea, + EvmNetwork::Sepolia, + ]; + + for network in ALL_NETWORKS { + let chunk = network.backfill_chunk_blocks(); + assert!(chunk >= 1, "{network}: zero-width chunk"); + assert!( + chunk <= network.max_block_lag(), + "{network}: chunk {chunk} exceeds backfill cap {}", + network.max_block_lag() + ); + } + } } diff --git a/src/services/rpc_client.rs b/src/services/rpc_client.rs index cd13b3a..b8639bd 100644 --- a/src/services/rpc_client.rs +++ b/src/services/rpc_client.rs @@ -35,6 +35,7 @@ use alloy::primitives::{Address, BlockNumber, U256}; use alloy::providers::{DynProvider, Dynamic, Failure, MulticallBuilder, MulticallError, Provider}; use alloy::rpc::types::{Filter, Log}; use alloy::sol_types::{SolCall, SolEvent}; +use alloy::transports::TransportError; use backon::{ExponentialBuilder, Retryable}; use futures::stream::{self, StreamExt}; use std::collections::HashMap; @@ -58,8 +59,9 @@ type DynMulticallBuilder = MulticallBuilder, Arc, Eth /// Error surface for this module. #[derive(Debug, Clone, thiserror::Error)] pub enum RpcError { - /// Multicall retry path: backoff exhausted, or short-circuited on a - /// permanent error by [`RpcClient::is_multicall_retryable`]. + /// Backoff exhausted, or short-circuited on a permanent error + /// ([`RpcClient::is_multicall_retryable`] / + /// [`RpcClient::is_get_logs_retryable`]). #[error("Provider exhausted after retries: {0}")] Exhausted(String), } @@ -147,12 +149,12 @@ impl RpcClient { self.get_logs_with_retries(filter).await } - /// Run `eth_getLogs` with retry/backoff. Retries every error unconditionally - /// — we construct all filters ourselves, so "permanent" errors (bad filter, - /// unknown topic) are not expected here. Real-world failures are transport - /// hiccups (5xx, connection reset, timeout) and "block not found" while - /// the node is briefly behind the head; both resolve within one backoff - /// window. + /// Run `eth_getLogs` with retry/backoff. Transient failures — transport + /// hiccups (5xx, connection reset, timeout), rate limiting, and "block + /// not found" while the node is briefly behind the head — resolve within + /// one backoff window and are retried. Deterministic query rejections + /// ([`Self::is_get_logs_retryable`] returns `false`) short-circuit so the + /// dispatcher's range bisect can react immediately. async fn get_logs_with_retries(&self, filter: Filter) -> Result, RpcError> { let metrics = Arc::clone(&self.metrics); (|| { @@ -160,6 +162,7 @@ impl RpcClient { async move { self.provider.get_logs(&filter).await } }) .retry(Self::backoff()) + .when(Self::is_get_logs_retryable) .notify(move |err, duration| { metrics.eth_get_logs_failed_total.increment(1); tracing::warn!( @@ -169,7 +172,53 @@ impl RpcClient { ); }) .await - .map_err(|err| RpcError::Exhausted(err.to_string())) + .map_err(|err| { + if !Self::is_get_logs_retryable(&err) { + tracing::warn!( + error = %err, + "eth_getLogs rejected the query permanently, not retrying" + ); + } + RpcError::Exhausted(err.to_string()) + }) + } + + /// Classify an `eth_getLogs` error: worth another backoff round, or a + /// deterministic rejection of this exact query? + /// + /// Providers cap ranged log queries in provider-specific ways — max + /// results (reth: "query exceeds max results", Alchemy: "query returned + /// more than 10000 results"), response size, or topic-only filters + /// refused outright beyond a single block (publicnode: "please specify + /// an address"). Those fail identically on every attempt; each futile + /// backoff round costs ~14s during which the dispatcher's range bisect + /// cannot start. Detection is message-based and heuristic by necessity + /// (error codes are provider-specific: reth -32602, Infura -32005, + /// publicnode -32701) — a missed pattern only costs the old + /// full-backoff behavior, a false hit skips retries for one range and + /// the bisect refetches its halves anyway. + fn is_get_logs_retryable(err: &TransportError) -> bool { + let Some(resp) = err.as_error_resp() else { + return true; + }; + + let message = resp.message.to_lowercase(); + // "rate limit exceeded" must keep backing off — don't let the + // "exceed" pattern below classify throttling as permanent. + if message.contains("rate") { + return true; + } + + const QUERY_TOO_BIG: [&str; 5] = [ + "exceed", + "too large", + "more than", + "response size", + "specify an address", + ]; + !QUERY_TOO_BIG + .iter() + .any(|pattern| message.contains(pattern)) } /// Read ERC20 balances for `owner` at `block_id`, one Multicall3 @@ -484,3 +533,60 @@ impl RpcClient { .with_jitter() } } + +#[cfg(test)] +mod tests { + use super::*; + use alloy::rpc::json_rpc::ErrorPayload; + use alloy::transports::TransportErrorKind; + + fn error_resp(code: i64, message: &str) -> TransportError { + TransportError::ErrorResp(ErrorPayload { + code, + message: message.to_string().into(), + data: None, + }) + } + + #[test] + fn transport_level_errors_are_retryable() { + assert!(RpcClient::is_get_logs_retryable( + &TransportErrorKind::backend_gone() + )); + } + + #[test] + fn rate_limiting_is_retryable() { + assert!(RpcClient::is_get_logs_retryable(&error_resp( + -32005, + "rate limit exceeded" + ))); + } + + #[test] + fn node_behind_head_is_retryable() { + assert!(RpcClient::is_get_logs_retryable(&error_resp( + -32000, + "block not found" + ))); + } + + #[test] + fn provider_query_caps_are_permanent() { + // reth + assert!(!RpcClient::is_get_logs_retryable(&error_resp( + -32602, + "query exceeds max results 20000, retry with the range 100-150" + ))); + // Alchemy + assert!(!RpcClient::is_get_logs_retryable(&error_resp( + -32602, + "Log response size exceeded. query returned more than 10000 results" + ))); + // publicnode-style topic-only restriction + assert!(!RpcClient::is_get_logs_retryable(&error_resp( + -32701, + "Please specify an address in your request" + ))); + } +} diff --git a/tests/common/env.rs b/tests/common/env.rs index 5e6b662..0d099f5 100644 --- a/tests/common/env.rs +++ b/tests/common/env.rs @@ -20,6 +20,7 @@ use super::api::start_token_list_server; use super::onchain::{ install_infrastructure, spawn_anvil, Weth9, CUSTOM_TOKEN_ADDRESS, WETH9_ADDRESS, }; +use super::rpc_proxy::RpcProxy; /// Prometheus recorder is a process-wide singleton — `install_recorder` /// after the first successful call returns Err, but there's no way to fish @@ -68,9 +69,39 @@ impl Env { /// 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 (env, _no_proxy) = Self::spawn_inner(network, false).await; + env + } + + /// Same as [`Self::spawn_with_network`], but the service's HTTP RPC is + /// routed through a programmable fault-injection [`RpcProxy`]; the WS + /// `newHeads` subscription stays wired straight to anvil. Lets tests + /// degrade `eth_getLogs` (delays, range caps, poisoned blocks) while + /// the chain keeps running honestly underneath — the shape of a real + /// provider incident, where heads keep flowing over WS while the HTTP + /// side struggles. + pub async fn spawn_with_rpc_proxy(network: EvmNetwork) -> (Self, RpcProxy) { + let (env, proxy) = Self::spawn_inner(network, true).await; + ( + env, + proxy.expect("spawn_inner(with_proxy = true) always builds a proxy"), + ) + } + + async fn spawn_inner(network: EvmNetwork, with_proxy: bool) -> (Self, Option) { let (anvil, provider, deployer, deployer_wallet) = spawn_anvil().await; install_infrastructure(&provider).await; + let proxy = if with_proxy { + Some(RpcProxy::spawn(anvil.endpoint()).await) + } else { + None + }; + let rpc_http_url = proxy + .as_ref() + .map(|proxy| proxy.url.clone()) + .unwrap_or_else(|| anvil.endpoint()); + let owner = deployer.address(); let (token_list_url, _token_list_cancel) = start_token_list_server(WETH9_ADDRESS).await; @@ -80,7 +111,7 @@ impl Env { let network_cfg = NetworkConfig { network, - rpc_http_url: anvil.endpoint(), + rpc_http_url, snapshot_interval: 5, max_watched_tokens_limit: 1500, // the in-process token-list server lives on 127.0.0.1 @@ -103,7 +134,7 @@ impl Env { // SSE `balance_update` races the connect. tokio::time::sleep(Duration::from_millis(1000)).await; - Self { + let env = Self { service_url: format!("http://{}", server.local_addr), anvil, provider, @@ -113,7 +144,8 @@ impl Env { token_list_url, _token_list_cancel, service_lifecycle, - } + }; + (env, proxy) } /// `WETH.deposit()` with `amount` wei attached. Triggers a Deposit log. diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f0d629d..d052095 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -18,6 +18,7 @@ pub mod api; pub mod env; pub mod onchain; +pub mod rpc_proxy; pub use api::{ fetch_metric, get_sse, open_sse, post_session, put_session, sse_stream, @@ -25,3 +26,4 @@ pub use api::{ }; pub use env::{Env, ServiceEnv}; pub use onchain::{Weth9, CUSTOM_TOKEN_ADDRESS, MULTICALL3_ADDRESS, WETH9_ADDRESS}; +pub use rpc_proxy::RpcProxy; diff --git a/tests/common/rpc_proxy.rs b/tests/common/rpc_proxy.rs new file mode 100644 index 0000000..8870238 --- /dev/null +++ b/tests/common/rpc_proxy.rs @@ -0,0 +1,199 @@ +//! Programmable JSON-RPC fault-injection proxy for the HTTP leg of the +//! service. Anvil always answers honestly and fast, so degraded-provider +//! scenarios — the 2026-07-29 Polygon incident class — cannot be produced +//! by the node itself. This proxy sits between the service and anvil and +//! selectively degrades `eth_getLogs` while passing every other method +//! through untouched: +//! +//! - **delay** — each successful `eth_getLogs` sleeps first, emulating a +//! provider slower than the chain's block time (the incident profile); +//! - **range cap** — ranges wider than N blocks are rejected with a +//! provider-style "query exceeds max results" error (reth/Alchemy +//! class), exercising the dispatcher's bisect; +//! - **poisoned block** — any range containing a given block is rejected, +//! exercising the single-block-loss path. +//! +//! Rejections short-circuit before the delay (a provider computes "too +//! big" cheaply), and every observed / rejected range is recorded so tests +//! can assert that the drain actually produced ranged queries. +//! +//! The WS `newHeads` leg is NOT proxied — tests wire it straight to anvil, +//! so heads keep flowing while HTTP degrades, exactly like production +//! (separate connections through the rpc-proxy). + +use axum::body::Bytes; +use axum::http::header::CONTENT_TYPE; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use axum::{Json, Router}; +use serde_json::{json, Value}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio_util::sync::{CancellationToken, DropGuard}; + +#[derive(Default)] +struct Rules { + get_logs_delay: Option, + max_get_logs_blocks: Option, + reject_range_containing: Option, + seen_get_logs_ranges: Vec<(u64, u64)>, + rejected_get_logs_ranges: Vec<(u64, u64)>, +} + +/// Handle to the running proxy. Dropping it shuts the listener down. +pub struct RpcProxy { + /// Base URL to hand the service as `rpc_http_url`. + pub url: String, + rules: Arc>, + _shutdown: DropGuard, +} + +impl RpcProxy { + /// Bind on `127.0.0.1:0` and forward everything to `upstream` (anvil's + /// HTTP endpoint), subject to the currently configured rules. + pub async fn spawn(upstream: String) -> Self { + let rules: Arc> = Arc::default(); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind rpc proxy"); + let addr = listener.local_addr().expect("rpc proxy local_addr"); + let cancel = CancellationToken::new(); + let client = reqwest::Client::new(); + + let router = Router::new().route( + "/", + post({ + let rules = Arc::clone(&rules); + move |body: Bytes| { + handle(Arc::clone(&rules), client.clone(), upstream.clone(), body) + } + }), + ); + + let shutdown = cancel.clone(); + tokio::spawn(async move { + axum::serve(listener, router) + .with_graceful_shutdown(async move { shutdown.cancelled().await }) + .await + .ok(); + }); + + Self { + url: format!("http://{addr}"), + rules, + _shutdown: cancel.drop_guard(), + } + } + + /// Sleep this long before answering each (non-rejected) `eth_getLogs`. + /// `None` restores full speed. + pub fn set_get_logs_delay(&self, delay: Option) { + self.rules.lock().unwrap().get_logs_delay = delay; + } + + /// Reject `eth_getLogs` ranges wider than `max` blocks with a + /// provider-style "query exceeds max results" error. `None` lifts the cap. + pub fn set_max_get_logs_blocks(&self, max: Option) { + self.rules.lock().unwrap().max_get_logs_blocks = max; + } + + /// Reject every `eth_getLogs` whose range contains `block`. `None` + /// heals the block. + pub fn set_reject_range_containing(&self, block: Option) { + self.rules.lock().unwrap().reject_range_containing = block; + } + + /// Every `eth_getLogs` `(from, to)` observed so far, in arrival order. + pub fn get_logs_ranges(&self) -> Vec<(u64, u64)> { + self.rules.lock().unwrap().seen_get_logs_ranges.clone() + } + + /// The subset of ranges that were rejected by the configured rules. + pub fn rejected_get_logs_ranges(&self) -> Vec<(u64, u64)> { + self.rules.lock().unwrap().rejected_get_logs_ranges.clone() + } +} + +async fn handle( + rules: Arc>, + client: reqwest::Client, + upstream: String, + body: Bytes, +) -> Response { + let request: Option = serde_json::from_slice(&body).ok(); + let id = request + .as_ref() + .and_then(|request| request.get("id")) + .cloned() + .unwrap_or(Value::Null); + + if let Some((from, to)) = request.as_ref().and_then(parse_get_logs_range) { + let (reject, delay) = { + let mut rules = rules.lock().unwrap(); + rules.seen_get_logs_ranges.push((from, to)); + let too_wide = rules + .max_get_logs_blocks + .is_some_and(|max| to - from + 1 > max); + let poisoned = rules + .reject_range_containing + .is_some_and(|block| (from..=to).contains(&block)); + if too_wide || poisoned { + rules.rejected_get_logs_ranges.push((from, to)); + } + (too_wide || poisoned, rules.get_logs_delay) + }; + + if reject { + return Json(json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32602, "message": "query exceeds max results 20000" } + })) + .into_response(); + } + if let Some(delay) = delay { + tokio::time::sleep(delay).await; + } + } + + let forwarded = client + .post(&upstream) + .header(CONTENT_TYPE, "application/json") + .body(body) + .send() + .await; + match forwarded { + Ok(response) => match response.bytes().await { + Ok(bytes) => ([(CONTENT_TYPE, "application/json")], bytes).into_response(), + Err(err) => upstream_error(id, err).into_response(), + }, + Err(err) => upstream_error(id, err).into_response(), + } +} + +/// Transport-style JSON-RPC error: the message deliberately avoids the +/// service's "query too big" patterns so it stays retryable. +fn upstream_error(id: Value, err: reqwest::Error) -> Json { + Json(json!({ + "jsonrpc": "2.0", + "id": id, + "error": { "code": -32000, "message": format!("proxy: upstream unreachable: {err}") } + })) +} + +/// `Some((from, to))` iff the request is an `eth_getLogs` with a concrete +/// numeric block range (the only shape the service sends). +fn parse_get_logs_range(request: &Value) -> Option<(u64, u64)> { + if request.get("method")?.as_str()? != "eth_getLogs" { + return None; + } + let filter = request.get("params")?.get(0)?; + let from = hex_block_number(filter.get("fromBlock")?)?; + let to = hex_block_number(filter.get("toBlock")?)?; + Some((from, to)) +} + +fn hex_block_number(value: &Value) -> Option { + u64::from_str_radix(value.as_str()?.strip_prefix("0x")?, 16).ok() +} diff --git a/tests/degraded_rpc.rs b/tests/degraded_rpc.rs new file mode 100644 index 0000000..1f01fa5 --- /dev/null +++ b/tests/degraded_rpc.rs @@ -0,0 +1,248 @@ +//! Degraded-provider scenarios for the event dispatcher, against a real +//! anvil chain with an [`RpcProxy`] fault-injection layer on the HTTP leg +//! (see `common::rpc_proxy` for why anvil alone cannot produce these). +//! +//! This is the regression suite for the 2026-07-29 Polygon crashloop: a +//! fallback node slightly slower than the chain's block time grew the +//! dispatcher's lag past `max_block_lag`, `/health` flipped, and k8s +//! crash-looped the pod. The fix — queue drain into ranged fetches plus +//! range bisect — is exercised here end-to-end: +//! +//! - positive: a backlog built under a slow provider is drained with +//! ranged `eth_getLogs` and survives a provider that rejects wide +//! ranges, with zero lost blocks; +//! - negative: a permanently failing block loses only itself and does not +//! flip `/health`; +//! - incident replay: a slow provider flips `/health`, and recovery +//! happens through the drain alone — no restart. +//! +//! Anvil mines every second (`--block-time 1`, see `common::onchain`), so +//! heads keep flowing over WS exactly like a live chain while the proxy +//! degrades only the HTTP side. +//! +//! `#[ignore]` per the suite convention; run with +//! `cargo test --test degraded_rpc -- --ignored --test-threads=1`. + +use alloy::primitives::U256; +use alloy::providers::Provider; +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() +} + +async fn wait_healthy(service_url: &str) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + if health_status(service_url).await == reqwest::StatusCode::OK { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "service never became healthy after spawn" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// Poll `metric` until it reaches at least `baseline + expected_delta`. +/// Panics with the observed value on timeout. +async fn wait_metric_delta( + service_url: &str, + metric: &str, + baseline: f64, + expected_delta: f64, + budget: Duration, +) { + let deadline = tokio::time::Instant::now() + budget; + loop { + let now = fetch_metric(service_url, metric).await; + if now >= baseline + expected_delta { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "{metric}: expected +{expected_delta} over {baseline}, still at {now} \ + after {budget:?}" + ); + tokio::time::sleep(Duration::from_millis(300)).await; + } +} + +/// A backlog built under a slow provider must be drained with ranged +/// fetches, and a provider that then rejects wide ranges must not cost a +/// single block: the bisect refetches the halves until the provider +/// accepts. Every Transfer emitted during the degradation reaches the +/// dispatcher; `missed_block_logs` stays flat. +#[tokio::test] +#[ignore] +async fn degraded_provider_backlog_is_bisected_without_losses() { + let (env, proxy) = Env::spawn_with_rpc_proxy(EvmNetwork::Polygon).await; + env.custom_deposit(U256::from(100)).await; + wait_healthy(&env.service_url).await; + + let erc20_before = fetch_metric(&env.service_url, "erc20_event_received_total").await; + let missed_before = + fetch_metric(&env.service_url, "event_dispatcher_missed_block_logs_total").await; + + // Slow provider: each eth_getLogs takes 5s against a 1s block time, so + // heads queue while cycles crawl — the incident profile. + proxy.set_get_logs_delay(Some(Duration::from_secs(5))); + + // Emit Transfers while degraded; each lands in some queued block. + let peer = env.peer_address(); + for _ in 0..6 { + env.custom_transfer(peer, U256::from(1)).await; + } + + // Provider "recovers" but with a range cap: the drained backlog comes + // out as a wide range, gets rejected, and must be bisected down. + proxy.set_max_get_logs_blocks(Some(2)); + proxy.set_get_logs_delay(None); + + wait_metric_delta( + &env.service_url, + "erc20_event_received_total", + erc20_before, + 6.0, + Duration::from_secs(30), + ) + .await; + + let missed_after = + fetch_metric(&env.service_url, "event_dispatcher_missed_block_logs_total").await; + assert_eq!( + missed_after, missed_before, + "no block may lose logs: wide-range rejections must be healed by the bisect" + ); + + let widest_seen = proxy + .get_logs_ranges() + .iter() + .map(|(from, to)| to - from + 1) + .max() + .unwrap_or(0); + assert!( + widest_seen >= 3, + "queue drain never produced a ranged fetch (widest getLogs range: {widest_seen} blocks)" + ); + assert!( + !proxy.rejected_get_logs_ranges().is_empty(), + "the range cap was never hit — bisect path untested" + ); + + assert_eq!( + health_status(&env.service_url).await, + reqwest::StatusCode::OK, + "a drained backlog must not leave health red" + ); +} + +/// A block whose `eth_getLogs` fails deterministically (every range +/// containing it is rejected) loses exactly its own logs — the bisect +/// isolates it, neighbours still deliver, `/health` stays green (per-block +/// fetch failures are logged, not fatal), and the service keeps processing +/// once the block heals. +#[tokio::test] +#[ignore] +async fn permanently_failing_block_loses_only_itself() { + let (env, proxy) = Env::spawn_with_rpc_proxy(EvmNetwork::Polygon).await; + env.custom_deposit(U256::from(100)).await; + wait_healthy(&env.service_url).await; + + let erc20_before = fetch_metric(&env.service_url, "erc20_event_received_total").await; + let missed_before = + fetch_metric(&env.service_url, "event_dispatcher_missed_block_logs_total").await; + + // Poison a near-future block. Anvil mines it within ~2s; both per-block + // fetches (erc20 + weth9) will fail on it and count one missed block each. + let poisoned = env.provider.get_block_number().await.expect("head") + 2; + proxy.set_reject_range_containing(Some(poisoned)); + + wait_metric_delta( + &env.service_url, + "event_dispatcher_missed_block_logs_total", + missed_before, + 2.0, + Duration::from_secs(15), + ) + .await; + + assert_eq!( + health_status(&env.service_url).await, + reqwest::StatusCode::OK, + "a single lost block must not flip /health" + ); + + // Heal the block; a fresh Transfer must flow through the same dispatcher. + proxy.set_reject_range_containing(None); + env.custom_transfer(env.peer_address(), U256::from(1)).await; + wait_metric_delta( + &env.service_url, + "erc20_event_received_total", + erc20_before, + 1.0, + Duration::from_secs(15), + ) + .await; + + let missed_after = + fetch_metric(&env.service_url, "event_dispatcher_missed_block_logs_total").await; + assert_eq!( + missed_after, + missed_before + 2.0, + "exactly the poisoned block may be lost (once per fetch path), nothing else" + ); +} + +/// Incident replay: a provider slower than the block time grows the lag +/// past `max_block_lag` and `/health` flips to 503 — and once the provider +/// recovers, the queue drain catches the dispatcher up and health returns +/// to 200 **without a restart**. Pre-fix, the second half fails: the +/// backlog is worked off one block per cycle and lag keeps growing as long +/// as blocks keep coming. +#[tokio::test] +#[ignore] +async fn slow_provider_flips_health_and_drain_recovers_without_restart() { + // Eth timings: max_block_lag = 3, so a 4s-per-fetch provider against + // 1s blocks overruns the budget within a couple of cycles. + let (env, proxy) = Env::spawn_with_rpc_proxy(EvmNetwork::Eth).await; + wait_healthy(&env.service_url).await; + + proxy.set_get_logs_delay(Some(Duration::from_secs(4))); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + if health_status(&env.service_url).await == reqwest::StatusCode::SERVICE_UNAVAILABLE { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "slow provider never flipped /health — lag guard broken" + ); + tokio::time::sleep(Duration::from_millis(300)).await; + } + + // Provider recovers. The dispatcher must drain the queued heads in one + // ranged plan and bring lag back under the budget on its own. + proxy.set_get_logs_delay(None); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + if health_status(&env.service_url).await == reqwest::StatusCode::OK { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "health never recovered after the provider healed — drain did not catch up" + ); + tokio::time::sleep(Duration::from_millis(300)).await; + } +}