//! Smart caching and deduplication of requests when using a forking provider.

use crate::{

cache::{

AccountFetchPolicy, BlockchainDb, FlushJsonBlockCacheDB, ForkBlockEnv, MemDb, StorageInfo,

},

error::{DatabaseError, DatabaseResult},

};

use alloy_chains::Chain;

use alloy_consensus::BlockHeader;

use alloy_primitives::{Address, B256, Bytes, U64, U256, keccak256, map::U256Map};

use alloy_provider::{

DynProvider, Network, Provider,

network::{AnyNetwork, BlockResponse, primitives::HeaderResponse},

};

use alloy_rpc_types::BlockId;

use eyre::WrapErr;

use futures::{

FutureExt,

channel::mpsc::{UnboundedReceiver, UnboundedSender, unbounded},

pin_mut,

stream::Stream,

task::{Context, Poll},

};

use revm::{

context::BlockEnv,

database::DatabaseRef,

primitives::{

KECCAK_EMPTY,

map::{AddressHashMap, HashMap, hash_map::Entry},

},

state::{AccountInfo, Bytecode},

};

use serde::Serialize;

use std::{

collections::VecDeque,

fmt,

path::Path,

pin::Pin,

sync::{

Arc,

atomic::{AtomicU8, Ordering},

mpsc::{Sender as OneshotSender, channel as oneshot_channel},

},

};

use tokio::select;

/// Keep state reads hash-addressed without relying on the bare-hash extension to EIP-1898.

/// Moonbeam parses bare hashes as block numbers, while Rootstock rejects a boolean

/// `requireCanonical`. An explicit hash object with the optional flag omitted supports both.

#[derive(Clone, Copy, Debug)]

struct StateBlockId(BlockId);

impl Serialize for StateBlockId {

fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {

if let BlockId::Hash(hash) = self.0

&& hash.require_canonical.is_none()

{

use serde::ser::SerializeStruct;

let mut object = serializer.serialize_struct("StateBlockId", 1)?;

object.serialize_field("blockHash", &hash.block_hash)?;

object.end()

} else {

self.0.serialize(serializer)

}

}

}

/// Logged when an error is indicative that the user is trying to fork from a non-archive node.

pub const NON_ARCHIVE_NODE_WARNING: &str = "\

It looks like you're trying to fork from an older block with a non-archive node which is not \

supported. Please try to change your RPC url to an archive node if the issue persists.";

// Various future/request type aliases

type AccountFuture<Err> =

Pin<Box<dyn Future<Output = (Result<(U256, u64, Bytes), Err>, Address)> + Send>>;

type StorageFuture<Err> = Pin<Box<dyn Future<Output = (Result<U256, Err>, Address, U256)> + Send>>;

type BlockHashFuture<Err> =

Pin<Box<dyn Future<Output = (Result<BlockHashData, Err>, u64, u64, bool)> + Send>>;

type FullBlockFuture<Err, N = AnyNetwork> = Pin<

Box<

dyn Future<

Output = (

FullBlockSender<N>,

Result<Option<<N as Network>::BlockResponse>, Err>,

BlockId,

),

> + Send,

>,

>;

type TransactionFuture<Err, N = AnyNetwork> = Pin<

Box<

dyn Future<

Output = (

TransactionSender<N>,

Result<<N as Network>::TransactionResponse, Err>,

B256,

),

> + Send,

>,

>;

type AccountInfoSender = OneshotSender<DatabaseResult<AccountInfo>>;

type StorageSender = OneshotSender<DatabaseResult<U256>>;

type BlockHashSender = OneshotSender<DatabaseResult<B256>>;

type FullBlockSender<N = AnyNetwork> = OneshotSender<DatabaseResult<<N as Network>::BlockResponse>>;

type TransactionSender<N = AnyNetwork> =

OneshotSender<DatabaseResult<<N as Network>::TransactionResponse>>;

type AddressData = AddressHashMap<AccountInfo>;

type StorageData = AddressHashMap<StorageInfo>;

type BlockHashData = U256Map<B256>;

/// States for tracking which account endpoints should be used when account info

