anvil/eth/

api.rs

use super::{
    backend::mem::{
        BlockRequest, DatabaseRef, GasEstimateCallOptions, MonadReplayContext, State,
        sanitize_simulation_blocks,
    },
    preserve_simulation_request_fields,
};
use crate::{
    ClientFork, LoggingManager, Miner, MiningMode, StorageInfo,
    eth::{
        backend::{
            self, db::SerializableState, mem::MIN_TRANSACTION_GAS,
            notifications::ChainNotifications, validate::TransactionValidator,
        },
        error::{
            BlockchainError, FeeHistoryError, InvalidTransactionError, Result, ToRpcResponseResult,
        },
        fees::{
            FeeDetails, FeeHistoryCache, FeeHistoryCacheItem, MIN_SUGGESTED_PRIORITY_FEE,
            REWARD_PERCENTILE_RESOLUTION, create_fee_history_cache_item,
        },
        macros::node_info,
        miner::FixedBlockTimeMiner,
        pool::{
            Pool, PoolSnapshot,
            transactions::{
                PoolTransaction, TransactionOrder, TransactionPriority, TxMarker, to_marker,
            },
        },
        sign::Signer,
    },
    filter::{EthFilter, Filters, LogsFilter},
    mem::transaction_build,
};
use alloy_consensus::{
    Blob, BlockHeader, Transaction, TrieAccount, TxEip4844Variant, TxReceipt, Typed2718,
    transaction::{Recovered, SignerRecoverable},
};
use alloy_dyn_abi::TypedData;
use alloy_eips::{
    eip2718::{EIP4844_TX_TYPE_ID, Encodable2718},
    eip4844::DATA_GAS_PER_BLOB,
    eip7910::{EthConfig, EthForkConfig},
};
use alloy_evm::overrides::{OverrideBlockHashes, apply_state_overrides};
use alloy_network::{
    AnyRpcBlock, AnyRpcHeader, AnyRpcTransaction, BlockResponse, Network,
    NetworkTransactionBuilder, ReceiptResponse, TransactionBuilder, TransactionBuilder4844,
    TransactionResponse, eip2718::Decodable2718,
};
use alloy_primitives::{
    Address, B64, B256, Bytes, TxHash, TxKind, U64, U256,
    map::{AddressSet, B256Set, HashMap, HashSet},
};
use alloy_rlp::{Encodable, Header, PayloadView};
use alloy_rpc_types::{
    AccessListResult, BlockId, BlockNumberOrTag as BlockNumber, BlockTransactions,
    EIP1186AccountProofResponse, FeeHistory, Filter, FilteredParams, Index, Log, Work,
    anvil::{
        ForkedNetwork, Forking, Metadata, MineOptions, NodeEnvironment, NodeForkConfig, NodeInfo,
    },
    debug::ExecutionWitness,
    erc4337::TransactionConditional,
    pubsub::TransactionReceiptsParams,
    request::TransactionRequest,
    simulate::{MAX_SIMULATE_BLOCKS, SimulatePayload, SimulatedBlock},
    state::{AccountOverride, EvmOverrides, StateOverride, StateOverridesBuilder},
    trace::{
        filter::TraceFilter,
        geth::{GethDebugTracingCallOptions, GethDebugTracingOptions, GethTrace, TraceResult},
        opcode::{BlockOpcodeGas, TransactionOpcodeGas},
        parity::{
            LocalizedTransactionTrace, TraceResults, TraceResultsWithTransactionHash, TraceType,
        },
    },
    txpool::{TxpoolContent, TxpoolContentFrom, TxpoolInspect, TxpoolInspectSummary, TxpoolStatus},
};
use alloy_rpc_types_eth::{AccountInfo, Bundle, EthCallResponse, FillTransaction, StateContext};
use alloy_rpc_types_mev::{EthCallBundle, EthCallBundleResponse};
use alloy_serde::WithOtherFields;
use alloy_sol_types::{SolCall, SolValue, sol};
use anvil_core::{
    eth::{
        EthRequest,
        block::{Block, BlockInfo, canonical_block},
        transaction::{MaybeImpersonatedTransaction, PendingTransaction},
    },
    types::{ReorgOptions, TransactionData},
};
use anvil_rpc::{
    error::{ErrorCode, RpcError},
    response::ResponseResult,
};
use foundry_common::{
    provider::redact_url,
    tempo::{PaymentLaneClassification, PaymentLaneReason, classify_payment_lane},
    version::{COMMIT_SHA, SEMVER_VERSION},
};
use foundry_evm::decode::RevertDecoder;
use foundry_primitives::{
    FoundryNetwork, FoundryReceiptEnvelope, FoundryTransactionRequest, FoundryTxEnvelope,
    FoundryTxReceipt, FoundryTypedTx,
};
use futures::{
    StreamExt, TryFutureExt,
    channel::{mpsc::Receiver, oneshot},
};
use parking_lot::{Mutex, RwLock};
use revm::{
    context::BlockEnv,
    context_interface::{
        block::BlobExcessGasAndPrice,
        result::{HaltReason, Output},
    },
    database::CacheDB,
    interpreter::{InstructionResult, SuccessOrHalt, return_ok, return_revert},
};
use std::{sync::Arc, time::Duration};
use tempo_hardfork::TempoHardfork;
use tempo_primitives::{AASigned, TEMPO_TX_TYPE_ID, transaction::FEE_PAYER_SIGNATURE_MARKER};
use tokio::{
    sync::mpsc::{self, UnboundedReceiver, unbounded_channel},
    try_join,
};

#[cfg(feature = "base")]
use base_common_consensus::Eip8130Constants;
#[cfg(feature = "base")]
use base_common_evm::EIP8130_TRANSACTION_TYPE;
#[cfg(feature = "base")]
use base_common_precompiles::NonceManagerStorage;
#[cfg(feature = "base")]
use base_common_rpc_types::{BaseTransactionRequest, Eip8130Nonce};
#[cfg(feature = "base")]
use base_execution_eip8130::{FeeCheck, IntrinsicGas};

/// The client version: `anvil/v{major}.{minor}.{patch}`
pub const CLIENT_VERSION: &str = concat!("anvil/v", env!("CARGO_PKG_VERSION"));

#[derive(Clone, Copy)]
struct TransactionFeeDefaults {
    gas_price: u128,
    blob_gas_price: u128,
}

/// The entry point for executing eth api RPC call - The Eth RPC interface.
///
/// This type is cheap to clone and can be used concurrently
pub struct EthApi<N: Network> {
    /// The transaction pool
    pool: Arc<Pool<N::TxEnvelope>>,
    /// Transaction pool state captured by `evm_snapshot`.
    pool_snapshots: Arc<Mutex<HashMap<U256, PoolSnapshot<N::TxEnvelope>>>>,
    /// Holds all blockchain related data
    /// In-Memory only for now
    pub backend: Arc<backend::mem::Backend<N>>,
    /// Whether this node is mining
    is_mining: bool,
    /// available signers
    signers: Arc<Vec<Box<dyn Signer<N>>>>,
    /// data required for `eth_feeHistory`
    fee_history_cache: FeeHistoryCache,
    /// max number of items kept in fee cache
    fee_history_limit: u64,
    /// access to the actual miner
    ///
    /// This access is required in order to adjust miner settings based on requests received from
    /// custom RPC endpoints
    miner: Miner<N::TxEnvelope>,
    /// allows to enabled/disable logging
    logger: LoggingManager,
    /// Tracks all active filters
    filters: Filters<N>,
    /// How transactions are ordered in the pool
    transaction_order: Arc<RwLock<TransactionOrder>>,
    /// Whether we're listening for RPC calls
    net_listening: bool,
    /// The instance ID. Changes on every reset.
    instance_id: Arc<RwLock<B256>>,
    /// Serializes endpoint identity reads with reset transitions.
    lifecycle_lock: Arc<tokio::sync::RwLock<()>>,
    /// Serializes reset preparation without blocking identity RPCs made by the target endpoint.
    reset_lock: Arc<tokio::sync::Mutex<()>>,
    /// Serializes automatic nonce assignment through pool insertion for each sender.
    nonce_locks: Arc<RwLock<HashMap<Address, Arc<tokio::sync::Mutex<()>>>>>,
}

impl<N: Network> Clone for EthApi<N> {
    fn clone(&self) -> Self {
        Self {
            pool: self.pool.clone(),
            pool_snapshots: self.pool_snapshots.clone(),
            backend: self.backend.clone(),
            is_mining: self.is_mining,
            signers: self.signers.clone(),
            fee_history_cache: self.fee_history_cache.clone(),
            fee_history_limit: self.fee_history_limit,
            miner: self.miner.clone(),
            logger: self.logger.clone(),
            filters: self.filters.clone(),
            transaction_order: self.transaction_order.clone(),
            net_listening: self.net_listening,
            instance_id: self.instance_id.clone(),
            lifecycle_lock: self.lifecycle_lock.clone(),
            reset_lock: self.reset_lock.clone(),
            nonce_locks: self.nonce_locks.clone(),
        }
    }
}

// == impl EthApi<N: Network> generic methods ==

