diff --git a/crates/interfaces/src/provider.rs b/crates/interfaces/src/provider.rs
index c8ca9f1d86..6ef7517d00 100644
--- a/crates/interfaces/src/provider.rs
+++ b/crates/interfaces/src/provider.rs
@@ -35,6 +35,9 @@ pub enum ProviderError {
     /// Thrown when required header related data was not found but was required.
     #[error("No header found for {0:?}")]
     HeaderNotFound(BlockHashOrNumber),
+    /// Thrown we were unable to find a specific block
+    #[error("Block does not exist {0:?}")]
+    BlockNotFound(BlockHashOrNumber),
     /// Thrown we were unable to find the best block
     #[error("Best block does not exist")]
     BestBlockNotFound,
diff --git a/crates/stages/src/stages/execution.rs b/crates/stages/src/stages/execution.rs
index 1759da6997..e57c944aea 100644
--- a/crates/stages/src/stages/execution.rs
+++ b/crates/stages/src/stages/execution.rs
@@ -16,11 +16,11 @@ use reth_primitives::{
     stage::{
         CheckpointBlockRange, EntitiesCheckpoint, ExecutionCheckpoint, StageCheckpoint, StageId,
     },
-    Block, BlockNumber, BlockWithSenders, Header, TransactionSigned, U256,
+    BlockNumber, Header, U256,
 };
 use reth_provider::{
     post_state::PostState, BlockExecutor, BlockProvider, DatabaseProviderRW, ExecutorFactory,
-    HeaderProvider, LatestStateProviderRef, ProviderError, WithdrawalsProvider,
+    HeaderProvider, LatestStateProviderRef, ProviderError,
 };
 use std::{ops::RangeInclusive, time::Instant};
 use tracing::*;
@@ -84,59 +84,6 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
         Self::new(executor_factory, ExecutionStageThresholds::default())
     }
 