const ACCOUNT_FETCH_UNCHECKED: u8 = 0;

/// Endpoints supports the non standard eth_getAccountInfo which is more efficient than sending 3

/// separate requests

const ACCOUNT_FETCH_SUPPORTS_ACC_INFO: u8 = 1;

/// Use regular individual getCode, getNonce, getBalance calls

const ACCOUNT_FETCH_SEPARATE_REQUESTS: u8 = 2;

struct AnyRequestFuture<T, Err> {

sender: OneshotSender<Result<T, Err>>,

future: Pin<Box<dyn Future<Output = Result<T, Err>> + Send>>,

}

impl<T, Err> fmt::Debug for AnyRequestFuture<T, Err> {

fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

f.debug_tuple("AnyRequestFuture").field(&self.sender).finish()

}

}

trait WrappedAnyRequest: Unpin + Send + fmt::Debug {

fn poll_inner(&mut self, cx: &mut Context<'_>) -> Poll<()>;

}

/// @dev Implements `WrappedAnyRequest` for `AnyRequestFuture`.

///

/// - `poll_inner` is similar to `Future` polling but intentionally consumes the Future<Output=T>

/// and return Future<Output=()>

/// - This design avoids storing `Future<Output = T>` directly, as its type may not be known at

/// compile time.

/// - Instead, the result (`Result<T, Err>`) is sent via the `sender` channel, which enforces type

/// safety.

impl<T, Err> WrappedAnyRequest for AnyRequestFuture<T, Err>

where

T: fmt::Debug + Send + 'static,

Err: fmt::Debug + Send + 'static,

{

fn poll_inner(&mut self, cx: &mut Context<'_>) -> Poll<()> {

match self.future.poll_unpin(cx) {

Poll::Ready(result) => {

let _ = self.sender.send(result);

Poll::Ready(())

}

Poll::Pending => Poll::Pending,

}

}

}

/// Request variants that are executed by the provider

enum ProviderRequest<Err, N: Network = AnyNetwork> {

Account(AccountFuture<Err>),

Storage(StorageFuture<Err>),

BlockHash(BlockHashFuture<Err>),

FullBlock(FullBlockFuture<Err, N>),

Transaction(TransactionFuture<Err, N>),

AnyRequest(Box<dyn WrappedAnyRequest>),

}

/// The Request type the Backend listens for

#[derive(Debug)]

enum BackendRequest<N: Network = AnyNetwork> {

/// Fetch the account info

Basic(Address, AccountInfoSender),

/// Fetch a storage slot

Storage(Address, U256, StorageSender),

/// Fetch a block hash

BlockHash(u64, BlockHashSender),

/// Fetch an entire block with transactions

FullBlock(BlockId, FullBlockSender<N>),

/// Fetch a transaction

Transaction(B256, TransactionSender<N>),

/// Sets the pinned block to fetch data from

SetPinnedBlock(BlockId),

/// Update Address data

UpdateAddress(AddressData),

/// Update Storage data

UpdateStorage(StorageData),

/// Update Block Hashes

UpdateBlockHash(BlockHashData),

/// Any other request

AnyRequest(Box<dyn WrappedAnyRequest>),

}

/// Handles an internal provider and listens for requests.

///

/// This handler will remain active as long as it is reachable (request channel still open) and

/// requests are in progress.

#[must_use = "futures do nothing unless polled"]

pub struct BackendHandler<N: Network = AnyNetwork, B = BlockEnv> {

provider: DynProvider<N>,

/// Stores all the data.

db: BlockchainDb<B>,

/// Requests currently in progress

pending_requests: Vec<ProviderRequest<eyre::Report, N>>,

/// Listeners that wait for a `get_account` related response

account_requests: HashMap<Address, Vec<AccountInfoSender>>,

/// Listeners that wait for a `get_storage_at` response

storage_requests: HashMap<(Address, U256), Vec<StorageSender>>,

/// Listeners that wait for a `get_block` response, keyed by pin generation and block number.

block_requests: HashMap<(u64, u64), Vec<BlockHashSender>>,

/// Block hashes whose value depends on the pinned block.

pinned_block_hashes: HashMap<(u64, u64), B256>,

/// Incoming commands.

incoming: UnboundedReceiver<BackendRequest<N>>,

/// unprocessed queued requests

queued_requests: VecDeque<BackendRequest<N>>,

/// The block to fetch data from.

// This is an `Option` so that we can have less code churn in the functions below

block_id: Option<BlockId>,

/// Exact block anchoring state reads and block ancestry.

block_anchor: Option<ForkBlock>,

/// Incremented whenever the pinned block changes.

block_generation: u64,

/// Whether block hashes are resolved in the remote EVM.

block_hash_via_evm: Option<bool>,

/// The mode for fetching account data

account_fetch_mode: Arc<AtomicU8>,

/// Account-loading policy fixed when this backend is created.

account_fetch_policy: AccountFetchPolicy,

}

impl<N: Network, B: ForkBlockEnv> BackendHandler<N, B> {

fn new(

provider: DynProvider<N>,

db: BlockchainDb<B>,

rx: UnboundedReceiver<BackendRequest<N>>,

block_id: Option<BlockId>,

block_anchor: Option<ForkBlock>,

) -> Self {

let block_hash_via_evm = db.meta().read().chain.and_then(|chain| {

if chain.is_arbitrum() {

Some(true)

} else if chain.is_named() {

Some(false)

} else {

None

}

});

let account_fetch_policy = db.meta().read().account_fetch_policy;

Self {

provider,

db,

account_fetch_policy,

pending_requests: Default::default(),

account_requests: Default::default(),

storage_requests: Default::default(),

block_requests: Default::default(),

pinned_block_hashes: Default::default(),

queued_requests: Default::default(),

incoming: rx,

block_id,

block_anchor,

block_generation: 0,

block_hash_via_evm,

account_fetch_mode: Arc::new(AtomicU8::new(ACCOUNT_FETCH_UNCHECKED)),

}

}

/// handle the request in queue in the future.

///

/// We always check:

/// 1. if the requested value is already stored in the cache, then answer the sender

/// 2. otherwise, fetch it via the provider but check if a request for that value is already in

/// progress (e.g. another Sender just requested the same account)

fn on_request(&mut self, req: BackendRequest<N>) {

match req {

BackendRequest::Basic(addr, sender) => {

trace!(target: "backendhandler", "received request basic address={:?}", addr);

let acc = self.db.accounts().read().get(&addr).cloned();

if let Some(basic) = acc {

self.db.cache().record_cache_hit();

let _ = sender.send(Ok(basic));

} else {

self.request_account(addr, sender);

}

}

BackendRequest::BlockHash(number, sender) => {

if self.block_anchor.is_some_and(|anchor| number > anchor.number) {

let _ = sender.send(Ok(B256::ZERO));

return;

}

let hash = self

.pinned_block_hashes

.get(&(self.block_generation, number))

.copied()

.or_else(|| self.db.block_hashes().read().get(&U256::from(number)).copied());

if let Some(hash) = hash {

self.db.cache().record_cache_hit();

let _ = sender.send(Ok(hash));

} else {

self.request_hash(number, sender);

}

}

BackendRequest::FullBlock(number, sender) => {

self.request_full_block(number, sender);

}

BackendRequest::Transaction(tx, sender) => {

self.request_transaction(tx, sender);

}

BackendRequest::Storage(addr, idx, sender) => {

// account is already stored in the cache

let value =

self.db.storage().read().get(&addr).and_then(|acc| acc.get(&idx).copied());

if let Some(value) = value {

self.db.cache().record_cache_hit();

let _ = sender.send(Ok(value));

} else {

// account present but not storage -> fetch storage

self.request_account_storage(addr, idx, sender);

}

}

BackendRequest::SetPinnedBlock(block_id) => {

self.block_id = Some(block_id);

self.block_anchor = None;

self.block_generation = self.block_generation.wrapping_add(1);

}

BackendRequest::UpdateAddress(address_data) => {

for (address, data) in address_data {

self.db.accounts().write().insert(address, data);

}

}

BackendRequest::UpdateStorage(storage_data) => {

for (address, data) in storage_data {

self.db.storage().write().insert(address, data);

}

}

BackendRequest::UpdateBlockHash(block_hash_data) => {

for (block, hash) in block_hash_data {

self.db.block_hashes().write().insert(block, hash);

}

}

BackendRequest::AnyRequest(fut) => {

self.pending_requests.push(ProviderRequest::AnyRequest(fut));

}

}

}

/// process a request for account's storage

fn request_account_storage(&mut self, address: Address, idx: U256, listener: StorageSender) {

match self.storage_requests.entry((address, idx)) {

Entry::Occupied(mut entry) => {

entry.get_mut().push(listener);

}

Entry::Vacant(entry) => {

trace!(target: "backendhandler", %address, %idx, "preparing storage request");

self.db.cache().record_cache_miss();

entry.insert(vec![listener]);

let provider = self.provider.clone();

let block_id = self.block_id.unwrap_or_default();

let fut = Box::pin(async move {

let storage = provider

.raw_request(

"eth_getStorageAt".into(),

(address, idx, StateBlockId(block_id)),

)

.await

.map_err(Into::into);

(storage, address, idx)

});

self.pending_requests.push(ProviderRequest::Storage(fut));

}

}

}

/// returns the future that fetches the account data

fn get_account_req(&self, address: Address) -> ProviderRequest<eyre::Report, N> {

trace!(target: "backendhandler", "preparing account request, address={:?}", address);

let provider = self.provider.clone();

let block_id = self.block_id.unwrap_or_default();

let mode = Arc::clone(&self.account_fetch_mode);

let policy = self.account_fetch_policy;

let fut = async move {

if policy == AccountFetchPolicy::RequireAccountInfo {

return Self::fetch_account_info(&provider, address, block_id)

.await

.map(|info| (info.balance, info.nonce, info.code))

.wrap_err("fork account policy requires eth_getAccountInfo");

}

// depending on the tracked mode we can dispatch requests.

let initial_mode = mode.load(Ordering::Relaxed);

match initial_mode {

ACCOUNT_FETCH_UNCHECKED => {

// single request for accountinfo object

let acc_info_fut = Self::fetch_account_info(&provider, address, block_id);

// tri request for account info

let triple_fut = Self::fetch_account_separately(&provider, address, block_id);

pin_mut!(acc_info_fut, triple_fut);

select! {

acc_info = &mut acc_info_fut => {

match acc_info {

Ok(info) => {

trace!(target: "backendhandler", "endpoint supports eth_getAccountInfo");

mode.store(ACCOUNT_FETCH_SUPPORTS_ACC_INFO, Ordering::Relaxed);

Ok((info.balance, info.nonce, info.code))

}

Err(err) => {

trace!(target: "backendhandler", ?err, "failed initial eth_getAccountInfo call");

mode.store(ACCOUNT_FETCH_SEPARATE_REQUESTS, Ordering::Relaxed);

Ok(triple_fut.await?)

}

}

}

triple = &mut triple_fut => {

match triple {

Ok((balance, nonce, code)) => {

mode.store(ACCOUNT_FETCH_SEPARATE_REQUESTS, Ordering::Relaxed);

Ok((balance, nonce, code))

}

Err(err) => Err(err)

}

}

}

}

ACCOUNT_FETCH_SUPPORTS_ACC_INFO => {

let mut res = Self::fetch_account_info(&provider, address, block_id)

.await

.map(|info| (info.balance, info.nonce, info.code));

// it's possible that the configured endpoint load balances requests to multiple

// instances and not all support that endpoint so we should reset here

if res.is_err() {

mode.store(ACCOUNT_FETCH_SEPARATE_REQUESTS, Ordering::Relaxed);

res = Self::fetch_account_separately(&provider, address, block_id).await;

}

Ok(res?)

}

ACCOUNT_FETCH_SEPARATE_REQUESTS => {

Self::fetch_account_separately(&provider, address, block_id).await

}

_ => unreachable!("Invalid account fetch mode"),

}

};

ProviderRequest::Account(Box::pin(async move {

let result = fut.await;

(result, address)

}))

}

async fn fetch_account_info(

provider: &DynProvider<N>,

address: Address,

block_id: BlockId,

) -> eyre::Result<alloy_rpc_types::AccountInfo> {

Ok(provider

.raw_request("eth_getAccountInfo".into(), (address, StateBlockId(block_id)))

.await?)

}

async fn fetch_account_separately(

provider: &DynProvider<N>,

address: Address,

block_id: BlockId,

) -> eyre::Result<(U256, u64, Bytes)> {

let params = (address, StateBlockId(block_id));

let (balance, nonce, code): (U256, U64, Bytes) = futures::future::try_join3(

provider.raw_request("eth_getBalance".into(), params),

provider.raw_request("eth_getTransactionCount".into(), params),

provider.raw_request("eth_getCode".into(), params),

)

.await?;

Ok((balance, nonce.to(), code))

}

/// process a request for an account

fn request_account(&mut self, address: Address, listener: AccountInfoSender) {

match self.account_requests.entry(address) {

Entry::Occupied(mut entry) => {

entry.get_mut().push(listener);

}

Entry::Vacant(entry) => {

self.db.cache().record_cache_miss();

entry.insert(vec![listener]);

self.pending_requests.push(self.get_account_req(address));

}

}

}

/// process a request for an entire block

fn request_full_block(&mut self, number: BlockId, sender: FullBlockSender<N>) {

let provider = self.provider.clone();

let fut = Box::pin(async move {

let block = provider

.get_block(number)

.full()

.await

.wrap_err(format!("could not fetch block {number:?}"));

(sender, block, number)

});

self.pending_requests.push(ProviderRequest::FullBlock(fut));

}

/// process a request for a transactions

fn request_transaction(&mut self, tx: B256, sender: TransactionSender<N>) {

let provider = self.provider.clone();

let fut = Box::pin(async move {

let block = provider

.get_transaction_by_hash(tx)

.await

.wrap_err_with(|| format!("could not get transaction {tx}"))

.and_then(|maybe| {

maybe.ok_or_else(|| eyre::eyre!("could not get transaction {tx}"))

});

(sender, block, tx)

});

self.pending_requests.push(ProviderRequest::Transaction(fut));

}

/// Resolves an Arbitrum `BLOCKHASH` request in the remote EVM.

///

/// Arbitrum's EVM block number is the L1 block number, while `eth_getBlockByNumber` uses the L2

/// block number. Executing `BLOCKHASH` lets ArbOS apply its L1-to-L2 block hash mapping.

async fn get_arbitrum_block_hash(

provider: &DynProvider<N>,

block_id: Option<BlockId>,

number: u64,

) -> eyre::Result<B256> {

let mut code = Vec::with_capacity(41);

code.push(0x7f); // PUSH32

code.extend_from_slice(&U256::from(number).to_be_bytes::<32>());

code.extend_from_slice(&[

0x40, // BLOCKHASH

0x60, 0x00, // PUSH1 0

0x52, // MSTORE

0x60, 0x20, // PUSH1 32

0x60, 0x00, // PUSH1 0

0xf3, // RETURN

]);

let address = Address::with_last_byte(0xde);

let code = Bytes::from(code);

let params = serde_json::json!([

{ "to": address, "data": code },

StateBlockId(block_id.unwrap_or_else(BlockId::latest)),

{ address.to_string(): { "code": code } },

]);

let output: Bytes = provider

.raw_request("eth_call".into(), params)

.await

.wrap_err("failed to get Arbitrum block hash")?;

eyre::ensure!(

output.len() == 32,

"invalid Arbitrum block hash response length: expected 32 bytes, got {}",

output.len()

);

Ok(B256::from_slice(&output))

}

/// Returns whether the standard ArbSys precompile is available at the pinned block.

async fn has_arbsys(

provider: &DynProvider<N>,

block_id: Option<BlockId>,

) -> eyre::Result<bool> {

let params = serde_json::json!([

{

"to": Address::with_last_byte(0x64),

"data": "0x051038f2", // arbOSVersion()

},

StateBlockId(block_id.unwrap_or_else(BlockId::latest)),

]);

let output = provider

.raw_request::<_, Bytes>("eth_call".into(), params)

.await

.wrap_err("failed to detect ArbSys")?;

Ok(output.len() == 32 && U256::from_be_slice(&output) >= U256::from(56))

}

/// process a request for a block hash

fn request_hash(&mut self, number: u64, listener: BlockHashSender) {

let generation = self.block_generation;

match self.block_requests.entry((generation, number)) {

Entry::Occupied(mut entry) => {

entry.get_mut().push(listener);

}

Entry::Vacant(entry) => {

trace!(target: "backendhandler", number, "preparing block hash request");

self.db.cache().record_cache_miss();

entry.insert(vec![listener]);

let provider = self.provider.clone();

let anchor = self.block_anchor;

let block_id = self.block_id;

let meta = Arc::clone(self.db.meta());

let known_block_hash_via_evm = self.block_hash_via_evm;

let fut = Box::pin(async move {

let result: eyre::Result<(BlockHashData, bool)> = async {

let cached_chain = meta.read().chain;

let chain = if let Some(chain) = cached_chain {

chain

} else {

let chain = Chain::from(

provider.get_chain_id().await.wrap_err("failed to get chain ID")?,

);

meta.write().chain = Some(chain);

chain

};

let via_evm = match known_block_hash_via_evm {

Some(value) => value,

None => {

chain.is_arbitrum()

|| (chain.is_id()

&& Self::has_arbsys(&provider, block_id).await?)

}

};

if via_evm {

let hash =

Self::get_arbitrum_block_hash(&provider, block_id, number).await?;

return Ok((

BlockHashData::from_iter([(U256::from(number), hash)]),

true,

));

}

let hashes = if let Some(anchor) = anchor

&& number < anchor.number

&& anchor.number == anchor.rpc_number

{

let mut hashes = BlockHashData::default();

let mut descendant = anchor;

while descendant.number > number {

let block = provider

.get_block_by_hash(descendant.hash)

.hashes()

.await

.wrap_err_with(|| {

format!(

"failed to get anchored block {} ({})",

descendant.rpc_number, descendant.hash

)

})?

.ok_or_else(|| {

eyre::eyre!(

"anchored block {} ({}) not found",

descendant.rpc_number,

descendant.hash

)

})?;

let header = block.header();

eyre::ensure!(

header.number() == descendant.rpc_number &&

header.hash() == descendant.hash,

"anchored block changed: expected {} ({}), got {} ({})",

descendant.rpc_number,

descendant.hash,

header.number(),

header.hash()

);

let rpc_number = descendant.rpc_number.checked_sub(1).ok_or_else(

|| eyre::eyre!("anchored ancestry precedes the genesis block"),

)?;

descendant = ForkBlock::with_rpc_number(

descendant.number - 1,

rpc_number,

header.parent_hash(),

);

hashes.insert(U256::from(descendant.number), descendant.hash);

}

hashes

} else if anchor.is_none_or(|anchor| anchor.number != anchor.rpc_number) {

let block = provider

.get_block_by_number(number.into())

.hashes()

.await

.wrap_err("failed to get block");

match block {

Ok(Some(block)) => BlockHashData::from_iter([(

U256::from(number),

block.header().hash(),

)]),

Ok(None) => {

warn!(target: "backendhandler", ?number, "block not found");

BlockHashData::from_iter([(

U256::from(number),

KECCAK_EMPTY,

)])

}

Err(err) => {

error!(target: "backendhandler", %err, ?number, "failed to get block");

return Err(err)

}

}

} else {

BlockHashData::from_iter([(U256::from(number), B256::ZERO)])

};

Ok((hashes, false))

}

.await;

let via_evm = result.as_ref().is_ok_and(|(_, via_evm)| *via_evm);

let block_hashes = result.map(|(hashes, _)| hashes);

(block_hashes, number, generation, via_evm)

});

self.pending_requests.push(ProviderRequest::BlockHash(fut));

}

}

}

}

impl<N: Network, B: ForkBlockEnv> Future for BackendHandler<N, B> {

type Output = ();

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

let pin = self.get_mut();

loop {

// Drain queued requests first.

while let Some(req) = pin.queued_requests.pop_front() {

pin.on_request(req)

}

// receive new requests to delegate to the underlying provider

loop {

match Pin::new(&mut pin.incoming).poll_next(cx) {

Poll::Ready(Some(req)) => {

pin.queued_requests.push_back(req);

}

Poll::Ready(None) => {

trace!(target: "backendhandler", "last sender dropped, ready to drop (&flush cache)");

return Poll::Ready(());

}

Poll::Pending => break,

}

}

// poll all requests in progress

for n in (0..pin.pending_requests.len()).rev() {

let mut request = pin.pending_requests.swap_remove(n);

match &mut request {

ProviderRequest::Account(fut) => {

if let Poll::Ready((resp, addr)) = fut.poll_unpin(cx) {

// get the response

let (balance, nonce, code) = match resp {

Ok(res) => res,

Err(err) => {

let err = Arc::new(err);

if let Some(listeners) = pin.account_requests.remove(&addr) {

for l in listeners {

let _ = l.send(Err(DatabaseError::GetAccount(

addr,

Arc::clone(&err),

)));

}

}

continue;

}

};

// convert it to revm-style types

let (code, code_hash) = if code.is_empty() {

(Bytes::default(), KECCAK_EMPTY)

} else {

(code.clone(), keccak256(&code))

};

// update the cache

let acc = AccountInfo {

nonce,

balance,

code: Some(Bytecode::new_raw(code)),

code_hash,

account_id: None,

};

pin.db.accounts().write().insert(addr, acc.clone());

// notify all listeners

if let Some(listeners) = pin.account_requests.remove(&addr) {

for l in listeners {

let _ = l.send(Ok(acc.clone()));

}

}

continue;

}

}

ProviderRequest::Storage(fut) => {

if let Poll::Ready((resp, addr, idx)) = fut.poll_unpin(cx) {

let value = match resp {

Ok(value) => value,

Err(err) => {

// notify all listeners

let err = Arc::new(err);

if let Some(listeners) =

pin.storage_requests.remove(&(addr, idx))

{

for l in listeners {

let _ = l.send(Err(DatabaseError::GetStorage(

addr,

idx,

Arc::clone(&err),

)));

}

}

continue;

}

};

// update the cache

pin.db.storage().write().entry(addr).or_default().insert(idx, value);

// notify all listeners

if let Some(listeners) = pin.storage_requests.remove(&(addr, idx)) {

for l in listeners {

let _ = l.send(Ok(value));

}

}

continue;

}

}

ProviderRequest::BlockHash(fut) => {

if let Poll::Ready((block_hash, number, generation, via_evm)) =

fut.poll_unpin(cx)

{

let request_key = (generation, number);

let hashes = match block_hash {

Ok(value) => value,

Err(err) => {

let err = Arc::new(err);

// notify all listeners

if let Some(listeners) = pin.block_requests.remove(&request_key)

{

for l in listeners {

let _ = l.send(Err(DatabaseError::GetBlockHash(

number,

Arc::clone(&err),

)));

}

}

continue;

}

};

// update the cache

let value = hashes[&U256::from(number)];

if via_evm {

pin.pinned_block_hashes.insert(request_key, value);

} else {

pin.db.block_hashes().write().extend(hashes);

}

pin.block_hash_via_evm = Some(via_evm);

// notify all listeners

if let Some(listeners) = pin.block_requests.remove(&request_key) {

for l in listeners {

let _ = l.send(Ok(value));

}

}

continue;

}

}

ProviderRequest::FullBlock(fut) => {

if let Poll::Ready((sender, resp, number)) = fut.poll_unpin(cx) {

let msg = match resp {

Ok(Some(block)) => Ok(block),

Ok(None) => Err(DatabaseError::BlockNotFound(number)),

Err(err) => {

let err = Arc::new(err);

Err(DatabaseError::GetFullBlock(number, err))

}

};

let _ = sender.send(msg);

continue;

}

}

ProviderRequest::Transaction(fut) => {

if let Poll::Ready((sender, tx, tx_hash)) = fut.poll_unpin(cx) {

let msg = match tx {

Ok(tx) => Ok(tx),

Err(err) => {

let err = Arc::new(err);

Err(DatabaseError::GetTransaction(tx_hash, err))

}

};

let _ = sender.send(msg);

continue;

}

}

ProviderRequest::AnyRequest(fut) => {

if fut.poll_inner(cx).is_ready() {

continue;

}

}

}

// not ready, insert and poll again

pin.pending_requests.push(request);

}

// If no new requests have been queued, break to

// be polled again later.

if pin.queued_requests.is_empty() {

return Poll::Pending;

}

}

}

}