impl<N: Network> EthApi<N> {
    /// Creates a new instance
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        pool: Arc<Pool<N::TxEnvelope>>,
        backend: Arc<backend::mem::Backend<N>>,
        signers: Arc<Vec<Box<dyn Signer<N>>>>,
        fee_history_cache: FeeHistoryCache,
        fee_history_limit: u64,
        miner: Miner<N::TxEnvelope>,
        logger: LoggingManager,
        filters: Filters<N>,
        transactions_order: TransactionOrder,
    ) -> Self {
        Self {
            pool,
            pool_snapshots: Arc::new(Mutex::new(HashMap::default())),
            backend,
            is_mining: true,
            signers,
            fee_history_cache,
            fee_history_limit,
            miner,
            logger,
            filters,
            net_listening: true,
            transaction_order: Arc::new(RwLock::new(transactions_order)),
            instance_id: Arc::new(RwLock::new(B256::random())),
            lifecycle_lock: Arc::new(tokio::sync::RwLock::new(())),
            reset_lock: Arc::new(tokio::sync::Mutex::new(())),
            nonce_locks: Default::default(),
        }
    }

    /// Returns the current gas price
    pub fn gas_price(&self) -> u128 {
        if self.backend.is_eip1559() {
            if self.backend.is_min_priority_fee_enforced() {
                (self.backend.base_fee() as u128).saturating_add(self.lowest_suggestion_tip())
            } else {
                self.backend.base_fee() as u128
            }
        } else {
            self.backend.fees().raw_gas_price()
        }
    }

    /// Returns the suggested fee cap.
    ///
    /// Returns at least [MIN_SUGGESTED_PRIORITY_FEE]
    fn lowest_suggestion_tip(&self) -> u128 {
        let block_number = self.backend.best_number();
        let cache = self.fee_history_cache.lock();
        let latest_tip = self
            .backend
            .get_block_with_hash(block_number)
            .and_then(|(_, hash)| cache.get(&block_number).filter(|item| item.block_hash == hash))
            .and_then(|item| item.rewards.iter().copied().min());

        latest_tip
            .or_else(|| {
                cache
                    .iter()
                    .filter(|(number, item)| {
                        self.backend
                            .get_block_with_hash(**number)
                            .is_some_and(|(_, hash)| item.block_hash == hash)
                    })
                    .flat_map(|(_, item)| item.rewards.iter().copied())
                    .min()
            })
            .map(|fee| fee.max(MIN_SUGGESTED_PRIORITY_FEE))
            .unwrap_or(MIN_SUGGESTED_PRIORITY_FEE)
    }

    /// Returns true if auto mining is enabled, and false.
    ///
    /// Handler for ETH RPC call: `anvil_getAutomine`
    pub fn anvil_get_auto_mine(&self) -> Result<bool> {
        node_info!("anvil_getAutomine");
        Ok(self.miner.is_auto_mine())
    }

    /// Returns the value of mining interval, if set.
    ///
    /// Handler for ETH RPC call: `anvil_getIntervalMining`.
    pub fn anvil_get_interval_mining(&self) -> Result<Option<u64>> {
        node_info!("anvil_getIntervalMining");
        Ok(self.miner.get_interval())
    }

    /// Enables or disables, based on the single boolean argument, the automatic mining of new
    /// blocks with each new transaction submitted to the network.
    ///
    /// Handler for ETH RPC call: `evm_setAutomine`
    pub async fn anvil_set_auto_mine(&self, enable_automine: bool) -> Result<()> {
        node_info!("evm_setAutomine");
        if self.miner.is_auto_mine() {
            if enable_automine {
                return Ok(());
            }
            self.miner.set_mining_mode(MiningMode::None);
        } else if enable_automine {
            let listener = self.pool.add_ready_listener();
            let window = self.backend.node_config.read().await.transaction_coalescing_window;
            let mode = MiningMode::instant(1_000, listener).with_coalescing_window(window);
            self.miner.set_mining_mode(mode);
        }
        Ok(())
    }

    /// Sets the mining behavior to interval with the given interval (seconds)
    ///
    /// Handler for ETH RPC call: `evm_setIntervalMining`
    pub fn anvil_set_interval_mining(&self, secs: u64) -> Result<()> {
        node_info!("evm_setIntervalMining");
        let mining_mode = if secs == 0 {
            MiningMode::None
        } else {
            let block_time = Duration::from_secs(secs);

            // This ensures that memory limits are stricter in interval-mine mode
            self.backend.update_interval_mine_block_time(block_time);

            MiningMode::FixedBlockTime(FixedBlockTimeMiner::new(block_time))
        };
        self.miner.set_mining_mode(mining_mode);
        Ok(())
    }

    /// Removes transactions from the pool
    ///
    /// Handler for RPC call: `anvil_dropTransaction`
    pub async fn anvil_drop_transaction(&self, tx_hash: B256) -> Result<Option<B256>> {
        node_info!("anvil_dropTransaction");
        Ok(self.pool.drop_transaction(tx_hash).map(|tx| tx.hash()))
    }

    /// Removes all transactions from the pool
    ///
    /// Handler for RPC call: `anvil_dropAllTransactions`
    pub async fn anvil_drop_all_transactions(&self) -> Result<()> {
        node_info!("anvil_dropAllTransactions");
        self.pool.clear();
        Ok(())
    }

    /// Clears all transactions from the pool.
    ///
    /// Handler for RPC call: `debug_clearTxpool`
    pub async fn debug_clear_txpool(&self) -> Result<()> {
        node_info!("debug_clearTxpool");
        self.pool.clear();
        Ok(())
    }

    pub async fn anvil_set_chain_id(&self, chain_id: u64) -> Result<()> {
        node_info!("anvil_setChainId");
        self.backend.set_chain_id(chain_id);
        #[cfg(feature = "base")]
        self.invalidate_base_eip8130_pool();
        Ok(())
    }

    /// Invalidates EIP-8130 admission state after out-of-band state mutation.
    #[cfg(feature = "base")]
    fn invalidate_base_eip8130_pool(&self) {
        if self.backend.is_base() {
            let _ = self.pool.clear_transaction_type(EIP8130_TRANSACTION_TYPE);
        }
    }

    /// Modifies the balance of an account.
    ///
    /// Handler for RPC call: `anvil_setBalance`
    pub async fn anvil_set_balance(&self, address: Address, balance: U256) -> Result<()> {
        node_info!("anvil_setBalance");
        self.backend.set_balance(address, balance).await?;
        #[cfg(feature = "base")]
        self.invalidate_base_eip8130_pool();
        Ok(())
    }

    /// Sets the code of a contract.
    ///
    /// Handler for RPC call: `anvil_setCode`
    pub async fn anvil_set_code(&self, address: Address, code: Bytes) -> Result<()> {
        node_info!("anvil_setCode");
        self.backend.set_code(address, code).await?;
        #[cfg(feature = "base")]
        self.invalidate_base_eip8130_pool();
        Ok(())
    }

    /// Sets the nonce of an address.
    ///
    /// Handler for RPC call: `anvil_setNonce`
    pub async fn anvil_set_nonce(&self, address: Address, nonce: U256) -> Result<()> {
        node_info!("anvil_setNonce");
        self.backend.set_nonce(address, nonce).await?;
        #[cfg(feature = "base")]
        self.invalidate_base_eip8130_pool();
        Ok(())
    }

    /// Writes a single slot of the account's storage.
    ///
    /// Handler for RPC call: `anvil_setStorageAt`
    pub async fn anvil_set_storage_at(
        &self,
        address: Address,
        slot: U256,
        val: B256,
    ) -> Result<bool> {
        node_info!("anvil_setStorageAt");
        self.backend.set_storage_at(address, slot, val).await?;
        #[cfg(feature = "base")]
        self.invalidate_base_eip8130_pool();
        Ok(true)
    }

    /// Enable or disable logging.
    ///
    /// Handler for RPC call: `anvil_setLoggingEnabled`
    pub async fn anvil_set_logging(&self, enable: bool) -> Result<()> {
        node_info!("anvil_setLoggingEnabled");
        self.logger.set_enabled(enable);
        Ok(())
    }

    /// Set the minimum gas price for the node.
    ///
    /// Handler for RPC call: `anvil_setMinGasPrice`
    pub async fn anvil_set_min_gas_price(&self, gas: U256) -> Result<()> {
        node_info!("anvil_setMinGasPrice");
        if self.backend.is_eip1559() {
            return Err(RpcError::invalid_params(
                "anvil_setMinGasPrice is not supported when EIP-1559 is active",
            )
            .into());
        }
        self.backend.set_gas_price(gas.saturating_to());
        Ok(())
    }

    /// Sets the base fee of the next block.
    ///
    /// Handler for RPC call: `anvil_setNextBlockBaseFeePerGas`
    pub async fn anvil_set_next_block_base_fee_per_gas(&self, basefee: U256) -> Result<()> {
        node_info!("anvil_setNextBlockBaseFeePerGas");
        if !self.backend.is_eip1559() {
            return Err(RpcError::invalid_params(
                "anvil_setNextBlockBaseFeePerGas is only supported when EIP-1559 is active",
            )
            .into());
        }
        self.backend.set_base_fee(basefee.saturating_to());
        Ok(())
    }

    /// Sets the coinbase address.
    ///
    /// Handler for RPC call: `anvil_setCoinbase`
    pub async fn anvil_set_coinbase(&self, address: Address) -> Result<()> {
        node_info!("anvil_setCoinbase");
        self.backend.set_coinbase(address);
        Ok(())
    }

    /// Sets the `prevrandao` value of the next block.
    ///
    /// This is a one-shot override: it applies to the next mined block only, after which anvil
    /// resumes deriving `prevrandao` from the parent hash and block number.
    ///
    /// Handler for RPC call: `anvil_setNextBlockPrevRandao`
    pub async fn anvil_set_next_block_prevrandao(&self, prevrandao: B256) -> Result<()> {
        node_info!("anvil_setNextBlockPrevRandao");
        self.backend.set_next_block_prevrandao(prevrandao);
        Ok(())
    }

    /// Sets the parent beacon block root of the next block.
    ///
    /// This is a one-shot override: it applies to the next mined block only, after which anvil
    /// resumes using the zero root. From Cancun the root is stored by the EIP-4788 beacon roots
    /// contract when the block is mined.
    ///
    /// Handler for RPC call: `anvil_setNextBlockParentBeaconBlockRoot`
    pub async fn anvil_set_next_block_parent_beacon_block_root(&self, root: B256) -> Result<()> {
        node_info!("anvil_setNextBlockParentBeaconBlockRoot");
        self.backend.set_next_block_parent_beacon_block_root(root);
        Ok(())
    }

    /// Retrieves the Anvil node configuration params.
    ///
    /// Handler for RPC call: `anvil_nodeInfo`
    pub async fn anvil_node_info(&self) -> Result<NodeInfo> {
        node_info!("anvil_nodeInfo");
        let _lifecycle = self.lifecycle_lock.read().await;

        let evm_env = self.backend.evm_env().read().clone();
        let fork_config = self.backend.get_fork();
        let tx_order = self.transaction_order.read();
        let hard_fork = self.backend.hardfork().name();

        Ok(NodeInfo {
            current_block_number: self.backend.best_number(),
            current_block_timestamp: evm_env.block_env.timestamp.saturating_to(),
            current_block_hash: self.backend.best_hash(),
            hard_fork,
            transaction_order: match *tx_order {
                TransactionOrder::Fifo => "fifo".to_string(),
                TransactionOrder::Fees => "fees".to_string(),
            },
            environment: NodeEnvironment {
                base_fee: self.backend.base_fee() as u128,
                chain_id: self.backend.chain_id().to::<u64>(),
                gas_limit: self.backend.gas_limit(),
                gas_price: self.gas_price(),
            },
            fork_config: fork_config
                .map(|fork| {
                    let config = fork.config.read();

                    NodeForkConfig {
                        fork_url: config.eth_rpc_url().map(redact_url),
                        fork_block_number: Some(config.block_number),
                        fork_retry_backoff: Some(config.backoff.as_millis()),
                    }
                })
                .unwrap_or_default(),
            // Always emit the execution profile. Older Anvil versions omitted plain Ethereum, so
            // consumers must still treat `None` as legacy unspecified metadata.
            network: Some(self.backend.execution_profile_name().to_string()),
        })
    }

    /// Retrieves metadata about the Anvil instance.
    ///
    /// Handler for RPC call: `anvil_metadata`
    pub async fn anvil_metadata(&self) -> Result<Metadata> {
        node_info!("anvil_metadata");
        let _lifecycle = self.lifecycle_lock.read().await;
        let fork_config = self.backend.get_fork();

        Ok(Metadata {
            client_version: CLIENT_VERSION.to_string(),
            client_semver: Some(SEMVER_VERSION.to_string()),
            client_commit_sha: Some(COMMIT_SHA.to_string()),
            chain_id: self.backend.chain_id().to::<u64>(),
            latest_block_hash: self.backend.best_hash(),
            latest_block_number: self.backend.best_number(),
            instance_id: *self.instance_id.read(),
            forked_network: fork_config.map(|cfg| ForkedNetwork {
                chain_id: cfg.chain_id(),
                fork_block_number: cfg.block_number(),
                fork_block_hash: cfg.block_hash(),
            }),
            snapshots: self.backend.list_state_snapshots(),
        })
    }

    pub async fn anvil_remove_pool_transactions(&self, address: Address) -> Result<()> {
        node_info!("anvil_removePoolTransactions");
        self.pool.remove_transactions_by_address(address);
        Ok(())
    }

    /// Snapshot the state of the blockchain at the current block.
    ///
    /// Handler for RPC call: `evm_snapshot`
    pub async fn evm_snapshot(&self) -> Result<U256> {
        node_info!("evm_snapshot");
        let _lifecycle = self.lifecycle_lock.read().await;
        let _mining = self.backend.lock_mining().await;
        let pool = self.pool.snapshot();
        let id = self.backend.create_state_snapshot().await;
        self.pool_snapshots.lock().insert(id, pool);
        Ok(id)
    }

    /// Jump forward in time by the given amount of time, in seconds.
    ///
    /// Handler for RPC call: `evm_increaseTime`
    pub async fn evm_increase_time(&self, seconds: U256) -> Result<i64> {
        node_info!("evm_increaseTime");
        Ok(self.backend.time().increase_time(seconds.try_into().unwrap_or(u64::MAX)) as i64)
    }

    /// Similar to `evm_increaseTime` but takes the exact timestamp that you want in the next block
    ///
    /// Handler for RPC call: `evm_setNextBlockTimestamp`
    pub fn evm_set_next_block_timestamp(&self, seconds: u64) -> Result<()> {
        node_info!("evm_setNextBlockTimestamp");
        self.backend.time().set_next_block_timestamp(seconds)
    }

    /// Sets the specific timestamp and returns the number of seconds between the given timestamp
    /// and the current time.
    ///
    /// The `timestamp` is in seconds. The `evm_setTime` JSON-RPC method accepts both seconds and
    /// milliseconds (values above 1e12 are treated as milliseconds); the RPC handler normalises
    /// the input to seconds before calling this function.
    ///
    /// Handler for RPC call: `evm_setTime`
    pub fn evm_set_time(&self, timestamp: u64) -> Result<u64> {
        node_info!("evm_setTime");
        let now = self.backend.time().current_call_timestamp();
        self.backend.time().set_time(timestamp);

        // number of seconds between the given timestamp and the current time.
        let offset = timestamp.saturating_sub(now);
        Ok(offset)
    }

    /// Set the next block gas limit
    ///
    /// Handler for RPC call: `evm_setBlockGasLimit`
    pub fn evm_set_block_gas_limit(&self, gas_limit: U256) -> Result<bool> {
        node_info!("evm_setBlockGasLimit");
        self.backend.set_gas_limit(gas_limit.saturating_to());
        Ok(true)
    }

    /// Sets an interval for the block timestamp
    ///
    /// Handler for RPC call: `anvil_setBlockTimestampInterval`
    pub fn evm_set_block_timestamp_interval(&self, seconds: u64) -> Result<()> {
        node_info!("anvil_setBlockTimestampInterval");
        self.backend.time().set_block_timestamp_interval(seconds);
        Ok(())
    }

    /// Sets an interval for the block timestamp
    ///
    /// Handler for RPC call: `anvil_removeBlockTimestampInterval`
    pub fn evm_remove_block_timestamp_interval(&self) -> Result<bool> {
        node_info!("anvil_removeBlockTimestampInterval");
        Ok(self.backend.time().remove_block_timestamp_interval())
    }

    /// Sets the backend rpc url
    ///
    /// Handler for ETH RPC call: `anvil_setRpcUrl`
    pub async fn anvil_set_rpc_url(&self, url: String) -> Result<()> {
        node_info!("anvil_setRpcUrl");
        let _reset = self.reset_lock.lock().await;
        let staged_fork = if let Some(fork) = self.backend.get_fork() {
            let mut validation_config = self.backend.node_config.read().await.clone();
            let (expected_identity, block_number, block_hash) = {
                let config = fork.config.read();
                (config.endpoint_identity, config.block_number, config.block_hash)
            };
            // A new URL replaces an offline discovery hint. Resolve the actual source identity so
            // unsupported networks cannot be hidden behind the previous endpoint's hint.
            validation_config.fork_chain_id = None;
            let (provider, endpoint_identity) = validation_config
                .replacement_fork_provider(
                    &url,
                    expected_identity,
                    block_number,
                    block_hash,
                    self.instance_id(),
                )
                .await?;
            Some((fork, provider, endpoint_identity))
        } else {
            None
        };

        let _lifecycle = self.lifecycle_lock.write().await;
        let _mining = self.backend.lock_mining().await;
        let mut node_config = self.backend.node_config.write().await;
        if let Some((fork, provider, endpoint_identity)) = staged_fork {
            let mut config = fork.config.write();
            trace!(target: "backend", "Updated fork rpc from \"{}\" to \"{}\"", config.eth_rpc_url().map(redact_url).unwrap_or_else(|| "none".to_string()), redact_url(&url));
            config.provider = provider;
            config.fork_urls = vec![url.clone()];
            config.fork_chain_id = None;
            config.endpoint_identity = endpoint_identity;
            node_config.fork_endpoint_is_anvil = endpoint_identity.is_authoritative();
        }
        // Keep node_config in sync so a subsequent URL-less fork reset uses the updated endpoint.
        node_config.fork_urls = vec![url];
        node_config.fork_chain_id = None;
        Ok(())
    }

    /// Returns the number of transactions currently pending for inclusion in the next block(s), as
    /// well as the ones that are being scheduled for future execution only.
    /// Ref: [Here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_status)
    ///
    /// Handler for ETH RPC call: `txpool_status`
    pub async fn txpool_status(&self) -> Result<TxpoolStatus> {
        node_info!("txpool_status");
        Ok(self.pool.txpool_status())
    }

    /// Executes the future on a new blocking task.
    async fn on_blocking_task<C, F, R>(&self, c: C) -> Result<R>
    where
        C: FnOnce(Self) -> F,
        F: Future<Output = Result<R>> + Send + 'static,
        R: Send + 'static,
    {
        let (tx, rx) = oneshot::channel();
        let this = self.clone();
        let f = c(this);
        tokio::task::spawn_blocking(move || {
            tokio::runtime::Handle::current().block_on(async move {
                let res = f.await;
                let _ = tx.send(res);
            })
        });
        rx.await.map_err(|_| BlockchainError::Internal("blocking task panicked".to_string()))?
    }

    /// Updates the `TransactionOrder`
    pub fn set_transaction_order(&self, order: TransactionOrder) {
        *self.transaction_order.write() = order;
    }

    /// Returns the chain ID used for transaction
    pub fn chain_id(&self) -> u64 {
        self.backend.chain_id().to::<u64>()
    }

    /// Returns the configured fork, if any.
    pub fn get_fork(&self) -> Option<ClientFork> {
        self.backend.get_fork()
    }

    /// Returns the current instance's ID.
    pub fn instance_id(&self) -> B256 {
        *self.instance_id.read()
    }

    /// Resets the instance ID.
    pub fn reset_instance_id(&self) {
        *self.instance_id.write() = B256::random();
    }

    /// Returns the first signer that can sign for the given address
    #[expect(clippy::borrowed_box)]
    pub fn get_signer(&self, address: Address) -> Option<&Box<dyn Signer<N>>> {
        self.signers.iter().find(|signer| signer.is_signer_for(address))
    }

    /// Returns a new listeners for ready transactions
    pub fn new_ready_transactions(&self) -> Receiver<TxHash> {
        self.pool.add_ready_listener()
    }

    /// Returns true if forked
    pub fn is_fork(&self) -> bool {
        self.backend.is_fork()
    }

    /// Returns the current state root
    pub async fn state_root(&self) -> Option<B256> {
        self.backend.get_db().read().await.maybe_state_root()
    }

    /// Returns true if the `addr` is currently impersonated
    pub fn is_impersonated(&self, addr: Address) -> bool {
        self.backend.cheats().is_impersonated(addr)
    }

    /// Returns a new accessor for certain storage elements
    pub fn storage_info(&self) -> StorageInfo<N> {
        StorageInfo::new(Arc::clone(&self.backend))
    }

    /// Handler for RPC call: `anvil_getBlobByHash`
    #[allow(clippy::large_stack_frames)]
    pub fn anvil_get_blob_by_versioned_hash(
        &self,
        hash: B256,
    ) -> Result<Option<alloy_consensus::Blob>> {
        node_info!("anvil_getBlobByHash");
        Ok(self.backend.get_blob_by_versioned_hash(hash)?)
    }

    /// Handler for RPC call: `anvil_getBlobsByBlockId`
    pub fn anvil_get_blobs_by_block_id(
        &self,
        block_id: impl Into<BlockId>,
        versioned_hashes: Vec<B256>,
    ) -> Result<Option<Vec<Blob>>> {
        node_info!("anvil_getBlobsByBlockId");
        Ok(self.backend.get_blobs_by_block_id(block_id, versioned_hashes)?)
    }

    /// Returns the genesis time for the Beacon chain
    ///
    /// Handler for Beacon API call: `GET /eth/v1/beacon/genesis`
    pub fn anvil_get_genesis_time(&self) -> Result<u64> {
        node_info!("anvil_getGenesisTime");
        Ok(self.backend.genesis_time())
    }

    /// Returns the UNIX wall time in milliseconds when the current head was installed.
    ///
    /// Handler for RPC call: `anvil_getLastBlockWallTime`
    pub fn anvil_get_last_block_wall_time(&self) -> Result<u64> {
        node_info!("anvil_getLastBlockWallTime");
        Ok(self.backend.time().last_block_wall_time())
    }

    /// Reset the fork to a fresh forked state, and optionally update the fork config.
    ///
    /// If `forking` is `None` then this will disable forking entirely.
    ///
    /// Handler for RPC call: `anvil_reset`
    pub async fn anvil_reset(&self, forking: Option<Forking>) -> Result<()> {
        node_info!("anvil_reset");
        let _reset = self.reset_lock.lock().await;
        if let Some(forking) = forking {
            let staged = self.backend.prepare_fork_reset(forking, self.instance_id()).await?;
            let _lifecycle = self.lifecycle_lock.write().await;
            // Stop an old-context block before publishing the replacement DB and environment.
            let _mining = self.backend.lock_mining().await;
            self.backend.commit_fork_reset(staged).await?;
            self.reset_instance_id();
            self.pool.clear();
            self.pool_snapshots.lock().clear();
            self.fee_history_cache.lock().clear();
        } else {
            let _lifecycle = self.lifecycle_lock.write().await;
            // Unlike fork preparation, memory preparation snapshots live local fee state and does
            // no remote I/O. Keep that snapshot and its publication in one lifecycle transition.
            let _mining = self.backend.lock_mining().await;
            let staged = self.backend.prepare_memory_reset().await?;
            self.backend.commit_memory_reset(staged).await?;
            self.reset_instance_id();
            self.pool.clear();
            self.pool_snapshots.lock().clear();
            self.fee_history_cache.lock().clear();
        }
        Ok(())
    }

    /// Revert the state of the blockchain to a previous snapshot.
    /// Takes a single parameter, which is the snapshot id to revert to.
    ///
    /// Handler for RPC call: `evm_revert`
    pub async fn evm_revert(&self, id: U256) -> Result<bool>
    where
        N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
    {
        node_info!("evm_revert");
        let _lifecycle = self.lifecycle_lock.read().await;
        let _mining = self.backend.lock_mining().await;
        let reverted = self.backend.revert_state_snapshot(id).await?;
        if reverted {
            let pool = {
                let mut snapshots = self.pool_snapshots.lock();
                let pool = snapshots.remove(&id);
                snapshots.retain(|snapshot_id, _| *snapshot_id < id);
                pool
            };
            if let Some(pool) = pool {
                self.pool.restore(pool);
            }
        }
        Ok(reverted)
    }

    /// Send transactions impersonating specific account and contract addresses.
    ///
    /// Handler for ETH RPC call: `anvil_impersonateAccount`
    pub async fn anvil_impersonate_account(&self, address: Address) -> Result<()> {
        node_info!("anvil_impersonateAccount");
        self.backend.impersonate(address);
        Ok(())
    }

    /// Stops impersonating an account if previously set with `anvil_impersonateAccount`.
    ///
    /// Handler for ETH RPC call: `anvil_stopImpersonatingAccount`
    pub async fn anvil_stop_impersonating_account(&self, address: Address) -> Result<()> {
        node_info!("anvil_stopImpersonatingAccount");
        self.backend.stop_impersonating(address);
        Ok(())
    }

    /// If set to true will make every account impersonated
    ///
    /// Handler for ETH RPC call: `anvil_autoImpersonateAccount`
    pub async fn anvil_auto_impersonate_account(&self, enabled: bool) -> Result<()> {
        node_info!("anvil_autoImpersonateAccount");
        self.backend.auto_impersonate_account(enabled);
        Ok(())
    }

    /// Registers a new address and signature pair to impersonate.
    pub async fn anvil_impersonate_signature(
        &self,
        signature: Bytes,
        address: Address,
    ) -> Result<()> {
        node_info!("anvil_impersonateSignature");
        self.backend.impersonate_signature(signature, address).await
    }

    /// Returns a new block event stream that yields Notifications when a new block was added or
    /// when logs were removed from the canonical chain due to a reorg
    pub fn new_block_notifications(&self) -> ChainNotifications {
        self.backend.new_block_notifications()
    }

    /// Returns the current client version.
    ///
    /// Handler for ETH RPC call: `web3_clientVersion`
    pub fn client_version(&self) -> Result<String> {
        node_info!("web3_clientVersion");
        Ok(CLIENT_VERSION.to_string())
    }

    /// Returns Keccak-256 (not the standardized SHA3-256) of the given data.
    ///
    /// Handler for ETH RPC call: `web3_sha3`
    pub fn sha3(&self, bytes: Bytes) -> Result<String> {
        node_info!("web3_sha3");
        let hash = alloy_primitives::keccak256(bytes.as_ref());
        Ok(alloy_primitives::hex::encode_prefixed(&hash[..]))
    }

    /// Returns protocol version encoded as a string (quotes are necessary).
    ///
    /// Handler for ETH RPC call: `eth_protocolVersion`
    pub fn protocol_version(&self) -> Result<u64> {
        node_info!("eth_protocolVersion");
        Ok(1)
    }

    /// Returns the number of hashes per second that the node is mining with.
    ///
    /// Handler for ETH RPC call: `eth_hashrate`
    pub fn hashrate(&self) -> Result<U256> {
        node_info!("eth_hashrate");
        Ok(U256::ZERO)
    }

    /// Returns the client coinbase address.
    ///
    /// Handler for ETH RPC call: `eth_coinbase`
    pub fn author(&self) -> Result<Address> {
        node_info!("eth_coinbase");
        Ok(self.backend.coinbase())
    }

    /// Returns true if client is actively mining new blocks.
    ///
    /// Handler for ETH RPC call: `eth_mining`
    pub fn is_mining(&self) -> Result<bool> {
        node_info!("eth_mining");
        Ok(self.is_mining)
    }

    /// Returns the chain ID used for transaction signing at the
    /// current best block. None is returned if not
    /// available.
    ///
    /// Handler for ETH RPC call: `eth_chainId`
    pub fn eth_chain_id(&self) -> Result<Option<U64>> {
        node_info!("eth_chainId");
        Ok(Some(self.backend.chain_id().to::<U64>()))
    }

    /// Returns the same as `chain_id`
    ///
    /// Handler for ETH RPC call: `eth_networkId`
    pub fn network_id(&self) -> Result<Option<String>> {
        node_info!("eth_networkId");
        let chain_id = self.backend.chain_id().to::<u64>();
        Ok(Some(format!("{chain_id}")))
    }

    /// Returns true if client is actively listening for network connections.
    ///
    /// Handler for ETH RPC call: `net_listening`
    pub fn net_listening(&self) -> Result<bool> {
        node_info!("net_listening");
        Ok(self.net_listening)
    }

    /// Returns the current gas price
    fn eth_gas_price(&self) -> Result<U256> {
        node_info!("eth_gasPrice");
        Ok(U256::from(self.gas_price()))
    }

    /// Returns the base fee for the next block, or null before London.
    ///
    /// Handler for ETH RPC call: `eth_baseFee`
    pub fn base_fee(&self) -> Result<Option<U256>> {
        node_info!("eth_baseFee");
        Ok(self.backend.is_eip1559().then(|| U256::from(self.backend.base_fee())))
    }

    /// Returns the excess blob gas and current blob gas price
    pub fn excess_blob_gas_and_price(&self) -> Result<Option<BlobExcessGasAndPrice>> {
        Ok(self.backend.excess_blob_gas_and_price())
    }

    /// Returns a fee per gas that is an estimate of how much you can pay as a priority fee, or
    /// 'tip', to get a transaction included in the current block.
    ///
    /// Handler for ETH RPC call: `eth_maxPriorityFeePerGas`
    pub fn gas_max_priority_fee_per_gas(&self) -> Result<U256> {
        self.max_priority_fee_per_gas()
    }

    /// Returns the base fee per blob required to send an EIP-4844 tx.
    ///
    /// Handler for ETH RPC call: `eth_blobBaseFee`
    pub fn blob_base_fee(&self) -> Result<U256> {
        Ok(U256::from(self.backend.fees().base_fee_per_blob_gas()))
    }

    /// Returns the block gas limit
    pub fn gas_limit(&self) -> U256 {
        U256::from(self.backend.gas_limit())
    }

    /// Returns the accounts list
    ///
    /// Handler for ETH RPC call: `eth_accounts`
    pub fn accounts(&self) -> Result<Vec<Address>> {
        node_info!("eth_accounts");
        let mut unique = AddressSet::default();
        let mut accounts: Vec<Address> = Vec::new();
        for signer in self.signers.iter() {
            accounts.extend(signer.accounts().into_iter().filter(|acc| unique.insert(*acc)));
        }
        accounts.extend(
            self.backend
                .cheats()
                .impersonated_accounts()
                .into_iter()
                .filter(|acc| unique.insert(*acc)),
        );
        Ok(accounts.into_iter().collect())
    }

    /// Returns the number of most recent block.
    ///
    /// Handler for ETH RPC call: `eth_blockNumber`
    pub fn block_number(&self) -> Result<U256> {
        node_info!("eth_blockNumber");
        Ok(U256::from(self.backend.best_number()))
    }

    /// Returns block with given hash.
    ///
    /// Handler for ETH RPC call: `eth_getBlockByHash`
    pub async fn block_by_hash(&self, hash: B256) -> Result<Option<AnyRpcBlock>> {
        node_info!("eth_getBlockByHash");
        self.backend.block_by_hash(hash).await
    }

    /// Returns block header with given hash.
    ///
    /// Handler for ETH RPC call: `eth_getHeaderByHash`
    pub async fn header_by_hash(
        &self,
        hash: B256,
    ) -> Result<Option<WithOtherFields<AnyRpcHeader>>> {
        node_info!("eth_getHeaderByHash");
        Ok(self.backend.block_by_hash(hash).await?.map(|block| {
            let WithOtherFields { inner: block, other } = block.0;
            WithOtherFields { inner: block.header, other }
        }))
    }

    /// Returns a _full_ block with given hash.
    ///
    /// Handler for ETH RPC call: `eth_getBlockByHash`
    pub async fn block_by_hash_full(&self, hash: B256) -> Result<Option<AnyRpcBlock>> {
        node_info!("eth_getBlockByHash");
        self.backend.block_by_hash_full(hash).await
    }

    /// Returns the number of transactions in a block with given hash.
    ///
    /// Handler for ETH RPC call: `eth_getBlockTransactionCountByHash`
    pub async fn block_transaction_count_by_hash(&self, hash: B256) -> Result<Option<U256>> {
        node_info!("eth_getBlockTransactionCountByHash");
        let block = self.backend.block_by_hash(hash).await?;
        let txs = block.map(|b| match b.transactions() {
            BlockTransactions::Full(txs) => U256::from(txs.len()),
            BlockTransactions::Hashes(txs) => U256::from(txs.len()),
            BlockTransactions::Uncle => U256::ZERO,
        });
        Ok(txs)
    }

    /// Returns the number of uncles in a block with given hash.
    ///
    /// Handler for ETH RPC call: `eth_getUncleCountByBlockHash`
    pub async fn block_uncles_count_by_hash(&self, hash: B256) -> Result<U256> {
        node_info!("eth_getUncleCountByBlockHash");
        let block =
            self.backend.block_by_hash(hash).await?.ok_or(BlockchainError::BlockNotFound)?;
        Ok(U256::from(block.uncles.len()))
    }

    /// Returns the number of uncles in a block with given block number.
    ///
    /// Handler for ETH RPC call: `eth_getUncleCountByBlockNumber`
    pub async fn block_uncles_count_by_number(&self, block_number: BlockNumber) -> Result<U256> {
        node_info!("eth_getUncleCountByBlockNumber");
        let block = self
            .backend
            .block_by_number(block_number)
            .await?
            .ok_or(BlockchainError::BlockNotFound)?;
        Ok(U256::from(block.uncles.len()))
    }

    /// Signs data via [EIP-712](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-712.md).
    ///
    /// Handler for ETH RPC call: `eth_signTypedData`
    pub async fn sign_typed_data(
        &self,
        _address: Address,
        _data: serde_json::Value,
    ) -> Result<String> {
        node_info!("eth_signTypedData");
        Err(BlockchainError::RpcUnimplemented)
    }

    /// Signs data via [EIP-712](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-712.md).
    ///
    /// Handler for ETH RPC call: `eth_signTypedData_v3`
    pub async fn sign_typed_data_v3(
        &self,
        _address: Address,
        _data: serde_json::Value,
    ) -> Result<String> {
        node_info!("eth_signTypedData_v3");
        Err(BlockchainError::RpcUnimplemented)
    }

    /// Signs data via [EIP-712](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-712.md), and includes full support of arrays and recursive data structures.
    ///
    /// Handler for ETH RPC call: `eth_signTypedData_v4`
    pub async fn sign_typed_data_v4(&self, address: Address, data: &TypedData) -> Result<String> {
        node_info!("eth_signTypedData_v4");
        let signer = self.get_signer(address).ok_or(BlockchainError::NoSignerAvailable)?;
        let signature = signer.sign_typed_data(address, data).await?;
        let signature = alloy_primitives::hex::encode(signature.as_bytes());
        Ok(format!("0x{signature}"))
    }

    /// The sign method calculates an Ethereum specific signature
    ///
    /// Handler for ETH RPC call: `eth_sign`
    pub async fn sign(&self, address: Address, content: impl AsRef<[u8]>) -> Result<String> {
        node_info!("eth_sign");
        let signer = self.get_signer(address).ok_or(BlockchainError::NoSignerAvailable)?;
        let signature =
            alloy_primitives::hex::encode(signer.sign(address, content.as_ref()).await?.as_bytes());
        Ok(format!("0x{signature}"))
    }

    /// Returns transaction at given block hash and index.
    ///
    /// Handler for ETH RPC call: `eth_getTransactionByBlockHashAndIndex`
    pub async fn transaction_by_block_hash_and_index(
        &self,
        hash: B256,
        index: Index,
    ) -> Result<Option<AnyRpcTransaction>> {
        node_info!("eth_getTransactionByBlockHashAndIndex");
        self.backend.transaction_by_block_hash_and_index(hash, index).await
    }

    /// Returns an uncles at given block and index.
    ///
    /// Handler for ETH RPC call: `eth_getUncleByBlockHashAndIndex`
    pub async fn uncle_by_block_hash_and_index(
        &self,
        block_hash: B256,
        idx: Index,
    ) -> Result<Option<AnyRpcBlock>> {
        node_info!("eth_getUncleByBlockHashAndIndex");
        let number = self.backend.ensure_block_number(Some(BlockId::hash(block_hash))).await?;
        if let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            return Ok(fork.uncle_by_block_hash_and_index(block_hash, idx.into()).await?);
        }
        // It's impossible to have uncles outside of fork mode
        Ok(None)
    }

    /// Returns an uncles at given block and index.
    ///
    /// Handler for ETH RPC call: `eth_getUncleByBlockNumberAndIndex`
    pub async fn uncle_by_block_number_and_index(
        &self,
        block_number: BlockNumber,
        idx: Index,
    ) -> Result<Option<AnyRpcBlock>> {
        node_info!("eth_getUncleByBlockNumberAndIndex");
        let number = self.backend.ensure_block_number(Some(BlockId::Number(block_number))).await?;
        if let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            return Ok(fork.uncle_by_block_number_and_index(number, idx.into()).await?);
        }
        // It's impossible to have uncles outside of fork mode
        Ok(None)
    }

    /// Returns the hash of the current block, the seedHash, and the boundary condition to be met.
    ///
    /// Handler for ETH RPC call: `eth_getWork`
    pub fn work(&self) -> Result<Work> {
        node_info!("eth_getWork");
        Err(BlockchainError::RpcUnimplemented)
    }

    /// Returns the sync status, always be fails.
    ///
    /// Handler for ETH RPC call: `eth_syncing`
    pub fn syncing(&self) -> Result<bool> {
        node_info!("eth_syncing");
        Ok(false)
    }

    /// Returns the current configuration of the chain.
    /// This is useful for finding out what precompiles and system contracts are available.
    ///
    /// Note: the activation timestamp is always 0 as the configuration is set at genesis.
    /// Note: the `fork_id` is always `0x00000000` as this node does not participate in any forking
    /// on the network.
    /// Note: the `next` and `last` fields are always `null` as this node does not participate in
    /// any forking on the network.
    ///
    /// Handler for ETH RPC call: `eth_config`
    pub fn config(&self) -> Result<EthConfig> {
        node_info!("eth_config");
        Ok(EthConfig {
            current: EthForkConfig {
                activation_time: 0,
                blob_schedule: self.backend.blob_params(),
                chain_id: self.backend.chain_id().to::<u64>(),
                fork_id: Bytes::from_static(&[0; 4]),
                precompiles: self.backend.precompiles(),
                system_contracts: self.backend.system_contracts(),
            },
            next: None,
            last: None,
        })
    }

    /// Used for submitting a proof-of-work solution.
    ///
    /// Handler for ETH RPC call: `eth_submitWork`
    pub fn submit_work(&self, _: B64, _: B256, _: B256) -> Result<bool> {
        node_info!("eth_submitWork");
        Err(BlockchainError::RpcUnimplemented)
    }

    /// Used for submitting mining hashrate.
    ///
    /// Handler for ETH RPC call: `eth_submitHashrate`
    pub fn submit_hashrate(&self, _: U256, _: B256) -> Result<bool> {
        node_info!("eth_submitHashrate");
        Err(BlockchainError::RpcUnimplemented)
    }

    /// Introduced in EIP-1559 for getting information on the appropriate priority fee to use.
    ///
    /// Handler for ETH RPC call: `eth_feeHistory`
    pub async fn fee_history(
        &self,
        block_count: U256,
        newest_block: BlockNumber,
        reward_percentiles: Vec<f64>,
    ) -> Result<FeeHistory>
    where
        N::ReceiptEnvelope: TxReceipt<Log = alloy_primitives::Log>,
    {
        node_info!("eth_feeHistory");

        if reward_percentiles.iter().any(|p| !(0.0..=100.0).contains(p))
            || reward_percentiles.windows(2).any(|pair| pair[0] >= pair[1])
        {
            return Err(FeeHistoryError::InvalidRewardPercentiles.into());
        }
        if block_count.is_zero() {
            return Ok(FeeHistory::default());
        }

        // max number of blocks in the requested range

        let number = self.backend.convert_block_number(Some(newest_block));

        // Snapshot the fork block once so the early return here and the pre-fork split below agree
        // on the same value. Reading it a second time lets an `anvil_reset` move the fork between
        // the reads: the newest block could clear this early return, then the split treats the
        // whole range as pre-fork and tacks on a stray local next-block fee.
        let fork = self.get_fork();
        let fork_block = fork.as_ref().map(|fork| fork.block_number());

        // check if the number predates the fork, if in fork mode
        if let (Some(fork), Some(fork_block)) = (fork.as_ref(), fork_block) {
            // if we're still at the forked block we don't have any history and can't compute it
            // efficiently, instead we fetch it from the fork
            if number <= fork_block {
                return fork
                    .fee_history(block_count.to(), BlockNumber::Number(number), &reward_percentiles)
                    .await
                    .map_err(BlockchainError::AlloyForkProvider);
            }
        }

        const MAX_BLOCK_COUNT: u64 = 1024u64;
        let block_count = block_count.saturating_to::<u64>().min(MAX_BLOCK_COUNT);

        // highest and lowest block num in the requested range
        let highest = number;
        let lowest = highest.saturating_sub(block_count.saturating_sub(1));

        // only support ranges that are in cache range
        if lowest < self.backend.best_number().saturating_sub(self.fee_history_limit) {
            return Err(FeeHistoryError::InvalidBlockRange.into());
        }

        let mut response = FeeHistory {
            oldest_block: lowest,
            base_fee_per_gas: Vec::new(),
            gas_used_ratio: Vec::new(),
            reward: Some(Default::default()),
            base_fee_per_blob_gas: Default::default(),
            blob_gas_used_ratio: Default::default(),
        };
        let mut rewards = Vec::new();

        // The early return above only delegates when the *newest* block predates the fork, but a
        // range can dip below the fork block while its newest block is post-fork. Local storage
        // has no pre-fork blocks, so resolving them via `backend.get_block` below would fail.
        // Serve the pre-fork portion from the fork provider and compute only post-fork locally.
        // Reuse the `fork_block` snapshot from above so this split stays consistent with the early
        // return; we didn't return, so `fork_block < highest` and only a crossing range splits.
        let local_lowest = if let (Some(fork), Some(fork_block)) = (fork.as_ref(), fork_block) {
            if lowest <= fork_block {
                let count_pre = fork_block - lowest + 1;
                let pre = fork
                    .fee_history(count_pre, BlockNumber::Number(fork_block), &reward_percentiles)
                    .await
                    .map_err(BlockchainError::AlloyForkProvider)?;
                merge_pre_fork_fee_history(&mut response, &mut rewards, pre, fork_block);
                // Compute only post-fork blocks locally; the pre-fork portion is served above.
                fork_block + 1
            } else {
                lowest
            }
        } else {
            lowest
        };

        {
            let storage_info = self.storage_info();
            let blob_params = self.backend.blob_params();

            // Snapshot the cached entries for the whole range under a single lock. The
            // `FeeHistoryService` populates the cache asynchronously and can lag the chain head,
            // so some entries may still be missing here.
            let cached: Vec<Option<FeeHistoryCacheItem>> = {
                let cache = self.fee_history_cache.lock();
                (local_lowest..=highest).map(|n| cache.get(&n).cloned()).collect()
            };

            // Resolve the whole range into concrete items. Cache entries include their block hash
            // so a reset or reorg that reuses a block number cannot return data from the old
            // canonical chain. Compute misses and stale entries on demand without holding the
            // lock. A block that can't be found is a hard error rather than a silently dropped
            // entry, so `items` always holds exactly one entry per requested block.
            let mut warmed: Vec<(u64, FeeHistoryCacheItem)> = Vec::new();
            let mut items: Vec<FeeHistoryCacheItem> = Vec::with_capacity(cached.len());
            for (cached_item, n) in cached.into_iter().zip(local_lowest..=highest) {
                let hash = self
                    .backend
                    .block_hash_by_number(n)
                    .ok_or(FeeHistoryError::BlockNotFound(BlockNumber::Number(n)))?;
                let item = match cached_item {
                    Some(item) if item.block_hash == hash => item,
                    _ => {
                        let block = self
                            .backend
                            .get_block_by_hash(hash)
                            .ok_or(FeeHistoryError::BlockNotFound(BlockNumber::Number(n)))?;
                        let header = block.header;
                        let (item, block_number) = create_fee_history_cache_item(
                            hash,
                            &header,
                            &storage_info,
                            blob_params,
                        );
                        // The block was found above, but missing stored block/receipts make the
                        // helper hand back a zero-filled item. Error out rather than serve
                        // synthetic data for a block we already confirmed exists.
                        let block_number = block_number
                            .ok_or(FeeHistoryError::BlockNotFound(BlockNumber::Number(n)))?;
                        warmed.push((block_number, item.clone()));
                        item
                    }
                };
                items.push(item);
            }

            // Warm the cache with the on-demand entries under a single lock, then trim to the
            // limit by dropping the oldest blocks so the fallback can't grow the cache unbounded.
            if !warmed.is_empty() {
                let mut cache = self.fee_history_cache.lock();
                for (block_number, item) in warmed {
                    cache.insert(block_number, item);
                }
                while cache.len() as u64 > self.fee_history_limit {
                    cache.pop_first();
                }
            }

            for item in items {
                response.base_fee_per_gas.push(item.base_fee);
                response.base_fee_per_blob_gas.push(item.base_fee_per_blob_gas.unwrap_or(0));
                response.blob_gas_used_ratio.push(item.blob_gas_used_ratio);
                response.gas_used_ratio.push(item.gas_used_ratio);

                // requested percentiles
                if !reward_percentiles.is_empty() {
                    let mut block_rewards = Vec::new();
                    for p in &reward_percentiles {
                        block_rewards.push(reward_at_percentile(&item.rewards, *p));
                    }
                    rewards.push(block_rewards);
                }
            }
        }

        response.reward = Some(rewards);

        // Both base-fee arrays include the block after the returned range. For a historical
        // request, use that block's header; the fee manager tracks the block after the current
        // chain head and would therefore append an unrelated value. Prefer an existing child even
        // when `highest` was the head at the start of this request, since mining can advance it.
        let next_number = highest
            .checked_add(1)
            .ok_or(FeeHistoryError::BlockNotFound(BlockNumber::Number(highest)))?;
        let (next_base_fee, next_blob_base_fee) = self
            .backend
            .fee_history_next_fees(highest)
            .await
            .ok_or(FeeHistoryError::BlockNotFound(BlockNumber::Number(next_number)))?;
        response.base_fee_per_gas.push(next_base_fee);
        response.base_fee_per_blob_gas.push(next_blob_base_fee);

        Ok(response)
    }

    /// Introduced in EIP-1159, a Geth-specific and simplified priority fee oracle.
    /// Leverages the already existing fee history cache.
    ///
    /// Returns a suggestion for a gas tip cap for dynamic fee transactions.
    ///
    /// Handler for ETH RPC call: `eth_maxPriorityFeePerGas`
    pub fn max_priority_fee_per_gas(&self) -> Result<U256> {
        node_info!("eth_maxPriorityFeePerGas");
        Ok(U256::from(self.lowest_suggestion_tip()))
    }

    /// Returns code by its hash
    ///
    /// Handler for RPC call: `debug_codeByHash`
    pub async fn debug_code_by_hash(
        &self,
        hash: B256,
        block_id: Option<BlockId>,
    ) -> Result<Option<Bytes>> {
        node_info!("debug_codeByHash");
        self.backend.debug_code_by_hash(hash, block_id).await
    }

    /// Returns the value associated with a key from the database
    /// Only supports bytecode lookups.
    ///
    /// Handler for RPC call: `debug_dbGet`
    pub async fn debug_db_get(&self, key: String) -> Result<Option<Bytes>> {
        node_info!("debug_dbGet");
        self.backend.debug_db_get(key).await
    }

    /// Handles reth's no-op `debug_getModifiedAccountsByNumber` endpoint.
    pub fn debug_get_modified_accounts_by_number(
        &self,
        _start_number: u64,
        _end_number: u64,
    ) -> Result<()> {
        node_info!("debug_getModifiedAccountsByNumber");
        Ok(())
    }

    /// Handles reth's no-op `debug_freeOSMemory` endpoint.
    pub fn debug_free_os_memory(&self) -> Result<()> {
        node_info!("debug_freeOSMemory");
        Ok(())
    }

    /// Executes a transaction and returns requested parity trace results.
    ///
    /// Handler for RPC call: `trace_call`
    pub async fn trace_call(
        &self,
        request: WithOtherFields<TransactionRequest>,
        trace_types: HashSet<TraceType>,
        block_id: Option<BlockId>,
    ) -> Result<TraceResults>
    where
        N: Network<TxEnvelope = FoundryTxEnvelope, ReceiptEnvelope = FoundryReceiptEnvelope>,
    {
        node_info!("trace_call");
        let block_id = block_id.unwrap_or_default();
        let block_request = match &block_id {
            BlockId::Number(BlockNumber::Pending) => {
                let pending_txs = self.pool.ready_transactions().collect();
                BlockRequest::Pending(pending_txs)
            }
            _ => {
                let number = self.backend.ensure_block_number(Some(block_id)).await?;
                BlockRequest::Number(number)
            }
        };
        let inner = request.as_ref();
        let fees = FeeDetails::new(
            inner.gas_price,
            inner.max_fee_per_gas,
            inner.max_priority_fee_per_gas,
            inner.max_fee_per_blob_gas,
        )?
        .or_zero_fees();

        self.backend.trace_call(request, fees, trace_types, block_request, block_id).await
    }

    /// Returns traces for the transaction hash via parity's tracing endpoint
    ///
    /// Handler for RPC call: `trace_transaction`
    pub async fn trace_transaction(
        &self,
        tx_hash: B256,
    ) -> Result<Option<Vec<LocalizedTransactionTrace>>> {
        node_info!("trace_transaction");
        self.backend.trace_transaction(tx_hash).await
    }

    /// Returns traces for the transaction hash via parity's tracing endpoint
    ///
    /// Handler for RPC call: `trace_block`
    pub async fn trace_block(&self, block: BlockNumber) -> Result<Vec<LocalizedTransactionTrace>> {
        node_info!("trace_block");
        self.backend.trace_block(block).await
    }

    /// Returns filtered traces over blocks
    ///
    /// Handler for RPC call: `trace_filter`
    pub async fn trace_filter(
        &self,
        filter: TraceFilter,
    ) -> Result<Vec<LocalizedTransactionTrace>> {
        node_info!("trace_filter");
        self.backend.trace_filter(filter).await
    }

    /// Returns the transaction trace at the given trace address.
    ///
    /// Handler for RPC call: `trace_get`.
    pub async fn trace_get(
        &self,
        hash: B256,
        indices: Vec<Index>,
    ) -> Result<Option<LocalizedTransactionTrace>> {
        node_info!("trace_get");
        self.backend.trace_get(hash, indices).await
    }

    /// Replays all transactions in a block returning the requested traces for each transaction
    ///
    /// Handler for RPC call: `trace_replayBlockTransactions`
    pub async fn trace_replay_block_transactions(
        &self,
        block: BlockNumber,
        trace_types: HashSet<TraceType>,
    ) -> Result<Option<Vec<TraceResultsWithTransactionHash>>> {
        node_info!("trace_replayBlockTransactions");
        self.backend.trace_replay_block_transactions(block, trace_types).await
    }

    /// Replays a transaction returning the requested traces.
    ///
    /// Handler for RPC call: `trace_replayTransaction`.
    pub async fn trace_replay_transaction(
        &self,
        transaction: B256,
        trace_types: HashSet<TraceType>,
    ) -> Result<Option<TraceResultsWithTransactionHash>> {
        node_info!("trace_replayTransaction");
        self.backend.trace_replay_transaction(transaction, trace_types).await
    }
}

