feat(storage): add EitherReader for routing history queries to MDBX or RocksDB

crates/storage/provider/src/providers/database/provider.rs

@@ -16,7 +16,7 @@ use crate::{
     HeaderSyncGapProvider, HistoricalStateProvider, HistoricalStateProviderRef, HistoryWriter,
     LatestStateProvider, LatestStateProviderRef, OriginalValuesKnown, ProviderError,
     PruneCheckpointReader, PruneCheckpointWriter, RawRocksDBBatch, RevertsInit, RocksBatchArg,
-    RocksDBProviderFactory, RocksTxRefArg, StageCheckpointReader, StateProviderBox, StateWriter,
+    RocksDBProviderFactory, StageCheckpointReader, StateProviderBox, StateWriter,
     StaticFileProviderFactory, StatsReader, StorageReader, StorageTrieWriter, TransactionVariant,
     TransactionsProvider, TransactionsProviderExt, TrieWriter,
 };
@@ -884,25 +884,6 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
     pub fn chain_spec(&self) -> &N::ChainSpec {
         &self.chain_spec
     }
-
-    /// Executes a closure with a `RocksDB` transaction for reading.
-    ///
-    /// This helper encapsulates all the cfg-gated `RocksDB` transaction handling for reads.
-    fn with_rocksdb_tx<F, R>(&self, f: F) -> ProviderResult<R>
-    where
-        F: FnOnce(RocksTxRefArg<'_>) -> ProviderResult<R>,
-    {
-        #[cfg(all(unix, feature = "rocksdb"))]
-        let rocksdb = self.rocksdb_provider();
-        #[cfg(all(unix, feature = "rocksdb"))]
-        let rocksdb_tx = rocksdb.tx();
-        #[cfg(all(unix, feature = "rocksdb"))]
-        let rocksdb_tx_ref = &rocksdb_tx;
-        #[cfg(not(all(unix, feature = "rocksdb")))]
-        let rocksdb_tx_ref = ();
-
-        f(rocksdb_tx_ref)
-    }
 }
 
 impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {

crates/storage/provider/src/providers/mod.rs

@@ -16,8 +16,8 @@ pub use static_file::{
 mod state;
 pub use state::{
     historical::{
-        history_info, needs_prev_shard_check, HistoricalStateProvider, HistoricalStateProviderRef,
-        HistoryInfo, LowestAvailableBlocks,
+        compute_history_rank, history_info, needs_prev_shard_check, HistoricalStateProvider,
+        HistoricalStateProviderRef, HistoryInfo, LowestAvailableBlocks,
     },
     latest::{LatestStateProvider, LatestStateProviderRef},
     overlay::{OverlayStateProvider, OverlayStateProviderFactory},

crates/storage/provider/src/providers/rocksdb/invariants.rs

@@ -164,16 +164,7 @@ impl RocksDBProvider {
                 self.prune_transaction_hash_numbers_in_range(provider, 0..=highest_tx)?;
             }
             (None, None) => {
-                // Both MDBX and static files are empty.
-                // If checkpoint says we should have data, that's an inconsistency.
-                if checkpoint > 0 {
-                    tracing::warn!(
-                        target: "reth::providers::rocksdb",
-                        checkpoint,
-                        "Checkpoint set but no transaction data exists, unwind needed"
-                    );
-                    return Ok(Some(0));
-                }
+                // Both MDBX and static files are empty, nothing to check.
             }
         }
 
@@ -263,16 +254,27 @@ impl RocksDBProvider {
                 }
 
                 // Find the max highest_block_number (excluding u64::MAX sentinel) across all
-                // entries
+                // entries. Also track if we found any non-sentinel entries.
                 let mut max_highest_block = 0u64;
+                let mut found_non_sentinel = false;
                 for result in self.iter::<tables::StoragesHistory>()? {
                     let (key, _) = result?;
                     let highest = key.sharded_key.highest_block_number;
-                    if highest != u64::MAX && highest > max_highest_block {
-                        max_highest_block = highest;
+                    if highest != u64::MAX {
+                        found_non_sentinel = true;
+                        if highest > max_highest_block {
+                            max_highest_block = highest;
+                        }
                     }
                 }
 
+                // If all entries are sentinel entries (u64::MAX), treat as first-run scenario.
+                // This means no completed shards exist (only sentinel shards with
+                // highest_block_number=u64::MAX), so no actual history has been indexed.
+                if !found_non_sentinel {
+                    return Ok(None);
+                }
+
                 // If any entry has highest_block > checkpoint, prune excess
                 if max_highest_block > checkpoint {
                     tracing::info!(
@@ -296,11 +298,7 @@ impl RocksDBProvider {
                 Ok(None)
             }
             None => {
-                // Empty RocksDB table
-                if checkpoint > 0 {
-                    // Stage says we should have data but we don't
-                    return Ok(Some(0));
-                }
+                // Empty RocksDB table, nothing to check.
                 Ok(None)
             }
         }
@@ -377,16 +375,27 @@ impl RocksDBProvider {
                 }
 
                 // Find the max highest_block_number (excluding u64::MAX sentinel) across all
-                // entries
+                // entries. Also track if we found any non-sentinel entries.
                 let mut max_highest_block = 0u64;
+                let mut found_non_sentinel = false;
                 for result in self.iter::<tables::AccountsHistory>()? {
                     let (key, _) = result?;
                     let highest = key.highest_block_number;
-                    if highest != u64::MAX && highest > max_highest_block {
-                        max_highest_block = highest;
+                    if highest != u64::MAX {
+                        found_non_sentinel = true;
+                        if highest > max_highest_block {
+                            max_highest_block = highest;
+                        }
                     }
                 }
 
+                // If all entries are sentinel entries (u64::MAX), treat as first-run scenario.
+                // This means no completed shards exist (only sentinel shards with
+                // highest_block_number=u64::MAX), so no actual history has been indexed.
+                if !found_non_sentinel {
+                    return Ok(None);
+                }
+
                 // If any entry has highest_block > checkpoint, prune excess
                 if max_highest_block > checkpoint {
                     tracing::info!(
@@ -413,11 +422,7 @@ impl RocksDBProvider {
                 Ok(None)
             }
             None => {
-                // Empty RocksDB table
-                if checkpoint > 0 {
-                    // Stage says we should have data but we don't
-                    return Ok(Some(0));
-                }
+                // Empty RocksDB table, nothing to check.
                 Ok(None)
             }
         }
@@ -542,7 +547,7 @@ mod tests {
     }
 
     #[test]
-    fn test_check_consistency_empty_rocksdb_with_checkpoint_needs_unwind() {
+    fn test_check_consistency_empty_rocksdb_with_checkpoint_is_first_run() {
         let temp_dir = TempDir::new().unwrap();
         let rocksdb = RocksDBBuilder::new(temp_dir.path())
             .with_table::<tables::TransactionHashNumbers>()
@@ -566,10 +571,10 @@ mod tests {
 
         let provider = factory.database_provider_ro().unwrap();
 
-        // RocksDB is empty but checkpoint says block 100 was processed
-        // This means RocksDB is missing data and we need to unwind to rebuild
+        // RocksDB is empty but checkpoint says block 100 was processed.
+        // This is treated as a first-run/migration scenario - no unwind needed.
         let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(result, Some(0), "Should require unwind to block 0 to rebuild RocksDB");
+        assert_eq!(result, None, "Empty data with checkpoint is treated as first run");
     }
 
     #[test]
@@ -650,7 +655,7 @@ mod tests {
     }
 
     #[test]
-    fn test_check_consistency_storages_history_empty_with_checkpoint_needs_unwind() {
+    fn test_check_consistency_storages_history_empty_with_checkpoint_is_first_run() {
         let temp_dir = TempDir::new().unwrap();
         let rocksdb = RocksDBBuilder::new(temp_dir.path())
             .with_table::<tables::StoragesHistory>()
@@ -674,9 +679,10 @@ mod tests {
 
         let provider = factory.database_provider_ro().unwrap();
 
-        // RocksDB is empty but checkpoint says block 100 was processed
+        // RocksDB is empty but checkpoint says block 100 was processed.
+        // This is treated as a first-run/migration scenario - no unwind needed.
         let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(result, Some(0), "Should require unwind to block 0 to rebuild StoragesHistory");
+        assert_eq!(result, None, "Empty RocksDB with checkpoint is treated as first run");
     }
 
     #[test]
@@ -978,6 +984,97 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_check_consistency_storages_history_sentinel_only_with_checkpoint_is_first_run() {
+        let temp_dir = TempDir::new().unwrap();
+        let rocksdb = RocksDBBuilder::new(temp_dir.path())
+            .with_table::<tables::StoragesHistory>()
+            .build()
+            .unwrap();
+
+        // Insert ONLY sentinel entries (highest_block_number = u64::MAX)
+        // This simulates a scenario where history tracking started but no shards were completed
+        let key_sentinel_1 = StorageShardedKey::new(Address::ZERO, B256::ZERO, u64::MAX);
+        let key_sentinel_2 = StorageShardedKey::new(Address::random(), B256::random(), u64::MAX);
+        let block_list = BlockNumberList::new_pre_sorted([10, 20, 30]);
+        rocksdb.put::<tables::StoragesHistory>(key_sentinel_1, &block_list).unwrap();
+        rocksdb.put::<tables::StoragesHistory>(key_sentinel_2, &block_list).unwrap();
+
+        // Verify entries exist (not empty table)
+        assert!(rocksdb.first::<tables::StoragesHistory>().unwrap().is_some());
+
+        // Create a test provider factory for MDBX
+        let factory = create_test_provider_factory();
+        factory.set_storage_settings_cache(
+            StorageSettings::legacy().with_storages_history_in_rocksdb(true),
+        );
+
+        // Set a checkpoint indicating we should have processed up to block 100
+        {
+            let provider = factory.database_provider_rw().unwrap();
+            provider
+                .save_stage_checkpoint(StageId::IndexStorageHistory, StageCheckpoint::new(100))
+                .unwrap();
+            provider.commit().unwrap();
+        }
+
+        let provider = factory.database_provider_ro().unwrap();
+
+        // RocksDB has only sentinel entries (no completed shards) but checkpoint is set.
+        // This is treated as a first-run/migration scenario - no unwind needed.
+        let result = rocksdb.check_consistency(&provider).unwrap();
+        assert_eq!(
+            result, None,
+            "Sentinel-only entries with checkpoint should be treated as first run"
+        );
+    }
+
+    #[test]
+    fn test_check_consistency_accounts_history_sentinel_only_with_checkpoint_is_first_run() {
+        use reth_db_api::models::ShardedKey;
+
+        let temp_dir = TempDir::new().unwrap();
+        let rocksdb = RocksDBBuilder::new(temp_dir.path())
+            .with_table::<tables::AccountsHistory>()
+            .build()
+            .unwrap();
+
+        // Insert ONLY sentinel entries (highest_block_number = u64::MAX)
+        let key_sentinel_1 = ShardedKey::new(Address::ZERO, u64::MAX);
+        let key_sentinel_2 = ShardedKey::new(Address::random(), u64::MAX);
+        let block_list = BlockNumberList::new_pre_sorted([10, 20, 30]);
+        rocksdb.put::<tables::AccountsHistory>(key_sentinel_1, &block_list).unwrap();
+        rocksdb.put::<tables::AccountsHistory>(key_sentinel_2, &block_list).unwrap();
+
+        // Verify entries exist (not empty table)
+        assert!(rocksdb.first::<tables::AccountsHistory>().unwrap().is_some());
+
+        // Create a test provider factory for MDBX
+        let factory = create_test_provider_factory();
+        factory.set_storage_settings_cache(
+            StorageSettings::legacy().with_account_history_in_rocksdb(true),
+        );
+
+        // Set a checkpoint indicating we should have processed up to block 100
+        {
+            let provider = factory.database_provider_rw().unwrap();
+            provider
+                .save_stage_checkpoint(StageId::IndexAccountHistory, StageCheckpoint::new(100))
+                .unwrap();
+            provider.commit().unwrap();
+        }
+
+        let provider = factory.database_provider_ro().unwrap();
+
+        // RocksDB has only sentinel entries (no completed shards) but checkpoint is set.
+        // This is treated as a first-run/migration scenario - no unwind needed.
+        let result = rocksdb.check_consistency(&provider).unwrap();
+        assert_eq!(
+            result, None,
+            "Sentinel-only entries with checkpoint should be treated as first run"
+        );
+    }
+
     #[test]
     fn test_check_consistency_storages_history_behind_checkpoint_single_entry() {
         use reth_db_api::models::storage_sharded_key::StorageShardedKey;
@@ -1135,7 +1232,7 @@ mod tests {
     }
 
     #[test]
-    fn test_check_consistency_accounts_history_empty_with_checkpoint_needs_unwind() {
+    fn test_check_consistency_accounts_history_empty_with_checkpoint_is_first_run() {
         let temp_dir = TempDir::new().unwrap();
         let rocksdb = RocksDBBuilder::new(temp_dir.path())
             .with_table::<tables::AccountsHistory>()
@@ -1159,9 +1256,10 @@ mod tests {
 
         let provider = factory.database_provider_ro().unwrap();
 
-        // RocksDB is empty but checkpoint says block 100 was processed
+        // RocksDB is empty but checkpoint says block 100 was processed.
+        // This is treated as a first-run/migration scenario - no unwind needed.
         let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(result, Some(0), "Should require unwind to block 0 to rebuild AccountsHistory");
+        assert_eq!(result, None, "Empty RocksDB with checkpoint is treated as first run");
     }
 
     #[test]