feat: add WithdrawalsProvider::withdrawals_by_block_range

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

@@ -457,6 +457,14 @@ 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>,
+        timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<Withdrawals>>> {
+        self.consistent_provider()?.withdrawals_by_block_range(range, timestamps)
+    }
 }
 
 impl<N: ProviderNodeTypes> OmmersProvider for BlockchainProvider<N> {

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

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

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

@@ -549,6 +549,30 @@ 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>,
+        timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<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),
+                    timestamps,
+                )
+            },
+            |range, _| {
+                self.provider()?.withdrawals_by_block_range(
+                    range.start..=range.end.saturating_sub(1),
+                    timestamps,
+                )
+            },
+            |_| true,
+        )
+    }
 }
 
 impl<N: ProviderNodeTypes> OmmersProvider for ProviderFactory<N> {

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

@@ -1593,6 +1593,58 @@ impl<TX: DbTx + 'static, N: NodeTypes<ChainSpec: EthereumHardforks>> Withdrawals
         }
         Ok(None)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+        timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<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),
+                    timestamps,
+                )
+            },
+            |range, _| {
+                let chain_spec = self.chain_spec();
+                let first_block = range.start;
+
+                let capacity = if range.start >= range.end {
+                    return Ok(Vec::new())
+                } else {
+                    (range.end - range.start) as usize
+                };
+
+                let mut withdrawals_cursor = self.tx.cursor_read::<tables::BlockWithdrawals>()?;
+                let mut withdrawals = Vec::with_capacity(capacity);
+
+                for (number, (_, timestamp)) in
+                    range.zip(timestamps.iter().skip_while(|(block, _)| *block < first_block))
+                {
+                    // If we are past shanghai, then all blocks should have a withdrawal list,
+                    // even if empty
+                    let block_withdrawals =
+                        if chain_spec.is_shanghai_active_at_timestamp(*timestamp) {
+                            withdrawals_cursor
+                                .seek_exact(number)?
+                                .map(|(_, w)| w.withdrawals)
+                                .unwrap_or_default()
+                                .into()
+                        } else {
+                            None
+                        };
+
+                    withdrawals.push(block_withdrawals);
+                }
+
+                Ok(withdrawals)
+            },
+            |_| true,
+        )
+    }
 }
 
 impl<TX: DbTx + 'static, N: NodeTypesForProvider> OmmersProvider for DatabaseProvider<TX, N> {

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

@@ -370,6 +370,27 @@ impl<N: NodePrimitives> WithdrawalsProvider for StaticFileJarProvider<'_, N> {
         // Only accepts block number queries
         Err(ProviderError::UnsupportedProvider)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+        _timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<Withdrawals>>> {
+        // Empty withdrawals post shangai are stored as Some([]), while pre shangai are stored as
+        // None
+        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,23 @@ impl<N: NodePrimitives> WithdrawalsProvider for StaticFileProvider<N> {
         // Only accepts block number queries
         Err(ProviderError::UnsupportedProvider)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+        _timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<Withdrawals>>> {
+        // Empty withdrawals post shangai are stored as Some([]), while pre shangai are stored as
+        // None
+        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,14 @@ impl WithdrawalsProvider for MockEthProvider {
     ) -> ProviderResult<Option<Withdrawals>> {
         Ok(None)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+        _timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<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,24 @@ where
         provider: &Provider,
         inputs: Vec<ReadBodyInput<'_, Self::Block>>,
     ) -> ProviderResult<Vec<<Self::Block as Block>::Body>> {
+        if inputs.is_empty() {
+            return Ok(vec![])
+        }
         // 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 block_range = inputs
+            .first()
+            .and_then(|(first, _)| inputs.last().map(|(last, _)| first.number()..=last.number()))
+            .expect("qed");
+        let timestamps =
+            inputs.iter().map(|(h, _)| (h.number(), h.timestamp())).collect::<Vec<_>>();
 
         let mut bodies = Vec::with_capacity(inputs.len());
 
-        for (header, transactions) in inputs {
-            // 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
-            };
+        for ((header, transactions), withdrawals) in
+            inputs.into_iter().zip(provider.withdrawals_by_block_range(block_range, &timestamps)?)
+        {
             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,14 @@ impl<C: Send + Sync, N: NodePrimitives> WithdrawalsProvider for NoopProvider<C,
     ) -> ProviderResult<Option<Withdrawals>> {
         Ok(None)
     }
+
+    fn withdrawals_by_block_range(
+        &self,
+        _range: RangeInclusive<BlockNumber>,
+        _timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<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,14 @@ pub trait WithdrawalsProvider: Send + Sync {
         id: BlockHashOrNumber,
         timestamp: u64,
     ) -> ProviderResult<Option<Withdrawals>>;
+
+    /// Returns the withdrawals per block within the requested block range.
+    ///
+    /// If it's dealing with a pre-shangai block, the withdrawal element will be `None`, otherwise
+    /// `Some`.
+    fn withdrawals_by_block_range(
+        &self,
+        range: RangeInclusive<BlockNumber>,
+        timestamps: &[(BlockNumber, u64)],
+    ) -> ProviderResult<Vec<Option<Withdrawals>>>;
 }