use super::{DatabaseProviderRO, ProviderFactory, ProviderNodeTypes};
use crate::{
providers::StaticFileProvider, AccountReader, BlockHashReader, BlockIdReader, BlockNumReader,
BlockReader, BlockReaderIdExt, BlockSource, ChainSpecProvider, ChangeSetReader, EvmEnvProvider,
HeaderProvider, ProviderError, PruneCheckpointReader, ReceiptProvider, ReceiptProviderIdExt,
StageCheckpointReader, StateReader, StaticFileProviderFactory, TransactionVariant,
TransactionsProvider, WithdrawalsProvider,
};
use alloy_consensus::Header;
use alloy_eips::{
eip4895::{Withdrawal, Withdrawals},
BlockHashOrNumber, BlockId, BlockNumHash, BlockNumberOrTag, HashOrNumber,
};
use alloy_primitives::{Address, BlockHash, BlockNumber, TxHash, TxNumber, B256, U256};
use reth_chain_state::{BlockState, CanonicalInMemoryState, MemoryOverlayStateProviderRef};
use reth_chainspec::{ChainInfo, EthereumHardforks};
use reth_db::models::BlockNumberAddress;
use reth_db_api::models::{AccountBeforeTx, StoredBlockBodyIndices};
use reth_evm::ConfigureEvmEnv;
use reth_execution_types::{BundleStateInit, ExecutionOutcome, RevertsInit};
use reth_primitives::{
Account, Block, BlockWithSenders, Receipt, SealedBlock, SealedBlockWithSenders, SealedHeader,
StorageEntry, TransactionMeta, TransactionSigned, TransactionSignedNoHash,
};
use reth_prune_types::{PruneCheckpoint, PruneSegment};
use reth_stages_types::{StageCheckpoint, StageId};
use reth_storage_api::{DatabaseProviderFactory, StateProvider, StorageChangeSetReader};
use reth_storage_errors::provider::ProviderResult;
use revm::{
db::states::PlainStorageRevert,
primitives::{BlockEnv, CfgEnvWithHandlerCfg},
};
use std::{
collections::{hash_map, HashMap},
ops::{Add, Bound, RangeBounds, RangeInclusive, Sub},
sync::Arc,
};
use tracing::trace;
#[derive(Debug)]
pub struct ConsistentProvider<N: ProviderNodeTypes> {
storage_provider: <ProviderFactory<N> as DatabaseProviderFactory>::Provider,
head_block: Option<Arc<BlockState>>,
canonical_in_memory_state: CanonicalInMemoryState,
}
impl<N: ProviderNodeTypes> ConsistentProvider<N> {
pub fn new(
storage_provider_factory: ProviderFactory<N>,
state: CanonicalInMemoryState,
) -> ProviderResult<Self> {
let head_block = state.head_state();
let storage_provider = storage_provider_factory.database_provider_ro()?;
Ok(Self { storage_provider, head_block, canonical_in_memory_state: state })
}
fn convert_range_bounds<T>(
&self,
range: impl RangeBounds<T>,
end_unbounded: impl FnOnce() -> T,
) -> (T, T)
where
T: Copy + Add<Output = T> + Sub<Output = T> + From<u8>,
{
let start = match range.start_bound() {
Bound::Included(&n) => n,
Bound::Excluded(&n) => n + T::from(1u8),
Bound::Unbounded => T::from(0u8),
};
let end = match range.end_bound() {
Bound::Included(&n) => n,
Bound::Excluded(&n) => n - T::from(1u8),
Bound::Unbounded => end_unbounded(),
};
(start, end)
}
fn latest_ref<'a>(&'a self) -> ProviderResult<Box<dyn StateProvider + 'a>> {
trace!(target: "providers::blockchain", "Getting latest block state provider");
if let Some(state) = &self.head_block {
trace!(target: "providers::blockchain", "Using head state for latest state provider");
Ok(self.block_state_provider_ref(state)?.boxed())
} else {
trace!(target: "providers::blockchain", "Using database state for latest state provider");
self.storage_provider.latest()
}
}
fn history_by_block_hash_ref<'a>(
&'a self,
block_hash: BlockHash,
) -> ProviderResult<Box<dyn StateProvider + 'a>> {
trace!(target: "providers::blockchain", ?block_hash, "Getting history by block hash");
self.get_in_memory_or_storage_by_block(
block_hash.into(),
|_| self.storage_provider.history_by_block_hash(block_hash),
|block_state| {
let state_provider = self.block_state_provider_ref(block_state)?;
Ok(Box::new(state_provider))
},
)
}
fn state_by_block_number_ref<'a>(
&'a self,
number: BlockNumber,
) -> ProviderResult<Box<dyn StateProvider + 'a>> {
let hash =
self.block_hash(number)?.ok_or_else(|| ProviderError::HeaderNotFound(number.into()))?;
self.history_by_block_hash_ref(hash)
}
pub fn get_state(
&self,
range: RangeInclusive<BlockNumber>,
) -> ProviderResult<Option<ExecutionOutcome>> {
if range.is_empty() {
return Ok(None)
}
let start_block_number = *range.start();
let end_block_number = *range.end();
let mut block_bodies = Vec::new();
for block_num in range.clone() {
let block_body = self
.block_body_indices(block_num)?
.ok_or(ProviderError::BlockBodyIndicesNotFound(block_num))?;
block_bodies.push((block_num, block_body))
}
let Some(from_transaction_num) = block_bodies.first().map(|body| body.1.first_tx_num())
else {
return Ok(None)
};
let Some(to_transaction_num) = block_bodies.last().map(|body| body.1.last_tx_num()) else {
return Ok(None)
};
let mut account_changeset = Vec::new();
for block_num in range.clone() {
let changeset =
self.account_block_changeset(block_num)?.into_iter().map(|elem| (block_num, elem));
account_changeset.extend(changeset);
}
let mut storage_changeset = Vec::new();
for block_num in range {
let changeset = self.storage_changeset(block_num)?;
storage_changeset.extend(changeset);
}
let (state, reverts) =
self.populate_bundle_state(account_changeset, storage_changeset, end_block_number)?;
let mut receipt_iter =
self.receipts_by_tx_range(from_transaction_num..=to_transaction_num)?.into_iter();
let mut receipts = Vec::with_capacity(block_bodies.len());
for (_, block_body) in block_bodies {
let mut block_receipts = Vec::with_capacity(block_body.tx_count as usize);
for tx_num in block_body.tx_num_range() {
let receipt = receipt_iter
.next()
.ok_or_else(|| ProviderError::ReceiptNotFound(tx_num.into()))?;
block_receipts.push(Some(receipt));
}
receipts.push(block_receipts);
}
Ok(Some(ExecutionOutcome::new_init(
state,
reverts,
Vec::new(),
receipts.into(),
start_block_number,
Vec::new(),
)))
}
fn populate_bundle_state(
&self,
account_changeset: Vec<(u64, AccountBeforeTx)>,
storage_changeset: Vec<(BlockNumberAddress, StorageEntry)>,
block_range_end: BlockNumber,
) -> ProviderResult<(BundleStateInit, RevertsInit)> {
let mut state: BundleStateInit = HashMap::new();
let mut reverts: RevertsInit = HashMap::new();
let state_provider = self.state_by_block_number_ref(block_range_end)?;
for (block_number, account_before) in account_changeset.into_iter().rev() {
let AccountBeforeTx { info: old_info, address } = account_before;
match state.entry(address) {
hash_map::Entry::Vacant(entry) => {
let new_info = state_provider.basic_account(address)?;
entry.insert((old_info, new_info, HashMap::new()));
}
hash_map::Entry::Occupied(mut entry) => {
entry.get_mut().0 = old_info;
}
}
reverts.entry(block_number).or_default().entry(address).or_default().0 = Some(old_info);
}
for (block_and_address, old_storage) in storage_changeset.into_iter().rev() {
let BlockNumberAddress((block_number, address)) = block_and_address;
let account_state = match state.entry(address) {
hash_map::Entry::Vacant(entry) => {
let present_info = state_provider.basic_account(address)?;
entry.insert((present_info, present_info, HashMap::new()))
}
hash_map::Entry::Occupied(entry) => entry.into_mut(),
};
match account_state.2.entry(old_storage.key) {
hash_map::Entry::Vacant(entry) => {
let new_storage_value =
state_provider.storage(address, old_storage.key)?.unwrap_or_default();
entry.insert((old_storage.value, new_storage_value));
}
hash_map::Entry::Occupied(mut entry) => {
entry.get_mut().0 = old_storage.value;
}
};
reverts
.entry(block_number)
.or_default()
.entry(address)
.or_default()
.1
.push(old_storage);
}
Ok((state, reverts))
}
fn get_in_memory_or_storage_by_block_range_while<T, F, G, P>(
&self,
range: impl RangeBounds<BlockNumber>,
fetch_db_range: F,
map_block_state_item: G,
mut predicate: P,
) -> ProviderResult<Vec<T>>
where
F: FnOnce(
&DatabaseProviderRO<N::DB, N>,
RangeInclusive<BlockNumber>,
&mut P,
) -> ProviderResult<Vec<T>>,
G: Fn(&BlockState, &mut P) -> Option<T>,
P: FnMut(&T) -> bool,
{
let mut in_memory_chain =
self.head_block.as_ref().map(|b| b.chain().collect::<Vec<_>>()).unwrap_or_default();
let db_provider = &self.storage_provider;
let (start, end) = self.convert_range_bounds(range, || {
in_memory_chain
.first()
.map(|b| b.number())
.unwrap_or_else(|| db_provider.last_block_number().unwrap_or_default())
});
if start > end {
return Ok(vec![])
}
let (in_memory, storage_range) = match in_memory_chain.last().as_ref().map(|b| b.number()) {
Some(lowest_memory_block) if lowest_memory_block <= end => {
let highest_memory_block =
in_memory_chain.first().as_ref().map(|b| b.number()).expect("qed");
let in_memory_range =
lowest_memory_block.max(start)..=end.min(highest_memory_block);
in_memory_chain.truncate(
in_memory_chain
.len()
.saturating_sub(start.saturating_sub(lowest_memory_block) as usize),
);
let storage_range =
(lowest_memory_block > start).then(|| start..=lowest_memory_block - 1);
(Some((in_memory_chain, in_memory_range)), storage_range)
}
_ => {
drop(in_memory_chain);
(None, Some(start..=end))
}
};
let mut items = Vec::with_capacity((end - start + 1) as usize);
if let Some(storage_range) = storage_range {
let mut db_items = fetch_db_range(db_provider, storage_range.clone(), &mut predicate)?;
items.append(&mut db_items);
if items.len() as u64 != storage_range.end() - storage_range.start() + 1 {
return Ok(items)
}
}
if let Some((in_memory_chain, in_memory_range)) = in_memory {
for (num, block) in in_memory_range.zip(in_memory_chain.into_iter().rev()) {
debug_assert!(num == block.number());
if let Some(item) = map_block_state_item(block, &mut predicate) {
items.push(item);
} else {
break
}
}
}
Ok(items)
}
fn block_state_provider_ref(
&self,
state: &BlockState,
) -> ProviderResult<MemoryOverlayStateProviderRef<'_>> {
let anchor_hash = state.anchor().hash;
let latest_historical = self.history_by_block_hash_ref(anchor_hash)?;
let in_memory = state.chain().map(|block_state| block_state.block()).collect();
Ok(MemoryOverlayStateProviderRef::new(latest_historical, in_memory))
}
fn get_in_memory_or_storage_by_tx_range<S, M, R>(
&self,
range: impl RangeBounds<BlockNumber>,
fetch_from_db: S,
fetch_from_block_state: M,
) -> ProviderResult<Vec<R>>
where
S: FnOnce(
&DatabaseProviderRO<N::DB, N>,
RangeInclusive<TxNumber>,
) -> ProviderResult<Vec<R>>,
M: Fn(RangeInclusive<usize>, &BlockState) -> ProviderResult<Vec<R>>,
{
let in_mem_chain = self.head_block.iter().flat_map(|b| b.chain()).collect::<Vec<_>>();
let provider = &self.storage_provider;
let last_database_block_number = in_mem_chain
.last()
.map(|b| Ok(b.anchor().number))
.unwrap_or_else(|| provider.last_block_number())?;
let last_block_body_index = provider
.block_body_indices(last_database_block_number)?
.ok_or(ProviderError::BlockBodyIndicesNotFound(last_database_block_number))?;
let mut in_memory_tx_num = last_block_body_index.next_tx_num();
let (start, end) = self.convert_range_bounds(range, || {
in_mem_chain
.iter()
.map(|b| b.block_ref().block().body.transactions.len() as u64)
.sum::<u64>() +
last_block_body_index.last_tx_num()
});
if start > end {
return Ok(vec![])
}
let mut tx_range = start..=end;
if *tx_range.end() < in_memory_tx_num {
return fetch_from_db(provider, tx_range);
}
let mut items = Vec::with_capacity((tx_range.end() - tx_range.start() + 1) as usize);
if *tx_range.start() < in_memory_tx_num {
let db_range = *tx_range.start()..=in_memory_tx_num.saturating_sub(1);
tx_range = in_memory_tx_num..=*tx_range.end();
items.extend(fetch_from_db(provider, db_range)?);
}
for block_state in in_mem_chain.iter().rev() {
let block_tx_count = block_state.block_ref().block().body.transactions.len();
let remaining = (tx_range.end() - tx_range.start() + 1) as usize;
if *tx_range.start() >= in_memory_tx_num + block_tx_count as u64 {
in_memory_tx_num += block_tx_count as u64;
continue
}
let skip = (tx_range.start() - in_memory_tx_num) as usize;
items.extend(fetch_from_block_state(
skip..=skip + (remaining.min(block_tx_count - skip) - 1),
block_state,
)?);
in_memory_tx_num += block_tx_count as u64;
if in_memory_tx_num > *tx_range.end() {
break
}
tx_range = in_memory_tx_num..=*tx_range.end();
}
Ok(items)
}
fn get_in_memory_or_storage_by_tx<S, M, R>(
&self,
id: HashOrNumber,
fetch_from_db: S,
fetch_from_block_state: M,
) -> ProviderResult<Option<R>>
where
S: FnOnce(&DatabaseProviderRO<N::DB, N>) -> ProviderResult<Option<R>>,
M: Fn(usize, TxNumber, &BlockState) -> ProviderResult<Option<R>>,
{
let in_mem_chain = self.head_block.iter().flat_map(|b| b.chain()).collect::<Vec<_>>();
let provider = &self.storage_provider;
let last_database_block_number = in_mem_chain
.last()
.map(|b| Ok(b.anchor().number))
.unwrap_or_else(|| provider.last_block_number())?;
let last_block_body_index = provider
.block_body_indices(last_database_block_number)?
.ok_or(ProviderError::BlockBodyIndicesNotFound(last_database_block_number))?;
let mut in_memory_tx_num = last_block_body_index.next_tx_num();
if let HashOrNumber::Number(id) = id {
if id < in_memory_tx_num {
return fetch_from_db(provider)
}
}
for block_state in in_mem_chain.iter().rev() {
let executed_block = block_state.block_ref();
let block = executed_block.block();
for tx_index in 0..block.body.transactions.len() {
match id {
HashOrNumber::Hash(tx_hash) => {
if tx_hash == block.body.transactions[tx_index].hash() {
return fetch_from_block_state(tx_index, in_memory_tx_num, block_state)
}
}
HashOrNumber::Number(id) => {
if id == in_memory_tx_num {
return fetch_from_block_state(tx_index, in_memory_tx_num, block_state)
}
}
}
in_memory_tx_num += 1;
}
}
if let HashOrNumber::Hash(_) = id {
return fetch_from_db(provider)
}
Ok(None)
}
pub(crate) fn get_in_memory_or_storage_by_block<S, M, R>(
&self,
id: BlockHashOrNumber,
fetch_from_db: S,
fetch_from_block_state: M,
) -> ProviderResult<R>
where
S: FnOnce(&DatabaseProviderRO<N::DB, N>) -> ProviderResult<R>,
M: Fn(&BlockState) -> ProviderResult<R>,
{
if let Some(Some(block_state)) = self.head_block.as_ref().map(|b| b.block_on_chain(id)) {
return fetch_from_block_state(block_state)
}
fetch_from_db(&self.storage_provider)
}
}
impl<N: ProviderNodeTypes> ConsistentProvider<N> {
#[inline]
pub(crate) fn ensure_canonical_block(&self, block_number: BlockNumber) -> ProviderResult<()> {
let latest = self.best_block_number()?;
if block_number > latest {
Err(ProviderError::HeaderNotFound(block_number.into()))
} else {
Ok(())
}
}
}
impl<N: ProviderNodeTypes> StaticFileProviderFactory for ConsistentProvider<N> {
fn static_file_provider(&self) -> StaticFileProvider {
self.storage_provider.static_file_provider()
}
}
impl<N: ProviderNodeTypes> HeaderProvider for ConsistentProvider<N> {
fn header(&self, block_hash: &BlockHash) -> ProviderResult<Option<Header>> {
self.get_in_memory_or_storage_by_block(
(*block_hash).into(),
|db_provider| db_provider.header(block_hash),
|block_state| Ok(Some(block_state.block_ref().block().header.header().clone())),
)
}
fn header_by_number(&self, num: BlockNumber) -> ProviderResult<Option<Header>> {
self.get_in_memory_or_storage_by_block(
num.into(),
|db_provider| db_provider.header_by_number(num),
|block_state| Ok(Some(block_state.block_ref().block().header.header().clone())),
)
}
fn header_td(&self, hash: &BlockHash) -> ProviderResult<Option<U256>> {
if let Some(num) = self.block_number(*hash)? {
self.header_td_by_number(num)
} else {
Ok(None)
}
}
fn header_td_by_number(&self, number: BlockNumber) -> ProviderResult<Option<U256>> {
let number = if self.head_block.as_ref().map(|b| b.block_on_chain(number.into())).is_some()
{
if let Some(last_finalized_num_hash) =
self.canonical_in_memory_state.get_finalized_num_hash()
{
last_finalized_num_hash.number
} else {
self.last_block_number()?
}
} else {
number
};
self.storage_provider.header_td_by_number(number)
}
fn headers_range(&self, range: impl RangeBounds<BlockNumber>) -> ProviderResult<Vec<Header>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, _| db_provider.headers_range(range),
|block_state, _| Some(block_state.block_ref().block().header.header().clone()),
|_| true,
)
}
fn sealed_header(&self, number: BlockNumber) -> ProviderResult<Option<SealedHeader>> {
self.get_in_memory_or_storage_by_block(
number.into(),
|db_provider| db_provider.sealed_header(number),
|block_state| Ok(Some(block_state.block_ref().block().header.clone())),
)
}
fn sealed_headers_range(
&self,
range: impl RangeBounds<BlockNumber>,
) -> ProviderResult<Vec<SealedHeader>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, _| db_provider.sealed_headers_range(range),
|block_state, _| Some(block_state.block_ref().block().header.clone()),
|_| true,
)
}
fn sealed_headers_while(
&self,
range: impl RangeBounds<BlockNumber>,
predicate: impl FnMut(&SealedHeader) -> bool,
) -> ProviderResult<Vec<SealedHeader>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, predicate| db_provider.sealed_headers_while(range, predicate),
|block_state, predicate| {
let header = &block_state.block_ref().block().header;
predicate(header).then(|| header.clone())
},
predicate,
)
}
}
impl<N: ProviderNodeTypes> BlockHashReader for ConsistentProvider<N> {
fn block_hash(&self, number: u64) -> ProviderResult<Option<B256>> {
self.get_in_memory_or_storage_by_block(
number.into(),
|db_provider| db_provider.block_hash(number),
|block_state| Ok(Some(block_state.hash())),
)
}
fn canonical_hashes_range(
&self,
start: BlockNumber,
end: BlockNumber,
) -> ProviderResult<Vec<B256>> {
self.get_in_memory_or_storage_by_block_range_while(
start..end,
|db_provider, inclusive_range, _| {
db_provider
.canonical_hashes_range(*inclusive_range.start(), *inclusive_range.end() + 1)
},
|block_state, _| Some(block_state.hash()),
|_| true,
)
}
}
impl<N: ProviderNodeTypes> BlockNumReader for ConsistentProvider<N> {
fn chain_info(&self) -> ProviderResult<ChainInfo> {
let best_number = self.best_block_number()?;
Ok(ChainInfo { best_hash: self.block_hash(best_number)?.unwrap_or_default(), best_number })
}
fn best_block_number(&self) -> ProviderResult<BlockNumber> {
self.head_block.as_ref().map(|b| Ok(b.number())).unwrap_or_else(|| self.last_block_number())
}
fn last_block_number(&self) -> ProviderResult<BlockNumber> {
self.storage_provider.last_block_number()
}
fn block_number(&self, hash: B256) -> ProviderResult<Option<BlockNumber>> {
self.get_in_memory_or_storage_by_block(
hash.into(),
|db_provider| db_provider.block_number(hash),
|block_state| Ok(Some(block_state.number())),
)
}
}
impl<N: ProviderNodeTypes> BlockIdReader for ConsistentProvider<N> {
fn pending_block_num_hash(&self) -> ProviderResult<Option<BlockNumHash>> {
Ok(self.canonical_in_memory_state.pending_block_num_hash())
}
fn safe_block_num_hash(&self) -> ProviderResult<Option<BlockNumHash>> {
Ok(self.canonical_in_memory_state.get_safe_num_hash())
}
fn finalized_block_num_hash(&self) -> ProviderResult<Option<BlockNumHash>> {
Ok(self.canonical_in_memory_state.get_finalized_num_hash())
}
}
impl<N: ProviderNodeTypes> BlockReader for ConsistentProvider<N> {
fn find_block_by_hash(&self, hash: B256, source: BlockSource) -> ProviderResult<Option<Block>> {
match source {
BlockSource::Any | BlockSource::Canonical => {
self.get_in_memory_or_storage_by_block(
hash.into(),
|db_provider| db_provider.find_block_by_hash(hash, source),
|block_state| Ok(Some(block_state.block_ref().block().clone().unseal())),
)
}
BlockSource::Pending => {
Ok(self.canonical_in_memory_state.pending_block().map(|block| block.unseal()))
}
}
}
fn block(&self, id: BlockHashOrNumber) -> ProviderResult<Option<Block>> {
self.get_in_memory_or_storage_by_block(
id,
|db_provider| db_provider.block(id),
|block_state| Ok(Some(block_state.block_ref().block().clone().unseal())),
)
}
fn pending_block(&self) -> ProviderResult<Option<SealedBlock>> {
Ok(self.canonical_in_memory_state.pending_block())
}
fn pending_block_with_senders(&self) -> ProviderResult<Option<SealedBlockWithSenders>> {
Ok(self.canonical_in_memory_state.pending_block_with_senders())
}
fn pending_block_and_receipts(&self) -> ProviderResult<Option<(SealedBlock, Vec<Receipt>)>> {
Ok(self.canonical_in_memory_state.pending_block_and_receipts())
}
fn ommers(&self, id: BlockHashOrNumber) -> ProviderResult<Option<Vec<Header>>> {
self.get_in_memory_or_storage_by_block(
id,
|db_provider| db_provider.ommers(id),
|block_state| {
if self.chain_spec().final_paris_total_difficulty(block_state.number()).is_some() {
return Ok(Some(Vec::new()))
}
Ok(Some(block_state.block_ref().block().body.ommers.clone()))
},
)
}
fn block_body_indices(
&self,
number: BlockNumber,
) -> ProviderResult<Option<StoredBlockBodyIndices>> {
self.get_in_memory_or_storage_by_block(
number.into(),
|db_provider| db_provider.block_body_indices(number),
|block_state| {
let last_storage_block_number = block_state.anchor().number;
let mut stored_indices = self
.storage_provider
.block_body_indices(last_storage_block_number)?
.ok_or(ProviderError::BlockBodyIndicesNotFound(last_storage_block_number))?;
stored_indices.first_tx_num = stored_indices.next_tx_num();
stored_indices.tx_count = 0;
for state in block_state.chain().collect::<Vec<_>>().into_iter().rev() {
let block_tx_count = state.block_ref().block.body.transactions.len() as u64;
if state.block_ref().block().number == number {
stored_indices.tx_count = block_tx_count;
} else {
stored_indices.first_tx_num += block_tx_count;
}
}
Ok(Some(stored_indices))
},
)
}
fn block_with_senders(
&self,
id: BlockHashOrNumber,
transaction_kind: TransactionVariant,
) -> ProviderResult<Option<BlockWithSenders>> {
self.get_in_memory_or_storage_by_block(
id,
|db_provider| db_provider.block_with_senders(id, transaction_kind),
|block_state| Ok(Some(block_state.block_with_senders())),
)
}
fn sealed_block_with_senders(
&self,
id: BlockHashOrNumber,
transaction_kind: TransactionVariant,
) -> ProviderResult<Option<SealedBlockWithSenders>> {
self.get_in_memory_or_storage_by_block(
id,
|db_provider| db_provider.sealed_block_with_senders(id, transaction_kind),
|block_state| Ok(Some(block_state.sealed_block_with_senders())),
)
}
fn block_range(&self, range: RangeInclusive<BlockNumber>) -> ProviderResult<Vec<Block>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, _| db_provider.block_range(range),
|block_state, _| Some(block_state.block_ref().block().clone().unseal()),
|_| true,
)
}
fn block_with_senders_range(
&self,
range: RangeInclusive<BlockNumber>,
) -> ProviderResult<Vec<BlockWithSenders>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, _| db_provider.block_with_senders_range(range),
|block_state, _| Some(block_state.block_with_senders()),
|_| true,
)
}
fn sealed_block_with_senders_range(
&self,
range: RangeInclusive<BlockNumber>,
) -> ProviderResult<Vec<SealedBlockWithSenders>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, _| db_provider.sealed_block_with_senders_range(range),
|block_state, _| Some(block_state.sealed_block_with_senders()),
|_| true,
)
}
}
impl<N: ProviderNodeTypes> TransactionsProvider for ConsistentProvider<N> {
fn transaction_id(&self, tx_hash: TxHash) -> ProviderResult<Option<TxNumber>> {
self.get_in_memory_or_storage_by_tx(
tx_hash.into(),
|db_provider| db_provider.transaction_id(tx_hash),
|_, tx_number, _| Ok(Some(tx_number)),
)
}
fn transaction_by_id(&self, id: TxNumber) -> ProviderResult<Option<TransactionSigned>> {
self.get_in_memory_or_storage_by_tx(
id.into(),
|provider| provider.transaction_by_id(id),
|tx_index, _, block_state| {
Ok(block_state.block_ref().block().body.transactions.get(tx_index).cloned())
},
)
}
fn transaction_by_id_no_hash(
&self,
id: TxNumber,
) -> ProviderResult<Option<TransactionSignedNoHash>> {
self.get_in_memory_or_storage_by_tx(
id.into(),
|provider| provider.transaction_by_id_no_hash(id),
|tx_index, _, block_state| {
Ok(block_state
.block_ref()
.block()
.body
.transactions
.get(tx_index)
.cloned()
.map(Into::into))
},
)
}
fn transaction_by_hash(&self, hash: TxHash) -> ProviderResult<Option<TransactionSigned>> {
if let Some(tx) = self.head_block.as_ref().and_then(|b| b.transaction_on_chain(hash)) {
return Ok(Some(tx))
}
self.storage_provider.transaction_by_hash(hash)
}
fn transaction_by_hash_with_meta(
&self,
tx_hash: TxHash,
) -> ProviderResult<Option<(TransactionSigned, TransactionMeta)>> {
if let Some((tx, meta)) =
self.head_block.as_ref().and_then(|b| b.transaction_meta_on_chain(tx_hash))
{
return Ok(Some((tx, meta)))
}
self.storage_provider.transaction_by_hash_with_meta(tx_hash)
}
fn transaction_block(&self, id: TxNumber) -> ProviderResult<Option<BlockNumber>> {
self.get_in_memory_or_storage_by_tx(
id.into(),
|provider| provider.transaction_block(id),
|_, _, block_state| Ok(Some(block_state.block_ref().block().number)),
)
}
fn transactions_by_block(
&self,
id: BlockHashOrNumber,
) -> ProviderResult<Option<Vec<TransactionSigned>>> {
self.get_in_memory_or_storage_by_block(
id,
|provider| provider.transactions_by_block(id),
|block_state| Ok(Some(block_state.block_ref().block().body.transactions.clone())),
)
}
fn transactions_by_block_range(
&self,
range: impl RangeBounds<BlockNumber>,
) -> ProviderResult<Vec<Vec<TransactionSigned>>> {
self.get_in_memory_or_storage_by_block_range_while(
range,
|db_provider, range, _| db_provider.transactions_by_block_range(range),
|block_state, _| Some(block_state.block_ref().block().body.transactions.clone()),
|_| true,
)
}
fn transactions_by_tx_range(
&self,
range: impl RangeBounds<TxNumber>,
) -> ProviderResult<Vec<TransactionSignedNoHash>> {
self.get_in_memory_or_storage_by_tx_range(
range,
|db_provider, db_range| db_provider.transactions_by_tx_range(db_range),
|index_range, block_state| {
Ok(block_state.block_ref().block().body.transactions[index_range]
.iter()
.cloned()
.map(Into::into)
.collect())
},
)
}
fn senders_by_tx_range(
&self,
range: impl RangeBounds<TxNumber>,
) -> ProviderResult<Vec<Address>> {
self.get_in_memory_or_storage_by_tx_range(
range,
|db_provider, db_range| db_provider.senders_by_tx_range(db_range),
|index_range, block_state| Ok(block_state.block_ref().senders[index_range].to_vec()),
)
}
fn transaction_sender(&self, id: TxNumber) -> ProviderResult<Option<Address>> {
self.get_in_memory_or_storage_by_tx(
id.into(),
|provider| provider.transaction_sender(id),
|tx_index, _, block_state| Ok(block_state.block_ref().senders.get(tx_index).copied()),
)
}
}
impl<N: ProviderNodeTypes> ReceiptProvider for ConsistentProvider<N> {
fn receipt(&self, id: TxNumber) -> ProviderResult<Option<Receipt>> {
self.get_in_memory_or_storage_by_tx(
id.into(),
|provider| provider.receipt(id),
|tx_index, _, block_state| {
Ok(block_state.executed_block_receipts().get(tx_index).cloned())
},
)
}
fn receipt_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Receipt>> {
for block_state in self.head_block.iter().flat_map(|b| b.chain()) {
let executed_block = block_state.block_ref();
let block = executed_block.block();
let receipts = block_state.executed_block_receipts();
debug_assert_eq!(
block.body.transactions.len(),
receipts.len(),
"Mismatch between transaction and receipt count"
);
if let Some(tx_index) = block.body.transactions.iter().position(|tx| tx.hash() == hash)
{
return Ok(receipts.get(tx_index).cloned());
}
}
self.storage_provider.receipt_by_hash(hash)
}
fn receipts_by_block(&self, block: BlockHashOrNumber) -> ProviderResult<Option<Vec<Receipt>>> {
self.get_in_memory_or_storage_by_block(
block,
|db_provider| db_provider.receipts_by_block(block),
|block_state| Ok(Some(block_state.executed_block_receipts())),
)
}
fn receipts_by_tx_range(
&self,
range: impl RangeBounds<TxNumber>,
) -> ProviderResult<Vec<Receipt>> {
self.get_in_memory_or_storage_by_tx_range(
range,
|db_provider, db_range| db_provider.receipts_by_tx_range(db_range),
|index_range, block_state| {
Ok(block_state.executed_block_receipts().drain(index_range).collect())
},
)
}
}
impl<N: ProviderNodeTypes> ReceiptProviderIdExt for ConsistentProvider<N> {
fn receipts_by_block_id(&self, block: BlockId) -> ProviderResult<Option<Vec<Receipt>>> {
match block {
BlockId::Hash(rpc_block_hash) => {
let mut receipts = self.receipts_by_block(rpc_block_hash.block_hash.into())?;
if receipts.is_none() && !rpc_block_hash.require_canonical.unwrap_or(false) {
if let Some(state) = self
.head_block
.as_ref()
.and_then(|b| b.block_on_chain(rpc_block_hash.block_hash.into()))
{
receipts = Some(state.executed_block_receipts());
}
}
Ok(receipts)
}
BlockId::Number(num_tag) => match num_tag {
BlockNumberOrTag::Pending => Ok(self
.canonical_in_memory_state
.pending_state()
.map(|block_state| block_state.executed_block_receipts())),
_ => {
if let Some(num) = self.convert_block_number(num_tag)? {
self.receipts_by_block(num.into())
} else {
Ok(None)
}
}
},
}
}
}
impl<N: ProviderNodeTypes> WithdrawalsProvider for ConsistentProvider<N> {
fn withdrawals_by_block(
&self,
id: BlockHashOrNumber,
timestamp: u64,
) -> ProviderResult<Option<Withdrawals>> {
if !self.chain_spec().is_shanghai_active_at_timestamp(timestamp) {
return Ok(None)
}
self.get_in_memory_or_storage_by_block(
id,
|db_provider| db_provider.withdrawals_by_block(id, timestamp),
|block_state| Ok(block_state.block_ref().block().body.withdrawals.clone()),
)
}
fn latest_withdrawal(&self) -> ProviderResult<Option<Withdrawal>> {
let best_block_num = self.best_block_number()?;
self.get_in_memory_or_storage_by_block(
best_block_num.into(),
|db_provider| db_provider.latest_withdrawal(),
|block_state| {
Ok(block_state
.block_ref()
.block()
.body
.withdrawals
.clone()
.and_then(|mut w| w.pop()))
},
)
}
}
impl<N: ProviderNodeTypes> StageCheckpointReader for ConsistentProvider<N> {
fn get_stage_checkpoint(&self, id: StageId) -> ProviderResult<Option<StageCheckpoint>> {
self.storage_provider.get_stage_checkpoint(id)
}
fn get_stage_checkpoint_progress(&self, id: StageId) -> ProviderResult<Option<Vec<u8>>> {
self.storage_provider.get_stage_checkpoint_progress(id)
}
fn get_all_checkpoints(&self) -> ProviderResult<Vec<(String, StageCheckpoint)>> {
self.storage_provider.get_all_checkpoints()
}
}
impl<N: ProviderNodeTypes> EvmEnvProvider for ConsistentProvider<N> {
fn fill_env_at<EvmConfig>(
&self,
cfg: &mut CfgEnvWithHandlerCfg,
block_env: &mut BlockEnv,
at: BlockHashOrNumber,
evm_config: EvmConfig,
) -> ProviderResult<()>
where
EvmConfig: ConfigureEvmEnv<Header = Header>,
{
let hash = self.convert_number(at)?.ok_or(ProviderError::HeaderNotFound(at))?;
let header = self.header(&hash)?.ok_or(ProviderError::HeaderNotFound(at))?;
self.fill_env_with_header(cfg, block_env, &header, evm_config)
}
fn fill_env_with_header<EvmConfig>(
&self,
cfg: &mut CfgEnvWithHandlerCfg,
block_env: &mut BlockEnv,
header: &Header,
evm_config: EvmConfig,
) -> ProviderResult<()>
where
EvmConfig: ConfigureEvmEnv<Header = Header>,
{
let total_difficulty = self
.header_td_by_number(header.number)?
.ok_or_else(|| ProviderError::HeaderNotFound(header.number.into()))?;
evm_config.fill_cfg_and_block_env(cfg, block_env, header, total_difficulty);
Ok(())
}
fn fill_cfg_env_at<EvmConfig>(
&self,
cfg: &mut CfgEnvWithHandlerCfg,
at: BlockHashOrNumber,
evm_config: EvmConfig,
) -> ProviderResult<()>
where
EvmConfig: ConfigureEvmEnv<Header = Header>,
{
let hash = self.convert_number(at)?.ok_or(ProviderError::HeaderNotFound(at))?;
let header = self.header(&hash)?.ok_or(ProviderError::HeaderNotFound(at))?;
self.fill_cfg_env_with_header(cfg, &header, evm_config)
}
fn fill_cfg_env_with_header<EvmConfig>(
&self,
cfg: &mut CfgEnvWithHandlerCfg,
header: &Header,
evm_config: EvmConfig,
) -> ProviderResult<()>
where
EvmConfig: ConfigureEvmEnv<Header = Header>,
{
let total_difficulty = self
.header_td_by_number(header.number)?
.ok_or_else(|| ProviderError::HeaderNotFound(header.number.into()))?;
evm_config.fill_cfg_env(cfg, header, total_difficulty);
Ok(())
}
}
impl<N: ProviderNodeTypes> PruneCheckpointReader for ConsistentProvider<N> {
fn get_prune_checkpoint(
&self,
segment: PruneSegment,
) -> ProviderResult<Option<PruneCheckpoint>> {
self.storage_provider.get_prune_checkpoint(segment)
}
fn get_prune_checkpoints(&self) -> ProviderResult<Vec<(PruneSegment, PruneCheckpoint)>> {
self.storage_provider.get_prune_checkpoints()
}
}
impl<N: ProviderNodeTypes> ChainSpecProvider for ConsistentProvider<N> {
type ChainSpec = N::ChainSpec;
fn chain_spec(&self) -> Arc<N::ChainSpec> {
ChainSpecProvider::chain_spec(&self.storage_provider)
}
}
impl<N: ProviderNodeTypes> BlockReaderIdExt for ConsistentProvider<N> {
fn block_by_id(&self, id: BlockId) -> ProviderResult<Option<Block>> {
match id {
BlockId::Number(num) => self.block_by_number_or_tag(num),
BlockId::Hash(hash) => {
if Some(true) == hash.require_canonical {
self.find_block_by_hash(hash.block_hash, BlockSource::Canonical)
} else {
self.block_by_hash(hash.block_hash)
}
}
}
}
fn header_by_number_or_tag(&self, id: BlockNumberOrTag) -> ProviderResult<Option<Header>> {
Ok(match id {
BlockNumberOrTag::Latest => {
Some(self.canonical_in_memory_state.get_canonical_head().unseal())
}
BlockNumberOrTag::Finalized => {
self.canonical_in_memory_state.get_finalized_header().map(|h| h.unseal())
}
BlockNumberOrTag::Safe => {
self.canonical_in_memory_state.get_safe_header().map(|h| h.unseal())
}
BlockNumberOrTag::Earliest => self.header_by_number(0)?,
BlockNumberOrTag::Pending => self.canonical_in_memory_state.pending_header(),
BlockNumberOrTag::Number(num) => self.header_by_number(num)?,
})
}
fn sealed_header_by_number_or_tag(
&self,
id: BlockNumberOrTag,
) -> ProviderResult<Option<SealedHeader>> {
match id {
BlockNumberOrTag::Latest => {
Ok(Some(self.canonical_in_memory_state.get_canonical_head()))
}
BlockNumberOrTag::Finalized => {
Ok(self.canonical_in_memory_state.get_finalized_header())
}
BlockNumberOrTag::Safe => Ok(self.canonical_in_memory_state.get_safe_header()),
BlockNumberOrTag::Earliest => self
.header_by_number(0)?
.map_or_else(|| Ok(None), |h| Ok(Some(SealedHeader::seal(h)))),
BlockNumberOrTag::Pending => Ok(self.canonical_in_memory_state.pending_sealed_header()),
BlockNumberOrTag::Number(num) => self
.header_by_number(num)?
.map_or_else(|| Ok(None), |h| Ok(Some(SealedHeader::seal(h)))),
}
}
fn sealed_header_by_id(&self, id: BlockId) -> ProviderResult<Option<SealedHeader>> {
Ok(match id {
BlockId::Number(num) => self.sealed_header_by_number_or_tag(num)?,
BlockId::Hash(hash) => self.header(&hash.block_hash)?.map(SealedHeader::seal),
})
}
fn header_by_id(&self, id: BlockId) -> ProviderResult<Option<Header>> {
Ok(match id {
BlockId::Number(num) => self.header_by_number_or_tag(num)?,
BlockId::Hash(hash) => self.header(&hash.block_hash)?,
})
}
fn ommers_by_id(&self, id: BlockId) -> ProviderResult<Option<Vec<Header>>> {
match id {
BlockId::Number(num) => self.ommers_by_number_or_tag(num),
BlockId::Hash(hash) => {
self.ommers(BlockHashOrNumber::Hash(hash.block_hash))
}
}
}
}
impl<N: ProviderNodeTypes> StorageChangeSetReader for ConsistentProvider<N> {
fn storage_changeset(
&self,
block_number: BlockNumber,
) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
if let Some(state) =
self.head_block.as_ref().and_then(|b| b.block_on_chain(block_number.into()))
{
let changesets = state
.block()
.execution_output
.bundle
.reverts
.clone()
.to_plain_state_reverts()
.storage
.into_iter()
.flatten()
.flat_map(|revert: PlainStorageRevert| {
revert.storage_revert.into_iter().map(move |(key, value)| {
(
BlockNumberAddress((block_number, revert.address)),
StorageEntry { key: key.into(), value: value.to_previous_value() },
)
})
})
.collect();
Ok(changesets)
} else {
let storage_history_exists = self
.storage_provider
.get_prune_checkpoint(PruneSegment::StorageHistory)?
.and_then(|checkpoint| {
checkpoint.block_number.map(|checkpoint| block_number > checkpoint)
})
.unwrap_or(true);
if !storage_history_exists {
return Err(ProviderError::StateAtBlockPruned(block_number))
}
self.storage_provider.storage_changeset(block_number)
}
}
}
impl<N: ProviderNodeTypes> ChangeSetReader for ConsistentProvider<N> {
fn account_block_changeset(
&self,
block_number: BlockNumber,
) -> ProviderResult<Vec<AccountBeforeTx>> {
if let Some(state) =
self.head_block.as_ref().and_then(|b| b.block_on_chain(block_number.into()))
{
let changesets = state
.block_ref()
.execution_output
.bundle
.reverts
.clone()
.to_plain_state_reverts()
.accounts
.into_iter()
.flatten()
.map(|(address, info)| AccountBeforeTx { address, info: info.map(Into::into) })
.collect();
Ok(changesets)
} else {
let account_history_exists = self
.storage_provider
.get_prune_checkpoint(PruneSegment::AccountHistory)?
.and_then(|checkpoint| {
checkpoint.block_number.map(|checkpoint| block_number > checkpoint)
})
.unwrap_or(true);
if !account_history_exists {
return Err(ProviderError::StateAtBlockPruned(block_number))
}
self.storage_provider.account_block_changeset(block_number)
}
}
}
impl<N: ProviderNodeTypes> AccountReader for ConsistentProvider<N> {
fn basic_account(&self, address: Address) -> ProviderResult<Option<Account>> {
let state_provider = self.latest_ref()?;
state_provider.basic_account(address)
}
}
impl<N: ProviderNodeTypes> StateReader for ConsistentProvider<N> {
fn get_state(&self, block: BlockNumber) -> ProviderResult<Option<ExecutionOutcome>> {
if let Some(state) = self.head_block.as_ref().and_then(|b| b.block_on_chain(block.into())) {
let state = state.block_ref().execution_outcome().clone();
Ok(Some(state))
} else {
Self::get_state(self, block..=block)
}
}
}
#[cfg(test)]
mod tests {
use crate::{
providers::blockchain_provider::BlockchainProvider2,
test_utils::create_test_provider_factory, BlockWriter,
};
use alloy_eips::BlockHashOrNumber;
use alloy_primitives::B256;
use itertools::Itertools;
use rand::Rng;
use reth_chain_state::{ExecutedBlock, NewCanonicalChain};
use reth_db::models::AccountBeforeTx;
use reth_execution_types::ExecutionOutcome;
use reth_primitives::SealedBlock;
use reth_storage_api::{BlockReader, BlockSource, ChangeSetReader};
use reth_testing_utils::generators::{
self, random_block_range, random_changeset_range, random_eoa_accounts, BlockRangeParams,
};
use revm::db::BundleState;
use std::{
ops::{Bound, Range, RangeBounds},
sync::Arc,
};
const TEST_BLOCKS_COUNT: usize = 5;
fn random_blocks(
rng: &mut impl Rng,
database_blocks: usize,
in_memory_blocks: usize,
requests_count: Option<Range<u8>>,
withdrawals_count: Option<Range<u8>>,
tx_count: impl RangeBounds<u8>,
) -> (Vec<SealedBlock>, Vec<SealedBlock>) {
let block_range = (database_blocks + in_memory_blocks - 1) as u64;
let tx_start = match tx_count.start_bound() {
Bound::Included(&n) | Bound::Excluded(&n) => n,
Bound::Unbounded => u8::MIN,
};
let tx_end = match tx_count.end_bound() {
Bound::Included(&n) | Bound::Excluded(&n) => n + 1,
Bound::Unbounded => u8::MAX,
};
let blocks = random_block_range(
rng,
0..=block_range,
BlockRangeParams {
parent: Some(B256::ZERO),
tx_count: tx_start..tx_end,
requests_count,
withdrawals_count,
},
);
let (database_blocks, in_memory_blocks) = blocks.split_at(database_blocks);
(database_blocks.to_vec(), in_memory_blocks.to_vec())
}
#[test]
fn test_block_reader_find_block_by_hash() -> eyre::Result<()> {
let mut rng = generators::rng();
let factory = create_test_provider_factory();
let blocks = random_block_range(
&mut rng,
0..=10,
BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
);
let (database_blocks, in_memory_blocks) = blocks.split_at(5);
let provider_rw = factory.provider_rw()?;
for block in database_blocks {
provider_rw.insert_historical_block(
block.clone().seal_with_senders().expect("failed to seal block with senders"),
)?;
}
provider_rw.commit()?;
let provider = BlockchainProvider2::new(factory)?;
let consistent_provider = provider.consistent_provider()?;
let first_db_block = database_blocks.first().unwrap();
let first_in_mem_block = in_memory_blocks.first().unwrap();
let last_in_mem_block = in_memory_blocks.last().unwrap();
assert_eq!(
consistent_provider.find_block_by_hash(first_in_mem_block.hash(), BlockSource::Any)?,
None
);
assert_eq!(
consistent_provider
.find_block_by_hash(first_in_mem_block.hash(), BlockSource::Canonical)?,
None
);
assert_eq!(
consistent_provider
.find_block_by_hash(first_in_mem_block.hash(), BlockSource::Pending)?,
None
);
let in_memory_block_senders =
first_in_mem_block.senders().expect("failed to recover senders");
let chain = NewCanonicalChain::Commit {
new: vec![ExecutedBlock::new(
Arc::new(first_in_mem_block.clone()),
Arc::new(in_memory_block_senders),
Default::default(),
Default::default(),
Default::default(),
)],
};
consistent_provider.canonical_in_memory_state.update_chain(chain);
let consistent_provider = provider.consistent_provider()?;
assert_eq!(
consistent_provider.find_block_by_hash(first_in_mem_block.hash(), BlockSource::Any)?,
Some(first_in_mem_block.clone().into())
);
assert_eq!(
consistent_provider
.find_block_by_hash(first_in_mem_block.hash(), BlockSource::Canonical)?,
Some(first_in_mem_block.clone().into())
);
assert_eq!(
consistent_provider.find_block_by_hash(first_db_block.hash(), BlockSource::Any)?,
Some(first_db_block.clone().into())
);
assert_eq!(
consistent_provider
.find_block_by_hash(first_db_block.hash(), BlockSource::Canonical)?,
Some(first_db_block.clone().into())
);
assert_eq!(
consistent_provider.find_block_by_hash(first_db_block.hash(), BlockSource::Pending)?,
None
);
provider.canonical_in_memory_state.set_pending_block(ExecutedBlock {
block: Arc::new(last_in_mem_block.clone()),
senders: Default::default(),
execution_output: Default::default(),
hashed_state: Default::default(),
trie: Default::default(),
});
assert_eq!(
consistent_provider
.find_block_by_hash(last_in_mem_block.hash(), BlockSource::Pending)?,
Some(last_in_mem_block.clone().into())
);
Ok(())
}
#[test]
fn test_block_reader_block() -> eyre::Result<()> {
let mut rng = generators::rng();
let factory = create_test_provider_factory();
let blocks = random_block_range(
&mut rng,
0..=10,
BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
);
let (database_blocks, in_memory_blocks) = blocks.split_at(5);
let provider_rw = factory.provider_rw()?;
for block in database_blocks {
provider_rw.insert_historical_block(
block.clone().seal_with_senders().expect("failed to seal block with senders"),
)?;
}
provider_rw.commit()?;
let provider = BlockchainProvider2::new(factory)?;
let consistent_provider = provider.consistent_provider()?;
let first_in_mem_block = in_memory_blocks.first().unwrap();
let first_db_block = database_blocks.first().unwrap();
assert_eq!(
consistent_provider.block(BlockHashOrNumber::Hash(first_in_mem_block.hash()))?,
None
);
assert_eq!(
consistent_provider.block(BlockHashOrNumber::Number(first_in_mem_block.number))?,
None
);
let in_memory_block_senders =
first_in_mem_block.senders().expect("failed to recover senders");
let chain = NewCanonicalChain::Commit {
new: vec![ExecutedBlock::new(
Arc::new(first_in_mem_block.clone()),
Arc::new(in_memory_block_senders),
Default::default(),
Default::default(),
Default::default(),
)],
};
consistent_provider.canonical_in_memory_state.update_chain(chain);
let consistent_provider = provider.consistent_provider()?;
assert_eq!(
consistent_provider.block(BlockHashOrNumber::Hash(first_in_mem_block.hash()))?,
Some(first_in_mem_block.clone().into())
);
assert_eq!(
consistent_provider.block(BlockHashOrNumber::Number(first_in_mem_block.number))?,
Some(first_in_mem_block.clone().into())
);
assert_eq!(
consistent_provider.block(BlockHashOrNumber::Hash(first_db_block.hash()))?,
Some(first_db_block.clone().into())
);
assert_eq!(
consistent_provider.block(BlockHashOrNumber::Number(first_db_block.number))?,
Some(first_db_block.clone().into())
);
Ok(())
}
#[test]
fn test_changeset_reader() -> eyre::Result<()> {
let mut rng = generators::rng();
let (database_blocks, in_memory_blocks) =
random_blocks(&mut rng, TEST_BLOCKS_COUNT, 1, None, None, 0..1);
let first_database_block = database_blocks.first().map(|block| block.number).unwrap();
let last_database_block = database_blocks.last().map(|block| block.number).unwrap();
let first_in_memory_block = in_memory_blocks.first().map(|block| block.number).unwrap();
let accounts = random_eoa_accounts(&mut rng, 2);
let (database_changesets, database_state) = random_changeset_range(
&mut rng,
&database_blocks,
accounts.into_iter().map(|(address, account)| (address, (account, Vec::new()))),
0..0,
0..0,
);
let (in_memory_changesets, in_memory_state) = random_changeset_range(
&mut rng,
&in_memory_blocks,
database_state
.iter()
.map(|(address, (account, storage))| (*address, (*account, storage.clone()))),
0..0,
0..0,
);
let factory = create_test_provider_factory();
let provider_rw = factory.provider_rw()?;
provider_rw.append_blocks_with_state(
database_blocks
.into_iter()
.map(|b| b.seal_with_senders().expect("failed to seal block with senders"))
.collect(),
ExecutionOutcome {
bundle: BundleState::new(
database_state.into_iter().map(|(address, (account, _))| {
(address, None, Some(account.into()), Default::default())
}),
database_changesets
.iter()
.map(|block_changesets| {
block_changesets.iter().map(|(address, account, _)| {
(*address, Some(Some((*account).into())), [])
})
})
.collect::<Vec<_>>(),
Vec::new(),
),
first_block: first_database_block,
..Default::default()
},
Default::default(),
Default::default(),
)?;
provider_rw.commit()?;
let provider = BlockchainProvider2::new(factory)?;
let in_memory_changesets = in_memory_changesets.into_iter().next().unwrap();
let chain = NewCanonicalChain::Commit {
new: vec![in_memory_blocks
.first()
.map(|block| {
let senders = block.senders().expect("failed to recover senders");
ExecutedBlock::new(
Arc::new(block.clone()),
Arc::new(senders),
Arc::new(ExecutionOutcome {
bundle: BundleState::new(
in_memory_state.into_iter().map(|(address, (account, _))| {
(address, None, Some(account.into()), Default::default())
}),
[in_memory_changesets.iter().map(|(address, account, _)| {
(*address, Some(Some((*account).into())), Vec::new())
})],
[],
),
first_block: first_in_memory_block,
..Default::default()
}),
Default::default(),
Default::default(),
)
})
.unwrap()],
};
provider.canonical_in_memory_state.update_chain(chain);
let consistent_provider = provider.consistent_provider()?;
assert_eq!(
consistent_provider.account_block_changeset(last_database_block).unwrap(),
database_changesets
.into_iter()
.last()
.unwrap()
.into_iter()
.sorted_by_key(|(address, _, _)| *address)
.map(|(address, account, _)| AccountBeforeTx { address, info: Some(account) })
.collect::<Vec<_>>()
);
assert_eq!(
consistent_provider.account_block_changeset(first_in_memory_block).unwrap(),
in_memory_changesets
.into_iter()
.sorted_by_key(|(address, _, _)| *address)
.map(|(address, account, _)| AccountBeforeTx { address, info: Some(account) })
.collect::<Vec<_>>()
);
Ok(())
}
}