feat: extend BlockBodyIndicesProvider with block_body_indices_range (#13829)

crates/stages/stages/src/stages/bodies.rs

@@ -112,8 +112,7 @@ impl<D: BodyDownloader> BodyStage<D> {
                 // fix the inconsistency right away.
                 if let Some(unwind_to) = unwind_block {
                     let next_tx_num_after_unwind = provider
-                        .tx_ref()
+                        .block_body_indices(unwind_to)?
-                        .get::<tables::BlockBodyIndices>(unwind_to)?
                         .map(|b| b.next_tx_num())
                         .ok_or(ProviderError::BlockBodyIndicesNotFound(unwind_to))?;
 

crates/stages/stages/src/stages/execution.rs

@@ -5,7 +5,6 @@ use alloy_primitives::BlockNumber;
 use num_traits::Zero;
 use reth_config::config::ExecutionConfig;
 use reth_db::{static_file::HeaderMask, tables};
-use reth_db_api::{cursor::DbCursorRO, transaction::DbTx};
 use reth_evm::{
     execute::{BatchExecutor, BlockExecutorProvider},
     metrics::ExecutorMetrics,
@@ -203,12 +202,8 @@ where
         }
 
         // Get next expected receipt number
-        let tx = provider.tx_ref();
+        let next_receipt_num =
-        let next_receipt_num = tx
+            provider.block_body_indices(checkpoint)?.map(|b| b.next_tx_num()).unwrap_or(0);
-            .cursor_read::<tables::BlockBodyIndices>()?
-            .seek_exact(checkpoint)?
-            .map(|(_, value)| value.next_tx_num())
-            .unwrap_or(0);
 
         let static_file_provider = provider.static_file_provider();
 
@@ -237,8 +232,7 @@ where
                 // fix the inconsistency right away.
                 if let Some(unwind_to) = unwind_to {
                     let next_receipt_num_after_unwind = provider
-                        .tx_ref()
+                        .block_body_indices(unwind_to)?
-                        .get::<tables::BlockBodyIndices>(unwind_to)?
                         .map(|b| b.next_tx_num())
                         .ok_or(ProviderError::BlockBodyIndicesNotFound(unwind_to))?;
 
@@ -645,6 +639,7 @@ mod tests {
     use alloy_rlp::Decodable;
     use assert_matches::assert_matches;
     use reth_chainspec::ChainSpecBuilder;
+    use reth_db::transaction::DbTx;
     use reth_db_api::{models::AccountBeforeTx, transaction::DbTxMut};
     use reth_evm::execute::BasicBlockExecutorProvider;
     use reth_evm_ethereum::execute::EthExecutionStrategyFactory;

crates/stages/stages/src/stages/tx_lookup.rs

@@ -5,7 +5,7 @@ use reth_config::config::{EtlConfig, TransactionLookupConfig};
 use reth_db::{table::Value, tables, RawKey, RawValue};
 use reth_db_api::{
     cursor::{DbCursorRO, DbCursorRW},
-    transaction::{DbTx, DbTxMut},
+    transaction::DbTxMut,
 };
 use reth_etl::Collector;
 use reth_primitives::NodePrimitives;
@@ -195,12 +195,16 @@ where
         let tx = provider.tx_ref();
         let (range, unwind_to, _) = input.unwind_block_range_with_threshold(self.chunk_size);
 
-        // Cursors to unwind tx hash to number
+        // Cursor to unwind tx hash to number
-        let mut body_cursor = tx.cursor_read::<tables::BlockBodyIndices>()?;
         let mut tx_hash_number_cursor = tx.cursor_write::<tables::TransactionHashNumbers>()?;
         let static_file_provider = provider.static_file_provider();
-        let mut rev_walker = body_cursor.walk_back(Some(*range.end()))?;
+        let rev_walker = provider
-        while let Some((number, body)) = rev_walker.next().transpose()? {
+            .block_body_indices_range(range.clone())?
+            .into_iter()
+            .zip(range.collect::<Vec<_>>())
+            .rev();
+
+        for (body, number) in rev_walker {
             if number <= unwind_to {
                 break;
             }
@@ -255,6 +259,7 @@ mod tests {
     };
     use alloy_primitives::{BlockNumber, B256};
     use assert_matches::assert_matches;
+    use reth_db::transaction::DbTx;
     use reth_primitives::SealedBlock;
     use reth_provider::{
         providers::StaticFileWriter, BlockBodyIndicesProvider, DatabaseProviderFactory,

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

@@ -472,6 +472,13 @@ impl<N: ProviderNodeTypes> BlockBodyIndicesProvider for BlockchainProvider<N> {
     ) -> ProviderResult<Option<StoredBlockBodyIndices>> {
         self.consistent_provider()?.block_body_indices(number)
     }
+
+    fn block_body_indices_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        self.consistent_provider()?.block_body_indices_range(range)
+    }
 }
 
 impl<N: ProviderNodeTypes> StageCheckpointReader for BlockchainProvider<N> {
@@ -882,10 +889,9 @@ mod tests {
         let static_file_provider = factory.static_file_provider();
 
         // Write transactions to static files with the right `tx_num``
-        let mut bodies_cursor = provider_rw.tx_ref().cursor_read::<tables::BlockBodyIndices>()?;
+        let mut tx_num = provider_rw
-        let mut tx_num = bodies_cursor
+            .block_body_indices(database_blocks.first().as_ref().unwrap().number.saturating_sub(1))?
-            .seek_exact(database_blocks.first().as_ref().unwrap().number.saturating_sub(1))?
+            .map(|indices| indices.next_tx_num())
-            .map(|(_, indices)| indices.next_tx_num())
             .unwrap_or_default();
 
         // Insert blocks into the database

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

@@ -1204,6 +1204,13 @@ impl<N: ProviderNodeTypes> BlockBodyIndicesProvider for ConsistentProvider<N> {
             },
         )
     }
+
+    fn block_body_indices_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        range.map_while(|b| self.block_body_indices(b).transpose()).collect()
+    }
 }
 
 impl<N: ProviderNodeTypes> StageCheckpointReader for ConsistentProvider<N> {

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

@@ -564,6 +564,13 @@ impl<N: ProviderNodeTypes> BlockBodyIndicesProvider for ProviderFactory<N> {
     ) -> ProviderResult<Option<StoredBlockBodyIndices>> {
         self.provider()?.block_body_indices(number)
     }
+
+    fn block_body_indices_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        self.provider()?.block_body_indices_range(range)
+    }
 }
 
 impl<N: ProviderNodeTypes> StageCheckpointReader for ProviderFactory<N> {

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

@@ -618,27 +618,23 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
         let len = range.end().saturating_sub(*range.start()) as usize;
         let mut blocks = Vec::with_capacity(len);
 
-        let headers = headers_range(range)?;
+        let headers = headers_range(range.clone())?;
         let mut tx_cursor = self.tx.cursor_read::<tables::Transactions<TxTy<N>>>()?;
-        let mut block_body_cursor = self.tx.cursor_read::<tables::BlockBodyIndices>()?;
 
-        let mut present_headers = Vec::new();
-        for header in headers {
         // If the body indices are not found, this means that the transactions either do
         // not exist in the database yet, or they do exit but are
         // not indexed. If they exist but are not indexed, we don't
         // have enough information to return the block anyways, so
         // we skip the block.
-            if let Some((_, block_body_indices)) =
+        let present_headers = self
-                block_body_cursor.seek_exact(header.as_ref().number())?
+            .block_body_indices_range(range)?
-            {
+            .into_iter()
-                let tx_range = block_body_indices.tx_num_range();
+            .map(|b| b.tx_num_range())
-                present_headers.push((header, tx_range));
+            .zip(headers)
-            }
+            .collect::<Vec<_>>();
-        }
 
         let mut inputs = Vec::new();
-        for (header, tx_range) in &present_headers {
+        for (tx_range, header) in &present_headers {
             let transactions = if tx_range.is_empty() {
                 Vec::new()
             } else {
@@ -650,7 +646,7 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
 
         let bodies = self.storage.reader().read_block_bodies(self, inputs)?;
 
-        for ((header, tx_range), body) in present_headers.into_iter().zip(bodies) {
+        for ((tx_range, header), body) in present_headers.into_iter().zip(bodies) {
             blocks.push(assemble_block(header, body, tx_range)?);
         }
 
@@ -1476,23 +1472,23 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> TransactionsProvider for Datab
         &self,
         range: impl RangeBounds<BlockNumber>,
     ) -> ProviderResult<Vec<Vec<Self::Transaction>>> {
+        let range = to_range(range);
         let mut tx_cursor = self.tx.cursor_read::<tables::Transactions<Self::Transaction>>()?;
-        let mut results = Vec::new();
+
-        let mut body_cursor = self.tx.cursor_read::<tables::BlockBodyIndices>()?;
+        self.block_body_indices_range(range.start..=range.end.saturating_sub(1))?
-        for entry in body_cursor.walk_range(range)? {
+            .into_iter()
-            let (_, body) = entry?;
+            .map(|body| {
                 let tx_num_range = body.tx_num_range();
                 if tx_num_range.is_empty() {
-                results.push(Vec::new());
+                    Ok(Vec::new())
                 } else {
-                results.push(
+                    Ok(self
-                    self.transactions_by_tx_range_with_cursor(tx_num_range, &mut tx_cursor)?
+                        .transactions_by_tx_range_with_cursor(tx_num_range, &mut tx_cursor)?
                         .into_iter()
-                        .collect(),
+                        .collect())
-                );
                 }
-        }
+            })
-        Ok(results)
+            .collect()
     }
 
     fn transactions_by_tx_range(
@@ -1620,6 +1616,18 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> BlockBodyIndicesProvider
     fn block_body_indices(&self, num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
         Ok(self.tx.get::<tables::BlockBodyIndices>(num)?)
     }
+
+    fn block_body_indices_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        Ok(self
+            .tx_ref()
+            .cursor_read::<tables::BlockBodyIndices>()?
+            .walk_range(range)?
+            .map(|r| r.map(|(_, b)| b))
+            .collect::<Result<_, _>>()?)
+    }
 }
 
 impl<TX: DbTx, N: NodeTypes> StageCheckpointReader for DatabaseProvider<TX, N> {
@@ -1756,13 +1764,31 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
         is_value_known: OriginalValuesKnown,
         write_receipts_to: StorageLocation,
     ) -> ProviderResult<()> {
+        let first_block = execution_outcome.first_block();
+        let block_count = execution_outcome.receipts.len() as u64;
+        let block_range = first_block..=first_block.saturating_add(block_count).saturating_sub(1);
+        let last_block = *block_range.end();
+
         let (plain_state, reverts) =
             execution_outcome.bundle.to_plain_state_and_reverts(is_value_known);
 
-        self.write_state_reverts(reverts, execution_outcome.first_block)?;
+        self.write_state_reverts(reverts, first_block)?;
         self.write_state_changes(plain_state)?;
 
-        let mut bodies_cursor = self.tx.cursor_read::<tables::BlockBodyIndices>()?;
+        // Fetch the first transaction number for each block in the range
+        let block_indices: Vec<_> = self
+            .block_body_indices_range(block_range)?
+            .into_iter()
+            .map(|b| b.first_tx_num)
+            .collect();
+
+        // Ensure all expected blocks are present.
+        if block_indices.len() < block_count as usize {
+            let missing_blocks = block_count - block_indices.len() as u64;
+            return Err(ProviderError::BlockBodyIndicesNotFound(
+                last_block.saturating_sub(missing_blocks - 1),
+            ));
+        }
 
         let has_receipts_pruning = self.prune_modes.has_receipts_pruning() ||
             execution_outcome.receipts.iter().flatten().any(|receipt| receipt.is_none());
@@ -1780,25 +1806,19 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
         // We are writing to static files if requested and if there's no receipt pruning configured
         let mut receipts_static_writer = (write_receipts_to.static_files() &&
             !has_receipts_pruning)
-            .then(|| {
+            .then(|| self.static_file_provider.get_writer(first_block, StaticFileSegment::Receipts))
-                self.static_file_provider
-                    .get_writer(execution_outcome.first_block, StaticFileSegment::Receipts)
-            })
             .transpose()?;
 
-        for (idx, receipts) in execution_outcome.receipts.iter().enumerate() {
+        for (idx, (receipts, first_tx_index)) in
-            let block_number = execution_outcome.first_block + idx as u64;
+            execution_outcome.receipts.iter().zip(block_indices).enumerate()
+        {
+            let block_number = first_block + idx as u64;
 
             // Increment block number for receipts static file writer
             if let Some(writer) = receipts_static_writer.as_mut() {
                 writer.increment_block(block_number)?;
             }
 
-            let first_tx_index = bodies_cursor
-                .seek_exact(block_number)?
-                .map(|(_, indices)| indices.first_tx_num())
-                .ok_or(ProviderError::BlockBodyIndicesNotFound(block_number))?;
-
             for (idx, receipt) in receipts.iter().enumerate() {
                 let receipt_idx = first_tx_index + idx as u64;
                 if let Some(receipt) = receipt {
@@ -2017,11 +2037,11 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
         }
 
         // We are not removing block meta as it is used to get block changesets.
-        let block_bodies = self.get::<tables::BlockBodyIndices>(range.clone())?;
+        let block_bodies = self.block_body_indices_range(range.clone())?;
 
         // get transaction receipts
         let from_transaction_num =
-            block_bodies.first().expect("already checked if there are blocks").1.first_tx_num();
+            block_bodies.first().expect("already checked if there are blocks").first_tx_num();
 
         let storage_range = BlockNumberAddress::range(range.clone());
 
@@ -2113,13 +2133,13 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
         let start_block_number = *range.start();
 
         // We are not removing block meta as it is used to get block changesets.
-        let block_bodies = self.get::<tables::BlockBodyIndices>(range.clone())?;
+        let block_bodies = self.block_body_indices_range(range.clone())?;
 
         // get transaction receipts
         let from_transaction_num =
-            block_bodies.first().expect("already checked if there are blocks").1.first_tx_num();
+            block_bodies.first().expect("already checked if there are blocks").first_tx_num();
         let to_transaction_num =
-            block_bodies.last().expect("already checked if there are blocks").1.last_tx_num();
+            block_bodies.last().expect("already checked if there are blocks").last_tx_num();
 
         let storage_range = BlockNumberAddress::range(range.clone());
 
@@ -2199,7 +2219,7 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
 
         let mut receipts = Vec::with_capacity(block_bodies.len());
         // loop break if we are at the end of the blocks.
-        for (_, block_body) in block_bodies {
+        for block_body in block_bodies {
             let mut block_receipts = Vec::with_capacity(block_body.tx_count as usize);
             for num in block_body.tx_num_range() {
                 if receipts_iter.peek().is_some_and(|(n, _)| *n == num) {
@@ -2920,8 +2940,7 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider + 'static> BlockWrite
 
         // First transaction to be removed
         let unwind_tx_from = self
-            .tx
+            .block_body_indices(block)?
-            .get::<tables::BlockBodyIndices>(block)?
             .map(|b| b.next_tx_num())
             .ok_or(ProviderError::BlockBodyIndicesNotFound(block))?;
 
@@ -2957,8 +2976,7 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider + 'static> BlockWrite
 
         // First transaction to be removed
         let unwind_tx_from = self
-            .tx
+            .block_body_indices(block)?
-            .get::<tables::BlockBodyIndices>(block)?
             .map(|b| b.next_tx_num())
             .ok_or(ProviderError::BlockBodyIndicesNotFound(block))?;
 

crates/storage/provider/src/providers/static_file/jar.rs

@@ -25,7 +25,7 @@ use reth_storage_api::{BlockBodyIndicesProvider, OmmersProvider, WithdrawalsProv
 use reth_storage_errors::provider::{ProviderError, ProviderResult};
 use std::{
     fmt::Debug,
-    ops::{Deref, RangeBounds},
+    ops::{Deref, RangeBounds, RangeInclusive},
     sync::Arc,
 };
 
@@ -386,4 +386,19 @@ impl<N: NodePrimitives> BlockBodyIndicesProvider for StaticFileJarProvider<'_, N
     fn block_body_indices(&self, num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
         self.cursor()?.get_one::<BodyIndicesMask>(num.into())
     }
+
+    fn block_body_indices_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        let mut cursor = self.cursor()?;
+        let mut indices = Vec::with_capacity((range.end() - range.start() + 1) as usize);
+
+        for num in range {
+            if let Some(block) = cursor.get_one::<BodyIndicesMask>(num.into())? {
+                indices.push(block)
+            }
+        }
+        Ok(indices)
+    }
 }

crates/storage/provider/src/providers/static_file/manager.rs

@@ -1729,6 +1729,14 @@ impl<N: NodePrimitives> BlockBodyIndicesProvider for StaticFileProvider<N> {
                 }
             })
     }
+
+    fn block_body_indices_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        // Required data not present in static_files
+        Err(ProviderError::UnsupportedProvider)
+    }
 }
 
 impl<N: NodePrimitives> StatsReader for StaticFileProvider<N> {

crates/storage/provider/src/test_utils/mock.rs

@@ -777,6 +777,12 @@ impl BlockBodyIndicesProvider for MockEthProvider {
     fn block_body_indices(&self, _num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
         Ok(None)
     }
+    fn block_body_indices_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        Ok(vec![])
+    }
 }
 
 impl ChangeSetReader for MockEthProvider {

crates/storage/storage-api/src/block_indices.rs

@@ -1,5 +1,7 @@
+use alloy_primitives::BlockNumber;
 use reth_db_models::StoredBlockBodyIndices;
 use reth_storage_errors::provider::ProviderResult;
+use std::ops::RangeInclusive;
 
 ///  Client trait for fetching block body indices related data.
 #[auto_impl::auto_impl(&, Arc)]
@@ -8,4 +10,10 @@ pub trait BlockBodyIndicesProvider: Send + Sync {
     ///
     /// Returns `None` if block is not found.
     fn block_body_indices(&self, num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>>;
+
+    /// Returns the block body indices within the requested range matching number from storage.
+    fn block_body_indices_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>>;
 }

crates/storage/storage-api/src/noop.rs

@@ -567,4 +567,11 @@ impl<C: Send + Sync, N: Send + Sync> BlockBodyIndicesProvider for NoopProvider<C
     fn block_body_indices(&self, _num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
         Ok(None)
     }
+
+    fn block_body_indices_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
+        Ok(vec![])
+    }
 }