-    // TODO(joshie): This should be in the block provider trait once we consolidate
-    fn read_block_with_senders<DB: Database>(
-        provider: &DatabaseProviderRW<'_, &DB>,
-        block_number: BlockNumber,
-    ) -> Result<(BlockWithSenders, U256), StageError> {
-        let header = provider
-            .header_by_number(block_number)?
-            .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;
-        let td = provider
-            .header_td_by_number(block_number)?
-            .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;
-        let ommers = provider.ommers(block_number.into())?.unwrap_or_default();
-        let withdrawals = provider.withdrawals_by_block(block_number.into(), header.timestamp)?;
-
-        // Get the block body
-        let body = provider.block_body_indices(block_number)?;
-        let tx_range = body.tx_num_range();
-
-        // Get the transactions in the body
-        let tx = provider.tx_ref();
-        let (transactions, senders) = if tx_range.is_empty() {
-            (Vec::new(), Vec::new())
-        } else {
-            let transactions = tx
-                .cursor_read::<tables::Transactions>()?
-                .walk_range(tx_range.clone())?
-                .map(|entry| entry.map(|tx| tx.1))
-                .collect::<Result<Vec<_>, _>>()?;
-
-            let senders = tx
-                .cursor_read::<tables::TxSenders>()?
-                .walk_range(tx_range)?
-                .map(|entry| entry.map(|sender| sender.1))
-                .collect::<Result<Vec<_>, _>>()?;
-
-            (transactions, senders)
-        };
-
-        let body = transactions
-            .into_iter()
-            .map(|tx| {
-                TransactionSigned {
-                    // TODO: This is the fastest way right now to make everything just work with
-                    // a dummy transaction hash.
-                    hash: Default::default(),
-                    signature: tx.signature,
-                    transaction: tx.transaction,
-                }
-            })
-            .collect();
-        Ok((Block { header, body, ommers, withdrawals }.with_senders(senders), td))
-    }
-
     /// Execute the stage.
     pub fn execute_inner<DB: Database>(
         &self,
@@ -162,7 +109,12 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
         // Execute block range
         let mut state = PostState::default();
         for block_number in start_block..=max_block {
-            let (block, td) = Self::read_block_with_senders(provider, block_number)?;
+            let td = provider
+                .header_td_by_number(block_number)?
+                .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;
+            let block = provider
+                .block_with_senders(block_number)?
+                .ok_or_else(|| ProviderError::BlockNotFound(block_number.into()))?;
 
             // Configure the executor to use the current state.
             trace!(target: "sync::stages::execution", number = block_number, txs = block.body.len(), "Executing block");
diff --git a/crates/storage/provider/src/providers/database/mod.rs b/crates/storage/provider/src/providers/database/mod.rs
index 30aa602986..c4370e7911 100644
--- a/crates/storage/provider/src/providers/database/mod.rs
+++ b/crates/storage/provider/src/providers/database/mod.rs
@@ -9,9 +9,9 @@ use reth_db::{database::Database, models::StoredBlockBodyIndices, tables, transa
 use reth_interfaces::Result;
 use reth_primitives::{
     stage::{StageCheckpoint, StageId},
-    Block, BlockHash, BlockHashOrNumber, BlockNumber, ChainInfo, ChainSpec, Header, Receipt,
-    SealedBlock, SealedHeader, TransactionMeta, TransactionSigned, TxHash, TxNumber, Withdrawal,
-    H256, U256,
+    Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders, ChainInfo,
+    ChainSpec, Header, Receipt, SealedBlock, SealedHeader, TransactionMeta, TransactionSigned,
+    TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
 };
 use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
 use std::{ops::RangeBounds, sync::Arc};
@@ -189,8 +189,12 @@ impl<DB: Database> BlockProvider for ProviderFactory<DB> {
         self.provider()?.ommers(id)
     }
 
-    fn block_body_indices(&self, num: u64) -> Result<Option<StoredBlockBodyIndices>> {
-        self.provider()?.block_body_indices(num)
+    fn block_body_indices(&self, number: BlockNumber) -> Result<Option<StoredBlockBodyIndices>> {
+        self.provider()?.block_body_indices(number)
+    }
+
+    fn block_with_senders(&self, number: BlockNumber) -> Result<Option<BlockWithSenders>> {
+        self.provider()?.block_with_senders(number)
     }
 }
 
@@ -231,6 +235,17 @@ impl<DB: Database> TransactionsProvider for ProviderFactory<DB> {
     ) -> Result<Vec<Vec<TransactionSigned>>> {
         self.provider()?.transactions_by_block_range(range)
     }
+
+    fn transactions_by_tx_range(
+        &self,
+        range: impl RangeBounds<TxNumber>,
+    ) -> Result<Vec<TransactionSignedNoHash>> {
+        self.provider()?.transactions_by_tx_range(range)
+    }
+
+    fn senders_by_tx_range(&self, range: impl RangeBounds<TxNumber>) -> Result<Vec<Address>> {
+        self.provider()?.senders_by_tx_range(range)
+    }
 }
 
 impl<DB: Database> ReceiptProvider for ProviderFactory<DB> {
diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs
index 77de5ac9f6..dec82dd68b 100644
--- a/crates/storage/provider/src/providers/database/provider.rs
+++ b/crates/storage/provider/src/providers/database/provider.rs
@@ -25,10 +25,10 @@ use reth_interfaces::Result;
 use reth_primitives::{
     keccak256,
     stage::{StageCheckpoint, StageId},
-    Account, Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, ChainInfo, ChainSpec,
-    Hardfork, Head, Header, Receipt, SealedBlock, SealedBlockWithSenders, SealedHeader,
-    StorageEntry, TransactionMeta, TransactionSigned, TransactionSignedEcRecovered, TxHash,
-    TxNumber, Withdrawal, H256, U256,
+    Account, Address, Block, BlockHash, BlockHashOrNumber, BlockNumber, BlockWithSenders,
+    ChainInfo, ChainSpec, Hardfork, Head, Header, Receipt, SealedBlock, SealedBlockWithSenders,
+    SealedHeader, StorageEntry, TransactionMeta, TransactionSigned, TransactionSignedEcRecovered,
+    TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
 };
 use reth_revm_primitives::{
     config::revm_spec,
@@ -1609,6 +1609,48 @@ impl<'this, TX: DbTx<'this>> BlockProvider for DatabaseProvider<'this, TX> {
     fn block_body_indices(&self, num: u64) -> Result<Option<StoredBlockBodyIndices>> {
         Ok(self.tx.get::<tables::BlockBodyIndices>(num)?)
     }
+
+    /// Returns the block with senders with matching number from database.
+    ///
+    /// **NOTE: The transactions have invalid hashes, since they would need to be calculated on the
+    /// spot, and we want fast querying.**
+    ///
+    /// Returns `None` if block is not found.
+    fn block_with_senders(&self, block_number: BlockNumber) -> Result<Option<BlockWithSenders>> {
+        let header = self
+            .header_by_number(block_number)?
+            .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;
+
+        let ommers = self.ommers(block_number.into())?.unwrap_or_default();
+        let withdrawals = self.withdrawals_by_block(block_number.into(), header.timestamp)?;
+
+        // Get the block body
+        let body = self
+            .block_body_indices(block_number)?
+            .ok_or(ProviderError::BlockBodyIndicesNotFound(block_number))?;
+        let tx_range = body.tx_num_range();
+
+        let (transactions, senders) = if tx_range.is_empty() {
+            (vec![], vec![])
+        } else {
+            (self.transactions_by_tx_range(tx_range.clone())?, self.senders_by_tx_range(tx_range)?)
+        };
+
+        let body = transactions
+            .into_iter()
+            .map(|tx| {
+                TransactionSigned {
+                    // TODO: This is the fastest way right now to make everything just work with
+                    // a dummy transaction hash.
+                    hash: Default::default(),
+                    signature: tx.signature,
+                    transaction: tx.transaction,
+                }
+            })
+            .collect();
+
+        Ok(Some(Block { header, body, ommers, withdrawals }.with_senders(senders)))
+    }
 }
 
 impl<'this, TX: DbTx<'this>> TransactionsProvider for DatabaseProvider<'this, TX> {
@@ -1716,6 +1758,27 @@ impl<'this, TX: DbTx<'this>> TransactionsProvider for DatabaseProvider<'this, TX
         }
         Ok(results)
     }
+
+    fn transactions_by_tx_range(
+        &self,
+        range: impl RangeBounds<TxNumber>,
+    ) -> Result<Vec<TransactionSignedNoHash>> {
+        Ok(self
+            .tx
+            .cursor_read::<tables::Transactions>()?
+            .walk_range(range)?
+            .map(|entry| entry.map(|tx| tx.1))
+            .collect::<std::result::Result<Vec<_>, _>>()?)
+    }
+
+    fn senders_by_tx_range(&self, range: impl RangeBounds<TxNumber>) -> Result<Vec<Address>> {
+        Ok(self
+            .tx
+            .cursor_read::<tables::TxSenders>()?
+            .walk_range(range)?
+            .map(|entry| entry.map(|sender| sender.1))
+            .collect::<std::result::Result<Vec<_>, _>>()?)
+    }
 }
 
 impl<'this, TX: DbTx<'this>> ReceiptProvider for DatabaseProvider<'this, TX> {
