add withdrawals_by_block_range

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

@@ -457,6 +457,13 @@ impl<N: ProviderNodeTypes> WithdrawalsProvider for BlockchainProvider<N> {
     ) -> ProviderResult<Option<Withdrawals>> {
         self.consistent_provider()?.withdrawals_by_block(id, timestamp)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        self.consistent_provider()?.withdrawals_by_block_range(range)
+    }
 }
 
 impl<N: ProviderNodeTypes> OmmersProvider for BlockchainProvider<N> {

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

@@ -1151,6 +1151,28 @@ impl<N: ProviderNodeTypes> WithdrawalsProvider for ConsistentProvider<N> {
             },
         )
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        self.get_in_memory_or_storage_by_block_range_while(
+            range,
+            |db_provider, range, _| db_provider.withdrawals_by_block_range(range),
+            |block_state, _| {
+                Some(
+                    block_state
+                        .block_ref()
+                        .recovered_block()
+                        .body()
+                        .withdrawals()
+                        .cloned()
+                        .unwrap_or_default(),
+                )
+            },
+            |_| true,
+        )
+    }
 }
 
 impl<N: ProviderNodeTypes> OmmersProvider for ConsistentProvider<N> {

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

@@ -549,6 +549,24 @@ impl<N: ProviderNodeTypes> WithdrawalsProvider for ProviderFactory<N> {
     ) -> ProviderResult<Option<Withdrawals>> {
         self.provider()?.withdrawals_by_block(id, timestamp)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        self.static_file_provider.get_range_with_static_file_or_database(
+            StaticFileSegment::BlockMeta,
+            *range.start()..*range.end() + 1,
+            |static_file, range, _| {
+                static_file.withdrawals_by_block_range(range.start..=range.end.saturating_sub(1))
+            },
+            |range, _| {
+                self.provider()?
+                    .withdrawals_by_block_range(range.start..=range.end.saturating_sub(1))
+            },
+            |_| true,
+        )
+    }
 }
 
 impl<N: ProviderNodeTypes> OmmersProvider for ProviderFactory<N> {

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

@@ -1593,6 +1593,24 @@ impl<TX: DbTx + 'static, N: NodeTypes<ChainSpec: EthereumHardforks>> Withdrawals
         }
         Ok(None)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        self.static_file_provider.get_range_with_static_file_or_database(
+            StaticFileSegment::BlockMeta,
+            *range.start()..*range.end() + 1,
+            |static_file, range, _| {
+                static_file.withdrawals_by_block_range(range.start..=range.end.saturating_sub(1))
+            },
+            |range, _| {
+                self.cursor_read_collect::<tables::BlockWithdrawals>(range)
+                    .map(|w| w.into_iter().map(|w| w.withdrawals).collect())
+            },
+            |_| true,
+        )
+    }
 }
 
 impl<TX: DbTx + 'static, N: NodeTypesForProvider> OmmersProvider for DatabaseProvider<TX, N> {

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

@@ -367,6 +367,24 @@ impl<N: NodePrimitives> WithdrawalsProvider for StaticFileJarProvider<'_, N> {
         // Only accepts block number queries
         Err(ProviderError::UnsupportedProvider)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        let mut cursor = self.cursor()?;
+        let mut withdrawals_by_block =
+            Vec::with_capacity((range.end() - range.start() + 1) as usize);
+
+        for num in range {
+            if let Some(withdrawals) =
+                cursor.get_one::<WithdrawalsMask>(num.into())?.map(|w| w.withdrawals)
+            {
+                withdrawals_by_block.push(withdrawals)
+            }
+        }
+        Ok(withdrawals_by_block)
+    }
 }
 
 impl<N: FullNodePrimitives<BlockHeader: Value>> OmmersProvider for StaticFileJarProvider<'_, N> {

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