impl<N: Network<ReceiptEnvelope = FoundryReceiptEnvelope>> EthApi<N> {
    /// Returns the current state
    pub async fn serialized_state(
        &self,
        preserve_historical_states: bool,
    ) -> Result<SerializableState> {
        self.backend.serialized_state(preserve_historical_states).await
    }
}

// == impl EthApi anvil endpoints ==

impl EthApi<FoundryNetwork> {
    /// Returns transaction by given block number and index.
    ///
    /// Handler for ETH RPC call: `eth_getTransactionByBlockNumberAndIndex`
    pub async fn transaction_by_block_number_and_index(
        &self,
        block: BlockNumber,
        idx: Index,
    ) -> Result<Option<AnyRpcTransaction>> {
        node_info!("eth_getTransactionByBlockNumberAndIndex");
        if block.is_pending() {
            return Ok(self.pending_block_full().await?.and_then(|block| {
                let WithOtherFields { inner: block, .. } = block.0;
                block.transactions.into_transactions().nth(idx.into())
            }));
        }

        self.backend.transaction_by_block_number_and_index(block, idx).await
    }

    /// Create a buffer that represents all state on the chain, which can be loaded to separate
    /// process by calling `anvil_loadState`
    ///
    /// Handler for RPC call: `anvil_dumpState`
    pub async fn anvil_dump_state(
        &self,
        preserve_historical_states: Option<bool>,
    ) -> Result<Bytes> {
        node_info!("anvil_dumpState");
        self.backend.dump_state(preserve_historical_states.unwrap_or(false)).await
    }

    /// Append chain state buffer to current chain. Will overwrite any conflicting addresses or
    /// storage.
    ///
    /// Handler for RPC call: `anvil_loadState`
    pub async fn anvil_load_state(&self, buf: Bytes) -> Result<bool> {
        node_info!("anvil_loadState");
        let loaded = self.backend.load_state_bytes(buf).await?;
        #[cfg(feature = "base")]
        if loaded {
            self.invalidate_base_eip8130_pool();
        }
        Ok(loaded)
    }

    async fn block_request(
        &self,
        block_number: Option<BlockId>,
    ) -> Result<BlockRequest<FoundryTxEnvelope>> {
        let block_request = match block_number {
            Some(BlockId::Number(BlockNumber::Pending)) => {
                let pending_txs = self.pool.ready_transactions().collect();
                BlockRequest::Pending(pending_txs)
            }
            _ => {
                let number = self.backend.ensure_block_number(block_number).await?;
                BlockRequest::Number(number)
            }
        };
        Ok(block_request)
    }

    /// Returns account information after replaying a block through a transaction index.
    ///
    /// Handler for RPC call: `debug_accountInfoAt`.
    pub async fn debug_account_info_at(
        &self,
        block_id: BlockId,
        tx_index: Index,
        address: Address,
    ) -> Result<Option<AccountInfo>> {
        node_info!("debug_accountInfoAt");
        self.backend.debug_account_info_at(block_id, tx_index, address).await
    }

    /// Returns a best-effort execution witness for the given block, built from the full parent
    /// state. See [`crate::eth::backend::mem::Backend::debug_execution_witness`] for the exact
    /// contents and limitations.
    ///
    /// Handler for RPC call: `debug_executionWitness`.
    pub async fn debug_execution_witness(&self, block: BlockNumber) -> Result<ExecutionWitness> {
        node_info!("debug_executionWitness");
        self.backend.debug_execution_witness(block).await
    }

    /// Returns opcode gas usage for a transaction.
    ///
    /// Handler for RPC call: `trace_transactionOpcodeGas`.
    pub async fn trace_transaction_opcode_gas(
        &self,
        tx_hash: B256,
    ) -> Result<Option<TransactionOpcodeGas>> {
        node_info!("trace_transactionOpcodeGas");
        self.backend.trace_transaction_opcode_gas(tx_hash).await
    }

    /// Returns opcode gas usage for all transactions in a block.
    ///
    /// Handler for RPC call: `trace_blockOpcodeGas`.
    pub async fn trace_block_opcode_gas(
        &self,
        block_id: BlockId,
    ) -> Result<Option<BlockOpcodeGas>> {
        node_info!("trace_blockOpcodeGas");
        self.backend.trace_block_opcode_gas(block_id).await
    }

    /// Traces a raw signed transaction without importing it into the mempool.
    ///
    /// Handler for RPC call: `trace_rawTransaction`.
    pub async fn trace_raw_transaction(
        &self,
        tx: Bytes,
        trace_types: HashSet<TraceType>,
        block_number: Option<BlockId>,
    ) -> Result<TraceResults> {
        node_info!("trace_rawTransaction");

        let mut data = tx.as_ref();
        if data.is_empty() {
            return Err(BlockchainError::EmptyRawTransactionData);
        }

        let transaction = FoundryTxEnvelope::decode_2718(&mut data)
            .map_err(|_| BlockchainError::FailedToDecodeSignedTransaction)?;
        self.ensure_typed_transaction_supported(&transaction)?;

        #[cfg(feature = "base")]
        let is_eip8130 = transaction.is_eip8130();
        let pending_transaction = PendingTransaction::new(transaction).map_err(|error| {
            #[cfg(feature = "base")]
            if is_eip8130 {
                return BlockchainError::Eip8130TransactionRejected(error.to_string());
            }
            BlockchainError::RecoveryError(error)
        })?;
        let block_request = self.block_request(block_number).await?;

        self.backend
            .trace_raw_transaction(pending_transaction, trace_types, Some(block_request))
            .await
    }

    /// Increases the balance of an account.
    ///
    /// Handler for RPC call: `anvil_addBalance`
    pub async fn anvil_add_balance(&self, address: Address, balance: U256) -> Result<()> {
        node_info!("anvil_addBalance");
        self.backend.add_balance(address, balance).await?;
        Ok(())
    }

    /// Returns a locally stored rollback ancestor, or the remote base of a Monad transaction-hash
    /// fork when that integration is enabled.
    async fn rollback_block(&self, number: u64) -> Result<Option<Block>> {
        if let Some(block) = self.backend.get_block(number) {
            return Ok(Some(block));
        }
        #[cfg(feature = "monad")]
        {
            if self.backend.is_monad() {
                self.backend.monad_rollback_block(number).await
            } else {
                Ok(None)
            }
        }
        #[cfg(not(feature = "monad"))]
        {
            Ok(None)
        }
    }

    /// Rewinds through the selected execution family's recovery path.
    async fn rollback_backend(
        &self,
        common_block: Block,
        mining_guard: &tokio::sync::MutexGuard<'_, ()>,
    ) -> Result<()> {
        #[cfg(feature = "monad")]
        if self.backend.is_monad() {
            return self.backend.rollback_monad(common_block, mining_guard).await;
        }
        self.backend.rollback(common_block, mining_guard).await
    }

    /// Rollback the chain to a specific depth.
    ///
    /// e.g depth = 3
    ///     A  -> B  -> C  -> D  -> E
    ///     A  -> B
    ///
    /// Depth specifies the height to rollback the chain back to. Depth must not exceed the current
    /// chain height, i.e. can't rollback past the genesis block.
    ///
    /// Handler for RPC call: `anvil_rollback`
    pub async fn anvil_rollback(&self, depth: Option<u64>) -> Result<()> {
        node_info!("anvil_rollback");
        let _lifecycle = self.lifecycle_lock.write().await;
        let mining_guard = self.backend.lock_mining().await;
        let depth = depth.unwrap_or(1);

        // Check reorg depth doesn't exceed current chain height
        let current_height = self.backend.best_number();
        let common_height = current_height.checked_sub(depth).ok_or(BlockchainError::RpcError(
            RpcError::invalid_params(format!(
                "Rollback depth must not exceed current chain height: current height {current_height}, depth {depth}"
            )),
        ))?;

        // Get the common ancestor block
        let common_block =
            self.rollback_block(common_height).await?.ok_or(BlockchainError::BlockNotFound)?;

        self.rollback_backend(common_block, &mining_guard).await?;
        Ok(())
    }

    /// Estimates the gas usage of the `request` with the state.
    ///
    /// This will execute the transaction request and find the best gas limit via binary search.
    fn do_estimate_gas_with_state(
        &self,
        mut request: FoundryTransactionRequest,
        state: &dyn DatabaseRef,
        block_env: BlockEnv,
        monad_context: Option<MonadReplayContext>,
    ) -> Result<u128> {
        // A blob request may carry only its sidecar. Take its versioned hashes first, so that the
        // funds check and every execution below price the blob fee.
        if request.as_ref().blob_versioned_hashes.is_none() {
            request.as_mut().populate_blob_hashes();
        }
        let inner = request.as_ref();
        let fees = FeeDetails::new(
            inner.gas_price,
            inner.max_fee_per_gas,
            inner.max_priority_fee_per_gas,
            inner.max_fee_per_blob_gas,
        )?
        .or_zero_fees();

        #[cfg(feature = "base")]
        if self.backend.is_base() && request.is_base() {
            self.backend.ensure_base_eip8130_active_at(block_env.timestamp.saturating_to())?;
            let gas_limit = block_env.gas_limit;
            let result = self
                .backend
                .call_with_state_typed_gas_limit(
                    state,
                    request,
                    fees,
                    block_env,
                    GasEstimateCallOptions::new(gas_limit, false, monad_context),
                )
                .try_into()?;
            return match result {
                GasEstimationCallResult::Success(gas) => Ok(gas),
                GasEstimationCallResult::OutOfGas => {
                    Err(InvalidTransactionError::BasicOutOfGas(u128::from(gas_limit)).into())
                }
                GasEstimationCallResult::Revert(output) => {
                    Err(InvalidTransactionError::Revert(output).into())
                }
                GasEstimationCallResult::EvmError(error) => Err(BlockchainError::EvmError(error)),
            };
        }

        // get the highest possible gas limit, either the request's set value or the currently
        // configured gas limit
        let mut highest_gas_limit = inner.gas.map_or(block_env.gas_limit.into(), |g| g as u128);

        // Tempo AA transactions pay fees in ERC-20 tokens, not ETH. Only treat requests as
        // Tempo AA in Tempo mode, otherwise `feeToken` should not bypass ETH funds checks.
        let is_tempo_aa_tx = self.backend.is_tempo() && request.is_tempo();
        let is_tempo_keychain = matches!(
            &request,
            FoundryTransactionRequest::Tempo(request) if request.key_id.is_some()
        );
        let disable_fee_charge = is_tempo_keychain || inner.from.is_none();

        // Requests without a sender historically estimate value transfers without requiring the
        // default caller to hold ETH. Fund only the simulation overlay so the execution probe
        // preserves that behavior, including transfers that need new-account state gas.
        let mut funded_state;
        let state = if !is_tempo_aa_tx
            && inner.from.is_none()
            && let Some(value) = inner.value
            && !value.is_zero()
        {
            funded_state = CacheDB::new(state);
            let mut caller = state.basic_ref(Address::ZERO)?.unwrap_or_default();
            caller.balance = caller.balance.max(value);
            funded_state.insert_account_info(Address::ZERO, caller);
            &funded_state as &dyn DatabaseRef
        } else {
            state
        };

        let gas_price = fees.gas_price.unwrap_or_default();
        // Check transfer value before any fast path, and cap gas limit by sender balance when the
        // request has a non-zero gas price. Only enforce this for explicit senders: calls without
        // `from` historically estimate against the default caller and must not be capped by the
        // zero address balance.
        if !is_tempo_aa_tx && let Some(from) = inner.from {
            let mut available_funds = self.backend.get_balance_with_state(state, from)?;
            if let Some(value) = inner.value {
                if value > available_funds {
                    return Err(InvalidTransactionError::InsufficientFunds.into());
                }
                // safe: value <= available_funds
                available_funds -= value;
            }
            if gas_price > 0 {
                // Blob gas is paid on top of execution gas, so reserve its maximum cost first. Like
                // the call itself, pay no blob fee when no cap is given.
                if inner.minimal_tx_type().is_eip4844()
                    && let Some(hashes) = &inner.blob_versioned_hashes
                {
                    let max_fee_per_blob_gas = fees.max_fee_per_blob_gas.unwrap_or_default();
                    let blob_gas = U256::from(hashes.len() as u64 * DATA_GAS_PER_BLOB);
                    let blob_cost = blob_gas.saturating_mul(U256::from(max_fee_per_blob_gas));
                    if blob_cost >= available_funds {
                        return Err(InvalidTransactionError::InsufficientFunds.into());
                    }
                    available_funds -= blob_cost;
                }
                // amount of gas the sender can afford with the `gas_price`
                let allowance =
                    available_funds.checked_div(U256::from(gas_price)).unwrap_or_default();
                highest_gas_limit = std::cmp::min(highest_gas_limit, allowance.saturating_to());
            }
        }

        // If the request is a simple native token transfer we can optimize
        // We assume it's a transfer if we have no input data.
        // Skip this optimization for Tempo mode since native ETH transfers are not allowed
        // and Tempo AA transactions have higher intrinsic gas costs (~46k).
        if !self.backend.is_tempo() {
            let to = inner.to.as_ref().and_then(TxKind::to);

            // check certain fields to see if the request could be a simple transfer
            let maybe_transfer = (inner.input.input().is_none()
                || inner.input.input().is_some_and(|data| data.is_empty()))
                && inner.authorization_list.is_none()
                && inner.access_list.is_none()
                && inner.blob_versioned_hashes.is_none();

            if maybe_transfer
                && let Some(to) = to
                && !self.backend.is_precompile(to, &block_env)
                && let Ok(target_code) = self.backend.get_code_with_state(&state, *to)
                && target_code.as_ref().is_empty()
            {
                // Execute the transfer to account for fork-dependent gas costs and validation.
                // With no bytecode or authorizations, a successful run receives no refunds and
                // its gas used is the exact gas required. A new-account state charge can make
                // this probe fail, in which case estimation continues below.
                let probe = self.backend.call_with_state_typed_gas_limit(
                    &state,
                    request.clone(),
                    fees.clone(),
                    block_env.clone(),
                    GasEstimateCallOptions::new(
                        highest_gas_limit.min(MIN_TRANSACTION_GAS) as u64,
                        disable_fee_charge,
                        monad_context.clone(),
                    ),
                );
                if let Ok(GasEstimationCallResult::Success(gas)) = probe.try_into() {
                    return Ok(gas);
                }
            }
        }

        // execute the call without writing to db
        let ethres = self.backend.call_with_state_typed_gas_limit(
            &state,
            request.clone(),
            fees.clone(),
            block_env.clone(),
            GasEstimateCallOptions::new(
                highest_gas_limit as u64,
                disable_fee_charge,
                monad_context.clone(),
            ),
        );

        let gas_used = match ethres.try_into()? {
            GasEstimationCallResult::Success(gas) => Ok(gas),
            GasEstimationCallResult::OutOfGas => {
                Err(InvalidTransactionError::BasicOutOfGas(highest_gas_limit).into())
            }
            GasEstimationCallResult::Revert(output) => {
                Err(InvalidTransactionError::Revert(output).into())
            }
            GasEstimationCallResult::EvmError(err) => {
                warn!(target: "node", "estimation failed due to {:?}", err);
                Err(BlockchainError::EvmError(err))
            }
        }?;

        // at this point we know the call succeeded but want to find the _best_ (lowest) gas the
        // transaction succeeds with. we find this by doing a binary search over the
        // possible range NOTE: this is the gas the transaction used, which is less than the
        // transaction requires to succeed

        // Gas used is a lower bound on the gas required. Keep it inside the search interval so
        // exact-cost transactions, including Amsterdam calls below 21,000 gas, are considered.
        let mut lowest_gas_limit = gas_used.saturating_sub(1);

        // pick a point that's close to the estimated gas
        let mut mid_gas_limit =
            std::cmp::min(gas_used * 3, (highest_gas_limit + lowest_gas_limit) / 2);

        // Binary search for the ideal gas limit
        while lowest_gas_limit + 1 < highest_gas_limit {
            let ethres = self.backend.call_with_state_typed_gas_limit(
                &state,
                request.clone(),
                fees.clone(),
                block_env.clone(),
                GasEstimateCallOptions::new(
                    mid_gas_limit as u64,
                    disable_fee_charge,
                    monad_context.clone(),
                ),
            );

            match ethres.try_into()? {
                GasEstimationCallResult::Success(_) => {
                    // If the transaction succeeded, we can set a ceiling for the highest gas limit
                    // at the current midpoint, as spending any more gas would
                    // make no sense (as the TX would still succeed).
                    highest_gas_limit = mid_gas_limit;
                }
                GasEstimationCallResult::OutOfGas
                | GasEstimationCallResult::Revert(_)
                | GasEstimationCallResult::EvmError(_) => {
                    // If the transaction failed, we can set a floor for the lowest gas limit at the
                    // current midpoint, as spending any less gas would make no
                    // sense (as the TX would still revert due to lack of gas).
                    //
                    // We don't care about the reason here, as we known that transaction is correct
                    // as it succeeded earlier
                    lowest_gas_limit = mid_gas_limit;
                }
            };
            // new midpoint
            mid_gas_limit = (highest_gas_limit + lowest_gas_limit) / 2;
        }

        trace!(target : "node", "Estimated Gas for call {:?}", highest_gas_limit);

        Ok(highest_gas_limit)
    }

