refactor: unified BlockStore for hash indexing and block caching

src/node/network/mod.rs

@@ -28,7 +28,7 @@ use reth_network_api::PeersInfo;
 use reth_payload_primitives::EngineApiMessageVersion;
 use reth_provider::StageCheckpointReader;
 use reth_stages_types::StageId;
-use reth_storage_api::{BlockHashReader, BlockNumReader};
+use reth_storage_api::BlockNumReader;
 use std::{
     net::{Ipv4Addr, SocketAddr},
     sync::Arc,
@@ -188,21 +188,18 @@ impl HlNetworkBuilder {
             // the pseudo peer's NewBlock announcements via the network layer don't
             // reliably generate forkchoice updates on this post-merge chain.
             if let Ok(target_hash) = fcu_trigger_rx.await {
-                let finalized_hash = consensus
-                    .provider
-                    .best_block_number()
-                    .ok()
-                    .and_then(|n| consensus.provider.block_hash(n).ok().flatten())
-                    .unwrap_or(target_hash);
+                // Use the target hash as finalized so reth's backfill_sync_target()
+                // sees an unknown finalized hash and triggers the pipeline.
+                // If finalized is set to the current best block (which is already
+                // known), reth concludes "fully synced" and never starts backfill.
                 let state = ForkchoiceState {
                     head_block_hash: target_hash,
-                    safe_block_hash: finalized_hash,
-                    finalized_block_hash: finalized_hash,
+                    safe_block_hash: target_hash,
+                    finalized_block_hash: target_hash,
                 };
                 info!(
                     target: "reth::cli",
                     head = %target_hash,
-                    finalized = %finalized_hash,
                     "Sending initial forkchoice update to trigger pipeline"
                 );
                 let _ = engine
@@ -270,28 +267,31 @@ where
                 .block_number +
                 1;
 
-            // Give the pseudo-peer a handle to the node's database so it can
+            // Give the block store a handle to the node's database so it can
             // resolve hash→number directly instead of scanning the block source.
             let provider = ctx.provider().clone();
             let db_block_number: DbBlockNumberFn = Arc::new(move |hash| {
                 provider.block_number(hash).ok().flatten()
             });
 
             let chain_spec = ctx.chain_spec();
-            let chain_id = chain_spec.inner.chain().id();
             ctx.task_executor().spawn_critical("pseudo peer", async move {
-                let block_source = block_source_config
-                    .create_cached_block_source((*chain_spec).clone(), next_block_number)
+                let block_store = block_source_config
+                    .create_block_store(
+                        (*chain_spec).clone(),
+                        next_block_number,
+                        Some(db_block_number),
+                    )
                     .await;
 
                 // Read the latest block and send its hash to trigger the pipeline
                 // via a direct forkchoice update. This must happen before
                 // start_pseudo_peer (which never returns).
-                if let Some(latest) = block_source.find_latest_block_number().await {
-                    match block_source.collect_block(latest).await {
+                // get_by_number auto-indexes the block's hash AND parent hash.
+                if let Some(latest) = block_store.find_latest_block_number().await {
+                    match block_store.get_by_number(latest).await {
                         Ok(block) => {
-                            let reth_block = block.to_reth_block(chain_id);
-                            let hash = alloy_primitives::Sealable::hash_slow(&reth_block.header);
+                            let hash = block.hash();
                             info!(
                                 target: "reth::cli",
                                 number = %latest,
@@ -313,9 +313,8 @@ where
                 start_pseudo_peer(
                     chain_spec.clone(),
                     local_node_record.to_string(),
-                    block_source,
+                    block_store,
                     debug_cutoff_height,
-                    Some(db_block_number),
                 )
                 .await
                 .unwrap();

src/node/types/mod.rs

@@ -158,6 +158,11 @@ impl BlockAndReceipts {
         let EvmBlock::Reth115(block) = &self.block;
         block.header.header.number
     }
+
+    pub fn parent_hash(&self) -> B256 {
+        let EvmBlock::Reth115(block) = &self.block;
+        block.header.header.parent_hash
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]

src/pseudo_peer/block_store.rs

@@ -0,0 +1,147 @@
+use super::{sources::BlockSourceBoxed, utils::LruBiMap};
+use crate::node::types::BlockAndReceipts;
+use alloy_primitives::B256;
+use futures::future::BoxFuture;
+use parking_lot::RwLock;
+use reth_network::cache::LruMap;
+use std::{
+    collections::HashMap,
+    sync::Arc,
+    time::Duration,
+};
+
+/// Function that resolves a block hash to its number via the node's database.
+/// Returns `None` if the hash is not yet in the database (e.g. headers not synced yet).
+/// Note: This queries the HeaderNumbers table which covers both database and static files.
+pub type DbBlockNumberFn = Arc<dyn Fn(B256) -> Option<u64> + Send + Sync>;
+
+const BLOCK_CACHE_LIMIT: u32 = 100_000;
+const HASH_INDEX_LIMIT: u32 = 1_000_000;
+
+/// Unified block store that combines block content caching, hash↔number indexing,
+/// and database fallback into a single abstraction.
+///
+/// Every block that passes through the store has its hash (and parent hash)
+/// automatically indexed, eliminating scattered cache population.
+pub struct BlockStore {
+    /// Block content cache: number → block
+    blocks: RwLock<LruMap<u64, BlockAndReceipts>>,
+    /// Hash index: hash ↔ number (bidirectional)
+    hash_index: RwLock<LruBiMap<B256, u64>>,
+    /// DB fallback for hash→number (HeaderNumbers table)
+    db_block_number: Option<DbBlockNumberFn>,
+    /// Underlying fetch source (S3, RPC, etc.)
+    source: BlockSourceBoxed,
+}
+
+impl std::fmt::Debug for BlockStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BlockStore").finish_non_exhaustive()
+    }
+}
+
+impl BlockStore {
+    pub fn new(source: BlockSourceBoxed, db_block_number: Option<DbBlockNumberFn>) -> Self {
+        Self {
+            blocks: RwLock::new(LruMap::new(BLOCK_CACHE_LIMIT)),
+            hash_index: RwLock::new(LruBiMap::new(HASH_INDEX_LIMIT)),
+            db_block_number,
+            source,
+        }
+    }
+
+    /// Index a block's hash and parent hash in the hash↔number map.
+    pub fn index_block(&self, block: &BlockAndReceipts) {
+        let mut idx = self.hash_index.write();
+        Self::index_block_inner(&mut idx, block);
+    }
+
+    /// Index a block into an already-held write guard. Avoids repeated lock acquisition.
+    fn index_block_inner(idx: &mut LruBiMap<B256, u64>, block: &BlockAndReceipts) {
+        let number = block.number();
+        idx.insert(block.hash(), number);
+        if number > 0 {
+            idx.insert(block.parent_hash(), number - 1);
+        }
+    }
+
+    /// Fetch a single block by number. Auto-indexes and caches.
+    pub async fn get_by_number(&self, n: u64) -> eyre::Result<BlockAndReceipts> {
+        if let Some(block) = self.blocks.write().get(&n) {
+            return Ok(block.clone());
+        }
+        let block = self.source.collect_block(n).await?;
+        self.blocks.write().insert(n, block.clone());
+        self.index_block(&block);
+        Ok(block)
+    }
+
+    /// Fetch multiple blocks by number. Auto-indexes and caches.
+    pub async fn get_by_numbers(
+        &self,
+        heights: Vec<u64>,
+    ) -> eyre::Result<Vec<BlockAndReceipts>> {
+        let mut cached: HashMap<u64, BlockAndReceipts> = HashMap::new();
+        let mut uncached_heights = Vec::new();
+        {
+            let mut c = self.blocks.write();
+            for &h in &heights {
+                if let Some(block) = c.get(&h) {
+                    cached.insert(h, block.clone());
+                } else {
+                    uncached_heights.push(h);
+                }
+            }
+        }
+
+        if !uncached_heights.is_empty() {
+            let fetched = self.source.collect_blocks(uncached_heights).await?;
+            let mut c = self.blocks.write();
+            for block in fetched {
+                let h = block.number();
+                c.insert(h, block.clone());
+                cached.insert(h, block);
+            }
+        }
+
+        // Batch-index all blocks under a single write lock
+        {
+            let mut idx = self.hash_index.write();
+            for block in cached.values() {
+                Self::index_block_inner(&mut idx, block);
+            }
+        }
+
+        heights
+            .iter()
+            .map(|h| cached.remove(h).ok_or_else(|| eyre::eyre!("Block {h} not found")))
+            .collect()
+    }
+
+    /// Resolve a block hash to a block number.
+    /// Checks the in-memory index first, then falls back to the database.
+    pub fn hash_to_number(&self, hash: B256) -> eyre::Result<u64> {
+        // Fast path: in-memory index
+        if let Some(n) = self.hash_index.read().get_by_left(&hash).copied() {
+            return Ok(n);
+        }
+
+        // Fallback: database lookup (MDBX is mmap'd, no need to re-cache)
+        if let Some(ref db_fn) = self.db_block_number
+            && let Some(n) = db_fn(hash) {
+                return Ok(n);
+            }
+
+        Err(eyre::eyre!("Hash not found in index or database: {hash:?}"))
+    }
+
+    // --- Delegated block source methods ---
+
+    pub fn find_latest_block_number(&self) -> BoxFuture<'static, Option<u64>> {
+        self.source.find_latest_block_number()
+    }
+
+    pub fn polling_interval(&self) -> Duration {
+        self.source.polling_interval()
+    }
+}

src/pseudo_peer/config.rs

@@ -1,8 +1,11 @@
 use crate::chainspec::HlChainSpec;
 
-use super::sources::{
-    BlockSourceBoxed, CachedBlockSource, HlNodeBlockSource, HlNodeBlockSourceArgs,
-    LocalBlockSource, RpcBlockSource, S3BlockSource,
+use super::{
+    block_store::{BlockStore, DbBlockNumberFn},
+    sources::{
+        BlockSourceBoxed, HlNodeBlockSource, HlNodeBlockSourceArgs,
+        LocalBlockSource, RpcBlockSource, S3BlockSource,
+    },
 };
 use aws_config::BehaviorVersion;
 use std::{env::home_dir, path::PathBuf, sync::Arc, time::Duration};
@@ -104,15 +107,16 @@ impl BlockSourceConfig {
         ))
     }
 
-    pub async fn create_cached_block_source(
+    pub async fn create_block_store(
         &self,
         chain_spec: HlChainSpec,
         next_block_number: u64,
-    ) -> BlockSourceBoxed {
+        db_block_number: Option<DbBlockNumberFn>,
+    ) -> Arc<BlockStore> {
         let block_source = self.create_block_source(chain_spec).await;
         let block_source =
             self.create_block_source_from_node(next_block_number, block_source).await;
-        Arc::new(Box::new(CachedBlockSource::new(block_source)))
+        Arc::new(BlockStore::new(block_source, db_block_number))
     }
 }
 

src/pseudo_peer/mod.rs

@@ -3,6 +3,7 @@
 //! This library exposes `start_pseudo_peer` to support reth-side NetworkState/StateFetcher
 //! to fetch blocks and feed it to its stages
 
+pub mod block_store;
 pub mod cli;
 pub mod config;
 pub mod network;
@@ -14,6 +15,7 @@ use std::sync::Arc;
 use tokio::sync::mpsc;
 use tracing::{error, info};
 
+pub use block_store::*;
 pub use cli::*;
 pub use config::*;
 pub use network::*;
@@ -23,11 +25,10 @@ pub use sources::*;
 /// Re-export commonly used types
 pub mod prelude {
     pub use super::{
+        block_store::BlockStore,
         config::BlockSourceConfig,
         service::{BlockPoller, PseudoPeer},
-        sources::{
-            BlockSource, CachedBlockSource, LocalBlockSource, RpcBlockSource, S3BlockSource,
-        },
+        sources::{BlockSource, LocalBlockSource, RpcBlockSource, S3BlockSource},
     };
 }
 
@@ -41,21 +42,18 @@ use std::str::FromStr;
 pub async fn start_pseudo_peer(
     chain_spec: Arc<HlChainSpec>,
     destination_peer: String,
-    block_source: BlockSourceBoxed,
+    block_store: Arc<BlockStore>,
     debug_cutoff_height: Option<u64>,
-    db_block_number: Option<DbBlockNumberFn>,
 ) -> eyre::Result<()> {
-    let blockhash_cache = new_blockhash_cache();
 
     // Parse the destination peer enode string
     let node_record = NodeRecord::from_str(&destination_peer)
         .map_err(|e| eyre::eyre!("Failed to parse destination peer: {e}"))?;
 
     // Create network manager (no boot_nodes — we add the peer directly)
-    let (mut network, start_tx) = create_network_manager::<BlockSourceBoxed>(
+    let (mut network, start_tx) = create_network_manager(
         (*chain_spec).clone(),
-        block_source.clone(),
-        blockhash_cache.clone(),
+        block_store.clone(),
         debug_cutoff_height,
     )
     .await?;
@@ -71,8 +69,7 @@ pub async fn start_pseudo_peer(
     let mut network_events = network_handle.event_listener();
     info!("Starting network manager...");
 
-    let mut service =
-        PseudoPeer::new(chain_spec, block_source, blockhash_cache.clone(), db_block_number);
+    let mut service = PseudoPeer::new(chain_spec, block_store);
     tokio::spawn(network);
 
     // Directly add the main node as a peer (bypasses discovery)