diff --git a/crates/storage/provider/src/providers/database/mod.rs b/crates/storage/provider/src/providers/database/mod.rs index 5d9d8ae6b..a2bf883d5 100644 --- a/crates/storage/provider/src/providers/database/mod.rs +++ b/crates/storage/provider/src/providers/database/mod.rs @@ -331,6 +331,13 @@ impl BlockReader for ProviderFactory { fn block_range(&self, range: RangeInclusive) -> ProviderResult> { self.provider()?.block_range(range) } + + fn block_with_senders_range( + &self, + range: RangeInclusive, + ) -> ProviderResult> { + self.provider()?.block_with_senders_range(range) + } } impl TransactionsProvider for ProviderFactory { diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index 98e3055e4..01c03e956 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -1286,6 +1286,67 @@ impl BlockNumReader for DatabaseProvider { } } +impl DatabaseProvider { + fn process_block_range( + &self, + range: RangeInclusive, + mut assemble_block: F, + ) -> ProviderResult> + where + F: FnMut(Range, Header, Vec
, Option) -> ProviderResult, + { + if range.is_empty() { + return Ok(Vec::new()) + } + + let len = range.end().saturating_sub(*range.start()) as usize; + let mut blocks = Vec::with_capacity(len); + + let headers = self.headers_range(range)?; + let mut ommers_cursor = self.tx.cursor_read::()?; + let mut withdrawals_cursor = self.tx.cursor_read::()?; + let mut block_body_cursor = self.tx.cursor_read::()?; + + 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)) = block_body_cursor.seek_exact(header.number)? { + let tx_range = block_body_indices.tx_num_range(); + + // If we are past shanghai, then all blocks should have a withdrawal list, + // even if empty + let withdrawals = + if self.chain_spec.is_shanghai_active_at_timestamp(header.timestamp) { + Some( + withdrawals_cursor + .seek_exact(header.number)? + .map(|(_, w)| w.withdrawals) + .unwrap_or_default(), + ) + } else { + None + }; + let ommers = + if self.chain_spec.final_paris_total_difficulty(header.number).is_some() { + Vec::new() + } else { + ommers_cursor + .seek_exact(header.number)? + .map(|(_, o)| o.ommers) + .unwrap_or_default() + }; + if let Ok(b) = assemble_block(tx_range, header, ommers, withdrawals) { + blocks.push(b); + } + } + } + Ok(blocks) + } +} + impl BlockReader for DatabaseProvider { fn find_block_by_hash(&self, hash: B256, source: BlockSource) -> ProviderResult> { if source.is_database() { @@ -1409,63 +1470,63 @@ impl BlockReader for DatabaseProvider { } fn block_range(&self, range: RangeInclusive) -> ProviderResult> { - if range.is_empty() { - return Ok(Vec::new()) - } - - let len = range.end().saturating_sub(*range.start()) as usize; - let mut blocks = Vec::with_capacity(len); - - let headers = self.headers_range(range)?; - let mut ommers_cursor = self.tx.cursor_read::()?; - let mut withdrawals_cursor = self.tx.cursor_read::()?; - let mut block_body_cursor = self.tx.cursor_read::()?; let mut tx_cursor = self.tx.cursor_read::()?; + self.process_block_range(range, |tx_range, header, ommers, withdrawals| { + let body = if tx_range.is_empty() { + Vec::new() + } else { + self.transactions_by_tx_range_with_cursor(tx_range, &mut tx_cursor)? + .into_iter() + .map(Into::into) + .collect() + }; + Ok(Block { header, body, ommers, withdrawals }) + }) + } - 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)) = block_body_cursor.seek_exact(header.number)? { - let tx_range = block_body_indices.tx_num_range(); - let body = if tx_range.is_empty() { - Vec::new() - } else { - self.transactions_by_tx_range_with_cursor(tx_range, &mut tx_cursor)? - .into_iter() - .map(Into::into) - .collect() - }; + fn block_with_senders_range( + &self, + range: RangeInclusive, + ) -> ProviderResult> { + let mut tx_cursor = self.tx.cursor_read::()?; + let mut senders_cursor = self.tx.cursor_read::()?; - // If we are past shanghai, then all blocks should have a withdrawal list, - // even if empty - let withdrawals = - if self.chain_spec.is_shanghai_active_at_timestamp(header.timestamp) { - Some( - withdrawals_cursor - .seek_exact(header.number)? - .map(|(_, w)| w.withdrawals) - .unwrap_or_default(), - ) - } else { - None - }; - let ommers = - if self.chain_spec.final_paris_total_difficulty(header.number).is_some() { - Vec::new() - } else { - ommers_cursor - .seek_exact(header.number)? - .map(|(_, o)| o.ommers) - .unwrap_or_default() - }; + self.process_block_range(range, |tx_range, header, ommers, withdrawals| { + let (body, senders) = if tx_range.is_empty() { + (Vec::new(), Vec::new()) + } else { + let body = self + .transactions_by_tx_range_with_cursor(tx_range.clone(), &mut tx_cursor)? + .into_iter() + .map(Into::into) + .collect::>(); + // fetch senders from the senders table + let known_senders = + senders_cursor + .walk_range(tx_range.clone())? + .collect::, _>>()?; - blocks.push(Block { header, body, ommers, withdrawals }); - } - } - Ok(blocks) + let mut senders = Vec::with_capacity(body.len()); + for (tx_num, tx) in tx_range.zip(body.iter()) { + match known_senders.get(&tx_num) { + None => { + // recover the sender from the transaction if not found + let sender = tx + .recover_signer_unchecked() + .ok_or_else(|| ProviderError::SenderRecoveryError)?; + senders.push(sender); + } + Some(sender) => senders.push(*sender), + } + } + + (body, senders) + }; + + Block { header, body, ommers, withdrawals } + .try_with_senders_unchecked(senders) + .map_err(|_| ProviderError::SenderRecoveryError) + }) } } diff --git a/crates/storage/provider/src/providers/mod.rs b/crates/storage/provider/src/providers/mod.rs index 4cd7fb6ac..73ea827d1 100644 --- a/crates/storage/provider/src/providers/mod.rs +++ b/crates/storage/provider/src/providers/mod.rs @@ -311,6 +311,13 @@ where fn block_range(&self, range: RangeInclusive) -> ProviderResult> { self.database.block_range(range) } + + fn block_with_senders_range( + &self, + range: RangeInclusive, + ) -> ProviderResult> { + self.database.block_with_senders_range(range) + } } impl TransactionsProvider for BlockchainProvider diff --git a/crates/storage/provider/src/providers/static_file/manager.rs b/crates/storage/provider/src/providers/static_file/manager.rs index e669275d9..7814a7097 100644 --- a/crates/storage/provider/src/providers/static_file/manager.rs +++ b/crates/storage/provider/src/providers/static_file/manager.rs @@ -1103,6 +1103,13 @@ impl BlockReader for StaticFileProvider { // Required data not present in static_files Err(ProviderError::UnsupportedProvider) } + + fn block_with_senders_range( + &self, + _range: RangeInclusive, + ) -> ProviderResult> { + Err(ProviderError::UnsupportedProvider) + } } impl WithdrawalsProvider for StaticFileProvider { diff --git a/crates/storage/provider/src/test_utils/mock.rs b/crates/storage/provider/src/test_utils/mock.rs index aa756f195..db490bd37 100644 --- a/crates/storage/provider/src/test_utils/mock.rs +++ b/crates/storage/provider/src/test_utils/mock.rs @@ -483,6 +483,13 @@ impl BlockReader for MockEthProvider { Ok(blocks) } + + fn block_with_senders_range( + &self, + _range: RangeInclusive, + ) -> ProviderResult> { + Ok(vec![]) + } } impl BlockReaderIdExt for MockEthProvider { diff --git a/crates/storage/provider/src/test_utils/noop.rs b/crates/storage/provider/src/test_utils/noop.rs index 13a2f3b40..626bd5351 100644 --- a/crates/storage/provider/src/test_utils/noop.rs +++ b/crates/storage/provider/src/test_utils/noop.rs @@ -12,8 +12,8 @@ use reth_interfaces::provider::ProviderResult; use reth_primitives::{ stage::{StageCheckpoint, StageId}, trie::AccountProof, - Account, Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumber, Bytecode, - ChainInfo, ChainSpec, Header, PruneCheckpoint, PruneSegment, Receipt, SealedBlock, + Account, Address, Block, BlockHash, BlockHashOrNumber, BlockId, BlockNumber, BlockWithSenders, + Bytecode, ChainInfo, ChainSpec, Header, PruneCheckpoint, PruneSegment, Receipt, SealedBlock, SealedBlockWithSenders, SealedHeader, StorageKey, StorageValue, TransactionMeta, TransactionSigned, TransactionSignedNoHash, TxHash, TxNumber, Withdrawal, Withdrawals, B256, MAINNET, U256, @@ -116,6 +116,13 @@ impl BlockReader for NoopProvider { fn block_range(&self, _range: RangeInclusive) -> ProviderResult> { Ok(vec![]) } + + fn block_with_senders_range( + &self, + _range: RangeInclusive, + ) -> ProviderResult> { + Ok(vec![]) + } } impl BlockReaderIdExt for NoopProvider { diff --git a/crates/storage/provider/src/traits/block.rs b/crates/storage/provider/src/traits/block.rs index f7af38166..b8ae1bdb1 100644 --- a/crates/storage/provider/src/traits/block.rs +++ b/crates/storage/provider/src/traits/block.rs @@ -133,6 +133,15 @@ pub trait BlockReader: /// /// Note: returns only available blocks fn block_range(&self, range: RangeInclusive) -> ProviderResult>; + + /// retrieves a range of blocks from the database, along with the senders of each + /// transaction in the blocks. + /// + /// The `transaction_kind` parameter determines whether to return its hash + fn block_with_senders_range( + &self, + range: RangeInclusive, + ) -> ProviderResult>; } /// Trait extension for `BlockReader`, for types that implement `BlockId` conversion.