    /// Executes the [EthRequest] and returns an RPC [ResponseResult].
    #[allow(clippy::large_stack_frames)]
    pub async fn execute(&self, request: EthRequest) -> ResponseResult {
        trace!(target: "rpc::api", "executing eth request");
        // Fork reset and RPC URL replacement take the write lock internally after their fallible
        // remote staging work. Memory reset takes it before staging because it snapshots live
        // state. Identity, snapshot, and chain-rewind methods also lock internally to keep direct
        // API callers safe without recursively acquiring this fair RwLock.
        let _lifecycle = if matches!(
            &request,
            EthRequest::Reset(_)
                | EthRequest::SetRpcUrl(_)
                | EthRequest::NodeInfo(_)
                | EthRequest::AnvilMetadata(_)
                | EthRequest::EvmSnapshot(_)
                | EthRequest::EvmRevert(_)
                | EthRequest::Reorg(_)
                | EthRequest::Rollback(_)
        ) {
            None
        } else {
            Some(self.lifecycle_lock.read().await)
        };
        let response = match request.clone() {
            EthRequest::EthProtocolVersion(()) => self.protocol_version().to_rpc_result(),
            EthRequest::Web3ClientVersion(()) => self.client_version().to_rpc_result(),
            EthRequest::Web3Sha3(content) => self.sha3(content).to_rpc_result(),
            EthRequest::EthGetAccount(addr, block) => {
                self.get_account(addr, block).await.to_rpc_result()
            }
            EthRequest::EthGetAccountInfo(addr, block) => {
                self.get_account_info(addr, block).await.to_rpc_result()
            }
            EthRequest::EthGetBalance(addr, block) => {
                self.balance(addr, block).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionByHash(hash) => {
                self.transaction_by_hash(hash).await.to_rpc_result()
            }
            EthRequest::EthPendingTransactions(_) => {
                self.pending_transactions().await.to_rpc_result()
            }
            EthRequest::EthSendTransaction(request) => {
                self.send_transaction(*request).await.to_rpc_result()
            }
            EthRequest::EthResend(request, gas_price, gas_limit) => {
                self.resend_transaction(*request, gas_price, gas_limit).await.to_rpc_result()
            }
            EthRequest::EthSendTransactionSync(request) => {
                self.send_transaction_sync(*request).await.to_rpc_result()
            }
            EthRequest::EthChainId(_) => self.eth_chain_id().to_rpc_result(),
            EthRequest::EthNetworkId(_) => self.network_id().to_rpc_result(),
            EthRequest::NetListening(_) => self.net_listening().to_rpc_result(),
            EthRequest::EthHashrate(()) => self.hashrate().to_rpc_result(),
            EthRequest::EthGasPrice(_) => self.eth_gas_price().to_rpc_result(),
            EthRequest::EthBaseFee(_) => self.base_fee().to_rpc_result(),
            EthRequest::EthMaxPriorityFeePerGas(_) => {
                self.gas_max_priority_fee_per_gas().to_rpc_result()
            }
            EthRequest::EthBlobBaseFee(_) => self.blob_base_fee().to_rpc_result(),
            EthRequest::EthAccounts(_) => self.accounts().to_rpc_result(),
            EthRequest::EthBlockNumber(_) => self.block_number().to_rpc_result(),
            EthRequest::EthCoinbase(()) => self.author().to_rpc_result(),
            EthRequest::EthGetStorageAt(addr, slot, block) => {
                self.storage_at(addr, slot, block).await.to_rpc_result()
            }
            EthRequest::EthGetStorageValues(requests, block) => {
                self.storage_values(requests, block).await.to_rpc_result()
            }
            EthRequest::EthGetBlockByHash(hash, full) => {
                if full {
                    self.block_by_hash_full(hash).await.to_rpc_result()
                } else {
                    self.block_by_hash(hash).await.to_rpc_result()
                }
            }
            EthRequest::EthGetHeaderByHash(hash) => self.header_by_hash(hash).await.to_rpc_result(),
            EthRequest::EthGetBlockByNumber(num, full) => {
                if full {
                    self.block_by_number_full(num).await.to_rpc_result()
                } else {
                    self.block_by_number(num).await.to_rpc_result()
                }
            }
            EthRequest::EthGetHeaderByNumber(num) => {
                self.header_by_number(num).await.to_rpc_result()
            }
            EthRequest::EthGetBlockAccessList(block_id) => {
                self.block_access_list(block_id).await.to_rpc_result()
            }
            EthRequest::EthGetBlockAccessListByBlockHash(block_hash) => {
                self.block_access_list_by_hash(block_hash).await.to_rpc_result()
            }
            EthRequest::EthGetBlockAccessListByBlockNumber(block_number) => {
                self.block_access_list_by_number(block_number).await.to_rpc_result()
            }
            EthRequest::EthGetBlockAccessListRaw(block_id) => {
                self.block_access_list_raw(block_id).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionCount(params) => {
                let (address, block, nonce_key) = params.into_parts();
                #[cfg(feature = "base")]
                {
                    self.transaction_count_with_key(address, block, nonce_key).await.to_rpc_result()
                }
                #[cfg(not(feature = "base"))]
                {
                    if nonce_key.is_some() {
                        return ResponseResult::Error(RpcError::invalid_params(
                            "nonce-key transaction counts require Base support",
                        ));
                    }
                    self.transaction_count(address, block).await.to_rpc_result()
                }
            }
            EthRequest::EthGetTransactionCountByHash(hash) => {
                self.block_transaction_count_by_hash(hash).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionCountByNumber(num) => {
                self.block_transaction_count_by_number(num).await.to_rpc_result()
            }
            EthRequest::EthGetUnclesCountByHash(hash) => {
                self.block_uncles_count_by_hash(hash).await.to_rpc_result()
            }
            EthRequest::EthGetUnclesCountByNumber(num) => {
                self.block_uncles_count_by_number(num).await.to_rpc_result()
            }
            EthRequest::EthGetCodeAt(addr, block) => {
                self.get_code(addr, block).await.to_rpc_result()
            }
            EthRequest::EthGetProof(addr, keys, block) => {
                self.get_proof(addr, keys, block).await.to_rpc_result()
            }
            EthRequest::EthSign(addr, content) => self.sign(addr, content).await.to_rpc_result(),
            EthRequest::PersonalSign(content, addr) => {
                self.sign(addr, content).await.to_rpc_result()
            }
            EthRequest::EthSignTransaction(request) => {
                self.sign_transaction(*request).await.to_rpc_result()
            }
            EthRequest::EthSignTypedData(addr, data) => {
                self.sign_typed_data(addr, data).await.to_rpc_result()
            }
            EthRequest::EthSignTypedDataV3(addr, data) => {
                self.sign_typed_data_v3(addr, data).await.to_rpc_result()
            }
            EthRequest::EthSignTypedDataV4(addr, data) => {
                self.sign_typed_data_v4(addr, &data).await.to_rpc_result()
            }
            EthRequest::EthSendRawTransaction(tx) => {
                self.send_raw_transaction(tx).await.to_rpc_result()
            }
            EthRequest::EthSendRawTransactionSync(tx, timeout_ms) => {
                self.send_raw_transaction_sync(tx, timeout_ms).await.to_rpc_result()
            }
            EthRequest::EthSendRawTransactionConditional(tx, condition) => {
                self.send_raw_transaction_conditional(tx, condition).await.to_rpc_result()
            }
            EthRequest::EthSignRawTransaction(tx) => {
                self.sign_raw_transaction(tx).await.to_rpc_result()
            }
            EthRequest::AnvilClassifyTransaction(tx) => {
                self.anvil_classify_transaction(tx).to_rpc_result()
            }
            EthRequest::EthCall(call, block, state_override, block_overrides) => self
                .call(call, block, EvmOverrides::new(state_override, block_overrides))
                .await
                .to_rpc_result(),
            EthRequest::EthCallMany(bundles, state_context, state_override) => {
                self.call_many(bundles, state_context, state_override).await.to_rpc_result()
            }
            EthRequest::EthCallBundle(bundle) => self.call_bundle(bundle).await.to_rpc_result(),
            EthRequest::EthSimulateV1(simulation, block) => {
                self.simulate_v1_raw(simulation, block).await.to_rpc_result()
            }
            EthRequest::EthCreateAccessList(call, block, state_override) => {
                self.create_access_list(call, block, state_override).await.to_rpc_result()
            }
            EthRequest::EthEstimateGas(call, block, state_override, block_overrides) => self
                .estimate_gas(call, block, EvmOverrides::new(state_override, block_overrides))
                .await
                .to_rpc_result(),
            EthRequest::EthFillTransaction(request) => {
                self.fill_transaction(request).await.to_rpc_result()
            }
            EthRequest::EthGetRawTransactionByHash(hash) => {
                self.raw_transaction(hash).await.to_rpc_result()
            }
            EthRequest::GetBlobByHash(hash) => {
                self.anvil_get_blob_by_versioned_hash(hash).to_rpc_result()
            }
            EthRequest::GetBlobByTransactionHash(hash) => {
                self.anvil_get_blob_by_tx_hash(hash).to_rpc_result()
            }
            EthRequest::GetGenesisTime(()) => self.anvil_get_genesis_time().to_rpc_result(),
            EthRequest::GetLastBlockWallTime(()) => {
                self.anvil_get_last_block_wall_time().to_rpc_result()
            }
            EthRequest::EthGetRawTransactionByBlockHashAndIndex(hash, index) => {
                self.raw_transaction_by_block_hash_and_index(hash, index).await.to_rpc_result()
            }
            EthRequest::EthGetRawTransactionByBlockNumberAndIndex(num, index) => {
                self.raw_transaction_by_block_number_and_index(num, index).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionByBlockHashAndIndex(hash, index) => {
                self.transaction_by_block_hash_and_index(hash, index).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionByBlockNumberAndIndex(num, index) => {
                self.transaction_by_block_number_and_index(num, index).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionReceipt(tx) => {
                self.transaction_receipt(tx).await.to_rpc_result()
            }
            EthRequest::EthGetBlockReceipts(number) => {
                self.block_receipts(number).await.to_rpc_result()
            }
            EthRequest::EthGetUncleByBlockHashAndIndex(hash, index) => {
                self.uncle_by_block_hash_and_index(hash, index).await.to_rpc_result()
            }
            EthRequest::EthGetUncleByBlockNumberAndIndex(num, index) => {
                self.uncle_by_block_number_and_index(num, index).await.to_rpc_result()
            }
            EthRequest::EthGetLogs(filter) => self.logs(filter).await.to_rpc_result(),
            EthRequest::EthGetWork(_) => self.work().to_rpc_result(),
            EthRequest::EthSyncing(_) => self.syncing().to_rpc_result(),
            EthRequest::EthConfig(_) => self.config().to_rpc_result(),
            EthRequest::EthSubmitWork(nonce, pow, digest) => {
                self.submit_work(nonce, pow, digest).to_rpc_result()
            }
            EthRequest::EthSubmitHashRate(rate, id) => {
                self.submit_hashrate(rate, id).to_rpc_result()
            }
            EthRequest::EthFeeHistory(count, newest, reward_percentiles) => {
                self.fee_history(count, newest, reward_percentiles).await.to_rpc_result()
            }
            // non eth-standard rpc calls
            EthRequest::DebugGetRawTransaction(hash) => {
                self.raw_transaction(hash).await.to_rpc_result()
            }
            EthRequest::DebugGetRawReceipts(block) => {
                self.raw_receipts(block).await.to_rpc_result()
            }
            EthRequest::DebugGetRawTransactions(block) => {
                self.raw_transactions(block).await.to_rpc_result()
            }
            EthRequest::DebugGetRawHeader(block) => self.raw_header(block).await.to_rpc_result(),
            EthRequest::DebugGetRawBlock(block) => self.raw_block(block).await.to_rpc_result(),
            // non eth-standard rpc calls
            EthRequest::DebugClearTxpool(_) => self.debug_clear_txpool().await.to_rpc_result(),
            // non eth-standard rpc calls
            EthRequest::DebugTraceTransaction(tx, opts) => {
                self.debug_trace_transaction(tx, opts).await.to_rpc_result()
            }
            // non eth-standard rpc calls
            EthRequest::DebugTraceCall(tx, block, opts) => {
                self.debug_trace_call(tx, block, opts).await.to_rpc_result()
            }
            EthRequest::DebugCodeByHash(hash, block) => {
                self.debug_code_by_hash(hash, block).await.to_rpc_result()
            }
            EthRequest::DebugDbGet(key) => self.debug_db_get(key).await.to_rpc_result(),
            EthRequest::DebugGetModifiedAccountsByNumber(start_number, end_number) => {
                self.debug_get_modified_accounts_by_number(start_number, end_number).to_rpc_result()
            }
            EthRequest::DebugFreeOsMemory(()) => self.debug_free_os_memory().to_rpc_result(),
            EthRequest::DebugAccountInfoAt(block_id, tx_index, address) => {
                self.debug_account_info_at(block_id, tx_index, address).await.to_rpc_result()
            }
            EthRequest::DebugExecutionWitness(block) => {
                self.debug_execution_witness(block).await.to_rpc_result()
            }
            EthRequest::DebugTraceBlock(rlp_block, opts) => {
                self.debug_trace_block(rlp_block, opts).await.to_rpc_result()
            }
            EthRequest::DebugTraceBlockByHash(block_hash, opts) => {
                self.debug_trace_block_by_hash(block_hash, opts).await.to_rpc_result()
            }
            EthRequest::DebugTraceBlockByNumber(block_number, opts) => {
                self.debug_trace_block_by_number(block_number, opts).await.to_rpc_result()
            }
            EthRequest::TraceCall(tx, trace_types, block) => {
                self.trace_call(tx, trace_types, block).await.to_rpc_result()
            }
            EthRequest::TraceTransaction(tx) => self.trace_transaction(tx).await.to_rpc_result(),
            EthRequest::TraceBlock(block) => self.trace_block(block).await.to_rpc_result(),
            EthRequest::TraceFilter(filter) => self.trace_filter(filter).await.to_rpc_result(),
            EthRequest::TraceGet(hash, indices) => {
                self.trace_get(hash, indices).await.to_rpc_result()
            }
            EthRequest::TraceReplayBlockTransactions(block, trace_types) => {
                self.trace_replay_block_transactions(block, trace_types).await.to_rpc_result()
            }
            EthRequest::TraceReplayTransaction(transaction, trace_types) => {
                self.trace_replay_transaction(transaction, trace_types).await.to_rpc_result()
            }
            EthRequest::TraceTransactionOpcodeGas(tx_hash) => {
                self.trace_transaction_opcode_gas(tx_hash).await.to_rpc_result()
            }
            EthRequest::TraceBlockOpcodeGas(block_id) => {
                self.trace_block_opcode_gas(block_id).await.to_rpc_result()
            }
            EthRequest::TraceRawTransaction(tx, trace_types, block_number) => {
                self.trace_raw_transaction(tx, trace_types, block_number).await.to_rpc_result()
            }
            EthRequest::TraceCallMany(calls, block_number) => {
                self.trace_call_many(calls, block_number).await.to_rpc_result()
            }
            EthRequest::ImpersonateAccount(addr) => {
                self.anvil_impersonate_account(addr).await.to_rpc_result()
            }
            EthRequest::StopImpersonatingAccount(addr) => {
                self.anvil_stop_impersonating_account(addr).await.to_rpc_result()
            }
            EthRequest::AutoImpersonateAccount(enable) => {
                self.anvil_auto_impersonate_account(enable).await.to_rpc_result()
            }
            EthRequest::ImpersonateSignature(signature, address) => {
                self.anvil_impersonate_signature(signature, address).await.to_rpc_result()
            }
            EthRequest::GetAutoMine(()) => self.anvil_get_auto_mine().to_rpc_result(),
            EthRequest::Mine(blocks, interval) => {
                self.anvil_mine(blocks, interval).await.to_rpc_result()
            }
            EthRequest::SetAutomine(enabled) => {
                self.anvil_set_auto_mine(enabled).await.to_rpc_result()
            }
            EthRequest::SetIntervalMining(interval) => {
                self.anvil_set_interval_mining(interval).to_rpc_result()
            }
            EthRequest::GetIntervalMining(()) => self.anvil_get_interval_mining().to_rpc_result(),
            EthRequest::DropTransaction(tx) => {
                self.anvil_drop_transaction(tx).await.to_rpc_result()
            }
            EthRequest::DropAllTransactions() => {
                self.anvil_drop_all_transactions().await.to_rpc_result()
            }
            EthRequest::Reset(fork) => {
                self.anvil_reset(fork.and_then(|p| p.params)).await.to_rpc_result()
            }
            EthRequest::SetBalance(addr, val) => {
                self.anvil_set_balance(addr, val).await.to_rpc_result()
            }
            EthRequest::AddBalance(addr, val) => {
                self.anvil_add_balance(addr, val).await.to_rpc_result()
            }
            EthRequest::DealERC20(addr, token_addr, val) => {
                self.anvil_deal_erc20(addr, token_addr, val).await.to_rpc_result()
            }
            EthRequest::DealTIP20(addr, token_addr, val) => {
                self.anvil_deal_tip20(addr, token_addr, val).await.to_rpc_result()
            }
            EthRequest::SetERC20Allowance(owner, spender, token_addr, val) => self
                .anvil_set_erc20_allowance(owner, spender, token_addr, val)
                .await
                .to_rpc_result(),
            EthRequest::SetCode(addr, code) => {
                self.anvil_set_code(addr, code).await.to_rpc_result()
            }
            EthRequest::SetNonce(addr, nonce) => {
                self.anvil_set_nonce(addr, nonce).await.to_rpc_result()
            }
            EthRequest::SetStorageAt(addr, slot, val) => {
                self.anvil_set_storage_at(addr, slot, val).await.to_rpc_result()
            }
            EthRequest::SetCoinbase(addr) => self.anvil_set_coinbase(addr).await.to_rpc_result(),
            EthRequest::SetNextBlockPrevRandao(prevrandao) => {
                self.anvil_set_next_block_prevrandao(prevrandao).await.to_rpc_result()
            }
            EthRequest::SetNextBlockParentBeaconBlockRoot(root) => {
                self.anvil_set_next_block_parent_beacon_block_root(root).await.to_rpc_result()
            }
            EthRequest::SetChainId(id) => self.anvil_set_chain_id(id).await.to_rpc_result(),
            EthRequest::SetLogging(log) => self.anvil_set_logging(log).await.to_rpc_result(),
            EthRequest::SetMinGasPrice(gas) => {
                self.anvil_set_min_gas_price(gas).await.to_rpc_result()
            }
            EthRequest::SetNextBlockBaseFeePerGas(gas) => {
                self.anvil_set_next_block_base_fee_per_gas(gas).await.to_rpc_result()
            }
            EthRequest::DumpState(preserve_historical_states) => self
                .anvil_dump_state(preserve_historical_states.and_then(|s| s.params))
                .await
                .to_rpc_result(),
            EthRequest::LoadState(buf) => self.anvil_load_state(buf).await.to_rpc_result(),
            EthRequest::NodeInfo(_) => self.anvil_node_info().await.to_rpc_result(),
            EthRequest::AnvilMetadata(_) => self.anvil_metadata().await.to_rpc_result(),
            EthRequest::EvmSnapshot(_) => self.evm_snapshot().await.to_rpc_result(),
            EthRequest::EvmRevert(id) => self.evm_revert(id).await.to_rpc_result(),
            EthRequest::EvmIncreaseTime(time) => self.evm_increase_time(time).await.to_rpc_result(),
            EthRequest::EvmSetNextBlockTimeStamp(time) => {
                if time >= U256::from(u64::MAX) {
                    return ResponseResult::Error(RpcError::invalid_params(
                        "The timestamp is too big",
                    ));
                }
                let time = time.to::<u64>();
                self.evm_set_next_block_timestamp(time).to_rpc_result()
            }
            EthRequest::EvmSetTime(timestamp) => {
                if timestamp >= U256::from(u64::MAX) {
                    return ResponseResult::Error(RpcError::invalid_params(
                        "The timestamp is too big",
                    ));
                }
                // evm_setTime accepts either seconds or milliseconds for Ganache compatibility.
                // Timestamps above 1e12 are interpreted as milliseconds and converted to
                // seconds; below that threshold they are treated as seconds directly.
                let raw = timestamp.to::<u64>();
                let time = if raw > 1_000_000_000_000 {
                    Duration::from_millis(raw).as_secs()
                } else {
                    raw
                };
                self.evm_set_time(time).to_rpc_result()
            }
            EthRequest::EvmSetBlockGasLimit(gas_limit) => {
                self.evm_set_block_gas_limit(gas_limit).to_rpc_result()
            }
            EthRequest::EvmSetBlockTimeStampInterval(time) => {
                self.evm_set_block_timestamp_interval(time).to_rpc_result()
            }
            EthRequest::EvmRemoveBlockTimeStampInterval(()) => {
                self.evm_remove_block_timestamp_interval().to_rpc_result()
            }
            EthRequest::EvmMine(mine) => {
                self.evm_mine(mine.and_then(|p| p.params)).await.to_rpc_result()
            }
            EthRequest::EvmMineDetailed(mine) => {
                self.evm_mine_detailed(mine.and_then(|p| p.params)).await.to_rpc_result()
            }
            EthRequest::SetRpcUrl(url) => self.anvil_set_rpc_url(url).await.to_rpc_result(),
            EthRequest::EthSendUnsignedTransaction(tx) => {
                self.eth_send_unsigned_transaction(*tx).await.to_rpc_result()
            }
            EthRequest::EthNewFilter(filter) => self.new_filter(filter).await.to_rpc_result(),
            EthRequest::EthGetFilterChanges(id) => self.get_filter_changes(&id).await,
            EthRequest::EthNewBlockFilter(_) => self.new_block_filter().await.to_rpc_result(),
            EthRequest::EthNewPendingTransactionFilter(full) => {
                self.new_pending_transaction_filter(full.unwrap_or(false)).await.to_rpc_result()
            }
            EthRequest::EthGetFilterLogs(id) => self.get_filter_logs(&id).await.to_rpc_result(),
            EthRequest::EthUninstallFilter(id) => self.uninstall_filter(&id).await.to_rpc_result(),
            EthRequest::TxPoolStatus(_) => self.txpool_status().await.to_rpc_result(),
            EthRequest::TxPoolInspect(_) => self.txpool_inspect().await.to_rpc_result(),
            EthRequest::TxPoolContent(_) => self.txpool_content().await.to_rpc_result(),
            EthRequest::TxPoolContentFrom(from) => {
                self.txpool_content_from(from).await.to_rpc_result()
            }
            EthRequest::ErigonGetHeaderByNumber(num) => {
                self.erigon_get_header_by_number(num).await.to_rpc_result()
            }
            EthRequest::OtsGetApiLevel(_) => self.ots_get_api_level().await.to_rpc_result(),
            EthRequest::OtsGetInternalOperations(hash) => {
                self.ots_get_internal_operations(hash).await.to_rpc_result()
            }
            EthRequest::OtsHasCode(addr, num) => self.ots_has_code(addr, num).await.to_rpc_result(),
            EthRequest::OtsTraceTransaction(hash) => {
                self.ots_trace_transaction(hash).await.to_rpc_result()
            }
            EthRequest::OtsGetTransactionError(hash) => {
                self.ots_get_transaction_error(hash).await.to_rpc_result()
            }
            EthRequest::OtsGetBlockDetails(num) => {
                self.ots_get_block_details(num).await.to_rpc_result()
            }
            EthRequest::OtsGetBlockDetailsByHash(hash) => {
                self.ots_get_block_details_by_hash(hash).await.to_rpc_result()
            }
            EthRequest::OtsGetBlockTransactions(num, page, page_size) => {
                self.ots_get_block_transactions(num, page, page_size).await.to_rpc_result()
            }
            EthRequest::OtsSearchTransactionsBefore(address, num, page_size) => {
                self.ots_search_transactions_before(address, num, page_size).await.to_rpc_result()
            }
            EthRequest::OtsSearchTransactionsAfter(address, num, page_size) => {
                self.ots_search_transactions_after(address, num, page_size).await.to_rpc_result()
            }
            EthRequest::OtsGetTransactionBySenderAndNonce(address, nonce) => {
                self.ots_get_transaction_by_sender_and_nonce(address, nonce).await.to_rpc_result()
            }
            EthRequest::EthGetTransactionBySenderAndNonce(sender, nonce) => {
                self.transaction_by_sender_and_nonce(sender, nonce).await.to_rpc_result()
            }
            EthRequest::OtsGetContractCreator(address) => {
                self.ots_get_contract_creator(address).await.to_rpc_result()
            }
            EthRequest::RemovePoolTransactions(address) => {
                self.anvil_remove_pool_transactions(address).await.to_rpc_result()
            }
            EthRequest::Reorg(reorg_options) => {
                self.anvil_reorg(reorg_options).await.to_rpc_result()
            }
            EthRequest::Rollback(depth) => self.anvil_rollback(depth).await.to_rpc_result(),
            EthRequest::SetFeeToken(user, token) => {
                self.anvil_set_fee_token(user, token).await.to_rpc_result()
            }
            EthRequest::SetValidatorFeeToken(validator, token) => {
                self.anvil_set_validator_fee_token(validator, token).await.to_rpc_result()
            }
            EthRequest::SetFeeAmmLiquidity(user_token, validator_token, amount) => self
                .anvil_set_fee_amm_liquidity(user_token, validator_token, amount)
                .await
                .to_rpc_result(),
        };

        if let ResponseResult::Error(err) = &response {
            node_info!("\nRPC request failed:");
            node_info!("    Request: {:?}", request);
            node_info!("    Error: {}\n", err);
        }

        response
    }

    fn sign_request(&self, from: &Address, typed_tx: FoundryTypedTx) -> Result<FoundryTxEnvelope> {
        match typed_tx {
            #[cfg(feature = "optimism")]
            FoundryTypedTx::Deposit(_) => return Ok(typed_tx.into_impersonated()),
            _ => {
                for signer in self.signers.iter() {
                    if signer.accounts().contains(from) {
                        return signer.sign_transaction_from(from, typed_tx);
                    }
                }
            }
        }
        Err(BlockchainError::NoSignerAvailable)
    }

    async fn inner_raw_transaction(&self, hash: B256) -> Result<Option<Bytes>> {
        match self.pool.get_transaction(hash) {
            Some(tx) => Ok(Some(tx.transaction.encoded_2718().into())),
            None => match self.backend.transaction_by_hash(hash).await? {
                Some(tx) => encode_rpc_transaction(&tx).map(Some),
                None => Ok(None),
            },
        }
    }

    /// Returns balance of the given account.
    ///
    /// Handler for ETH RPC call: `eth_getBalance`
    pub async fn balance(&self, address: Address, block_number: Option<BlockId>) -> Result<U256> {
        node_info!("eth_getBalance");
        let block_request = self.block_request(block_number).await?;

        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(fork.get_balance(address, number).await?);
        }

        self.backend.get_balance(address, Some(block_request)).await
    }

    /// Returns the ethereum account.
    ///
    /// Handler for ETH RPC call: `eth_getAccount`
    pub async fn get_account(
        &self,
        address: Address,
        block_number: Option<BlockId>,
    ) -> Result<TrieAccount> {
        node_info!("eth_getAccount");
        let block_request = self.block_request(block_number).await?;

        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(fork.get_account(address, number).await?);
        }

        self.backend.get_account_at_block(address, Some(block_request)).await
    }

    /// Returns the account information including balance, nonce, code and storage
    ///
    /// Note: This isn't support by all providers
    pub async fn get_account_info(
        &self,
        address: Address,
        block_number: Option<BlockId>,
    ) -> Result<alloy_rpc_types::eth::AccountInfo> {
        node_info!("eth_getAccountInfo");

        let block_request = self.block_request(block_number).await?;
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
        {
            trace!(target: "node", "get_account_info: fork block {}, requested block {number}", fork.block_number());
            if block_number.is_some_and(|block| !block.is_latest())
                && fork.predates_fork_inclusive(number)
            {
                if fork.requires_account_info() {
                    return Ok(fork.get_account_info(address, number).await?);
                }
                let balance = fork.get_balance(address, number).map_err(BlockchainError::from);
                let code = fork.get_code(address, number).map_err(BlockchainError::from);
                let nonce = fork.get_nonce(address, number).map_err(BlockchainError::from);
                let (balance, code, nonce) = try_join!(balance, code, nonce)?;
                return Ok(alloy_rpc_types::eth::AccountInfo { balance, nonce, code });
            }
        }

        self.backend.get_account_info_at_block(address, Some(block_request)).await
    }
    /// Returns content of the storage at given address.
    ///
    /// Handler for ETH RPC call: `eth_getStorageAt`
    pub async fn storage_at(
        &self,
        address: Address,
        index: U256,
        block_number: Option<BlockId>,
    ) -> Result<B256> {
        node_info!("eth_getStorageAt");
        let block_request = self.block_request(block_number).await?;

        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(B256::from(
                fork.storage_at(address, index, Some(BlockNumber::Number(number))).await?,
            ));
        }

        self.backend.storage_at(address, index, Some(block_request)).await
    }

    /// Returns storage values for multiple accounts and slots in a single call.
    ///
    /// Handler for ETH RPC call: `eth_getStorageValues`
    pub async fn storage_values(
        &self,
        requests: HashMap<Address, Vec<B256>>,
        block_number: Option<BlockId>,
    ) -> Result<HashMap<Address, Vec<B256>>> {
        node_info!("eth_getStorageValues");

        if requests.is_empty() {
            return Err(RpcError::invalid_params("empty request").into());
        }

        let total_slots: usize = requests.values().map(|s| s.len()).sum();
        if total_slots > 1024 {
            return Err(BlockchainError::RpcError(RpcError::invalid_params(format!(
                "total slot count {total_slots} exceeds limit 1024"
            ))));
        }

        let block_request = self.block_request(block_number).await?;

        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            let mut result: HashMap<Address, Vec<B256>> = HashMap::default();
            for (address, slots) in requests {
                let mut values = Vec::with_capacity(slots.len());
                for slot in &slots {
                    let val = fork
                        .storage_at(address, (*slot).into(), Some(BlockNumber::Number(number)))
                        .await?;
                    values.push(B256::from(val));
                }
                result.insert(address, values);
            }
            return Ok(result);
        }

        self.backend.storage_values(requests, Some(block_request)).await
    }

    /// Returns block with given number.
    ///
    /// Handler for ETH RPC call: `eth_getBlockByNumber`
    pub async fn block_by_number(&self, number: BlockNumber) -> Result<Option<AnyRpcBlock>> {
        node_info!("eth_getBlockByNumber");
        if number.is_pending() {
            return Ok(Some(self.pending_block().await?));
        }

        self.backend.block_by_number(number).await
    }

    /// Returns block header with given number.
    ///
    /// Handler for ETH RPC call: `eth_getHeaderByNumber`
    pub async fn header_by_number(
        &self,
        number: BlockNumber,
    ) -> Result<Option<WithOtherFields<AnyRpcHeader>>> {
        node_info!("eth_getHeaderByNumber");
        if number.is_pending() {
            let WithOtherFields { inner: block, other } = self.pending_block().await?.0;
            return Ok(Some(WithOtherFields { inner: block.header, other }));
        }

        Ok(self.backend.block_by_number(number).await?.map(|block| {
            let WithOtherFields { inner: block, other } = block.0;
            WithOtherFields { inner: block.header, other }
        }))
    }

    /// Returns a _full_ block with given number
    ///
    /// Handler for ETH RPC call: `eth_getBlockByNumber`
    pub async fn block_by_number_full(&self, number: BlockNumber) -> Result<Option<AnyRpcBlock>> {
        node_info!("eth_getBlockByNumber");
        if number.is_pending() {
            return self.pending_block_full().await;
        }
        self.backend.block_by_number_full(number).await
    }

    /// Returns the EIP-7928 block access list for a block.
    ///
    /// Handler for ETH RPC call: `eth_getBlockAccessList`
    pub async fn block_access_list(&self, block_id: BlockId) -> Result<Option<serde_json::Value>> {
        node_info!("eth_getBlockAccessList");
        let block_request = self.block_request(Some(block_id)).await?;
        let BlockRequest::Number(number) = block_request else { return Ok(None) };

        if let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            return Ok(fork.block_access_list(block_id).await?);
        }

        let block_access_list = match block_id {
            BlockId::Hash(hash) => self.backend.block_access_list_by_hash(hash.block_hash),
            BlockId::Number(_) => self.backend.block_access_list_by_number(number),
        };
        Ok(block_access_list
            .map(|bal| serde_json::to_value(bal).expect("BAL serialization is infallible")))
    }

    /// Returns the EIP-7928 block access list for a block hash.
    ///
    /// Handler for ETH RPC call: `eth_getBlockAccessListByBlockHash`
    pub async fn block_access_list_by_hash(
        &self,
        block_hash: B256,
    ) -> Result<Option<serde_json::Value>> {
        node_info!("eth_getBlockAccessListByBlockHash");
        if let Some(block_access_list) = self.backend.block_access_list_by_hash(block_hash) {
            return Ok(Some(
                serde_json::to_value(block_access_list).expect("BAL serialization is infallible"),
            ));
        }

        let Some(fork) = self.get_fork() else { return Ok(None) };

        // Only blocks we know to be mined after the fork point are guaranteed to be unknown
        // upstream. Anything else, including hashes we cannot resolve locally, is left to the fork.
        if let Some(block) = self.backend.get_block_by_hash(block_hash)
            && !fork.predates_fork_inclusive(block.header.number)
        {
            return Ok(None);
        }

        Ok(fork.block_access_list_by_hash(block_hash).await?)
    }

    /// Returns the EIP-7928 block access list for a block number.
    ///
    /// Handler for ETH RPC call: `eth_getBlockAccessListByBlockNumber`
    pub async fn block_access_list_by_number(
        &self,
        block_number: BlockNumber,
    ) -> Result<Option<serde_json::Value>> {
        node_info!("eth_getBlockAccessListByBlockNumber");
        let block_request = self.block_request(Some(BlockId::Number(block_number))).await?;
        let BlockRequest::Number(number) = block_request else { return Ok(None) };

        if let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            return Ok(fork.block_access_list_by_number(block_number).await?);
        }

        Ok(self
            .backend
            .block_access_list_by_number(number)
            .map(|bal| serde_json::to_value(bal).expect("BAL serialization is infallible")))
    }