/// Mode for the `SharedBackend` how to block in the non-async [`DatabaseRef`] when interacting with

/// [`BackendHandler`].

#[derive(Default, Clone, Debug, PartialEq, Eq)]

pub enum BlockingMode {

/// This mode use `tokio::task::block_in_place()` to block in place.

///

/// This should be used when blocking on the call site is disallowed.

#[default]

BlockInPlace,

/// The mode blocks the current task

///

/// This can be used if blocking on the call site is allowed, e.g. on a tokio blocking task.

Block,

}

/// A block number and hash that identify the exact root of a fork.

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub struct ForkBlock {

/// EVM-visible block number used to key `BLOCKHASH` requests.

pub number: u64,

/// RPC block number used to validate fetched headers.

pub rpc_number: u64,

/// Block hash of the fork root.

pub hash: B256,

}

impl ForkBlock {

/// Creates a new exact fork root.

pub const fn new(number: u64, hash: B256) -> Self {

Self { number, rpc_number: number, hash }

}

/// Creates an exact fork root whose RPC and EVM-visible block numbers differ.

pub const fn with_rpc_number(number: u64, rpc_number: u64, hash: B256) -> Self {

Self { number, rpc_number, hash }

}

}

impl BlockingMode {

/// run process logic with the blocking mode

pub fn run<F, R>(&self, f: F) -> R

where

F: FnOnce() -> R,

{

match self {

Self::BlockInPlace => tokio::task::block_in_place(f),

Self::Block => f(),

}

}

}

