refactor(host-rpc): rewrite notifier with hash-based chain walking

Replace fetch-by-number with hash-anchored backward walk, eliminating
TOCTOU race conditions in reorg detection. The subscription is now a
wake-up signal; on each event the notifier walks from the hint hash
backward until it finds overlap with a lightweight (u64, B256) ring
buffer. Full blocks are fetched by hash only for the new segment.

Key changes:
- Hash-based walk algorithm (WalkResult enum: Advance/Reorg/Exhausted)
- Parallel block+receipt fetches via tokio::try_join! and FuturesOrdered
- Self-healing buffer exhaustion: clears state and re-enters backfill
- Backfill stops at (latest - buffer_capacity), frontfill bridges gap
- set_backfill_thresholds(None) resets to default per trait contract
- fetch_range guards against from > to underflow
- Segment: non-empty invariant enforced, pub(crate) visibility
- Error: removed dead SubscriptionClosed/ReorgTooDeep variants
- Builder: genesis_timestamp validation, uses pub(crate) constructor
- Shared constants in lib.rs, private notifier fields
- Tracing instrumentation on walk and notification methods

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: prestwich

crates/host-rpc/Cargo.toml

@@ -17,4 +17,5 @@ signet-types.workspace = true
 alloy.workspace = true
 futures-util.workspace = true
 thiserror.workspace = true
+tokio.workspace = true
 tracing.workspace = true

crates/host-rpc/src/builder.rs

@@ -1,11 +1,6 @@
 use crate::{RpcHostError, RpcHostNotifier};
 use alloy::providers::Provider;
-use std::collections::VecDeque;
-
-/// Default block buffer capacity.
-const DEFAULT_BUFFER_CAPACITY: usize = 64;
-/// Default backfill batch size.
-const DEFAULT_BACKFILL_BATCH_SIZE: u64 = 32;
+use tracing::warn;
 
 /// Builder for [`RpcHostNotifier`].
 ///
@@ -15,6 +10,7 @@ const DEFAULT_BACKFILL_BATCH_SIZE: u64 = 32;
 /// let notifier = RpcHostNotifierBuilder::new(provider)
 ///     .with_buffer_capacity(128)
 ///     .with_backfill_batch_size(64)
+///     .with_genesis_timestamp(1_606_824_023)
 ///     .build()
 ///     .await?;
 /// ```
@@ -34,8 +30,8 @@ where
     pub const fn new(provider: P) -> Self {
         Self {
             provider,
-            buffer_capacity: DEFAULT_BUFFER_CAPACITY,
-            backfill_batch_size: DEFAULT_BACKFILL_BATCH_SIZE,
+            buffer_capacity: crate::DEFAULT_BUFFER_CAPACITY,
+            backfill_batch_size: crate::DEFAULT_BACKFILL_BATCH_SIZE,
             genesis_timestamp: 0,
         }
     }
@@ -60,20 +56,18 @@ where
 
     /// Build the notifier, establishing the `newHeads` WebSocket subscription.
     pub async fn build(self) -> Result<RpcHostNotifier<P>, RpcHostError> {
+        if self.genesis_timestamp == 0 {
+            warn!("genesis_timestamp not set; epoch calculations will use Unix epoch");
+        }
         let sub = self.provider.subscribe_blocks().await?;
         let header_sub = sub.into_stream();
 
-        Ok(RpcHostNotifier {
-            provider: self.provider,
+        Ok(RpcHostNotifier::new(
+            self.provider,
             header_sub,
-            block_buffer: VecDeque::with_capacity(self.buffer_capacity),
-            buffer_capacity: self.buffer_capacity,
-            cached_safe: None,
-            cached_finalized: None,
-            last_tag_epoch: None,
-            backfill_from: None,
-            backfill_batch_size: self.backfill_batch_size,
-            genesis_timestamp: self.genesis_timestamp,
-        })
+            self.buffer_capacity,
+            self.backfill_batch_size,
+            self.genesis_timestamp,
+        ))
     }
 }

crates/host-rpc/src/error.rs

@@ -1,26 +1,21 @@
-use alloy::transports::{RpcError, TransportErrorKind};
+use alloy::{
+    eips::BlockNumberOrTag,
+    primitives::B256,
+    transports::{RpcError, TransportErrorKind},
+};
 
 /// Errors from the RPC host notifier.
 #[derive(Debug, thiserror::Error)]
 pub enum RpcHostError {
-    /// The WebSocket subscription was dropped unexpectedly.
-    #[error("subscription closed")]
-    SubscriptionClosed,
-
     /// An RPC call failed.
     #[error("rpc error: {0}")]
     Rpc(#[from] RpcError<TransportErrorKind>),
 
-    /// The RPC node returned no block for the requested number.
-    #[error("missing block {0}")]
-    MissingBlock(u64),
+    /// The RPC node returned no block for the requested hash.
+    #[error("missing block with hash {0}")]
+    MissingBlockByHash(B256),
 
-    /// Reorg deeper than the block buffer.
-    #[error("reorg depth {depth} exceeds buffer capacity {capacity}")]
-    ReorgTooDeep {
-        /// The detected reorg depth.
-        depth: u64,
-        /// The configured buffer capacity.
-        capacity: usize,
-    },
+    /// The RPC node returned no block for the requested number or tag.
+    #[error("missing block {0}")]
+    MissingBlock(BlockNumberOrTag),
 }

crates/host-rpc/src/lib.rs

@@ -11,6 +11,11 @@
 #![deny(unused_must_use, rust_2018_idioms)]
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
+/// Default block buffer capacity.
+pub(crate) const DEFAULT_BUFFER_CAPACITY: usize = 64;
+/// Default backfill batch size.
+pub(crate) const DEFAULT_BACKFILL_BATCH_SIZE: u64 = 32;
+
 mod builder;
 pub use builder::RpcHostNotifierBuilder;