@@ -18,7 +18,7 @@ use reth_db::{
     lockfile::StorageLock,
     static_file::{
         iter_static_files, BlockHashMask, BodyIndicesMask, HeaderMask, HeaderWithHashMask,
-        ReceiptMask, StaticFileCursor, TDWithHashMask, TransactionMask,
+        ReceiptMask, StaticFileCursor, TDWithHashMask, TransactionMask, WithdrawalsMask,
     },
     table::{Decompress, Value},
     tables,
@@ -1696,6 +1696,20 @@ impl<N: NodePrimitives> WithdrawalsProvider for StaticFileProvider<N> {
         // Only accepts block number queries
         Err(ProviderError::UnsupportedProvider)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        self.fetch_range_with_predicate(
+            StaticFileSegment::BlockMeta,
+            *range.start()..*range.end() + 1,
+            |cursor, number| {
+                cursor.get_one::<WithdrawalsMask>(number.into()).map(|w| w.map(|w| w.withdrawals))
+            },
+            |_| true,
+        )
+    }
 }
 
 impl<N: FullNodePrimitives<BlockHeader: Value>> OmmersProvider for StaticFileProvider<N> {

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

@@ -765,6 +765,13 @@ impl WithdrawalsProvider for MockEthProvider {
     ) -> ProviderResult<Option<Withdrawals>> {
         Ok(None)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        Ok(vec![])
+    }
 }
 
 impl OmmersProvider for MockEthProvider {

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

@@ -1,12 +1,12 @@
-use crate::{DBProvider, OmmersProvider, StorageLocation};
-use alloy_consensus::Header;
+use crate::{DBProvider, OmmersProvider, StorageLocation, WithdrawalsProvider};
+use alloy_consensus::{BlockHeader, Header};
 use alloy_primitives::BlockNumber;
 use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
 use reth_db::{
-    cursor::{DbCursorRO, DbCursorRW},
+    cursor::DbCursorRW,
     models::{StoredBlockOmmers, StoredBlockWithdrawals},
     tables,
-    transaction::{DbTx, DbTxMut},
+    transaction::DbTxMut,
     DbTxUnwindExt,
 };
 use reth_primitives::TransactionSigned;
@@ -141,7 +141,8 @@ impl<Provider, T> BlockBodyReader<Provider> for EthStorage<T>
 where
     Provider: DBProvider
         + ChainSpecProvider<ChainSpec: EthereumHardforks>
-        + OmmersProvider<Header = Header>,
+        + OmmersProvider<Header = Header>
+        + WithdrawalsProvider,
     T: SignedTransaction,
 {
     type Block = reth_primitives::Block<T>;
@@ -151,25 +152,28 @@ where
         provider: &Provider,
         inputs: Vec<ReadBodyInput<'_, Self::Block>>,
     ) -> ProviderResult<Vec<<Self::Block as Block>::Body>> {
+        if inputs.is_empty() {
+            return Ok(vec![])
+        }
+
+        let block_range = inputs
+            .first()
+            .and_then(|(first, _)| inputs.last().map(|(last, _)| first.number()..=last.number()))
+            .expect("qed");
+
         // TODO: Ideally storage should hold its own copy of chain spec
         let chain_spec = provider.chain_spec();
 
-        let mut withdrawals_cursor = provider.tx_ref().cursor_read::<tables::BlockWithdrawals>()?;
-
         let mut bodies = Vec::with_capacity(inputs.len());
 
-        for (header, transactions) in inputs {
+        for ((header, transactions), withdrawals) in
+            inputs.into_iter().zip(provider.withdrawals_by_block_range(block_range)?)
+        {
             // If we are past shanghai, then all blocks should have a withdrawal list,
             // even if empty
-            let withdrawals = if chain_spec.is_shanghai_active_at_timestamp(header.timestamp) {
-                withdrawals_cursor
-                    .seek_exact(header.number)?
-                    .map(|(_, w)| w.withdrawals)
-                    .unwrap_or_default()
-                    .into()
-            } else {
-                None
-            };
+            let withdrawals =
+                chain_spec.is_shanghai_active_at_timestamp(header.timestamp).then_some(withdrawals);
+
             let ommers = if chain_spec.final_paris_total_difficulty(header.number).is_some() {
                 Vec::new()
             } else {

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

@@ -538,6 +538,13 @@ impl<C: Send + Sync, N: NodePrimitives> WithdrawalsProvider for NoopProvider<C,
     ) -> ProviderResult<Option<Withdrawals>> {
         Ok(None)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>> {
+        Ok(vec![])
+    }
 }
 
 impl<C: Send + Sync, N: NodePrimitives> OmmersProvider for NoopProvider<C, N> {

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

@@ -1,5 +1,7 @@
 use alloy_eips::{eip4895::Withdrawals, BlockHashOrNumber};
+use alloy_primitives::BlockNumber;
 use reth_storage_errors::provider::ProviderResult;
+use std::ops::RangeInclusive;
 
 ///  Client trait for fetching [`alloy_eips::eip4895::Withdrawal`] related data.
 #[auto_impl::auto_impl(&, Arc)]
@@ -10,4 +12,10 @@ pub trait WithdrawalsProvider: Send + Sync {
         id: BlockHashOrNumber,
         timestamp: u64,
     ) -> ProviderResult<Option<Withdrawals>>;
+
+    /// Returns the withdrawals per block within the requested block range.
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+    ) -> ProviderResult<Vec<Withdrawals>>;
 }