diff --git a/crates/storage/provider/src/providers/mod.rs b/crates/storage/provider/src/providers/mod.rs
index 7bb70825e4..9594f81736 100644
--- a/crates/storage/provider/src/providers/mod.rs
+++ b/crates/storage/provider/src/providers/mod.rs
@@ -13,9 +13,10 @@ use reth_interfaces::{
 };
 use reth_primitives::{
     stage::{StageCheckpoint, StageId},
-    Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumHash, BlockNumber, BlockNumberOrTag,
-    ChainInfo, Header, Receipt, SealedBlock, SealedBlockWithSenders, SealedHeader, TransactionMeta,
-    TransactionSigned, TxHash, TxNumber, Withdrawal, H256, U256,
+    Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumHash, BlockNumber,
+    BlockNumberOrTag, BlockWithSenders, ChainInfo, Header, Receipt, SealedBlock,
+    SealedBlockWithSenders, SealedHeader, TransactionMeta, TransactionSigned,
+    TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, H256, U256,
 };
 use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
 pub use state::{
@@ -234,8 +235,18 @@ where
         self.database.provider()?.ommers(id)
     }
 
-    fn block_body_indices(&self, num: u64) -> Result<Option<StoredBlockBodyIndices>> {
-        self.database.provider()?.block_body_indices(num)
+    fn block_body_indices(&self, number: BlockNumber) -> Result<Option<StoredBlockBodyIndices>> {
+        self.database.provider()?.block_body_indices(number)
+    }
+
+    /// Returns the block with senders with matching number from database.
+    ///
+    /// **NOTE: The transactions have invalid hashes, since they would need to be calculated on the
+    /// spot, and we want fast querying.**
+    ///
+    /// Returns `None` if block is not found.
+    fn block_with_senders(&self, number: BlockNumber) -> Result<Option<BlockWithSenders>> {
+        self.database.provider()?.block_with_senders(number)
     }
 }
 