    /// Returns the raw EIP-7928 block access list for a block.
    ///
    /// Handler for ETH RPC call: `eth_getBlockAccessListRaw`
    pub async fn block_access_list_raw(&self, block_id: BlockId) -> Result<Option<Bytes>> {
        node_info!("eth_getBlockAccessListRaw");
        let block_request = self.block_request(Some(block_id)).await?;
        let BlockRequest::Number(number) = block_request else { return Ok(None) };

        if let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            return Ok(fork.block_access_list_raw(block_id).await?);
        }

        let block_access_list = match block_id {
            BlockId::Hash(hash) => self.backend.block_access_list_by_hash(hash.block_hash),
            BlockId::Number(_) => self.backend.block_access_list_by_number(number),
        };
        Ok(block_access_list.map(|bal| {
            let mut encoded = Vec::new();
            alloy_rlp::encode_list(&bal, &mut encoded);
            encoded.into()
        }))
    }

    /// Returns the number of transactions sent from given address at given time (block number).
    ///
    /// Also checks the pending transactions if `block_number` is
    /// `BlockId::Number(BlockNumber::Pending)`
    ///
    /// Handler for ETH RPC call: `eth_getTransactionCount`
    pub async fn transaction_count(
        &self,
        address: Address,
        block_number: Option<BlockId>,
    ) -> Result<U256> {
        node_info!("eth_getTransactionCount");
        self.get_transaction_count(address, block_number).await.map(U256::from)
    }

    /// Returns a protocol or EIP-8130 channel nonce.
    #[cfg(feature = "base")]
    pub async fn transaction_count_with_key(
        &self,
        address: Address,
        block_number: Option<BlockId>,
        nonce_key: Option<U256>,
    ) -> Result<U256> {
        let Some(nonce_key) = nonce_key else {
            return self.transaction_count(address, block_number).await;
        };
        if nonce_key.is_zero() {
            return self.transaction_count(address, block_number).await;
        }

        let block_request = self.block_request(block_number).await?;
        let timestamp = self.backend.block_request_timestamp(&block_request).await?;
        self.backend.ensure_base_eip8130_active_at(timestamp)?;
        if nonce_key == Eip8130Constants::NONCE_KEY_MAX {
            return Err(RpcError::invalid_params(
                "nonce_key NONCE_KEY_MAX selects the expiring-nonce channel which has no \
                 per-channel counter",
            )
            .into());
        }
        let slot = NonceManagerStorage::nonce_slot(address, nonce_key)
            .map_err(|error| RpcError::invalid_params(error.to_string()))?;
        let word = self
            .backend
            .storage_at(NonceManagerStorage::ADDRESS, slot, Some(block_request))
            .await?;
        Ok(Eip8130Nonce::decode_channel_nonce(word.into()))
    }

    /// Returns the number of transactions in a block with given block number.
    ///
    /// Handler for ETH RPC call: `eth_getBlockTransactionCountByNumber`
    pub async fn block_transaction_count_by_number(
        &self,
        block_number: BlockNumber,
    ) -> Result<Option<U256>> {
        node_info!("eth_getBlockTransactionCountByNumber");
        if block_number.is_pending() {
            let txs = self.pool.ready_transactions().collect();
            let block = self.backend.pending_block(txs).await?;
            return Ok(Some(U256::from(block.block.body.transactions.len())));
        }

        let block = self.backend.block_by_number(block_number).await?;
        let txs = block.map(|b| match b.transactions() {
            BlockTransactions::Full(txs) => U256::from(txs.len()),
            BlockTransactions::Hashes(txs) => U256::from(txs.len()),
            BlockTransactions::Uncle => U256::ZERO,
        });
        Ok(txs)
    }

    /// Returns the code at given address at given time (block number).
    ///
    /// Handler for ETH RPC call: `eth_getCode`
    pub async fn get_code(&self, address: Address, block_number: Option<BlockId>) -> Result<Bytes> {
        node_info!("eth_getCode");
        let block_request = self.block_request(block_number).await?;
        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(fork.get_code(address, number).await?);
        }
        self.backend.get_code(address, Some(block_request)).await
    }

    /// Returns the account and storage values of the specified account including the Merkle-proof.
    /// This call can be used to verify that the data you are pulling from is not tampered with.
    ///
    /// Handler for ETH RPC call: `eth_getProof`
    pub async fn get_proof(
        &self,
        address: Address,
        keys: Vec<B256>,
        block_number: Option<BlockId>,
    ) -> Result<EIP1186AccountProofResponse> {
        node_info!("eth_getProof");
        let block_request = self.block_request(block_number).await?;

        // If we're in forking mode, or still on the forked block (no blocks mined yet) then we can
        // delegate the call.
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            return Ok(fork.get_proof(address, keys, Some(number.into())).await?);
        }

        let proof = self.backend.prove_account_at(address, keys, Some(block_request)).await?;
        Ok(proof)
    }

    /// Signs a transaction
    ///
    /// Handler for ETH RPC call: `eth_signTransaction`
    pub async fn sign_transaction(
        &self,
        request: WithOtherFields<TransactionRequest>,
    ) -> Result<String> {
        node_info!("eth_signTransaction");
        let request = self.parse_transaction_request(request)?;

        let from = request.from().map(Ok).unwrap_or_else(|| {
            self.accounts()?.first().copied().ok_or(BlockchainError::NoSignerAvailable)
        })?;

        let nonce = self.request_nonce_for_transaction(&request, from).await?;

        let request = self.build_tx_request(request, nonce).await?;

        let signed_transaction = self.sign_request(&from, request)?.encoded_2718();
        Ok(alloy_primitives::hex::encode_prefixed(signed_transaction))
    }

    /// Sends a transaction
    ///
    /// Handler for ETH RPC call: `eth_sendTransaction`
    pub async fn send_transaction(
        &self,
        request: WithOtherFields<TransactionRequest>,
    ) -> Result<TxHash> {
        node_info!("eth_sendTransaction");
        let request = self.parse_transaction_request(request)?;

        let from = request.from().map(Ok).unwrap_or_else(|| {
            self.accounts()?.first().copied().ok_or(BlockchainError::NoSignerAvailable)
        })?;
        // Keep nonce selection and pool insertion atomic with respect to other requests from
        // this sender. Explicit nonces retain the pool's normal replacement behavior.
        let _nonce_guard = if request.nonce().is_none() {
            let lock = self.nonce_locks.write().entry(from).or_default().clone();
            Some(lock.lock_owned().await)
        } else {
            None
        };
        let nonce = self.request_nonce_for_transaction(&request, from).await?;

        let typed_tx = self.build_tx_request(request, nonce).await?;

        // if the sender is currently impersonated we need to "bypass" signing
        let pending_transaction = if self.is_impersonated(from) {
            let transaction = typed_tx.into_impersonated();
            self.ensure_typed_transaction_supported(&transaction)?;
            trace!(target : "node", ?from, "eth_sendTransaction: impersonating");
            PendingTransaction::with_impersonated(transaction, from)
        } else {
            let transaction = self.sign_request(&from, typed_tx)?;
            self.ensure_typed_transaction_supported(&transaction)?;
            PendingTransaction::new(transaction)?
        };
        self.add_pending_transaction(pending_transaction).await
    }

    /// Resends a pending transaction with an updated gas price or gas limit.
    ///
    /// Handler for ETH RPC call: `eth_resend`
    pub async fn resend_transaction(
        &self,
        request: WithOtherFields<TransactionRequest>,
        gas_price: Option<U256>,
        gas_limit: Option<U64>,
    ) -> Result<TxHash> {
        node_info!("eth_resend");
        let mut request = self.parse_transaction_request(request)?;

        let from = request.from().map(Ok).unwrap_or_else(|| {
            self.accounts()?.first().copied().ok_or(BlockchainError::NoSignerAvailable)
        })?;
        let nonce = request.nonce().ok_or_else(|| {
            BlockchainError::InvalidTransactionRequest(
                "missing transaction nonce in transaction spec".to_string(),
            )
        })?;

        if !self.pool.contains_sender_nonce(from, nonce) {
            return Err(BlockchainError::TransactionNotFound);
        }

        if let Some(gas_price) = gas_price.filter(|gas_price| !gas_price.is_zero()) {
            request.set_gas_price(gas_price.saturating_to());
        }

        if let Some(gas_limit) = gas_limit.filter(|gas_limit| *gas_limit != U64::ZERO) {
            request.set_gas_limit(gas_limit.to());
        }

        let typed_tx = self.build_tx_request(request, nonce).await?;

        let pending_transaction = if self.is_impersonated(from) {
            let transaction = typed_tx.into_impersonated();
            self.ensure_typed_transaction_supported(&transaction)?;
            trace!(target : "node", ?from, "eth_resend: impersonating");
            PendingTransaction::with_impersonated(transaction, from)
        } else {
            let transaction = self.sign_request(&from, typed_tx)?;
            self.ensure_typed_transaction_supported(&transaction)?;
            PendingTransaction::new(transaction)?
        };

        self.add_pending_transaction(pending_transaction).await
    }

    /// Waits for a transaction to be included in a block and returns its receipt (no timeout).
    async fn await_transaction_inclusion(&self, hash: TxHash) -> Result<FoundryTxReceipt> {
        let mut stream = self.new_block_notifications();
        // Check if the transaction is already included before listening for new blocks.
        if let Some(receipt) = self.backend.transaction_receipt(hash).await? {
            return Ok(receipt);
        }
        while let Some(notification) = stream.next().await {
            if let Some(new_block) = notification.as_new_block()
                && let Some(block) = self.backend.get_block_by_hash(new_block.hash)
                && block.body.transactions.iter().any(|tx| tx.hash() == hash)
                && let Some(receipt) = self.backend.transaction_receipt(hash).await?
            {
                return Ok(receipt);
            }
        }

        Err(BlockchainError::Message("Failed to await transaction inclusion".to_string()))
    }

    fn transaction_confirmation_timeout(timeout_ms: Option<u64>) -> Duration {
        const TIMEOUT_DURATION: Duration = Duration::from_secs(30);
        timeout_ms
            .filter(|timeout_ms| *timeout_ms > 0)
            .map(Duration::from_millis)
            .map(|timeout| timeout.min(TIMEOUT_DURATION))
            .unwrap_or(TIMEOUT_DURATION)
    }

    /// Waits for a transaction to be included in a block and returns its receipt, with timeout.
    async fn check_transaction_inclusion(
        &self,
        hash: TxHash,
        timeout_ms: Option<u64>,
    ) -> Result<FoundryTxReceipt> {
        let timeout_duration = Self::transaction_confirmation_timeout(timeout_ms);
        tokio::time::timeout(timeout_duration, self.await_transaction_inclusion(hash))
            .await
            .unwrap_or_else(|_elapsed| {
                Err(BlockchainError::TransactionConfirmationTimeout {
                    hash,
                    duration: timeout_duration,
                })
            })
    }

    /// Sends a transaction and waits for receipt
    ///
    /// Handler for ETH RPC call: `eth_sendTransactionSync`
    pub async fn send_transaction_sync(
        &self,
        request: WithOtherFields<TransactionRequest>,
    ) -> Result<FoundryTxReceipt> {
        node_info!("eth_sendTransactionSync");
        let hash = self.send_transaction(request).await?;

        let receipt = self.check_transaction_inclusion(hash, None).await?;

        Ok(receipt)
    }

    /// Sends signed transaction, returning its hash.
    ///
    /// Handler for ETH RPC call: `eth_sendRawTransaction`
    pub async fn send_raw_transaction(&self, tx: Bytes) -> Result<TxHash> {
        node_info!("eth_sendRawTransaction");
        if tx.is_empty() {
            return Err(BlockchainError::EmptyRawTransactionData);
        }

        // Built-in Tempo fee payer: raw transactions submitted in the fee payer service encoding
        // or carrying the sponsorship placeholder ask the node to sponsor them before submission
        // (the sign-and-relay mode of the fee payer service contract).
        let service_encoded = if self.backend.is_tempo() {
            normalize_fee_payer_service_encoding(tx.as_ref())
        } else {
            None
        };
        let raw = service_encoded.map(Bytes::from).unwrap_or(tx);

        let transaction = if raw.first() == Some(&EIP4844_TX_TYPE_ID) {
            // Pooled EIP-4844 decoding uses large stack frames for inline blobs. Isolate it from
            // the already-large RPC dispatcher without increasing every worker's stack.
            let raw = raw.clone();
            tokio::task::spawn_blocking(move || FoundryTxEnvelope::decode_2718(&mut raw.as_ref()))
                .await
                .map_err(|_| {
                    BlockchainError::Internal("transaction decoding task panicked".into())
                })?
        } else {
            FoundryTxEnvelope::decode_2718(&mut raw.as_ref())
        }
        .map_err(|_| BlockchainError::FailedToDecodeSignedTransaction)?;

        self.ensure_typed_transaction_supported(&transaction)?;

        let transaction = match transaction {
            FoundryTxEnvelope::Tempo(aa_tx)
                if self.backend.is_tempo()
                    && aa_tx.tx().fee_payer_signature == Some(FEE_PAYER_SIGNATURE_MARKER) =>
            {
                FoundryTxEnvelope::Tempo(self.sponsor_sign_tempo_transaction(aa_tx).await?)
            }
            transaction => transaction,
        };

        if self.backend.is_tempo() && TempoHardfork::from(self.backend.hardfork()).is_t5() {
            let classification = classify_payment_lane(raw.as_ref());
            trace!(target: "node", tx = ?transaction.hash(), ?classification, "classified transaction lane");
        }

        #[cfg(feature = "base")]
        let is_eip8130 = transaction.is_eip8130();
        let pending_transaction = PendingTransaction::new(transaction).map_err(|error| {
            #[cfg(feature = "base")]
            if is_eip8130 {
                return BlockchainError::Eip8130TransactionRejected(error.to_string());
            }
            BlockchainError::RecoveryError(error)
        })?;

        self.add_pending_transaction(pending_transaction).await
    }

    /// Sends a signed transaction with an ignored transaction condition.
    ///
    /// Handler for ETH RPC call: `eth_sendRawTransactionConditional`
    pub async fn send_raw_transaction_conditional(
        &self,
        tx: Bytes,
        _condition: TransactionConditional,
    ) -> Result<TxHash> {
        node_info!("eth_sendRawTransactionConditional");
        self.send_raw_transaction(tx).await
    }

    /// Classifies a raw transaction with the active Anvil Tempo/T5 payment-lane classifier.
    pub fn anvil_classify_transaction(&self, tx: Bytes) -> Result<PaymentLaneClassification> {
        node_info!("anvil_classifyTransaction");
        let mut data = tx.as_ref();
        if data.is_empty() {
            return Err(BlockchainError::EmptyRawTransactionData);
        }

        FoundryTxEnvelope::decode_2718(&mut data)
            .map_err(|_| BlockchainError::FailedToDecodeSignedTransaction)?;

        Ok(self.classify_transaction_lane(tx.as_ref()))
    }

    fn classify_transaction_lane(&self, raw: &[u8]) -> PaymentLaneClassification {
        if !self.backend.is_tempo() {
            return PaymentLaneClassification::general(PaymentLaneReason::NotTempo);
        }

        if !TempoHardfork::from(self.backend.hardfork()).is_t5() {
            return PaymentLaneClassification::general(PaymentLaneReason::T5NotActive);
        }

        classify_payment_lane(raw)
    }

    /// Sends signed transaction, returning its receipt.
    ///
    /// Handler for ETH RPC call: `eth_sendRawTransactionSync`
    pub async fn send_raw_transaction_sync(
        &self,
        tx: Bytes,
        timeout_ms: Option<u64>,
    ) -> Result<FoundryTxReceipt> {
        node_info!("eth_sendRawTransactionSync");

        let hash = self.send_raw_transaction(tx).await?;
        let receipt = self.check_transaction_inclusion(hash, timeout_ms).await?;

        Ok(receipt)
    }

    /// Signs a raw Tempo transaction as the node's fee payer (sponsor) and returns the fully
    /// signed raw transaction without broadcasting it.
    ///
    /// This is the sign-only mode of the Tempo fee payer service contract, so
    /// `cast send --sponsor-url` and the tempo SDK relay transport can point directly at this
    /// node. Only available on Tempo networks.
    ///
    /// Handler for ETH RPC call: `eth_signRawTransaction`
    pub async fn sign_raw_transaction(&self, tx: Bytes) -> Result<Bytes> {
        node_info!("eth_signRawTransaction");
        if !self.backend.is_tempo() {
            return Err(BlockchainError::RpcUnimplemented);
        }

        if tx.is_empty() {
            return Err(BlockchainError::EmptyRawTransactionData);
        }

        // Fee payer service clients submit the request in the service encoding, which carries a
        // `0x00` placeholder in the fee payer signature field.
        let service_encoded = normalize_fee_payer_service_encoding(tx.as_ref());
        let mut data = service_encoded.as_deref().unwrap_or(tx.as_ref());

        let transaction = FoundryTxEnvelope::decode_2718(&mut data)
            .map_err(|_| BlockchainError::FailedToDecodeSignedTransaction)?;

        let FoundryTxEnvelope::Tempo(aa_tx) = transaction else {
            return Err(RpcError::invalid_params(
                "only Tempo (0x76) transactions can be fee-payer signed",
            )
            .into());
        };

        let signed = self.sponsor_sign_tempo_transaction(aa_tx).await?;
        Ok(FoundryTxEnvelope::Tempo(signed).encoded_2718().into())
    }

    /// Fills the sponsor-owned fields of a sender-signed Tempo AA transaction.
    ///
    /// Mirrors the hosted Tempo fee payer service: the node's fee payer account selects the fee
    /// token when the sender left it open and signs the fee payer digest. Only the fee token and
    /// fee payer signature are touched, so the sender signature stays valid.
    async fn sponsor_sign_tempo_transaction(&self, tx: AASigned) -> Result<AASigned> {
        // The sender must have signed in the sponsored state (fee payer placeholder set): the
        // sender signature commits to whether a fee payer is present, so sponsoring a
        // transaction signed without the placeholder would invalidate it.
        match tx.tx().fee_payer_signature {
            Some(FEE_PAYER_SIGNATURE_MARKER) => {}
            Some(_) => {
                return Err(
                    RpcError::invalid_params("transaction is already fee-payer signed").into()
                );
            }
            None => {
                return Err(RpcError::invalid_params(
                    "transaction does not request sponsorship; sign it with the fee payer \
                     signature placeholder",
                )
                .into());
            }
        }

        let sender = tx.recover_signer().map_err(|_| {
            BlockchainError::RpcError(RpcError::invalid_params(
                "transaction must be signed by the sender before fee-payer signing",
            ))
        })?;

        let Some(sponsor) = self.backend.tempo_fee_payer().await else {
            return Err(RpcError::invalid_params("no Tempo fee payer account available").into());
        };
        if sponsor == sender {
            return Err(RpcError::invalid_params(format!(
                "Tempo fee payer {sponsor} must not equal the transaction sender"
            ))
            .into());
        }
        let signer = self.get_signer(sponsor).ok_or(BlockchainError::NoSignerAvailable)?;

        let (mut tx, sender_signature, _) = tx.into_parts();
        // The fee payer digest commits to the fee token, so it must be resolved first. The
        // service encoding omits the sender's fee token preference, letting the sponsor pick the
        // token it pays with.
        if tx.fee_token.is_none() {
            tx.fee_token = Some(self.backend.tempo_user_fee_token(sponsor).await?);
        }
        let digest = tx.fee_payer_signature_hash(sender);
        tx.fee_payer_signature = Some(signer.sign_hash(sponsor, digest).await?);

        Ok(tx.into_signed(sender_signature))
    }

    /// Call contract, returning the output data.
    ///
    /// Handler for ETH RPC call: `eth_call`
    pub async fn call(
        &self,
        request: WithOtherFields<TransactionRequest>,
        block_number: Option<BlockId>,
        overrides: EvmOverrides,
    ) -> Result<Bytes> {
        node_info!("eth_call");
        let block_request = self.block_request(block_number).await?;
        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            if overrides.has_state() || overrides.has_block() {
                return Err(BlockchainError::EvmOverrideError(
                    "not available on past forked blocks".to_string(),
                ));
            }
            return Ok(fork.call_raw(&request, Some(number.into())).await?);
        }

        #[cfg(feature = "base")]
        if self.backend.is_base()
            && serde_json::to_value(&request)
                .and_then(serde_json::from_value::<BaseTransactionRequest>)
                .is_ok_and(|request| request.as_eip8130().is_some())
        {
            let timestamp = self.backend.block_request_timestamp(&block_request).await?;
            self.backend.ensure_base_eip8130_active_at(timestamp)?;
        }

        let fees = FeeDetails::new(
            request.gas_price,
            request.max_fee_per_gas,
            request.max_priority_fee_per_gas,
            request.max_fee_per_blob_gas,
        )?
        .or_zero_fees();
        // this can be blocking for a bit, especially in forking mode
        // <https://github.com/foundry-rs/foundry/issues/6036>
        self.on_blocking_task(|this| async move {
            let (exit, out, gas, _) =
                this.backend.call(request, fees, Some(block_request), overrides).await?;
            trace!(target : "node", "Call status {:?}, gas {}", exit, gas);

            ensure_return_ok(exit, &out)
        })
        .await
    }

    pub async fn call_many(
        &self,
        bundles: Vec<Bundle<WithOtherFields<TransactionRequest>>>,
        state_context: Option<StateContext>,
        state_override: Option<StateOverride>,
    ) -> Result<Vec<Vec<EthCallResponse>>> {
        node_info!("eth_callMany");
        let StateContext { transaction_index, block_number } = state_context.unwrap_or_default();
        if transaction_index.is_some_and(|index| index.index().is_some()) {
            return Err(BlockchainError::RpcError(RpcError::invalid_params(
                "transactionIndex is not supported for eth_callMany yet".to_string(),
            )));
        }

        let block_request = self.block_request(block_number).await?;
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(fork
                .call_many(
                    bundles,
                    Some(StateContext { transaction_index, block_number: Some(number.into()) }),
                    state_override,
                )
                .await?);
        }

        self.on_blocking_task(|this| async move {
            this.backend.call_many(bundles, Some(block_request), state_override).await
        })
        .await
    }

    /// Simulates a bundle of signed transactions against the requested state.
    ///
    /// Handler for ETH RPC call: `eth_callBundle`.
    pub async fn call_bundle(&self, bundle: EthCallBundle) -> Result<EthCallBundleResponse> {
        node_info!("eth_callBundle");
        if bundle.txs.is_empty() {
            return Err(BlockchainError::RpcError(RpcError::invalid_params(
                "bundle missing txs".to_string(),
            )));
        }
        if bundle.block_number == 0 {
            return Err(BlockchainError::RpcError(RpcError::invalid_params(
                "bundle missing blockNumber".to_string(),
            )));
        }

        let block_request = self.block_request(Some(bundle.state_block_number.into())).await?;
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(fork.call_bundle(bundle).await?);
        }

        let transactions = bundle
            .txs
            .iter()
            .map(|raw| {
                let mut data = raw.as_ref();
                if data.is_empty() {
                    return Err(BlockchainError::EmptyRawTransactionData);
                }
                let transaction = FoundryTxEnvelope::decode_2718(&mut data)
                    .map_err(|_| BlockchainError::FailedToDecodeSignedTransaction)?;
                self.ensure_typed_transaction_supported(&transaction)?;
                PendingTransaction::new(transaction).map_err(Into::into)
            })
            .collect::<Result<Vec<_>>>()?;

        self.on_blocking_task(|this| async move {
            this.backend.call_bundle(bundle, transactions, Some(block_request)).await
        })
        .await
    }

    pub async fn simulate_v1(
        &self,
        request: SimulatePayload,
        block_number: Option<BlockId>,
    ) -> Result<Vec<SimulatedBlock<AnyRpcBlock>>> {
        self.simulate_v1_raw(preserve_simulation_request_fields(request), block_number).await
    }

    pub(crate) async fn simulate_v1_raw(
        &self,
        mut request: SimulatePayload<WithOtherFields<TransactionRequest>>,
        block_number: Option<BlockId>,
    ) -> Result<Vec<SimulatedBlock<AnyRpcBlock>>> {
        const DEFAULT_BLOCK_INTERVAL_SECS: u64 = 12;

        node_info!("eth_simulateV1");
        if request.block_state_calls.is_empty() {
            return Err(BlockchainError::RpcError(RpcError::invalid_params("empty input")));
        }
        if request.block_state_calls.len() > MAX_SIMULATE_BLOCKS as usize {
            return Err(BlockchainError::RpcError(RpcError {
                code: ErrorCode::ServerError(-38026),
                message: "too many blocks".into(),
                data: None,
            }));
        }
        let block_id = block_number;
        let block_request =
            self.block_request(block_number).await.map_err(|error| match error {
                BlockchainError::BlockOutOfRange(_, _) | BlockchainError::BlockNotFound => {
                    BlockchainError::RpcError(RpcError {
                        code: ErrorCode::ServerError(-32000),
                        message: "header not found".into(),
                        data: None,
                    })
                }
                error => error,
            })?;
        let block_interval = self.backend.time().block_timestamp_interval().unwrap_or_else(|| {
            self.miner
                .block_interval()
                .map(|duration| {
                    duration
                        .as_secs()
                        .saturating_add(u64::from(duration.subsec_nanos() != 0))
                        .max(1)
                })
                .unwrap_or(DEFAULT_BLOCK_INTERVAL_SECS)
        });

        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            let block_id = match block_id {
                Some(BlockId::Hash(hash)) => BlockId::Hash(hash),
                _ => number.into(),
            };
            let base_block = fork.fetch_block(block_id).await?.ok_or_else(|| {
                BlockchainError::RpcError(RpcError {
                    code: ErrorCode::ServerError(-32000),
                    message: "header not found".into(),
                    data: None,
                })
            })?;
            request.block_state_calls = sanitize_simulation_blocks(
                request.block_state_calls,
                base_block.header.number(),
                base_block.header.timestamp(),
                block_interval,
            )?;
            return Ok(fork.simulate_v1(&request, Some(base_block.header.hash.into())).await?);
        }

        // this can be blocking for a bit, especially in forking mode
        // <https://github.com/foundry-rs/foundry/issues/6036>
        self.on_blocking_task(|this| async move {
            let simulated_blocks =
                this.backend.simulate_raw(request, Some(block_request), block_interval).await?;
            trace!(target : "node", "Simulate status {:?}", simulated_blocks);

            Ok(simulated_blocks)
        })
        .await
    }

    /// This method creates an EIP2930 type accessList based on a given Transaction. The accessList
    /// contains all storage slots and addresses read and written by the transaction, except for the
    /// sender account and the precompiles.
    ///
    /// It returns list of addresses and storage keys used by the transaction, plus the gas
    /// consumed when the access list is added. That is, it gives you the list of addresses and
    /// storage keys that will be used by that transaction, plus the gas consumed if the access
    /// list is included. Like eth_estimateGas, this is an estimation; the list could change
    /// when the transaction is actually mined. Adding an accessList to your transaction does
    /// not necessary result in lower gas usage compared to a transaction without an access
    /// list.
    ///
    /// Handler for ETH RPC call: `eth_createAccessList`
    pub async fn create_access_list(
        &self,
        request: WithOtherFields<TransactionRequest>,
        block_number: Option<BlockId>,
        state_override: Option<StateOverride>,
    ) -> Result<AccessListResult> {
        node_info!("eth_createAccessList");
        let block_request = self.block_request(block_number).await?;
        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            if state_override.is_some() {
                return Err(BlockchainError::EvmOverrideError(
                    "not available on past forked blocks".to_string(),
                ));
            }
            return Ok(fork.create_access_list_raw(&request, Some(number.into())).await?);
        }
        let typed_request = self.parse_transaction_request(request.clone())?;

        #[cfg(feature = "base")]
        if self.backend.is_base()
            && serde_json::to_value(&request)
                .and_then(serde_json::from_value::<BaseTransactionRequest>)
                .is_ok_and(|request| request.as_eip8130().is_some())
        {
            let timestamp = self.backend.block_request_timestamp(&block_request).await?;
            self.backend.ensure_base_eip8130_active_at(timestamp)?;
            return Err(BlockchainError::InvalidTransactionRequest(
                "eth_createAccessList does not support EIP-8130 transaction requests".to_string(),
            ));
        }

        self.backend
            .with_database_at_and_context(Some(block_request), |state, block_env, monad_context| {
                let mut cache_db = CacheDB::new(state);
                if let Some(state_override) = state_override {
                    apply_state_overrides(state_override.into_iter().collect(), &mut cache_db)?;
                }

                let (_, _, _, access_list) =
                    self.backend.build_access_list_with_state_and_context(
                        &cache_db,
                        request.clone(),
                        FeeDetails::zero(),
                        block_env.clone(),
                        monad_context.clone(),
                    )?;

                // Re-execute with the access list applied to get the post-AL gas usage.
                // EVM failures (including reverts) are surfaced in the result's `error`
                // field per the execution-apis `eth_createAccessList` spec, so callers
                // can still inspect the traced slots when execution fails.
                let (exit, _, gas_used, _) = self.backend.call_with_state_typed_access_list(
                    &cache_db,
                    typed_request,
                    FeeDetails::zero(),
                    block_env,
                    access_list.clone(),
                    monad_context,
                )?;

                Ok(AccessListResult {
                    access_list,
                    gas_used: U256::from(gas_used),
                    error: execution_error(exit),
                })
            })
            .await
    }

    /// Estimate gas needed for execution of given contract.
    /// If no block parameter is given, it will use the pending block by default
    ///
    /// Handler for ETH RPC call: `eth_estimateGas`
    pub async fn estimate_gas(
        &self,
        request: WithOtherFields<TransactionRequest>,
        block_number: Option<BlockId>,
        overrides: EvmOverrides,
    ) -> Result<U256> {
        node_info!("eth_estimateGas");
        self.do_estimate_gas(request, block_number.or_else(|| Some(BlockId::pending())), overrides)
            .await
            .map(U256::from)
    }

    /// Fills a transaction request with default values for missing fields.
    ///
    /// This method populates missing transaction fields like nonce, gas limit,
    /// chain ID, and fee parameters with appropriate defaults.
    ///
    /// Handler for ETH RPC call: `eth_fillTransaction`
    pub async fn fill_transaction(
        &self,
        request: WithOtherFields<TransactionRequest>,
    ) -> Result<FillTransaction<AnyRpcTransaction>> {
        node_info!("eth_fillTransaction");
        let mut request = self.parse_transaction_request(request)?;

        let from = match request.from() {
            Some(from) => from,
            None => self.accounts()?.first().copied().ok_or(BlockchainError::NoSignerAvailable)?,
        };

        let nonce = self.request_nonce_for_transaction(&request, from).await?;

        // Prefill gas limit with estimated gas and bubble up estimation errors directly.
        if request.gas_limit().is_none() {
            let estimated_gas =
                self.do_estimate_gas_typed(request.clone(), Some(BlockId::pending())).await?;
            request.set_gas_limit(estimated_gas as u64);
        }

        let typed_tx = self.build_tx_request(request, nonce).await?;
        let tx = typed_tx.into_impersonated();

        let raw = tx.encoded_2718().into();

        let mut tx =
            transaction_build(None, MaybeImpersonatedTransaction::new(tx), None, None, None);

        // Set the correct `from` address (overrides the recovered zero address from dummy
        // signature)
        tx.0.inner.inner = Recovered::new_unchecked(tx.0.inner.inner.into_inner(), from);

        Ok(FillTransaction { raw, tx })
    }

    /// Handler for RPC call: `anvil_getBlobsByTransactionHash`
    pub fn anvil_get_blob_by_tx_hash(&self, hash: B256) -> Result<Option<Vec<Blob>>> {
        node_info!("anvil_getBlobsByTransactionHash");
        Ok(self.backend.get_blob_by_tx_hash(hash)?)
    }

    /// Get transaction by its hash.
    ///
    /// This will check the storage for a matching transaction, if no transaction exists in storage
    /// this will also scan the mempool for a matching pending transaction
    ///
    /// Handler for ETH RPC call: `eth_getTransactionByHash`
    pub async fn transaction_by_hash(&self, hash: B256) -> Result<Option<AnyRpcTransaction>> {
        node_info!("eth_getTransactionByHash");
        let mut tx =
            self.pool.get_transaction(hash).map(|pending| self.build_pool_transaction(pending));
        if tx.is_none() {
            tx = self.backend.transaction_by_hash(hash).await?
        }

        Ok(tx)
    }

    fn build_pool_transaction(
        &self,
        pending: PendingTransaction<FoundryTxEnvelope>,
    ) -> AnyRpcTransaction {
        let from = *pending.sender();
        let tx = transaction_build(
            Some(*pending.hash()),
            pending.transaction,
            None,
            None,
            Some(self.backend.base_fee()),
        );

        let WithOtherFields { inner: mut tx, other } = tx.0;
        // we set the from field here explicitly to the set sender of the pending transaction,
        // in case the transaction is impersonated.
        tx.inner = Recovered::new_unchecked(tx.inner.into_inner(), from);

        AnyRpcTransaction(WithOtherFields { inner: tx, other })
    }

    /// Returns all ready transactions from the local pending pool.
    ///
    /// Handler for ETH RPC call: `eth_pendingTransactions`
    pub async fn pending_transactions(&self) -> Result<Vec<AnyRpcTransaction>> {
        node_info!("eth_pendingTransactions");
        Ok(self
            .pool
            .ready_transactions()
            .map(|pending| self.build_pool_transaction(pending.pending_transaction.clone()))
            .collect())
    }

    /// Returns the transaction by sender and nonce.
    ///
    /// This will check the mempool for pending transactions first, then perform a binary search
    /// over mined blocks to find the transaction.
    ///
    /// Handler for ETH RPC call: `eth_getTransactionBySenderAndNonce`
    pub async fn transaction_by_sender_and_nonce(
        &self,
        sender: Address,
        nonce: U256,
    ) -> Result<Option<AnyRpcTransaction>> {
        node_info!("eth_getTransactionBySenderAndNonce");

        // check pending txs first
        for pending_tx in self.pool.ready_transactions().chain(self.pool.pending_transactions()) {
            if U256::from(pending_tx.pending_transaction.nonce()) == nonce
                && *pending_tx.pending_transaction.sender() == sender
            {
                let tx = transaction_build(
                    Some(*pending_tx.pending_transaction.hash()),
                    pending_tx.pending_transaction.transaction.clone(),
                    None,
                    None,
                    Some(self.backend.base_fee()),
                );

                let WithOtherFields { inner: mut tx, other } = tx.0;
                // we set the from field here explicitly to the set sender of the pending
                // transaction, in case the transaction is impersonated.
                let from = *pending_tx.pending_transaction.sender();
                tx.inner = Recovered::new_unchecked(tx.inner.into_inner(), from);

                return Ok(Some(AnyRpcTransaction(WithOtherFields { inner: tx, other })));
            }
        }

        let highest_nonce = self.transaction_count(sender, None).await?.saturating_to::<u64>();
        let target_nonce = nonce.saturating_to::<u64>();

        // if the nonce is higher or equal to the highest nonce, the transaction doesn't exist
        if target_nonce >= highest_nonce {
            return Ok(None);
        }

        // no mined blocks yet
        let latest_block = self.backend.best_number();
        if latest_block == 0 {
            return Ok(None);
        }

        // binary search for the block containing the transaction
        let mut low = 1u64;
        let mut high = latest_block;

        while low <= high {
            let mid = low + (high - low) / 2;
            let mid_nonce =
                self.transaction_count(sender, Some(mid.into())).await?.saturating_to::<u64>();

            if mid_nonce > target_nonce {
                high = mid - 1;
            } else {
                low = mid + 1;
            }
        }

        // search in the target block
        let target_block = low;
        if target_block <= latest_block
            && let Some(txs) =
                self.backend.mined_transactions_by_block_number(target_block.into()).await
        {
            for tx in txs {
                if tx.from() == sender && tx.nonce() == target_nonce {
                    return Ok(Some(tx));
                }
            }
        }

        Ok(None)
    }

    /// Returns transaction receipt by transaction hash.
    ///
    /// Handler for ETH RPC call: `eth_getTransactionReceipt`
    pub async fn transaction_receipt(&self, hash: B256) -> Result<Option<FoundryTxReceipt>> {
        node_info!("eth_getTransactionReceipt");
        self.backend.transaction_receipt(hash).await
    }

    /// Returns block receipts by block number.
    ///
    /// Handler for ETH RPC call: `eth_getBlockReceipts`
    pub async fn block_receipts(&self, number: BlockId) -> Result<Option<Vec<FoundryTxReceipt>>> {
        node_info!("eth_getBlockReceipts");
        if number == BlockId::pending() {
            let transactions = self.pool.ready_transactions().collect::<Vec<_>>();
            return Ok(Some(self.backend.pending_block_receipts(transactions).await?));
        }

        self.backend.block_receipts(number).await
    }

    /// Returns logs matching given filter object.
    ///
    /// Handler for ETH RPC call: `eth_getLogs`
    pub async fn logs(&self, filter: Filter) -> Result<Vec<Log>> {
        node_info!("eth_getLogs");
        let best = self.backend.best_number();
        let to_block =
            self.backend.convert_block_number(filter.block_option.get_to_block().copied());
        if to_block > best {
            return Err(BlockchainError::BlockOutOfRange(best, to_block));
        }
        let from_block =
            self.backend.convert_block_number(filter.block_option.get_from_block().copied());
        if from_block > to_block {
            return Err(RpcError::invalid_params("invalid block range params").into());
        }

        self.backend.logs(filter).await
    }

    /// Creates a filter object, based on filter options, to notify when the state changes (logs).
    ///
    /// Handler for ETH RPC call: `eth_newFilter`
    pub async fn new_filter(&self, filter: Filter) -> Result<String> {
        node_info!("eth_newFilter");
        // Valid future ranges have no historic logs to seed yet.
        let historic = match filter.block_option.get_from_block() {
            Some(BlockNumber::Number(from))
                if *from > self.backend.best_number()
                    && filter.block_option.get_to_block().is_none_or(|to| {
                        to.is_latest() || to.as_number().is_some_and(|to| to >= *from)
                    }) =>
            {
                vec![]
            }
            Some(_) => self.backend.logs(filter.clone()).await?,
            None => vec![],
        };
        let filter = EthFilter::Logs(Box::new(LogsFilter {
            blocks: self.new_block_notifications(),
            storage: self.storage_info(),
            filter: FilteredParams::new(Some(filter)),
            historic: Some(historic),
        }));
        Ok(self.filters.add_filter(filter).await)
    }

    /// Creates a filter in the node, to notify when a new block arrives.
    ///
    /// Handler for ETH RPC call: `eth_newBlockFilter`
    pub async fn new_block_filter(&self) -> Result<String> {
        node_info!("eth_newBlockFilter");
        let filter = EthFilter::Blocks(self.new_block_notifications());
        Ok(self.filters.add_filter(filter).await)
    }

    /// Creates a filter in the node, to notify when new pending transactions arrive.
    ///
    /// Handler for ETH RPC call: `eth_newPendingTransactionFilter`
    pub async fn new_pending_transaction_filter(&self, full: bool) -> Result<String> {
        node_info!("eth_newPendingTransactionFilter");
        let filter = if full {
            EthFilter::FullPendingTransactions(self.full_pending_transactions_filter())
        } else {
            EthFilter::PendingTransactions(self.new_ready_transactions())
        };
        Ok(self.filters.add_filter(filter).await)
    }

    /// Polling method for a filter, which returns an array of logs which occurred since last poll.
    ///
    /// Handler for ETH RPC call: `eth_getFilterChanges`
    pub async fn get_filter_changes(&self, id: &str) -> ResponseResult {
        node_info!("eth_getFilterChanges");
        self.filters.get_filter_changes(id).await
    }

    /// Returns an array of all logs matching filter with given id.
    ///
    /// Handler for ETH RPC call: `eth_getFilterLogs`
    pub async fn get_filter_logs(&self, id: &str) -> Result<Vec<Log>> {
        node_info!("eth_getFilterLogs");
        if let Some(filter) = self.filters.get_log_filter(id).await {
            self.backend.logs(filter).await
        } else {
            Err(BlockchainError::FilterNotFound)
        }
    }

    /// Handler for ETH RPC call: `eth_uninstallFilter`
    pub async fn uninstall_filter(&self, id: &str) -> Result<bool> {
        node_info!("eth_uninstallFilter");
        Ok(self.filters.uninstall_filter(id).await.is_some())
    }

    /// Returns EIP-2718 encoded raw transaction
    ///
    /// Handler for RPC call: `debug_getRawTransaction`
    pub async fn raw_transaction(&self, hash: B256) -> Result<Option<Bytes>> {
        node_info!("debug_getRawTransaction");
        self.inner_raw_transaction(hash).await
    }

    /// Returns EIP-2718 encoded raw receipts for the block.
    ///
    /// Handler for RPC call: `debug_getRawReceipts`.
    pub async fn raw_receipts(&self, block: BlockId) -> Result<Vec<Bytes>> {
        node_info!("debug_getRawReceipts");

        // In fork mode, serve pre-fork blocks from the upstream provider.
        if let BlockRequest::Number(number) = self.block_request(Some(block)).await?
            && let Some(fork) = self.get_fork()
            && fork.predates_fork_inclusive(number)
        {
            let receipts = fork.block_receipts(number).await?.unwrap_or_default();
            return Ok(receipts
                .into_iter()
                .map(|receipt| {
                    receipt.0.inner.inner.map_logs(|log| log.inner).encoded_2718().into()
                })
                .collect());
        }

        let block = self.backend.get_block(block).ok_or(BlockchainError::BlockNotFound)?;
        let receipts = self
            .backend
            .mined_receipts(block.header.hash_slow())
            .ok_or(BlockchainError::BlockNotFound)?;
        Ok(receipts.into_iter().map(|receipt| receipt.encoded_2718().into()).collect())
    }

    /// Returns EIP-2718 encoded raw transactions for the block.
    ///
    /// Handler for RPC call: `debug_getRawTransactions`.
    pub async fn raw_transactions(&self, block: BlockId) -> Result<Vec<Bytes>> {
        node_info!("debug_getRawTransactions");

        if let Some(block) = self.backend.get_block(block) {
            return Ok(block
                .body
                .transactions
                .into_iter()
                .map(|tx| tx.into_inner().into_canonical().encoded_2718().into())
                .collect());
        }

        // In fork mode, serve pre-fork blocks from the upstream provider. Genuinely unknown or
        // out-of-range blocks yield an empty result, mirroring reth; transport errors propagate.
        let Some(fork) = self.get_fork() else {
            return Ok(Vec::new());
        };
        let block = match block {
            BlockId::Number(BlockNumber::Pending) => None,
            BlockId::Number(number) => {
                let number = self.backend.convert_block_number(Some(number));
                if !fork.predates_fork_inclusive(number) {
                    return Ok(Vec::new());
                }
                fork.block_by_number_full(number).await?
            }
            BlockId::Hash(hash) => fork
                .block_by_hash_full(hash.block_hash)
                .await?
                .filter(|block| fork.predates_fork_inclusive(block.header().number())),
        };
        let Some(block) = block else {
            return Ok(Vec::new());
        };
        let BlockTransactions::Full(txs) = block.transactions() else {
            return Err(BlockchainError::Internal(
                "fork provider returned a non-full block for a full block request".to_string(),
            ));
        };
        txs.iter().map(encode_rpc_transaction).collect()
    }

    /// Returns RLP encoded raw block header.
    ///
    /// Handler for RPC call: `debug_getRawHeader`.
    pub async fn raw_header(&self, block: BlockId) -> Result<Bytes> {
        node_info!("debug_getRawHeader");
        let block = self.backend.get_block(block).ok_or(BlockchainError::BlockNotFound)?;
        Ok(alloy_rlp::encode(&block.header).into())
    }

    /// Returns RLP encoded raw block.
    ///
    /// Handler for RPC call: `debug_getRawBlock`.
    pub async fn raw_block(&self, block: BlockId) -> Result<Bytes> {
        node_info!("debug_getRawBlock");
        let block = self.backend.get_block(block).ok_or(BlockchainError::BlockNotFound)?;
        Ok(alloy_rlp::encode(canonical_block(block)).into())
    }

    /// Returns EIP-2718 encoded raw transaction by block hash and index
    ///
    /// Handler for RPC call: `eth_getRawTransactionByBlockHashAndIndex`
    pub async fn raw_transaction_by_block_hash_and_index(
        &self,
        block_hash: B256,
        index: Index,
    ) -> Result<Option<Bytes>> {
        node_info!("eth_getRawTransactionByBlockHashAndIndex");
        match self.backend.transaction_by_block_hash_and_index(block_hash, index).await? {
            Some(tx) => self.inner_raw_transaction(tx.tx_hash()).await,
            None => Ok(None),
        }
    }

    /// Returns EIP-2718 encoded raw transaction by block number and index
    ///
    /// Handler for RPC call: `eth_getRawTransactionByBlockNumberAndIndex`
    pub async fn raw_transaction_by_block_number_and_index(
        &self,
        block_number: BlockNumber,
        index: Index,
    ) -> Result<Option<Bytes>> {
        node_info!("eth_getRawTransactionByBlockNumberAndIndex");
        match self.transaction_by_block_number_and_index(block_number, index).await? {
            Some(tx) => self.inner_raw_transaction(tx.tx_hash()).await,
            None => Ok(None),
        }
    }

    /// Returns traces for the transaction hash for geth's tracing endpoint
    ///
    /// Handler for RPC call: `debug_traceTransaction`
    pub async fn debug_trace_transaction(
        &self,
        tx_hash: B256,
        opts: GethDebugTracingOptions,
    ) -> Result<GethTrace> {
        node_info!("debug_traceTransaction");
        self.backend.debug_trace_transaction(tx_hash, opts).await
    }

    /// Returns traces for all transactions in an RLP-encoded block.
    ///
    /// Handler for RPC call: `debug_traceBlock`
    pub async fn debug_trace_block(
        &self,
        rlp_block: Bytes,
        opts: GethDebugTracingOptions,
    ) -> Result<Vec<TraceResult>> {
        node_info!("debug_traceBlock");
        self.backend.debug_trace_block(rlp_block, opts).await
    }

    /// Returns traces for all transactions in a block by hash.
    ///
    /// Handler for RPC call: `debug_traceBlockByHash`
    pub async fn debug_trace_block_by_hash(
        &self,
        block_hash: B256,
        opts: GethDebugTracingOptions,
    ) -> Result<Vec<TraceResult>> {
        node_info!("debug_traceBlockByHash");
        self.backend.debug_trace_block_by_hash(block_hash, opts).await
    }

    /// Returns traces for all transactions in a block by number.
    ///
    /// Handler for RPC call: `debug_traceBlockByNumber`
    pub async fn debug_trace_block_by_number(
        &self,
        block_number: BlockNumber,
        opts: GethDebugTracingOptions,
    ) -> Result<Vec<TraceResult>> {
        node_info!("debug_traceBlockByNumber");
        self.backend.debug_trace_block_by_number(block_number, opts).await
    }

    /// Returns traces for the transaction for geth's tracing endpoint
    ///
    /// Handler for RPC call: `debug_traceCall`
    pub async fn debug_trace_call(
        &self,
        request: WithOtherFields<TransactionRequest>,
        block_number: Option<BlockId>,
        opts: GethDebugTracingCallOptions,
    ) -> Result<GethTrace> {
        node_info!("debug_traceCall");
        let block_request = self.block_request(block_number).await?;
        let fees = FeeDetails::new(
            request.gas_price,
            request.max_fee_per_gas,
            request.max_priority_fee_per_gas,
            request.max_fee_per_blob_gas,
        )?
        .or_zero_fees();

        let result: std::result::Result<GethTrace, BlockchainError> =
            self.backend.call_with_tracing(request, fees, Some(block_request), opts).await;
        result
    }

    /// Traces calls sequentially on top of the same block.
    ///
    /// Handler for RPC call: `trace_callMany`.
    pub async fn trace_call_many(
        &self,
        calls: Vec<(WithOtherFields<TransactionRequest>, HashSet<TraceType>)>,
        block_number: Option<BlockId>,
    ) -> Result<Vec<TraceResults>> {
        node_info!("trace_callMany");
        let block_request = self.block_request(block_number).await?;

        self.backend.trace_call_many(calls, Some(block_request)).await
    }
}