/// A cloneable backend type that shares access to the backend data with all its clones.

///

/// This backend type is connected to the `BackendHandler` via a mpsc unbounded channel. The

/// `BackendHandler` is spawned on a tokio task and listens for incoming commands on the receiver

/// half of the channel. A `SharedBackend` holds a sender for that channel, which is `Clone`, so

/// there can be multiple `SharedBackend`s communicating with the same `BackendHandler`, hence this

/// `Backend` type is thread safe.

///

/// All `Backend` trait functions are delegated as a `BackendRequest` via the channel to the

/// `BackendHandler`. All `BackendRequest` variants include a sender half of an additional channel

/// that is used by the `BackendHandler` to send the result of an executed `BackendRequest` back to

/// `SharedBackend`.

///

/// The `BackendHandler` holds a `Provider` to look up missing accounts or storage slots

/// from remote (e.g. infura). It detects duplicate requests from multiple `SharedBackend`s and

/// bundles them together, so that always only one provider request is executed. For example, there

/// are two `SharedBackend`s, `A` and `B`, both request the basic account info of account

/// `0xasd9sa7d...` at the same time. After the `BackendHandler` receives the request from `A`, it

/// sends a new provider request to the provider's endpoint, then it reads the identical request

/// from `B` and simply adds it as an additional listener for the request already in progress,

/// instead of sending another one. So that after the provider returns the response all listeners

/// (`A` and `B`) get notified.

// **Note**: the implementation makes use of [tokio::task::block_in_place()] when interacting with

// the underlying [BackendHandler] which runs on a separate spawned tokio task.

// [tokio::task::block_in_place()]

// > Runs the provided blocking function on the current thread without blocking the executor.

// This prevents issues (hangs) we ran into were the [SharedBackend] itself is called from a spawned

// task.

#[derive(Clone, Debug)]

pub struct SharedBackend<N: Network = AnyNetwork, B: Serialize + Clone = BlockEnv> {

/// channel used for sending commands related to database operations

backend: UnboundedSender<BackendRequest<N>>,

/// Ensures that the underlying cache gets flushed once the last `SharedBackend` is dropped.