fix: use rpc fallback to handle chains with dynamic block time

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"),

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<RpcClient>,
     on_connect_tx: watch::Sender<bool>,
     latest_block_tx: mpsc::Sender<BlockNumber>,
 }
@@ -64,6 +91,7 @@ impl BlockWatcher {
         metrics: Arc<Metrics>,
         lifecycle: LifeCycle,
         ws_connection: WsConnection,
+        rpc_client: Arc<RpcClient>,
     ) -> (Arc<Self>, mpsc::Receiver<BlockNumber>) {
         let (on_connect_tx, _) = watch::channel(false);
         let (latest_block_tx, latest_block_rx) = mpsc::channel::<BlockNumber>(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 }
+        );
+    }
 }