// == impl EthApi anvil endpoints ==

impl EthApi<FoundryNetwork> {
    /// Mines a series of blocks.
    ///
    /// Handler for ETH RPC call: `anvil_mine`
    pub async fn anvil_mine(&self, num_blocks: Option<U256>, interval: Option<U256>) -> Result<()> {
        node_info!("anvil_mine");
        let interval = interval.map(|i| i.saturating_to::<u64>());
        let blocks = num_blocks.unwrap_or(U256::ONE);
        if blocks.is_zero() {
            return Ok(());
        }

        self.on_blocking_task(|this| async move {
            // mine all the blocks
            for _ in 0..blocks.saturating_to::<u64>() {
                // If we have an interval, jump forwards in time to the "next" timestamp
                let pending_increase =
                    interval.map(|interval| this.backend.time().apply_time_increase(interval));
                if let Err(error) = this.mine_one().await {
                    if let Some(pending) = pending_increase {
                        this.backend.time().revert_time_increase(pending);
                    }
                    return Err(error);
                }
            }
            Ok(())
        })
        .await?;

        Ok(())
    }

    /// Helper function to find the storage slot for an ERC20 function call by testing slots
    /// from an access list until one produces the expected result.
    ///
    /// Rather than trying to reverse-engineer the storage layout, this function uses a
    /// "trial and error" approach: try overriding each slot that the function accesses,
    /// and see which one actually affects the function's return value.
    ///
    /// ## Parameters
    /// - `token_address`: The ERC20 token contract address
    /// - `calldata`: The encoded function call (e.g., `balanceOf(user)` or `allowance(owner,
    ///   spender)`)
    /// - `expected_value`: The value we want to set (balance or allowance amount)
    ///
    /// ## Returns
    /// The storage slot (B256) that contains the target ERC20 data, or an error if no slot is
    /// found.
    async fn find_erc20_storage_slot(
        &self,
        token_address: Address,
        calldata: Bytes,
        expected_value: U256,
    ) -> Result<B256> {
        let tx = TransactionRequest::default().with_to(token_address).with_input(calldata.clone());

        // first collect all the slots that are used by the function call
        let access_list_result =
            self.create_access_list(WithOtherFields::new(tx.clone()), None, None).await?;
        let access_list = access_list_result.access_list;

        // iterate over all the accessed slots and try to find the one that contains the
        // target value by overriding the slot and checking the function call result
        for item in access_list.0 {
            if item.address != token_address {
                continue;
            };
            for slot in &item.storage_keys {
                let account_override = AccountOverride::default()
                    .with_state_diff(std::iter::once((*slot, B256::from(expected_value))));

                let state_override = StateOverridesBuilder::default()
                    .append(token_address, account_override)
                    .build();

                let evm_override = EvmOverrides::state(Some(state_override));

                let Ok(result) =
                    self.call(WithOtherFields::new(tx.clone()), None, evm_override).await
                else {
                    // overriding this slot failed
                    continue;
                };

                let Ok(result_value) = U256::abi_decode(&result) else {
                    // response returned something other than a U256
                    continue;
                };

                if result_value == expected_value {
                    return Ok(*slot);
                }
            }
        }

        Err(BlockchainError::Message("Unable to find storage slot".to_string()))
    }

    /// Deals ERC20 tokens to a address
    ///
    /// Handler for RPC call: `anvil_dealERC20`
    pub async fn anvil_deal_erc20(
        &self,
        address: Address,
        token_address: Address,
        balance: U256,
    ) -> Result<()> {
        node_info!("anvil_dealERC20");

        if self.backend.is_tempo()
            && self.backend.try_set_tip20_balance(address, token_address, balance).await?
        {
            return Ok(());
        }

        sol! {
            #[sol(rpc)]
            contract IERC20 {
                function balanceOf(address target) external view returns (uint256);
            }
        }

        let calldata = IERC20::balanceOfCall { target: address }.abi_encode().into();

        // Find the storage slot that contains the balance
        let slot =
            self.find_erc20_storage_slot(token_address, calldata, balance).await.map_err(|_| {
                BlockchainError::Message("Unable to set ERC20 balance, no slot found".to_string())
            })?;

        // Set the storage slot to the desired balance
        self.anvil_set_storage_at(token_address, slot.into(), balance.into()).await?;

        Ok(())
    }

    /// Deals TIP-20 tokens to an address.
    ///
    /// Handler for RPC call: `anvil_dealTIP20`.
    pub async fn anvil_deal_tip20(
        &self,
        address: Address,
        token_address: Address,
        balance: U256,
    ) -> Result<()> {
        node_info!("anvil_dealTIP20");
        self.ensure_tempo_mode()?;
        self.backend.set_tip20_balance(address, token_address, balance).await?;
        Ok(())
    }

    /// Sets the ERC20 allowance for a spender
    ///
    /// Handler for RPC call: `anvil_set_erc20_allowance`
    pub async fn anvil_set_erc20_allowance(
        &self,
        owner: Address,
        spender: Address,
        token_address: Address,
        amount: U256,
    ) -> Result<()> {
        node_info!("anvil_setERC20Allowance");

        sol! {
            #[sol(rpc)]
            contract IERC20 {
                function allowance(address owner, address spender) external view returns (uint256);
            }
        }

        let calldata = IERC20::allowanceCall { owner, spender }.abi_encode().into();

        // Find the storage slot that contains the allowance
        let slot =
            self.find_erc20_storage_slot(token_address, calldata, amount).await.map_err(|_| {
                BlockchainError::Message("Unable to set ERC20 allowance, no slot found".to_string())
            })?;

        // Set the storage slot to the desired allowance
        self.anvil_set_storage_at(token_address, slot.into(), amount.into()).await?;

        Ok(())
    }

    /// Recognizes a canonical Monad protocol envelope without requiring its reserved sender to be
    /// globally impersonated or recoverable from the envelope signature.
    #[cfg(feature = "monad")]
    fn monad_protocol_reorg_transaction(
        &self,
        transaction: FoundryTxEnvelope,
    ) -> Option<PendingTransaction<FoundryTxEnvelope>> {
        if !self.backend.is_monad() {
            return None;
        }

        let pending = PendingTransaction::with_sender(
            MaybeImpersonatedTransaction::new(transaction),
            monad_revm::staking::constants::SYSTEM_ADDRESS,
        );
        let tx_env: revm::context::TxEnv = crate::eth::backend::executor::build_tx_env_for_pending(
            &pending,
            self.backend.cheats(),
        );
        foundry_evm::core::evm::protocol_system_call(&tx_env)
            .is_ok_and(|call| call.is_some())
            .then_some(pending)
    }