@@ -280,6 +291,17 @@ where
     ) -> Result<Vec<Vec<TransactionSigned>>> {
         self.database.provider()?.transactions_by_block_range(range)
     }
+
+    fn transactions_by_tx_range(
+        &self,
+        range: impl RangeBounds<TxNumber>,
+    ) -> Result<Vec<TransactionSignedNoHash>> {
+        self.database.provider()?.transactions_by_tx_range(range)
+    }
+
+    fn senders_by_tx_range(&self, range: impl RangeBounds<TxNumber>) -> Result<Vec<Address>> {
+        self.database.provider()?.senders_by_tx_range(range)
+    }
 }
 
 impl<DB, Tree> ReceiptProvider for BlockchainProvider<DB, Tree>
diff --git a/crates/storage/provider/src/test_utils/mock.rs b/crates/storage/provider/src/test_utils/mock.rs
index a94e26a4f1..9c029cd112 100644
--- a/crates/storage/provider/src/test_utils/mock.rs
+++ b/crates/storage/provider/src/test_utils/mock.rs
@@ -3,14 +3,15 @@ use crate::{
     AccountProvider, BlockHashProvider, BlockIdProvider, BlockNumProvider, BlockProvider,
     BlockProviderIdExt, EvmEnvProvider, HeaderProvider, PostState, PostStateDataProvider,
     StateProvider, StateProviderBox, StateProviderFactory, StateRootProvider, TransactionsProvider,
+    WithdrawalsProvider,
 };
 use parking_lot::Mutex;
 use reth_db::models::StoredBlockBodyIndices;
 use reth_interfaces::{provider::ProviderError, Result};
 use reth_primitives::{
     keccak256, Account, Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumber,
-    Bytecode, Bytes, ChainInfo, Header, Receipt, SealedBlock, SealedHeader, StorageKey,
-    StorageValue, TransactionMeta, TransactionSigned, TxHash, TxNumber, H256, U256,
+    BlockWithSenders, Bytecode, Bytes, ChainInfo, Header, Receipt, SealedBlock, SealedHeader,
+    StorageKey, StorageValue, TransactionMeta, TransactionSigned, TxHash, TxNumber, H256, U256,
 };
 use reth_revm_primitives::primitives::{BlockEnv, CfgEnv};
 use std::{
@@ -207,6 +208,17 @@ impl TransactionsProvider for MockEthProvider {
 
         Ok(map.into_values().collect())
     }
+
+    fn senders_by_tx_range(&self, _range: impl RangeBounds<TxNumber>) -> Result<Vec<Address>> {
+        unimplemented!()
+    }
+
+    fn transactions_by_tx_range(
+        &self,
+        _range: impl RangeBounds<TxNumber>,
+    ) -> Result<Vec<reth_primitives::TransactionSignedNoHash>> {
+        unimplemented!()
+    }
 }
 
 impl ReceiptProvider for MockEthProvider {
@@ -313,6 +325,10 @@ impl BlockProvider for MockEthProvider {
     fn block_body_indices(&self, _num: u64) -> Result<Option<StoredBlockBodyIndices>> {
         Ok(None)
     }
+
+    fn block_with_senders(&self, _number: BlockNumber) -> Result<Option<BlockWithSenders>> {
+        Ok(None)
+    }
 }
 
 impl BlockProviderIdExt for MockEthProvider {
@@ -478,3 +494,16 @@ impl StateProviderFactory for Arc<MockEthProvider> {
         todo!()
     }
 }
+
+impl WithdrawalsProvider for MockEthProvider {
+    fn latest_withdrawal(&self) -> Result<Option<reth_primitives::Withdrawal>> {
+        unimplemented!()
+    }
+    fn withdrawals_by_block(
+        &self,
+        _id: BlockHashOrNumber,
+        _timestamp: u64,
+    ) -> Result<Option<Vec<reth_primitives::Withdrawal>>> {
+        unimplemented!()
+    }
+}
diff --git a/crates/storage/provider/src/test_utils/noop.rs b/crates/storage/provider/src/test_utils/noop.rs
index 24310c620a..97d6933b2d 100644
--- a/crates/storage/provider/src/test_utils/noop.rs
+++ b/crates/storage/provider/src/test_utils/noop.rs
@@ -3,6 +3,7 @@ use crate::{
     AccountProvider, BlockHashProvider, BlockIdProvider, BlockNumProvider, BlockProvider,
     BlockProviderIdExt, EvmEnvProvider, HeaderProvider, PostState, StageCheckpointProvider,
     StateProvider, StateProviderBox, StateProviderFactory, StateRootProvider, TransactionsProvider,
+    WithdrawalsProvider,
 };
 use reth_db::models::StoredBlockBodyIndices;
 use reth_interfaces::Result;
@@ -69,6 +70,13 @@ impl BlockProvider for NoopProvider {
     fn block_body_indices(&self, _num: u64) -> Result<Option<StoredBlockBodyIndices>> {
         Ok(None)
     }
+
+    fn block_with_senders(
+        &self,
+        _number: BlockNumber,
+    ) -> Result<Option<reth_primitives::BlockWithSenders>> {
+        Ok(None)
+    }
 }
 
 impl BlockProviderIdExt for NoopProvider {
@@ -140,6 +148,17 @@ impl TransactionsProvider for NoopProvider {
     ) -> Result<Vec<Vec<TransactionSigned>>> {
         Ok(Vec::default())
     }
+
+    fn senders_by_tx_range(&self, _range: impl RangeBounds<TxNumber>) -> Result<Vec<Address>> {
+        Ok(Vec::default())
+    }
+
+    fn transactions_by_tx_range(
+        &self,
+        _range: impl RangeBounds<TxNumber>,
+    ) -> Result<Vec<reth_primitives::TransactionSignedNoHash>> {
+        Ok(Vec::default())
+    }
 }
 
 impl ReceiptProvider for NoopProvider {
@@ -293,3 +312,16 @@ impl StageCheckpointProvider for NoopProvider {
         Ok(None)
     }
 }
+
+impl WithdrawalsProvider for NoopProvider {
+    fn latest_withdrawal(&self) -> Result<Option<reth_primitives::Withdrawal>> {
+        Ok(None)
+    }
+    fn withdrawals_by_block(
+        &self,
+        _id: BlockHashOrNumber,
+        _timestamp: u64,
+    ) -> Result<Option<Vec<reth_primitives::Withdrawal>>> {
+        Ok(None)
+    }
+}
diff --git a/crates/storage/provider/src/traits/block.rs b/crates/storage/provider/src/traits/block.rs
index 52c4de5d36..f2ffe87675 100644
--- a/crates/storage/provider/src/traits/block.rs
+++ b/crates/storage/provider/src/traits/block.rs
@@ -1,10 +1,12 @@
 use crate::{
     BlockIdProvider, BlockNumProvider, HeaderProvider, ReceiptProvider, TransactionsProvider,
+    WithdrawalsProvider,
 };
 use reth_db::models::StoredBlockBodyIndices;
 use reth_interfaces::Result;
 use reth_primitives::{
-    Block, BlockHashOrNumber, BlockId, BlockNumberOrTag, Header, SealedBlock, SealedHeader, H256,
+    Block, BlockHashOrNumber, BlockId, BlockNumber, BlockNumberOrTag, BlockWithSenders, Header,
+    SealedBlock, SealedHeader, H256,
 };
 
 /// A helper enum that represents the origin of the requested block.
@@ -44,7 +46,13 @@ impl BlockSource {
 /// the database.
 #[auto_impl::auto_impl(&, Arc)]
 pub trait BlockProvider:
-    BlockNumProvider + HeaderProvider + TransactionsProvider + ReceiptProvider + Send + Sync
+    BlockNumProvider
+    + HeaderProvider
+    + TransactionsProvider
+    + ReceiptProvider
+    + WithdrawalsProvider
+    + Send
+    + Sync
 {
     /// Tries to find in the given block source.
     ///
@@ -87,6 +95,11 @@ pub trait BlockProvider:
     ///
     /// Returns `None` if block is not found.
     fn block_body_indices(&self, num: u64) -> Result<Option<StoredBlockBodyIndices>>;
+
+    /// Returns the block with senders with matching number from database.
+    ///
+    /// Returns `None` if block is not found.
+    fn block_with_senders(&self, number: BlockNumber) -> Result<Option<BlockWithSenders>>;
 }
 
 /// Trait extension for `BlockProvider`, for types that implement `BlockId` conversion.
diff --git a/crates/storage/provider/src/traits/transactions.rs b/crates/storage/provider/src/traits/transactions.rs
index f80bf9db5b..ca5e15a88c 100644
--- a/crates/storage/provider/src/traits/transactions.rs
+++ b/crates/storage/provider/src/traits/transactions.rs
@@ -1,7 +1,8 @@
 use crate::BlockNumProvider;
 use reth_interfaces::Result;
 use reth_primitives::{
-    BlockHashOrNumber, BlockNumber, TransactionMeta, TransactionSigned, TxHash, TxNumber,
+    Address, BlockHashOrNumber, BlockNumber, TransactionMeta, TransactionSigned,
+    TransactionSignedNoHash, TxHash, TxNumber,
 };
 use std::ops::RangeBounds;
 
@@ -41,4 +42,13 @@ pub trait TransactionsProvider: BlockNumProvider + Send + Sync {
         &self,
         range: impl RangeBounds<BlockNumber>,
     ) -> Result<Vec<Vec<TransactionSigned>>>;
+
+    /// Get transactions by tx range.
+    fn transactions_by_tx_range(
+        &self,
+        range: impl RangeBounds<TxNumber>,
+    ) -> Result<Vec<TransactionSignedNoHash>>;
+
+    /// Get Senders from a tx range.
+    fn senders_by_tx_range(&self, range: impl RangeBounds<TxNumber>) -> Result<Vec<Address>>;
 }
diff --git a/crates/storage/provider/src/traits/withdrawals.rs b/crates/storage/provider/src/traits/withdrawals.rs
index ffa33febab..d5d65808f5 100644
--- a/crates/storage/provider/src/traits/withdrawals.rs
+++ b/crates/storage/provider/src/traits/withdrawals.rs
@@ -2,6 +2,7 @@ use reth_interfaces::Result;
 use reth_primitives::{BlockHashOrNumber, Withdrawal};
 
 ///  Client trait for fetching [Withdrawal] related data.
+#[auto_impl::auto_impl(&, Arc)]
 pub trait WithdrawalsProvider: Send + Sync {
     /// Get withdrawals by block id.
     fn withdrawals_by_block(