    /// Reorg the chain to a specific depth and mine new blocks back to the canonical height.
    ///
    /// e.g depth = 3
    ///     A  -> B  -> C  -> D  -> E
    ///     A  -> B  -> C' -> D' -> E'
    ///
    /// Depth specifies the height to reorg the chain back to. Depth must not exceed the current
    /// chain height, i.e. can't reorg past the genesis block.
    ///
    /// Optionally supply a list of transaction and block pairs that will populate the reorged
    /// blocks. The maximum block number of the pairs must not exceed the specified depth.
    ///
    /// Handler for RPC call: `anvil_reorg`
    pub async fn anvil_reorg(&self, options: ReorgOptions) -> Result<()> {
        node_info!("anvil_reorg");
        let _lifecycle = self.lifecycle_lock.write().await;
        let mining_guard = self.backend.lock_mining().await;
        let depth = options.depth;
        let mut tx_block_pairs = options.tx_block_pairs;

        if let Some((_, num)) = tx_block_pairs.iter().find(|(_, num)| *num >= depth) {
            // A reorg of depth 0 mines no blocks, so there is nowhere to place the transactions.
            let Some(last_block) = depth.checked_sub(1) else {
                return Err(BlockchainError::RpcError(RpcError::invalid_params(
                    "Reorg depth must be at least 1 to include transactions",
                )));
            };
            return Err(BlockchainError::RpcError(RpcError::invalid_params(format!(
                "Block number for reorg tx will exceed the reorged chain height. Block number {num} must not exceed (depth-1) {last_block}"
            ))));
        }
        tx_block_pairs.sort_by_key(|a| a.1);

        // Check reorg depth doesn't exceed current chain height
        let current_height = self.backend.best_number();
        let common_height = current_height.checked_sub(depth).ok_or(BlockchainError::RpcError(
            RpcError::invalid_params(format!(
                "Reorg depth must not exceed current chain height: current height {current_height}, depth {depth}"
            )),
        ))?;

        // Get the common ancestor block
        let common_block =
            self.rollback_block(common_height).await?.ok_or(BlockchainError::BlockNotFound)?;

        #[cfg(feature = "monad")]
        let fee_defaults = if self.backend.is_monad() {
            self.backend
                .monad_rollback_fee_defaults(&common_block, self.lowest_suggestion_tip())
                .await
                .map(|(gas_price, blob_gas_price)| TransactionFeeDefaults {
                    gas_price,
                    blob_gas_price,
                })
        } else {
            None
        };
        #[cfg(not(feature = "monad"))]
        let fee_defaults = None;

        // Convert the transaction requests to pool transactions if they exist, otherwise use empty
        // hashmap
        let block_pool_txs = if tx_block_pairs.is_empty() {
            HashMap::default()
        } else {
            // Manage nonces for each signer
            // address -> cumulative nonce
            let mut nonces: HashMap<Address, u64> = HashMap::default();

            let mut txs: HashMap<u64, Vec<Arc<PoolTransaction<FoundryTxEnvelope>>>> =
                HashMap::default();
            for pair in tx_block_pairs {
                let (tx_data, block_index) = pair;

                let pending = match tx_data {
                    TransactionData::Raw(bytes) => {
                        let mut data = bytes.as_ref();
                        let decoded = FoundryTxEnvelope::decode_2718(&mut data)
                            .map_err(|_| BlockchainError::FailedToDecodeSignedTransaction)?;
                        if matches!(&decoded, FoundryTxEnvelope::Celo(_)) {
                            self.ensure_typed_transaction_supported(&decoded)?;
                        }
                        let protocol_pending = {
                            #[cfg(feature = "monad")]
                            {
                                self.monad_protocol_reorg_transaction(decoded.clone())
                            }
                            #[cfg(not(feature = "monad"))]
                            {
                                None
                            }
                        };
                        if let Some(pending) = protocol_pending {
                            pending
                        } else {
                            PendingTransaction::new(decoded)?
                        }
                    }

                    TransactionData::JSON(request) => {
                        let request = if request.transaction_type
                            == Some(foundry_primitives::CIP64_TX_TYPE)
                            || request
                                .other
                                .get("feeCurrency")
                                .is_some_and(|value| !value.is_null())
                        {
                            self.parse_transaction_request(request)?
                        } else {
                            request.inner.into()
                        };
                        let from = request.from().map(Ok).unwrap_or_else(|| {
                            self.accounts()?
                                .first()
                                .copied()
                                .ok_or(BlockchainError::NoSignerAvailable)
                        })?;

                        // Get the nonce at the common block
                        let curr_nonce = nonces.entry(from).or_insert(
                            self.get_transaction_count(
                                from,
                                Some(common_block.header.number().into()),
                            )
                            .await?,
                        );

                        // Build typed transaction request
                        let typed_tx = self
                            .build_tx_request_with_fee_defaults(request, *curr_nonce, fee_defaults)
                            .await?;

                        // Increment nonce
                        *curr_nonce += 1;

                        let protocol_pending = {
                            #[cfg(feature = "monad")]
                            {
                                if from == monad_revm::staking::constants::SYSTEM_ADDRESS {
                                    self.monad_protocol_reorg_transaction(
                                        typed_tx.clone().into_impersonated(),
                                    )
                                } else {
                                    None
                                }
                            }
                            #[cfg(not(feature = "monad"))]
                            {
                                None
                            }
                        };
                        if let Some(pending) = protocol_pending {
                            pending
                        } else if self.is_impersonated(from) {
                            let transaction = typed_tx.into_impersonated();
                            self.ensure_typed_transaction_supported(&transaction)?;
                            PendingTransaction::with_impersonated(transaction, from)
                        } else {
                            let transaction = self.sign_request(&from, typed_tx)?;
                            self.ensure_typed_transaction_supported(&transaction)?;
                            PendingTransaction::new(transaction)?
                        }
                    }
                };

                let pooled = PoolTransaction::new(pending).with_replay();
                txs.entry(block_index).or_default().push(Arc::new(pooled));
            }

            txs
        };

        self.rollback_backend(common_block, &mining_guard).await?;
        self.backend.reorg(depth, block_pool_txs, &mining_guard).await?;
        Ok(())
    }

    /// Mine blocks, instantly.
    ///
    /// Handler for RPC call: `evm_mine`
    ///
    /// This will mine the blocks regardless of the configured mining mode.
    /// **Note**: ganache returns `0x0` here as placeholder for additional meta-data in the future.
    pub async fn evm_mine(&self, opts: Option<MineOptions>) -> Result<String> {
        node_info!("evm_mine");

        self.do_evm_mine(opts).await?;

        Ok("0x0".to_string())
    }

    /// Mine blocks, instantly and return the mined blocks.
    ///
    /// Handler for RPC call: `evm_mine_detailed`
    ///
    /// This will mine the blocks regardless of the configured mining mode.
    ///
    /// **Note**: This behaves exactly as [Self::evm_mine] but returns different output, for
    /// compatibility reasons, this is a separate call since `evm_mine` is not an anvil original.
    /// and `ganache` may change the `0x0` placeholder.
    pub async fn evm_mine_detailed(&self, opts: Option<MineOptions>) -> Result<Vec<AnyRpcBlock>> {
        node_info!("evm_mine_detailed");

        let mined_blocks = self.do_evm_mine(opts).await?;

        let mut blocks = Vec::with_capacity(mined_blocks as usize);

        let latest = self.backend.best_number();
        for offset in (0..mined_blocks).rev() {
            let block_num = latest - offset;
            if let Some(mut block) =
                self.backend.block_by_number_full(BlockNumber::Number(block_num)).await?
            {
                let block_txs = match block.transactions_mut() {
                    BlockTransactions::Full(txs) => txs,
                    BlockTransactions::Hashes(_) | BlockTransactions::Uncle => unreachable!(),
                };
                for tx in block_txs.iter_mut() {
                    if let Some(receipt) = self.backend.mined_transaction_receipt(tx.tx_hash())
                        && let Some(output) = receipt.out
                    {
                        // insert revert reason if failure
                        if !receipt.inner.as_ref().status()
                            && let Some(reason) = RevertDecoder::new().maybe_decode(&output, None)
                        {
                            tx.other.insert(
                                "revertReason".to_string(),
                                serde_json::to_value(reason).expect("Infallible"),
                            );
                        }
                        tx.other.insert(
                            "output".to_string(),
                            serde_json::to_value(output).expect("Infallible"),
                        );
                    }
                }
                block.transactions = BlockTransactions::Full(block_txs.clone());
                blocks.push(block);
            }
        }

        Ok(blocks)
    }

    /// Execute a transaction regardless of signature status
    ///
    /// Handler for ETH RPC call: `eth_sendUnsignedTransaction`
    pub async fn eth_send_unsigned_transaction(
        &self,
        request: WithOtherFields<TransactionRequest>,
    ) -> Result<TxHash> {
        node_info!("eth_sendUnsignedTransaction");
        let request = self.parse_transaction_request(request)?;
        // either use the impersonated account of the request's `from` field
        let from = request.from().ok_or(BlockchainError::NoSignerAvailable)?;

        let nonce = self.request_nonce_for_transaction(&request, from).await?;

        let typed_tx = self.build_tx_request(request, nonce).await?;

        let transaction = typed_tx.into_impersonated();

        self.ensure_typed_transaction_supported(&transaction)?;

        let pending_transaction = PendingTransaction::with_impersonated(transaction, from);

        self.add_pending_transaction(pending_transaction).await
    }

    /// Returns a summary of all the transactions currently pending for inclusion in the next
    /// block(s), as well as the ones that are being scheduled for future execution only.
    ///
    /// See [here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_inspect) for more details
    ///
    /// Handler for ETH RPC call: `txpool_inspect`
    pub async fn txpool_inspect(&self) -> Result<TxpoolInspect> {
        node_info!("txpool_inspect");
        let mut inspect = TxpoolInspect::default();

        fn convert(tx: Arc<PoolTransaction<FoundryTxEnvelope>>) -> TxpoolInspectSummary {
            let tx = &tx.pending_transaction.transaction;
            let to = tx.to();
            let gas_price = tx.max_fee_per_gas();
            let value = tx.value();
            let gas = tx.gas_limit();
            TxpoolInspectSummary { to, value, gas, gas_price }
        }

        // Note: naming differs geth vs anvil:
        //
        // _Pending transactions_ are transactions that are ready to be processed and included in
        // the block. _Queued transactions_ are transactions where the transaction nonce is
        // not in sequence. The transaction nonce is an incrementing number for each transaction
        // with the same From address.
        for pending in self.pool.ready_transactions() {
            let entry = inspect.pending.entry(*pending.pending_transaction.sender()).or_default();
            let key = txpool_transaction_key(&pending.pending_transaction);
            entry.insert(key, convert(pending));
        }
        for queued in self.pool.pending_transactions() {
            let entry = inspect.queued.entry(*queued.pending_transaction.sender()).or_default();
            let key = txpool_transaction_key(&queued.pending_transaction);
            entry.insert(key, convert(queued));
        }
        Ok(inspect)
    }

    /// Returns the details of all transactions currently pending for inclusion in the next
    /// block(s), as well as the ones that are being scheduled for future execution only.
    ///
    /// See [here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_content) for more details
    ///
    /// Handler for ETH RPC call: `txpool_content`
    pub async fn txpool_content(&self) -> Result<TxpoolContent<AnyRpcTransaction>> {
        node_info!("txpool_content");
        self.build_txpool_content(None)
    }

    /// Builds the txpool content, optionally filtered to a single sender.
    ///
    /// When `filter` is `Some`, only transactions from that sender are converted, avoiding the
    /// cost of building RPC transactions for unrelated senders.
    fn build_txpool_content(
        &self,
        filter: Option<Address>,
    ) -> Result<TxpoolContent<AnyRpcTransaction>> {
        let mut content = TxpoolContent::<AnyRpcTransaction>::default();
        fn convert(tx: Arc<PoolTransaction<FoundryTxEnvelope>>) -> Result<AnyRpcTransaction> {
            let from = *tx.pending_transaction.sender();
            let tx = transaction_build(
                Some(tx.hash()),
                tx.pending_transaction.transaction.clone(),
                None,
                None,
                None,
            );

            let WithOtherFields { inner: mut tx, other } = tx.0;

            // we set the from field here explicitly to the set sender of the pending transaction,
            // in case the transaction is impersonated.
            tx.inner = Recovered::new_unchecked(tx.inner.into_inner(), from);

            let tx = AnyRpcTransaction(WithOtherFields { inner: tx, other });

            Ok(tx)
        }

        for pending in self.pool.ready_transactions() {
            let sender = *pending.pending_transaction.sender();
            if filter.is_some_and(|from| from != sender) {
                continue;
            }
            let entry = content.pending.entry(sender).or_default();
            let key = txpool_transaction_key(&pending.pending_transaction);
            entry.insert(key, convert(pending)?);
        }
        for queued in self.pool.pending_transactions() {
            let sender = *queued.pending_transaction.sender();
            if filter.is_some_and(|from| from != sender) {
                continue;
            }
            let entry = content.queued.entry(sender).or_default();
            let key = txpool_transaction_key(&queued.pending_transaction);
            entry.insert(key, convert(queued)?);
        }

        Ok(content)
    }

    /// Returns the details of all transactions currently pending for inclusion in the next
    /// block(s), as well as the ones that are being scheduled for future execution only, filtered
    /// by sender.
    ///
    /// See [here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_contentFrom) for more details
    ///
    /// Handler for ETH RPC call: `txpool_contentFrom`
    pub async fn txpool_content_from(
        &self,
        from: Address,
    ) -> Result<TxpoolContentFrom<AnyRpcTransaction>> {
        node_info!("txpool_contentFrom");
        let mut content = self.build_txpool_content(Some(from))?;
        Ok(content.remove_from(&from))
    }
}

impl EthApi<FoundryNetwork> {
    /// Executes the `evm_mine` and returns the number of blocks mined
    async fn do_evm_mine(&self, opts: Option<MineOptions>) -> Result<u64> {
        let mut blocks_to_mine = 1u64;

        if let Some(opts) = opts {
            let timestamp = match opts {
                MineOptions::Timestamp(timestamp) => timestamp,
                MineOptions::Options { timestamp, blocks } => {
                    if let Some(blocks) = blocks {
                        blocks_to_mine = blocks;
                    }
                    timestamp
                }
            };
            if let Some(timestamp) = timestamp {
                // timestamp was explicitly provided to be the next timestamp
                self.evm_set_next_block_timestamp(timestamp)?;
            }
        }

        // this can be blocking for a bit, especially in forking mode
        // <https://github.com/foundry-rs/foundry/issues/6036>
        self.on_blocking_task(|this| async move {
            // mine all the blocks
            for _ in 0..blocks_to_mine {
                this.mine_one().await?;
            }
            Ok(())
        })
        .await?;

        Ok(blocks_to_mine)
    }

    async fn do_estimate_gas(
        &self,
        request: WithOtherFields<TransactionRequest>,
        block_number: Option<BlockId>,
        overrides: EvmOverrides,
    ) -> Result<u128> {
        let block_request = self.block_request(block_number).await?;
        // check if the number predates the fork, if in fork mode
        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            if overrides.has_state() || overrides.has_block() {
                return Err(BlockchainError::EvmOverrideError(
                    "not available on past forked blocks".to_string(),
                ));
            }
            return Ok(fork.estimate_gas_raw(&request, Some(number.into())).await?);
        }

        // this can be blocking for a bit, especially in forking mode
        // <https://github.com/foundry-rs/foundry/issues/6036>
        self.on_blocking_task(|this| async move {
            let request = this.parse_transaction_request(request)?;
            this.backend
                .with_database_at_and_context(
                    Some(block_request),
                    |state, mut block, monad_context| {
                        let mut cache_db = CacheDB::new(state);
                        if let Some(state_overrides) = overrides.state {
                            apply_state_overrides(
                                state_overrides.into_iter().collect(),
                                &mut cache_db,
                            )?;
                        }
                        if let Some(block_overrides) = overrides.block {
                            cache_db.apply_block_overrides(*block_overrides, &mut block);
                        }
                        this.do_estimate_gas_with_state(request, &cache_db, block, monad_context)
                    },
                )
                .await
        })
        .await
    }

    async fn do_estimate_gas_typed(
        &self,
        request: FoundryTransactionRequest,
        block_number: Option<BlockId>,
    ) -> Result<u128> {
        let block_request = self.block_request(block_number).await?;
        self.on_blocking_task(|this| async move {
            this.backend
                .with_database_at_and_context(Some(block_request), |state, block, monad_context| {
                    this.do_estimate_gas_with_state(request, &state, block, monad_context)
                })
                .await
        })
        .await
    }

    /// Returns the priority of the transaction based on the current `TransactionOrder`
    fn transaction_priority(&self, tx: &FoundryTxEnvelope) -> TransactionPriority {
        self.transaction_order.read().priority(tx)
    }

    /// Returns a listener for pending transactions, yielding full transactions
    pub fn full_pending_transactions(&self) -> UnboundedReceiver<AnyRpcTransaction> {
        let (tx, rx) = unbounded_channel();
        let mut hashes = self.new_ready_transactions();

        let this = self.clone();

        tokio::spawn(async move {
            while let Some(hash) = hashes.next().await {
                if let Ok(Some(txn)) = this.transaction_by_hash(hash).await
                    && tx.send(txn).is_err()
                {
                    break;
                }
            }
        });

        rx
    }

    /// Returns a bounded stream of full pending transactions for poll-based filters.
    ///
    /// Unlike [`Self::full_pending_transactions`], this applies backpressure via a bounded channel,
    /// so an unpolled filter cannot buffer transactions without bound.
    pub fn full_pending_transactions_filter(&self) -> mpsc::Receiver<AnyRpcTransaction> {
        // Mirror the pool's ready-listener buffer so a full filter is bounded like a hash filter.
        let (tx, rx) = mpsc::channel(2048);
        let mut hashes = self.new_ready_transactions();
        let this = self.clone();

        tokio::spawn(async move {
            while let Some(hash) = hashes.next().await {
                if let Ok(Some(txn)) = this.transaction_by_hash(hash).await
                    && tx.send(txn).await.is_err()
                {
                    break;
                }
            }
        });

        rx
    }

    /// Returns a listener for new block receipts.
    pub fn transaction_receipts_subscription(
        &self,
        filter: TransactionReceiptsParams,
    ) -> UnboundedReceiver<Vec<FoundryTxReceipt>> {
        let (tx, rx) = unbounded_channel();
        let mut blocks = self.new_block_notifications();
        let this = self.clone();

        tokio::spawn(async move {
            // Precompute the hash filter once.
            let hash_filter = filter
                .transaction_hashes
                .filter(|hashes| !hashes.is_empty())
                .map(|hashes| hashes.into_iter().collect::<B256Set>());

            loop {
                let notification = tokio::select! {
                    biased;
                    // Exit when the subscriber unsubscribes, even while awaiting the next block.
                    _ = tx.closed() => break,
                    maybe_block = blocks.next() => match maybe_block {
                        Some(block) => block,
                        None => break,
                    },
                };

                let Some(block) = notification.as_new_block() else {
                    continue;
                };

                let receipts = match this.block_receipts(BlockId::hash(block.hash)).await {
                    Ok(Some(mut receipts)) => {
                        if let Some(hashes) = &hash_filter {
                            receipts.retain(|receipt| hashes.contains(&receipt.transaction_hash()));
                        }
                        receipts
                    }
                    Ok(None) => continue,
                    Err(err) => {
                        trace!(target: "node", %err, "failed to build block receipts for subscription");
                        continue;
                    }
                };

                if receipts.is_empty() {
                    continue;
                }

                if tx.send(receipts).is_err() {
                    break;
                }
            }
        });

        rx
    }

    /// Mines exactly one block
    pub async fn mine_one(&self) -> Result<()> {
        let _mining = self.backend.lock_mining().await;
        let transactions = self.pool.ready_transactions().collect::<Vec<_>>();
        let outcome = self.backend.mine_block_locked(transactions).await?;

        trace!(target: "node", blocknumber = ?outcome.block_number, "mined block");
        if self.pool.on_mined_block(outcome) {
            self.miner.retry_ready_transactions();
        }
        Ok(())
    }

    /// Returns the pending block with tx hashes
    async fn pending_block(&self) -> Result<AnyRpcBlock> {
        let transactions = self.pool.ready_transactions().collect::<Vec<_>>();
        let info = self.backend.pending_block(transactions).await?;
        Ok(self.backend.convert_block(info.block))
    }

    /// Returns the full pending block with `Transaction` objects
    async fn pending_block_full(&self) -> Result<Option<AnyRpcBlock>> {
        let transactions = self.pool.ready_transactions().collect::<Vec<_>>();
        let BlockInfo { block, transactions, receipts: _ } =
            self.backend.pending_block(transactions).await?;

        let mut partial_block = self.backend.convert_block(block.clone());

        let mut block_transactions = Vec::with_capacity(block.body.transactions.len());
        let base_fee = self.backend.base_fee();

        for info in transactions {
            let Some(tx) = block.body.transactions.get(info.transaction_index as usize).cloned()
            else {
                return Ok(None);
            };

            let tx = transaction_build(
                Some(info.transaction_hash),
                tx,
                Some(&block),
                Some(info),
                Some(base_fee),
            );
            block_transactions.push(tx);
        }

        partial_block.transactions = BlockTransactions::from(block_transactions);

        Ok(Some(partial_block))
    }

    /// Prepares transaction request by filling missing fields using Anvil's API, then attempts
    /// to build a [`FoundryTypedTx`].
    async fn build_tx_request(
        &self,
        request: FoundryTransactionRequest,
        nonce: u64,
    ) -> Result<FoundryTypedTx> {
        self.build_tx_request_with_fee_defaults(request, nonce, None).await
    }

    async fn build_tx_request_with_fee_defaults(
        &self,
        mut request: FoundryTransactionRequest,
        nonce: u64,
        fee_defaults: Option<TransactionFeeDefaults>,
    ) -> Result<FoundryTypedTx> {
        let from = request.from().or(self.accounts()?.first().copied());
        if let Some(from) = from {
            request.set_from(from);
        }

        // Fill common fields for all tx types
        request.chain_id().is_none().then(|| request.set_chain_id(self.chain_id()));
        request.nonce().is_none().then(|| request.set_nonce(nonce));
        let is_tempo_batch =
            matches!(&request, FoundryTransactionRequest::Tempo(tx) if !tx.calls.is_empty());
        if request.kind().is_none() && !is_tempo_batch {
            request.set_kind(TxKind::default());
        }
        if request.gas_limit().is_none() {
            let fallback_gas_limit = {
                let evm_env = self.backend.evm_env().read();
                self.backend.fallback_tx_gas_limit(&evm_env)
            };
            let estimated_gas = self
                .do_estimate_gas_typed(request.clone(), None)
                .await
                .map(|v| v as u64)
                .unwrap_or_else(|_| {
                    if is_simple_transfer_request(request.as_ref()) {
                        MIN_TRANSACTION_GAS as u64
                    } else {
                        fallback_gas_limit
                    }
                });
            request.set_gas_limit(estimated_gas);
        }

        // Fill missing tx type specific fields
        if let Err((tx_type, _)) = request.missing_keys() {
            let gas_price = || fee_defaults.map_or_else(|| self.gas_price(), |fees| fees.gas_price);
            if tx_type.is_legacy() || tx_type.is_eip2930() {
                request.gas_price().is_none().then(|| request.set_gas_price(gas_price()));
            }
            if tx_type.is_eip2930() {
                request
                    .access_list()
                    .is_none()
                    .then(|| request.set_access_list(Default::default()));
            }
            if tx_type.is_eip1559()
                || tx_type.is_eip4844()
                || tx_type.is_eip7702()
                || tx_type.is_tempo()
                || tx_type.is_celo()
            {
                request
                    .max_fee_per_gas()
                    .is_none()
                    .then(|| request.set_max_fee_per_gas(gas_price()));
                request
                    .max_priority_fee_per_gas()
                    .is_none()
                    .then(|| request.set_max_priority_fee_per_gas(MIN_SUGGESTED_PRIORITY_FEE));
            }
            if tx_type.is_eip4844() {
                request.as_ref().max_fee_per_blob_gas().is_none().then(|| {
                    let blob_gas_price = fee_defaults.map_or_else(
                        || self.backend.fees().get_next_block_blob_base_fee_per_gas(),
                        |fees| fees.blob_gas_price,
                    );
                    request.as_mut().set_max_fee_per_blob_gas(blob_gas_price)
                });
            }
        }

        match request
            .build_unsigned()
            .map_err(|e| BlockchainError::InvalidTransactionRequest(e.to_string()))?
        {
            FoundryTypedTx::Eip4844(TxEip4844Variant::TxEip4844(_))
                if !self.backend.skip_blob_validation(from) =>
            {
                // If blob validation is not skipped, reject TxEip4844 variant without sidecar.
                Err(BlockchainError::FailedToDecodeTransaction)
            }
            res => Ok(res),
        }
    }

    /// Returns the nonce of the `address` depending on the `block_number`
    async fn get_transaction_count(
        &self,
        address: Address,
        block_number: Option<BlockId>,
    ) -> Result<u64> {
        let block_request = self.block_request(block_number).await?;

        if let BlockRequest::Number(number) = block_request
            && let Some(fork) = self.get_fork()
            && fork.predates_fork(number)
        {
            return Ok(fork.get_nonce(address, number).await?);
        }

        self.backend.get_nonce(address, block_request).await
    }

    /// Returns the nonce for this request
    ///
    /// If the nonce field of the `request` is `None`, this accounts for pending transactions from
    /// the sender.
    async fn request_nonce(&self, request: &TransactionRequest, from: Address) -> Result<u64> {
        if let Some(nonce) = request.nonce {
            Ok(nonce)
        } else {
            self.get_transaction_count(from, Some(BlockId::pending())).await
        }
    }

    /// Adds the given transaction to the pool
    async fn add_pending_transaction(
        &self,
        pending_transaction: PendingTransaction<FoundryTxEnvelope>,
    ) -> Result<TxHash> {
        let _mining = self.backend.lock_mining().await;
        self.backend.validate_pool_transaction(&pending_transaction).await?;

        let from = *pending_transaction.sender();
        #[cfg(feature = "base")]
        let base_markers = self.eip8130_nonce_markers(&pending_transaction).await?;
        #[cfg(not(feature = "base"))]
        let base_markers = None;

        let markers = if base_markers.is_some() {
            base_markers
        } else {
            self.tempo_nonce_markers(&pending_transaction).await?
        };
        let (requires, provides) = if let Some(markers) = markers {
            markers
        } else {
            let nonce = pending_transaction.transaction.nonce();
            let on_chain_nonce = self.backend.current_nonce(from).await?;
            (required_marker(nonce, on_chain_nonce, from), vec![to_marker(nonce, from)])
        };

        debug_assert!(requires != provides);
        let priority = self.transaction_priority(&pending_transaction.transaction);
        let pool_transaction =
            PoolTransaction { requires, provides, pending_transaction, priority, is_replay: false };
        let tx = self.pool.add_transaction(pool_transaction)?;
        trace!(target: "node", "Added transaction: [{:?}] sender={:?}", tx.hash(), from);
        Ok(*tx.hash())
    }

    /// Returns Tempo lane/replay markers when the transaction does not use the protocol nonce.
    async fn tempo_nonce_markers(
        &self,
        pending: &PendingTransaction<FoundryTxEnvelope>,
    ) -> Result<Option<(Vec<TxMarker>, Vec<TxMarker>)>> {
        let FoundryTxEnvelope::Tempo(signed) = pending.transaction.as_ref() else {
            return Ok(None);
        };
        let tx = signed.tx();
        if tx.nonce_key.is_zero() {
            return Ok(None);
        }
        if tx.nonce_key == U256::MAX {
            return Ok(Some((Vec::new(), vec![pending.hash().to_vec()])));
        }

        let state_nonce = self.backend.tempo_nonce(*pending.sender(), tx.nonce_key, None).await?;
        if tx.nonce < state_nonce {
            return Err(InvalidTransactionError::NonceTooLow.into());
        }

        let requires = if tx.nonce == state_nonce {
            Vec::new()
        } else {
            vec![tempo_nonce_marker(*pending.sender(), tx.nonce_key, tx.nonce - 1)]
        };
        let provides = vec![tempo_nonce_marker(*pending.sender(), tx.nonce_key, tx.nonce)];
        Ok(Some((requires, provides)))
    }

    /// Returns Base EIP-8130 channel/replay markers when the transaction does not use the
    /// protocol nonce lane.
    #[cfg(feature = "base")]
    async fn eip8130_nonce_markers(
        &self,
        pending: &PendingTransaction<FoundryTxEnvelope>,
    ) -> Result<Option<(Vec<TxMarker>, Vec<TxMarker>)>> {
        if let FoundryTxEnvelope::Eip8130(signed) = pending.transaction.as_ref() {
            let tx = signed.tx();
            let now = self.backend.eip8130_pool_timestamp_ms();
            signed
                .validate_admission_static(self.backend.chain_id().to())
                .map_err(|error| BlockchainError::Eip8130TransactionRejected(error.to_string()))?;
            signed
                .validate_timestamp(now)
                .map_err(|error| BlockchainError::Eip8130TransactionRejected(error.to_string()))?;
            if tx.nonce_key.is_zero() {
                let state_nonce = self.backend.current_nonce(*pending.sender()).await?;
                if tx.nonce_sequence < state_nonce {
                    return Err(InvalidTransactionError::NonceTooLow.into());
                }
                let provides = vec![to_marker(tx.nonce_sequence, *pending.sender())];
                self.ensure_eip8130_replacement_price(pending, &provides)?;
                self.ensure_eip8130_payer_reservation(pending, &provides).await?;
                return Ok(None);
            }
            if tx.nonce_key == Eip8130Constants::NONCE_KEY_MAX {
                let replay_id = tx.replay_id(*pending.sender());
                let slot = NonceManagerStorage::expiring_nonce_seen_slot(replay_id);
                let word =
                    self.backend.storage_at(NonceManagerStorage::ADDRESS, slot, None).await?;
                let recorded_expiry = (Into::<U256>::into(word) & U256::from(u64::MAX)).to::<u64>();
                if recorded_expiry > now {
                    return Err(BlockchainError::Eip8130TransactionRejected(
                        "expiring nonce replay has already been recorded".to_string(),
                    ));
                }
                let provides = vec![eip8130_replay_marker(replay_id)];
                self.ensure_eip8130_replacement_price(pending, &provides)?;
                self.ensure_eip8130_payer_reservation(pending, &provides).await?;
                return Ok(Some((Vec::new(), provides)));
            }

            let slot = NonceManagerStorage::nonce_slot(*pending.sender(), tx.nonce_key)
                .map_err(|error| BlockchainError::InvalidTransactionRequest(error.to_string()))?;
            let word = self.backend.storage_at(NonceManagerStorage::ADDRESS, slot, None).await?;
            let state_nonce = Eip8130Nonce::decode_channel_nonce(word.into()).to::<u64>();
            if tx.nonce_sequence < state_nonce {
                return Err(InvalidTransactionError::NonceTooLow.into());
            }

            let requires = if tx.nonce_sequence == state_nonce {
                Vec::new()
            } else {
                vec![eip8130_channel_marker(*pending.sender(), tx.nonce_key, tx.nonce_sequence - 1)]
            };
            let provides =
                vec![eip8130_channel_marker(*pending.sender(), tx.nonce_key, tx.nonce_sequence)];
            self.ensure_eip8130_replacement_price(pending, &provides)?;
            self.ensure_eip8130_payer_reservation(pending, &provides).await?;
            return Ok(Some((requires, provides)));
        }

        Ok(None)
    }

    /// Enforces Base's default ten-percent EIP-8130 replacement price bump.
    #[cfg(feature = "base")]
    fn ensure_eip8130_replacement_price(
        &self,
        pending: &PendingTransaction<FoundryTxEnvelope>,
        provides: &[TxMarker],
    ) -> Result<()> {
        let Some(existing) = self.pool.transaction_with_markers(*pending.sender(), provides) else {
            return Ok(());
        };
        let FoundryTxEnvelope::Eip8130(incoming) = pending.transaction.as_ref() else {
            return Ok(());
        };
        let FoundryTxEnvelope::Eip8130(current) = existing.pending_transaction.transaction.as_ref()
        else {
            return Ok(());
        };

        const PRICE_BUMP_PERCENT: u128 = 10;
        let bumped =
            |price: u128| price.saturating_mul(100 + PRICE_BUMP_PERCENT).saturating_add(99) / 100;
        if incoming.tx().max_fee_per_gas < bumped(current.tx().max_fee_per_gas)
            || incoming.tx().max_priority_fee_per_gas
                < bumped(current.tx().max_priority_fee_per_gas)
        {
            return Err(
                crate::eth::error::PoolError::ReplacementUnderpriced(*pending.hash()).into()
            );
        }
        Ok(())
    }

    /// Reserves each payer's maximum EIP-8130 fee across pending transactions.
    #[cfg(feature = "base")]
    async fn ensure_eip8130_payer_reservation(
        &self,
        pending: &PendingTransaction<FoundryTxEnvelope>,
        replacing_markers: &[TxMarker],
    ) -> Result<()> {
        let FoundryTxEnvelope::Eip8130(incoming) = pending.transaction.as_ref() else {
            return Ok(());
        };
        let payer = incoming.tx().payer.unwrap_or(*pending.sender());
        let incoming_cost = FeeCheck::max_fee_charge(
            incoming.tx().gas_limit,
            IntrinsicGas::max_payer_auth_cost(incoming)
                .map_err(|error| BlockchainError::Eip8130TransactionRejected(error.to_string()))?,
            incoming.tx().max_fee_per_gas,
        );
        let mut reserved = U256::ZERO;
        for pooled in self.pool.all_transactions() {
            if pooled.provides == replacing_markers {
                continue;
            }
            let FoundryTxEnvelope::Eip8130(signed) =
                pooled.pending_transaction.transaction.as_ref()
            else {
                continue;
            };
            if signed.tx().payer.unwrap_or(*pooled.pending_transaction.sender()) != payer {
                continue;
            }
            let payer_auth = IntrinsicGas::max_payer_auth_cost(signed)
                .map_err(|error| BlockchainError::Eip8130TransactionRejected(error.to_string()))?;
            reserved = reserved.saturating_add(FeeCheck::max_fee_charge(
                signed.tx().gas_limit,
                payer_auth,
                signed.tx().max_fee_per_gas,
            ));
        }
        let balance = self.backend.get_account(payer).await?.balance;
        let required = reserved.saturating_add(incoming_cost);
        if balance < required {
            return Err(BlockchainError::Eip8130TransactionRejected(format!(
                "gas payer balance {balance} is below pending reservation {required}"
            )));
        }
        Ok(())
    }

    /// additional validation against hardfork
    fn ensure_typed_transaction_supported(&self, tx: &FoundryTxEnvelope) -> Result<()> {
        match &tx {
            FoundryTxEnvelope::Eip2930(_) => self.backend.ensure_eip2930_active(),
            FoundryTxEnvelope::Eip1559(_) => self.backend.ensure_eip1559_active(),
            FoundryTxEnvelope::Eip4844(_) => self.backend.ensure_eip4844_active(),
            FoundryTxEnvelope::Eip7702(_) => self.backend.ensure_eip7702_active(),
            #[cfg(any(feature = "base", feature = "optimism"))]
            FoundryTxEnvelope::Deposit(_) => self.backend.ensure_deposits_active(),
            #[cfg(feature = "optimism")]
            FoundryTxEnvelope::PostExec(_) => Err(BlockchainError::InvalidTransactionRequest(
                "not implemented for post-exec tx".to_string(),
            )),
            #[cfg(feature = "base")]
            FoundryTxEnvelope::Eip8130(_) => self.backend.ensure_base_eip8130_submission_active(),
            FoundryTxEnvelope::Celo(_) => self.backend.ensure_cip64_active(),
            FoundryTxEnvelope::Legacy(_) => Ok(()),
            FoundryTxEnvelope::Tempo(_) => self.backend.ensure_tempo_active(),
        }
    }

    /// Sets the fee token for a user address.
    ///
    /// Handler for RPC call: `anvil_setFeeToken`
    ///
    /// Only supported when running in Tempo mode (`--tempo`).
    pub async fn anvil_set_fee_token(&self, user: Address, token: Address) -> Result<()> {
        node_info!("anvil_setFeeToken");
        self.ensure_tempo_mode()?;
        self.backend.set_fee_token(user, token).await?;
        Ok(())
    }

    /// Sets the fee token for a validator address.
    ///
    /// Handler for RPC call: `anvil_setValidatorFeeToken`
    ///
    /// Only supported when running in Tempo mode (`--tempo`).
    pub async fn anvil_set_validator_fee_token(
        &self,
        validator: Address,
        token: Address,
    ) -> Result<()> {
        node_info!("anvil_setValidatorFeeToken");
        self.ensure_tempo_mode()?;
        self.backend.set_validator_fee_token(validator, token).await?;
        Ok(())
    }

    /// Mints FeeAMM liquidity for a token pair.
    ///
    /// Handler for RPC call: `anvil_setFeeAmmLiquidity`
    ///
    /// Only supported when running in Tempo mode (`--tempo`).
    pub async fn anvil_set_fee_amm_liquidity(
        &self,
        user_token: Address,
        validator_token: Address,
        amount: U256,
    ) -> Result<()> {
        node_info!("anvil_setFeeAmmLiquidity");
        self.ensure_tempo_mode()?;
        self.backend.set_fee_amm_liquidity(user_token, validator_token, amount).await?;
        Ok(())
    }

    /// Ensures anvil runs in Tempo mode (`--tempo`), the RPC method is unavailable otherwise.
    fn ensure_tempo_mode(&self) -> Result<()> {
        if self.backend.is_tempo() { Ok(()) } else { Err(BlockchainError::RpcUnimplemented) }
    }

    fn parse_transaction_request(
        &self,
        request: WithOtherFields<TransactionRequest>,
    ) -> Result<FoundryTransactionRequest> {
        self.backend.parse_transaction_request(request)
    }

    async fn request_nonce_for_transaction(
        &self,
        request: &FoundryTransactionRequest,
        from: Address,
    ) -> Result<u64> {
        if let FoundryTransactionRequest::Tempo(request) = request
            && let Some(nonce_key) = request.nonce_key.filter(|key| !key.is_zero())
        {
            if let Some(nonce) = request.nonce() {
                return Ok(nonce);
            }
            let nonce = self.backend.tempo_nonce(from, nonce_key, None).await?;
            return Ok(nonce);
        }
        self.request_nonce(request.as_ref(), from).await
    }
}

fn is_simple_transfer_request(request: &TransactionRequest) -> bool {
    request.to.as_ref().and_then(TxKind::to).is_some()
        && (request.input.input().is_none()
            || request.input.input().is_some_and(|data| data.is_empty()))
        && request.authorization_list.is_none()
        && request.access_list.is_none()
        && request.blob_versioned_hashes.is_none()
}

fn required_marker(provided_nonce: u64, on_chain_nonce: u64, from: Address) -> Vec<TxMarker> {
    if provided_nonce == on_chain_nonce {
        return Vec::new();
    }
    let prev_nonce = provided_nonce.saturating_sub(1);
    if on_chain_nonce <= prev_nonce { vec![to_marker(prev_nonce, from)] } else { Vec::new() }
}

fn tempo_nonce_marker(sender: Address, nonce_key: U256, nonce: u64) -> TxMarker {
    let mut marker = b"tempo-nonce".to_vec();
    marker.extend_from_slice(sender.as_slice());
    marker.extend_from_slice(&nonce_key.to_be_bytes::<32>());
    marker.extend_from_slice(&nonce.to_be_bytes());
    marker
}

#[cfg(feature = "base")]
fn eip8130_channel_marker(sender: Address, nonce_key: U256, nonce_sequence: u64) -> TxMarker {
    let mut marker = b"base-eip8130-channel".to_vec();
    marker.extend_from_slice(sender.as_slice());
    marker.extend_from_slice(&nonce_key.to_be_bytes::<32>());
    marker.extend_from_slice(&nonce_sequence.to_be_bytes());
    marker
}

#[cfg(feature = "base")]
fn eip8130_replay_marker(replay_id: B256) -> TxMarker {
    let mut marker = b"base-eip8130-replay".to_vec();
    marker.extend_from_slice(replay_id.as_slice());
    marker
}

/// Rewrites the Tempo fee payer service encoding of a raw transaction into the standard envelope
/// encoding.
///
/// Fee payer service clients submit sponsorship requests with a `0x00` placeholder in the fee
/// payer signature field (and the fee token left empty for the sponsor to fill), which the
/// standard envelope decoder rejects. The placeholder is rewritten to the standard encoding of
/// [`FEE_PAYER_SIGNATURE_MARKER`], preserving the sponsored signing state the sender committed
/// to. Returns the normalized raw transaction when the input carries the placeholder and `None`
/// for any other encoding.
fn normalize_fee_payer_service_encoding(raw: &[u8]) -> Option<Vec<u8>> {
    let (tx_type, mut encoded_fields) = raw.split_first()?;
    if *tx_type != TEMPO_TX_TYPE_ID {
        return None;
    }
    let PayloadView::List(fields) = Header::decode_raw(&mut encoded_fields).ok()? else {
        return None;
    };
    if !encoded_fields.is_empty() {
        return None;
    }
    // The fee payer signature is the twelfth field of a Tempo AA transaction.
    if fields.get(11).is_none_or(|field| *field != [0x00]) {
        return None;
    }

    // The standard encoding of the marker signature, mirroring the fee payer signature closure of
    // `TempoTransaction::rlp_encode_fields_default`.
    let marker = FEE_PAYER_SIGNATURE_MARKER;
    let mut marker_field = Vec::new();
    Header { list: true, payload_length: marker.rlp_rs_len() + marker.v().length() }
        .encode(&mut marker_field);
    marker.write_rlp_vrs(&mut marker_field, marker.v());

    let mut payload = Vec::new();
    for (index, field) in fields.into_iter().enumerate() {
        if index == 11 {
            payload.extend_from_slice(&marker_field);
        } else {
            payload.extend_from_slice(field);
        }
    }
    let mut normalized = vec![TEMPO_TX_TYPE_ID];
    Header { list: true, payload_length: payload.len() }.encode(&mut normalized);
    normalized.extend_from_slice(&payload);
    Some(normalized)
}

/// EIP-2718 encodes a transaction held in its JSON-RPC form, reporting the types anvil cannot
/// encode as [`BlockchainError::UnsupportedTransactionEncoding`].
fn encode_rpc_transaction(transaction: &AnyRpcTransaction) -> Result<Bytes> {
    FoundryTxEnvelope::encode_rpc_2718(transaction)
        .map_err(|_| BlockchainError::UnsupportedTransactionEncoding(transaction.ty()))
}

fn txpool_transaction_key(pending_transaction: &PendingTransaction<FoundryTxEnvelope>) -> String {
    match pending_transaction.transaction.as_ref() {
        #[cfg(feature = "base")]
        FoundryTxEnvelope::Eip8130(signed) if !signed.tx().nonce_key.is_zero() => {
            if signed.tx().nonce_key == Eip8130Constants::NONCE_KEY_MAX {
                return signed.tx().replay_id(*pending_transaction.sender()).to_string();
            }
            format!("{}:{}", signed.tx().nonce_key, signed.tx().nonce_sequence)
        }
        FoundryTxEnvelope::Tempo(tx) if !tx.tx().nonce_key.is_zero() => {
            let tx = tx.tx();
            format!("{}:{}", tx.nonce_key, tx.nonce)
        }
        _ => pending_transaction.nonce().to_string(),
    }
}

fn convert_transact_out(out: &Option<Output>) -> Bytes {
    out.as_ref().map(Output::data).cloned().unwrap_or_default()
}

/// Returns an error if the `exit` code is _not_ ok
fn ensure_return_ok(exit: InstructionResult, out: &Option<Output>) -> Result<Bytes> {
    let out = convert_transact_out(out);
    match exit {
        return_ok!() => Ok(out),
        return_revert!() => Err(InvalidTransactionError::Revert(Some(out)).into()),
        reason => Err(BlockchainError::EvmError(reason)),
    }
}

/// Maps an EVM exit code to the optional `error` string reported in
/// `eth_createAccessList` results.
fn execution_error(exit: InstructionResult) -> Option<String> {
    match SuccessOrHalt::<HaltReason>::from(exit) {
        SuccessOrHalt::Success(_) => None,
        SuccessOrHalt::Revert => Some("execution reverted".to_string()),
        SuccessOrHalt::Halt(reason) => Some(reason.to_string()),
        SuccessOrHalt::FatalExternalError => Some("fatal external error".to_string()),
        SuccessOrHalt::Internal(_) => Some("internal EVM error".to_string()),
    }
}

/// Keeps result of a call to revm EVM used for gas estimation
enum GasEstimationCallResult {
    Success(u128),
    OutOfGas,
    Revert(Option<Bytes>),
    EvmError(InstructionResult),
}

/// Converts the result of a call to revm EVM into a [`GasEstimationCallResult`].
///
/// Expected to stay up to date with: <https://github.com/bluealloy/revm/blob/main/crates/interpreter/src/instruction_result.rs>
impl TryFrom<Result<(InstructionResult, Option<Output>, u128, State)>> for GasEstimationCallResult {
    type Error = BlockchainError;

    fn try_from(res: Result<(InstructionResult, Option<Output>, u128, State)>) -> Result<Self> {
        match res {
            // Exceptional case: init used too much gas, treated as out of gas error
            Err(BlockchainError::InvalidTransaction(InvalidTransactionError::GasTooHigh(_))) => {
                Ok(Self::OutOfGas)
            }
            // Tempo intrinsic gas errors come through as Message variants
            Err(BlockchainError::Message(ref msg))
                if msg.contains("insufficient gas for intrinsic cost") =>
            {
                Ok(Self::OutOfGas)
            }
            Err(err) => Err(err),
            Ok((exit, output, gas, _)) => match exit {
                return_ok!() => Ok(Self::Success(gas)),

                // Revert opcodes:
                InstructionResult::Revert => {
                    Ok(Self::Revert(Some(output.map(|o| o.into_data()).unwrap_or_default())))
                }
                InstructionResult::CallTooDeep
                | InstructionResult::OutOfFunds
                | InstructionResult::CreateInitCodeStartingEF00
                | InstructionResult::InvalidEOFInitCode
                | InstructionResult::InvalidExtDelegateCallTarget => Ok(Self::EvmError(exit)),

                // Out of gas errors:
                InstructionResult::OutOfGas
                | InstructionResult::MemoryOOG
                | InstructionResult::MemoryLimitOOG
                | InstructionResult::PrecompileOOG
                | InstructionResult::InvalidOperandOOG
                | InstructionResult::ReentrancySentryOOG => Ok(Self::OutOfGas),

                // Other errors:
                InstructionResult::OpcodeNotFound
                | InstructionResult::CallNotAllowedInsideStatic
                | InstructionResult::StateChangeDuringStaticCall
                | InstructionResult::InvalidFEOpcode
                | InstructionResult::InvalidJump
                | InstructionResult::NotActivated
                | InstructionResult::StackUnderflow
                | InstructionResult::StackOverflow
                | InstructionResult::OutOfOffset
                | InstructionResult::CreateCollision
                | InstructionResult::OverflowPayment
                | InstructionResult::PrecompileError
                | InstructionResult::NonceOverflow
                | InstructionResult::CreateContractSizeLimit
                | InstructionResult::CreateContractStartingWithEF
                | InstructionResult::CreateInitCodeSizeLimit
                | InstructionResult::InvalidImmediateEncoding
                | InstructionResult::FatalExternalError => Ok(Self::EvmError(exit)),
            },
        }
    }
}

/// Appends the upstream part of a fee-history range that crosses the fork boundary.
fn merge_pre_fork_fee_history(
    response: &mut FeeHistory,
    rewards: &mut Vec<Vec<u128>>,
    pre: FeeHistory,
    fork_block: u64,
) {
    // Upstream may return fewer blocks than requested when its available history starts later.
    response.oldest_block = pre.oldest_block;
    let count = fork_block.checked_sub(pre.oldest_block).map_or(0, |count| count.saturating_add(1))
        as usize;

    // Base-fee arrays include a trailing next-block entry. The caller computes that block locally.
    response.base_fee_per_gas.extend(pre.base_fee_per_gas.into_iter().take(count));
    response.gas_used_ratio.extend(pre.gas_used_ratio.into_iter().take(count));
    if let Some(reward) = pre.reward {
        rewards.extend(reward.into_iter().take(count));
    }

    // EIP-4844 fields may be omitted for pre-Cancun blocks. Pad to the actual returned count.
    response.base_fee_per_blob_gas.extend(pre.base_fee_per_blob_gas.into_iter().take(count));
    response.base_fee_per_blob_gas.resize(count, 0);
    response.blob_gas_used_ratio.extend(pre.blob_gas_used_ratio.into_iter().take(count));
    response.blob_gas_used_ratio.resize(count, 0.0);
}

fn reward_at_percentile(rewards: &[u128], percentile: f64) -> u128 {
    let index = (percentile * REWARD_PERCENTILE_RESOLUTION).round() as usize;
    rewards.get(index).copied().unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{NodeConfig, spawn};

    #[cfg(feature = "base")]
    #[tokio::test(flavor = "multi_thread")]
    async fn base_requests_are_rejected_without_execution() {
        let (api, _handle) = spawn(NodeConfig::test()).await;
        for request in [serde_json::json!({ "type": "0x79" }), serde_json::json!({ "calls": [[]] })]
        {
            let request = serde_json::from_value(request).unwrap();
            assert!(matches!(
                api.parse_transaction_request(request),
                Err(BlockchainError::BaseTransactionUnsupported)
            ));
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn set_rpc_url_installs_context_equivalent_identity_with_new_instance() {
        let genesis_timestamp = 1_700_000_000u64;
        let (_origin_api, origin_handle) =
            spawn(NodeConfig::test().with_genesis_timestamp(Some(genesis_timestamp))).await;
        let (api, _handle) =
            spawn(NodeConfig::test().with_eth_rpc_url(Some(origin_handle.http_endpoint()))).await;
        let (target_api, target_handle) =
            spawn(NodeConfig::test().with_genesis_timestamp(Some(genesis_timestamp))).await;
        let target_url = target_handle.http_endpoint();
        let fork = api.backend.get_fork().unwrap();
        let identity_before = fork.config.read().endpoint_identity;

        api.anvil_set_rpc_url(target_url.clone()).await.unwrap();

        let fork = api.backend.get_fork().unwrap();
        let config = fork.config.read();
        assert!(config.endpoint_identity.context_eq(identity_before));
        assert_ne!(config.endpoint_identity.instance_id, identity_before.instance_id);
        assert_eq!(config.endpoint_identity.instance_id, Some(target_api.instance_id()));
        assert_eq!(config.eth_rpc_url(), Some(target_url.as_str()));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn memory_reset_stages_live_fees_after_active_mining() {
        let (api, _handle) = spawn(NodeConfig::test()).await;
        api.backend.set_base_fee(123);

        // Model an in-flight miner that publishes its final fee update before completing. Reset
        // must wait to snapshot that state until mining finishes.
        let mining = api.backend.lock_mining().await;
        let reset_api = api.clone();
        let reset = tokio::spawn(async move { reset_api.anvil_reset(None).await });

        tokio::time::timeout(Duration::from_secs(5), async {
            while api.lifecycle_lock.try_read().is_ok() {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();

        api.backend.set_base_fee(456);
        drop(mining);
        tokio::time::timeout(Duration::from_secs(5), reset).await.unwrap().unwrap().unwrap();

        assert_eq!(api.backend.evm_env().read().block_env.basefee, 456);
        assert_eq!(api.base_fee().unwrap(), Some(U256::from(crate::eth::fees::INITIAL_BASE_FEE)));
    }

    // Regression test for <https://github.com/foundry-rs/foundry/issues/13680>.
    #[tokio::test(flavor = "multi_thread")]
    async fn fee_history_is_complete_when_cache_entries_are_missing() {
        let (api, _handle) = spawn(NodeConfig::test()).await;
        let count = 10u64;
        api.anvil_mine(Some(U256::from(count)), None).await.unwrap();

        // Wait until the asynchronous service has consumed all block notifications, then clear
        // its cache. With no notifications left to repopulate it, this deterministically exercises
        // the RPC handler's fallback for canonical blocks whose cache entries are absent.
        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if (1..=count).all(|number| api.fee_history_cache.lock().contains_key(&number)) {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
        api.fee_history_cache.lock().clear();

        let fee_history =
            api.fee_history(U256::from(count), BlockNumber::Latest, vec![50.0]).await.unwrap();

        assert_eq!(fee_history.oldest_block, 1);
        assert_eq!(fee_history.gas_used_ratio.len(), count as usize);
        assert_eq!(fee_history.blob_gas_used_ratio.len(), count as usize);
        assert_eq!(fee_history.base_fee_per_gas.len(), count as usize + 1);
        assert_eq!(fee_history.base_fee_per_blob_gas.len(), count as usize + 1);
        let rewards = fee_history.reward.unwrap();
        assert_eq!(rewards.len(), count as usize);
        assert!(rewards.iter().all(|reward| reward.len() == 1));
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn fee_history_rejects_invalid_reward_percentiles() {
        let (api, _handle) = spawn(NodeConfig::test()).await;

        for percentiles in [vec![-0.5], vec![100.5], vec![50.0, 25.0], vec![50.0, 50.0]] {
            let err =
                api.fee_history(U256::ONE, BlockNumber::Latest, percentiles).await.unwrap_err();
            assert!(matches!(
                err,
                BlockchainError::FeeHistory(FeeHistoryError::InvalidRewardPercentiles)
            ));
        }

        for percentiles in [vec![], vec![0.0, 100.0]] {
            api.fee_history(U256::ONE, BlockNumber::Latest, percentiles).await.unwrap();
        }
    }

    #[test]
    fn fractional_reward_percentiles_use_cache_resolution() {
        let rewards = (0..=200).collect::<Vec<_>>();

        assert_eq!(reward_at_percentile(&rewards, 0.0), 0);
        assert_eq!(reward_at_percentile(&rewards, 0.5), 1);
        assert_eq!(reward_at_percentile(&rewards, 1.0), 2);
        assert_eq!(reward_at_percentile(&rewards, 100.0), 200);
    }

    #[test]
    fn shortened_pre_fork_fee_history_uses_upstream_start() {
        let requested_lowest = 4;
        let fork_block = 10;
        let pre = FeeHistory {
            oldest_block: 5,
            base_fee_per_gas: vec![1; 7],
            gas_used_ratio: vec![0.0; 6],
            reward: Some(vec![vec![]; 6]),
            base_fee_per_blob_gas: vec![],
            blob_gas_used_ratio: vec![],
        };

        let mut response = FeeHistory { oldest_block: requested_lowest, ..Default::default() };
        let mut rewards = Vec::new();

        merge_pre_fork_fee_history(&mut response, &mut rewards, pre, fork_block);

        assert_eq!(response.oldest_block, 5);
        assert_eq!(response.gas_used_ratio.len(), 6);
        assert_eq!(response.base_fee_per_gas.len(), 6);
        assert_eq!(response.blob_gas_used_ratio.len(), 6);
        assert_eq!(response.base_fee_per_blob_gas.len(), 6);
        assert_eq!(rewards.len(), 6);
    }
}