reth_rpc/

debug.rs

use alloy_consensus::{transaction::TxHashRef, BlockHeader};
use alloy_eips::{eip2718::Encodable2718, BlockId, BlockNumberOrTag};
use alloy_genesis::ChainConfig;
use alloy_primitives::{hex::decode, uint, Address, Bytes, B256, U256, U64};
use alloy_rlp::{Decodable, Encodable};
use alloy_rpc_types::BlockTransactionsKind;
use alloy_rpc_types_debug::ExecutionWitness;
use alloy_rpc_types_eth::{
    state::EvmOverrides, Account, AccountInfo, BlockError, Bundle, Index, StateContext,
};
use alloy_rpc_types_trace::geth::{
    ChainBlockTraceResult, GethDebugTracingCallOptions, GethDebugTracingOptions, GethTrace,
    TraceResult,
};
use evm2::evm::StateChangeSource;
use evm2_inspectors::tracing::{DebugInspector, TransactionContext};
use futures::Stream;
use jsonrpsee::{core::RpcResult, PendingSubscriptionSink, SubscriptionMessage};
use parking_lot::RwLock;
use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
use reth_engine_primitives::ConsensusEngineEvent;
use reth_errors::RethError;
use reth_evm::{
    execute::{BlockExecutor, BlockTransactionResult, Executor},
    witness::ExecutionWitnessRecord,
    BlockExecutionError, ConfigureEvm, DynDatabase, Evm, EvmEnv, EvmEnvFor,
};
use reth_primitives_traits::{
    Block as BlockTrait, BlockBody, BlockTy, ReceiptWithBloom, RecoveredBlock,
};
use reth_rpc_api::DebugApiServer;
use reth_rpc_convert::RpcTxReq;
use reth_rpc_eth_api::{
    helpers::{EthTransactions, TraceExt},
    AsEthApiError, FromEthApiError, FromEvmError, RpcConvert, RpcNodeCore,
};
use reth_rpc_eth_types::{utils::CallManyBudget, EthApiError, StateCacheDb};
use reth_rpc_server_types::{result::internal_rpc_err, ToRpcResult};
use reth_storage_api::{
    BlockIdReader, BlockReaderIdExt, HashedPostStateProvider, HeaderProvider, ProviderBlock,
    ReceiptProviderIdExt, StateProviderFactory, StateRootProvider, StorageRootProvider,
    TransactionVariant,
};
use reth_tasks::{cancel::is_cancelled, pool::BlockingTaskGuard, Runtime};
use reth_transaction_pool::TransactionPool;
use reth_trie_common::{
    root::storage_root_unsorted, updates::TrieUpdates, ExecutionWitnessMode, HashedPostState,
    HashedStorage,
};
use serde::{Deserialize, Serialize};
use std::{collections::VecDeque, sync::Arc, time::Duration};
use tokio::sync::{AcquireError, OwnedSemaphorePermit};
use tokio_stream::StreamExt;

#[cfg(feature = "js-tracer")]
use alloy_rpc_types_trace::geth::{
    GethDebugBuiltInTracerType, GethDebugTracerConfig, GethDebugTracerType,
};
#[cfg(feature = "js-tracer")]
use evm2_inspectors::tracing::js::JsInspectorInterrupt;

/// `debug` API implementation.
///
/// This type provides the functionality for handling `debug` related requests.
pub struct DebugApi<Eth: RpcNodeCore> {
    inner: Arc<DebugApiInner<Eth>>,
}

impl<Eth> DebugApi<Eth>
where
    Eth: RpcNodeCore,
{
    /// Create a new instance of the [`DebugApi`]
    pub fn new(
        eth_api: Eth,
        blocking_task_guard: BlockingTaskGuard,
        executor: &Runtime,
        mut stream: impl Stream<Item = ConsensusEngineEvent<Eth::Primitives>> + Send + Unpin + 'static,
    ) -> Self {
        let bad_block_store = BadBlockStore::default();
        let inner = Arc::new(DebugApiInner {
            eth_api,
            blocking_task_guard,
            task_spawner: executor.clone(),
            bad_block_store: bad_block_store.clone(),
        });

        // Spawn a task caching bad blocks
        executor.spawn_task(async move {
            while let Some(event) = stream.next().await {
                if let ConsensusEngineEvent::InvalidBlock { block, error } = event &&
                    let Ok(recovered) = RecoveredBlock::try_recover_sealed(*block)
                {
                    bad_block_store.insert(recovered, error);
                }
            }
        });

        Self { inner }
    }

    /// Access the underlying `Eth` API.
    pub fn eth_api(&self) -> &Eth {
        &self.inner.eth_api
    }

    /// Access the underlying provider.
    pub fn provider(&self) -> &Eth::Provider {
        self.inner.eth_api.provider()
    }
}

// === impl DebugApi ===

impl<Eth> DebugApi<Eth>
where
    Eth: TraceExt,
{
    /// Acquires a permit to execute a tracing call.
    async fn acquire_trace_permit(&self) -> Result<OwnedSemaphorePermit, AcquireError> {
        self.inner.blocking_task_guard.clone().acquire_owned().await
    }

    /// Traces every transaction of a block through the block executor, see
    /// [`Call::block_executor`](reth_rpc_eth_api::helpers::Call::block_executor).
    ///
    /// A supplied permit stays with the worker if the waiting request times out or is dropped.
    async fn trace_block(
        &self,
        block: Arc<RecoveredBlock<ProviderBlock<Eth::Provider>>>,
        evm_env: EvmEnvFor<Eth::Evm>,
        opts: GethDebugTracingOptions,
        permit: Option<OwnedSemaphorePermit>,
    ) -> Result<Vec<TraceResult>, Eth::Error> {
        self.eth_api()
            .spawn_with_state_at_block(block.parent_hash(), move |eth_api, mut db| {
                let _permit = permit;
                let mut results = Vec::with_capacity(block.body().transactions().len());

                eth_api.apply_pre_execution_changes(&block, &mut db)?;

                let block_env = *evm_env.block_env();

                let mut transactions = block.transactions_recovered().enumerate().peekable();
                let mut inspector = DebugInspector::new(opts).map_err(Eth::Error::from_eth_err)?;
                let mut executor = eth_api.block_executor(&block, &mut db, evm_env)?;
                while let Some((index, tx)) = transactions.next() {
                    if is_cancelled() {
                        return Err(EthApiError::InternalEthError.into())
                    }
                    let tx_hash = *tx.tx_hash();
                    let tx_env = eth_api.evm_config().tx_env(tx.cloned());

                    let (next_inspector, output) = executor
                        .inspect_transaction_without_commit(tx, inspector)
                        .map_err(Eth::Error::from_evm_err)?;
                    inspector = next_inspector;
                    let result = inspector
                        .get_result(
                            Some(TransactionContext {
                                block_hash: Some(block.hash()),
                                tx_hash: Some(tx_hash),
                                tx_index: Some(index),
                            }),
                            &tx_env,
                            &block_env,
                            output.result(),
                            executor.evm_mut().state_db_mut(),
                        )
                        .map_err(Eth::Error::from_eth_err)?;

                    results.push(TraceResult::Success { result, tx_hash: Some(tx_hash) });
                    if transactions.peek().is_some() {
                        inspector.fuse().map_err(Eth::Error::from_eth_err)?;
                        // need to apply the state changes of this transaction before executing the
                        // next transaction
                        executor.commit_transaction(output).map_err(Eth::Error::from_evm_err)?;
                    }
                }

                Ok(results)
            })
            .await
    }

    /// Traces a block for `traceChain`, preserving transaction-level tracing failures in the
    /// subscription result instead of failing the entire range.
    async fn trace_chain_block(
        &self,
        block: Arc<RecoveredBlock<ProviderBlock<Eth::Provider>>>,
        opts: GethDebugTracingOptions,
    ) -> Result<Vec<Option<TraceResult>>, Eth::Error> {
        let evm_env = self.eth_api().evm_env_for_header(block.sealed_block().sealed_header())?;

        self.eth_api()
            .spawn_with_state_at_block(block.parent_hash(), move |eth_api, mut db| {
                let tx_count = block.body().transactions().len();
                let mut results = Vec::with_capacity(tx_count);

                eth_api.apply_pre_execution_changes(&block, &mut db)?;

                let block_env = *evm_env.block_env();
                let mut transactions = block.transactions_recovered().enumerate().peekable();
                let mut inspector = match DebugInspector::new(opts) {
                    Ok(inspector) => inspector,
                    Err(err) => {
                        if let Some((_, tx)) = transactions.peek() {
                            results.push(Some(TraceResult::Error {
                                error: err.to_string(),
                                tx_hash: Some(*tx.tx_hash()),
                            }));
                        }
                        results.resize(tx_count, None);
                        return Ok(results)
                    }
                };
                let mut executor = eth_api.block_executor(&block, &mut db, evm_env)?;

                while let Some((index, tx)) = transactions.next() {
                    let tx_hash = *tx.tx_hash();
                    let tx_env = eth_api.evm_config().tx_env(tx.cloned());
                    let output = match executor.inspect_transaction_without_commit(tx, inspector) {
                        Ok((next_inspector, output)) => {
                            inspector = next_inspector;
                            output
                        }
                        Err(err) => {
                            results.push(Some(TraceResult::Error {
                                error: err.to_string(),
                                tx_hash: Some(tx_hash),
                            }));
                            break
                        }
                    };

                    let result = match inspector.get_result(
                        Some(TransactionContext {
                            block_hash: Some(block.hash()),
                            tx_hash: Some(tx_hash),
                            tx_index: Some(index),
                        }),
                        &tx_env,
                        &block_env,
                        output.result(),
                        executor.evm_mut().state_db_mut(),
                    ) {
                        Ok(result) => result,
                        Err(err) => {
                            results.push(Some(TraceResult::Error {
                                error: err.to_string(),
                                tx_hash: Some(tx_hash),
                            }));
                            break
                        }
                    };

                    results.push(Some(TraceResult::Success { result, tx_hash: Some(tx_hash) }));
                    if let Some((_, next_tx)) = transactions.peek() {
                        if let Err(err) = inspector.fuse() {
                            results.push(Some(TraceResult::Error {
                                error: err.to_string(),
                                tx_hash: Some(*next_tx.tx_hash()),
                            }));
                            break
                        }
                        if let Err(err) = executor.commit_transaction(output) {
                            results.push(Some(TraceResult::Error {
                                error: err.to_string(),
                                tx_hash: Some(*next_tx.tx_hash()),
                            }));
                            break
                        }
                    }
                }

                results.resize(tx_count, None);
                Ok(results)
            })
            .await
    }

    /// Replays the given block and returns the trace of each transaction.
    ///
    /// This expects a rlp encoded block
    ///
    /// Note, the parent of this block must be present, or it will fail.
    ///
    /// The block is supplied by the caller, so its replay is bounded by a timeout of 30 seconds: no
    /// transaction starts after it expires, the request then fails with
    /// [`EthApiError::ExecutionTimedOut`].
    /// The replay retains its tracing permit until the running transaction finishes.
    pub async fn debug_trace_raw_block(
        &self,
        rlp_block: Bytes,
        opts: GethDebugTracingOptions,
    ) -> Result<Vec<TraceResult>, Eth::Error> {
        let permit =
            self.acquire_trace_permit().await.map_err(|_| EthApiError::InternalEthError)?;
        let block: ProviderBlock<Eth::Provider> = Decodable::decode(&mut rlp_block.as_ref())
            .map_err(BlockError::RlpDecodeRawBlock)
            .map_err(Eth::Error::from_eth_err)?;

        let evm_env = self
            .eth_api()
            .evm_config()
            .evm_env(block.header())
            .map_err(RethError::other)
            .map_err(Eth::Error::from_eth_err)?;

        // Depending on EIP-2 we need to recover the transactions differently
        let senders =
            if self.provider().chain_spec().is_homestead_active_at_block(block.header().number()) {
                block.body().recover_signers()
            } else {
                block.body().recover_signers_unchecked()
            }
            .map_err(Eth::Error::from_eth_err)?;

        // Dropping the replay cancels it before its next transaction.
        tokio::time::timeout(
            RAW_BLOCK_TRACE_TIMEOUT,
            self.trace_block(
                Arc::new(block.into_recovered_with_signers(senders)),
                evm_env,
                opts,
                Some(permit),
            ),
        )
        .await
        .map_err(|_| {
            Eth::Error::from_eth_err(EthApiError::ExecutionTimedOut(RAW_BLOCK_TRACE_TIMEOUT))
        })?
    }

    /// Replays a block and returns the trace of each transaction.
    pub async fn debug_trace_block(
        &self,
        block_id: BlockId,
        opts: GethDebugTracingOptions,
    ) -> Result<Vec<TraceResult>, Eth::Error> {
        let block = self
            .eth_api()
            .recovered_block(block_id)
            .await?
            .ok_or(EthApiError::TracingBlockNotFound(block_id))?;
        // Tracing requires the parent state, which does not exist for the genesis block.
        if block.number() == 0 {
            return Err(EthApiError::GenesisNotTraceable.into())
        }
        let evm_env = self.eth_api().evm_env_for_header(block.sealed_block().sealed_header())?;

        self.trace_block(block, evm_env, opts, None).await
    }

    /// Trace the transaction according to the provided options.
    ///
    /// Ref: <https://geth.ethereum.org/docs/developers/evm-tracing/built-in-tracers>
    pub async fn debug_trace_transaction(
        &self,
        tx_hash: B256,
        opts: GethDebugTracingOptions,
    ) -> Result<GethTrace, Eth::Error> {
        let (transaction, block, bal) =
            match self.eth_api().transaction_and_block_and_maybe_bal(tx_hash).await? {
                None => return Err(EthApiError::TracingTransactionNotFound.into()),
                Some(res) => res,
            };

        let interrupt = JsTraceInterrupt::new(&opts)?;
        let tracer_interrupt = interrupt.clone();
        let trace = self.eth_api().spawn_with_state_at_block(
            block.parent_hash(),
            move |eth_api, mut db| {
                // configure env for the target transaction
                let (tx, tx_info) = transaction.split();
                let tx_hash = *tx.tx_hash();

                // index should always be available because `transaction_and_block` only
                // returns transactions included in a block
                let index =
                    tx_info.index.expect("transaction_and_block only returns block transactions")
                        as usize;

                let inspector = tracer_interrupt
                    .attach(DebugInspector::new(opts).map_err(Eth::Error::from_eth_err)?);
                let tx_env = eth_api.evm_config().tx_env(tx);
                let (mut inspector, res, evm_env) = eth_api.inspect_transaction_in_block(
                    &block,
                    &mut db,
                    inspector,
                    index,
                    bal.as_deref(),
                )?;
                let trace = inspector
                    .get_result(
                        Some(TransactionContext {
                            block_hash: Some(block.hash()),
                            tx_index: Some(index),
                            tx_hash: Some(tx_hash),
                        }),
                        &tx_env,
                        evm_env.block_env(),
                        &res,
                        &mut db,
                    )
                    .map_err(Eth::Error::from_eth_err)?;

                Ok(trace)
            },
        );
        interrupt.run(trace).await
    }

    /// The `debug_traceCall` method lets you run an `eth_call` within the context of the given
    /// block execution using the final state of parent block as the base.
    ///
    /// If `tx_index` is provided in opts, the call will be traced at the state after executing
    /// transactions up to the specified index within the block (0-indexed).
    /// If not provided, then uses the post-state (default behavior).
    ///
    /// Differences compare to `eth_call`:
    ///  - `debug_traceCall` executes with __enabled__ basefee check, `eth_call` does not: <https://github.com/paradigmxyz/reth/issues/6240>
    pub async fn debug_trace_call(
        &self,
        call: RpcTxReq<Eth::NetworkTypes>,
        block_id: Option<BlockId>,
        opts: GethDebugTracingCallOptions,
    ) -> Result<GethTrace, Eth::Error> {
        let at = block_id.unwrap_or_default();
        let GethDebugTracingCallOptions {
            tracing_options,
            state_overrides,
            block_overrides,
            tx_index,
        } = opts;
        let overrides = EvmOverrides::new(state_overrides, block_overrides.map(Box::new));

        // Check if we need to replay transactions for a specific tx_index
        if let Some(tx_idx) = tx_index {
            return self
                .debug_trace_call_at_tx_index(call, at, tx_idx as usize, tracing_options, overrides)
                .await;
        }

        let interrupt = JsTraceInterrupt::new(&tracing_options)?;
        let tracer_interrupt = interrupt.clone();
        let this = self.clone();
        let trace =
            self.eth_api().spawn_with_call_at(call, at, overrides, move |db, evm_env, tx_env| {
                let inspector = tracer_interrupt.attach(
                    DebugInspector::new(tracing_options).map_err(Eth::Error::from_eth_err)?,
                );
                let (mut inspector, res) =
                    this.eth_api().inspect(&mut *db, evm_env.clone(), &tx_env, inspector)?;
                let trace = inspector
                    .get_result(None, &tx_env, evm_env.block_env(), &res, db)
                    .map_err(Eth::Error::from_eth_err)?;
                Ok(trace)
            });
        interrupt.run(trace).await.map_err(|err| match err.as_err() {
            Some(EthApiError::HeaderNotFound(id)) if *id == at => {
                // Unknown blocks use -32000: https://github.com/ethereum/execution-apis/pull/855
                EthApiError::TracingBlockNotFound(at).into()
            }
            _ => err,
        })
    }

    /// Helper method to execute `debug_trace_call` at a specific transaction index within a block.
    /// This replays transactions up to the specified index, then executes the trace call in that
    /// state.
    async fn debug_trace_call_at_tx_index(
        &self,
        call: RpcTxReq<Eth::NetworkTypes>,
        block_id: BlockId,
        tx_index: usize,
        tracing_options: GethDebugTracingOptions,
        overrides: EvmOverrides,
    ) -> Result<GethTrace, Eth::Error> {
        // Get the target block to check transaction count
        let (block, bal) = self
            .eth_api()
            .recovered_block_and_maybe_bal(block_id)
            .await?
            .ok_or(EthApiError::TracingBlockNotFound(block_id))?;

        if tx_index >= block.transaction_count() {
            // tx_index out of bounds
            return Err(EthApiError::InvalidParams(format!(
                "tx_index {} out of bounds for block with {} transactions",
                tx_index,
                block.transaction_count()
            ))
            .into())
        }

        // state overrides commit changes to the database, which an attached BAL would take read
        // precedence over, so only position via BAL if there are none
        let bal = bal.filter(|_| !overrides.has_state());

        let evm_env = self.eth_api().evm_env_for_header(block.sealed_block().sealed_header())?;

        let interrupt = JsTraceInterrupt::new(&tracing_options)?;
        let tracer_interrupt = interrupt.clone();
        let trace = self.eth_api().spawn_with_state_at_block(
            block.parent_hash(),
            move |eth_api, mut db| {
                // 1. position the state before the transaction at the index
                eth_api.replay_block_until(&mut db, &block, tx_index, bal.as_deref())?;

                // 2. now execute the trace call on this state
                let (evm_env, tx_env) =
                    eth_api.prepare_call_env(evm_env, call, &mut db, overrides)?;

                let inspector = tracer_interrupt.attach(
                    DebugInspector::new(tracing_options).map_err(Eth::Error::from_eth_err)?,
                );
                let (mut inspector, res) =
                    eth_api.inspect(&mut db, evm_env.clone(), &tx_env, inspector)?;
                let trace = inspector
                    .get_result(None, &tx_env, evm_env.block_env(), &res, &mut db)
                    .map_err(Eth::Error::from_eth_err)?;

                Ok(trace)
            },
        );
        interrupt.run(trace).await
    }

    /// The `debug_traceCallMany` method lets you run an `eth_callMany` within the context of the
    /// given block execution using the first n transactions in the given block as base.
    /// Each following bundle increments block number by 1 and block timestamp by 12 seconds
    pub async fn debug_trace_call_many(
        &self,
        bundles: Vec<Bundle<RpcTxReq<Eth::NetworkTypes>>>,
        state_context: Option<StateContext>,
        opts: Option<GethDebugTracingCallOptions>,
    ) -> Result<Vec<Vec<GethTrace>>, Eth::Error> {
        if bundles.is_empty() {
            return Err(EthApiError::InvalidParams(String::from("bundles are empty.")).into())
        }
        let mut budget = CallManyBudget::for_bundles(&bundles, self.eth_api().call_gas_limit())?;

        let StateContext { transaction_index, block_number } = state_context.unwrap_or_default();
        let transaction_index = transaction_index.unwrap_or_default();

        let target_block = block_number.unwrap_or_default();
        let block = self
            .eth_api()
            .recovered_block(target_block)
            .await?
            .ok_or(EthApiError::HeaderNotFound(target_block))?;
        let mut evm_env =
            self.eth_api().evm_env_for_header(block.sealed_block().sealed_header())?;

        let opts = opts.unwrap_or_default();
        let GethDebugTracingCallOptions { tracing_options, mut state_overrides, .. } = opts;

        // we're essentially replaying the transactions in the block here, hence we need the state
        // that points to the beginning of the block, which is the state at the parent block
        let mut at = block.parent_hash();
        let mut replay_block_txs = true;

        // if a transaction index is provided, we need to replay the transactions until the index
        let num_txs =
            transaction_index.index().unwrap_or_else(|| block.body().transactions().len());
        // but if all transactions are to be replayed, we can use the state at the block itself
        // this works with the exception of the PENDING block, because its state might not exist if
        // built locally
        if !target_block.is_pending() && num_txs == block.body().transactions().len() {
            at = block.hash();
            replay_block_txs = false;
        }

        self.eth_api()
            .spawn_with_state_at_block(at, move |eth_api, mut db| {
                // the outer vec for the bundles
                let mut all_bundles = Vec::with_capacity(bundles.len());

                if replay_block_txs {
                    // only need to replay the transactions in the block if not all transactions are
                    // to be replayed
                    // Execute all transactions until index. No BAL positioning here: bundle
                    // transactions commit state on top, and an attached BAL would take read
                    // precedence over the committed changes
                    eth_api.replay_block_until(&mut db, &block, num_txs, None)?;
                }

                // Trace all bundles
                let mut bundles = bundles.into_iter().peekable();
                let mut inspector = DebugInspector::new(tracing_options.clone())
                    .map_err(Eth::Error::from_eth_err)?;
                while let Some(bundle) = bundles.next() {
                    let mut results = Vec::with_capacity(bundle.transactions.len());
                    let Bundle { transactions, block_override } = bundle;

                    let block_overrides = block_override.map(Box::new);

                    let mut transactions = transactions.into_iter().peekable();
                    while let Some(tx) = transactions.next() {
                        if is_cancelled() {
                            return Err(EthApiError::InternalEthError.into())
                        }
                        // apply state overrides only once, before the first transaction
                        let state_overrides = state_overrides.take();
                        let overrides = EvmOverrides::new(state_overrides, block_overrides.clone());

                        let (evm_env, mut tx_env) =
                            eth_api.prepare_call_env(evm_env.clone(), tx, &mut db, overrides)?;
                        budget.admit(&mut tx_env)?;

                        let (next_inspector, res) = eth_api
                            .inspect(&mut db, evm_env.clone(), &tx_env, inspector)
                            .map_err(|err| budget.map_call_err(err))?;
                        budget.charge(&res.result)?;
                        inspector = next_inspector;
                        let trace = inspector
                            .get_result(None, &tx_env, evm_env.block_env(), &res, &mut db)
                            .map_err(Eth::Error::from_eth_err)?;

                        // If there is more transactions, commit the database
                        // If there is no transactions, but more bundles, commit to the database too
                        if transactions.peek().is_some() || bundles.peek().is_some() {
                            inspector.fuse().map_err(Eth::Error::from_eth_err)?;
                            db.commit_source(&res.pending_state);
                        }
                        results.push(trace);
                    }
                    // Increment block_env number and timestamp for the next bundle
                    evm_env.block_env_mut().number += uint!(1_U256);
                    evm_env.block_env_mut().timestamp += uint!(12_U256);

                    all_bundles.push(results);
                }
                Ok(all_bundles)
            })
            .await
    }

    /// Generates an execution witness for the given block hash. see
    /// [`Self::debug_execution_witness`] for more info.
    pub async fn debug_execution_witness_by_block_hash(
        &self,
        hash: B256,
        mode: Option<ExecutionWitnessMode>,
    ) -> Result<ExecutionWitness, Eth::Error> {
        let this = self.clone();
        let block = this
            .eth_api()
            .recovered_block(hash.into())
            .await?
            .ok_or(EthApiError::HeaderNotFound(hash.into()))?;

        self.debug_execution_witness_for_block(block, mode.unwrap_or_default()).await
    }

    /// The `debug_executionWitness` method allows for re-execution of a block with the purpose of
    /// generating an execution witness. The witness comprises of a map of all hashed trie nodes to
    /// their preimages that were required during the execution of the block, including during state
    /// root recomputation.
    pub async fn debug_execution_witness(
        &self,
        block_id: BlockId,
        mode: Option<ExecutionWitnessMode>,
    ) -> Result<ExecutionWitness, Eth::Error> {
        let this = self.clone();
        let block = this
            .eth_api()
            .recovered_block(block_id)
            .await?
            .ok_or(EthApiError::HeaderNotFound(block_id))?;

        self.debug_execution_witness_for_block(block, mode.unwrap_or_default()).await
    }

    /// Generates an execution witness, using the given recovered block.
    pub async fn debug_execution_witness_for_block(
        &self,
        block: Arc<RecoveredBlock<ProviderBlock<Eth::Provider>>>,
        mode: ExecutionWitnessMode,
    ) -> Result<ExecutionWitness, Eth::Error> {
        let block_number = block.header().number();
        self.eth_api()
            .spawn_with_state_at_block(block.parent_hash(), move |eth_api, mut db| {
                let output = eth_api
                    .evm_config()
                    .executor(&mut db)
                    .execute(&block)
                    .map_err(|err| EthApiError::Internal(err.into()))?;
                let codes = ExecutionWitnessRecord::commit_execution(&mut db, &output.state);

                Ok(ExecutionWitnessRecord::new(&db)
                    .with_canonical_codes(codes)
                    .into_execution_witness(
                        &db.db.inner().0 .0,
                        eth_api.provider(),
                        block_number,
                        mode,
                    )
                    .map_err(EthApiError::from)?)
            })
            .await
    }

    /// Returns account information, including the storage root, after replaying the block through
    /// the transaction at the given index.
    pub async fn debug_account_at(
        &self,
        block_id: BlockId,
        tx_index: Index,
        address: Address,
    ) -> Result<Option<Account>, Eth::Error> {
        self.replay_block_until(block_id, tx_index, move |db| Self::account(db, address))
            .await
            .map(Option::flatten)
    }

    /// Returns account information after replaying the block through the transaction at the given
    /// index.
    pub async fn debug_account_info_at(
        &self,
        block_id: BlockId,
        tx_index: Index,
        address: Address,
    ) -> Result<Option<AccountInfo>, Eth::Error> {
        self.replay_block_until(block_id, tx_index, move |db| Self::account_info(db, address)).await
    }

    /// Replays a block through the transaction at the given index and calls `f` with the resulting
    /// state.
    async fn replay_block_until<F, R>(
        &self,
        block_id: BlockId,
        tx_index: Index,
        f: F,
    ) -> Result<Option<R>, Eth::Error>
    where
        F: FnOnce(&mut StateCacheDb) -> Result<R, Eth::Error> + Send + 'static,
        R: Send + 'static,
    {
        let block = self
            .eth_api()
            .recovered_block(block_id)
            .await?
            .ok_or(EthApiError::HeaderNotFound(block_id))?;
        let tx_index = usize::from(tx_index);
        let transaction_count = block.transaction_count();
        if tx_index >= transaction_count {
            return Err(EthApiError::InvalidParams(format!(
                "tx_index {tx_index} out of bounds for block with {transaction_count} transactions"
            ))
            .into())
        }

        self.eth_api()
            .spawn_with_state_at_block(block.parent_hash(), move |eth_api, mut db| {
                eth_api.replay_block_until(&mut db, &block, tx_index + 1, None)?;

                f(&mut db)
            })
            .await
            .map(Some)
    }

    /// Retrieves the account's balance, nonce, code hash, and storage root from the given state.
    fn account(db: &mut StateCacheDb, address: Address) -> Result<Option<Account>, Eth::Error> {
        let account = db.get_account(&address).map_err(Eth::Error::from_eth_err)?;
        let Some(account) = account else { return Ok(None) };

        let storage = db.cache.storage.get(&address);
        let hashed_storage = storage.map_or_else(HashedStorage::default, |s| {
            HashedStorage::from_plain_storage(s.slots.iter())
        });
        let storage_root = if storage.is_some_and(|s| s.wiped) {
            storage_root_unsorted(
                hashed_storage.storage.into_iter().filter(|(_, value)| !value.is_zero()),
            )
        } else {
            db.db.inner().storage_root(address, hashed_storage).map_err(Eth::Error::from_eth_err)?
        };

        Ok(Some(reth_primitives_traits::Account::from(&account).into_trie_account(storage_root)))
    }

    /// Retrieves the account's balance, nonce, and code from the given state.
    #[allow(clippy::needless_update)]
    fn account_info(db: &mut StateCacheDb, address: Address) -> Result<AccountInfo, Eth::Error> {
        let account =
            db.get_account(&address).map_err(Eth::Error::from_eth_err)?.unwrap_or_default();
        let code = if account.is_empty_code_hash() {
            Default::default()
        } else if let Some(code) = account.code {
            code.original_bytes()
        } else {
            db.get_code_by_hash(&account.code_hash)
                .map_err(Eth::Error::from_eth_err)?
                .original_bytes()
        };

        Ok(AccountInfo {
            balance: account.balance,
            nonce: account.nonce,
            code,
            #[cfg(feature = "account-ext")]
            extension: reth_primitives_traits::AccountExtension::from_shared(
                account.extension.into_shared(),
            ),
            ..Default::default()
        })
    }

    /// Returns the code associated with a given hash at the specified block ID. If no code is
    /// found, it returns None. If no block ID is provided, it defaults to the latest block.
    pub async fn debug_code_by_hash(
        &self,
        hash: B256,
        block_id: Option<BlockId>,
    ) -> Result<Option<Bytes>, Eth::Error> {
        Ok(self
            .provider()
            .state_by_block_id(block_id.unwrap_or_default())
            .map_err(Eth::Error::from_eth_err)?
            .bytecode_by_hash(&hash)
            .map_err(Eth::Error::from_eth_err)?
            .map(|b| b.original_bytes()))
    }

    /// Returns the state root of the `HashedPostState` on top of the state for the given block with
    /// trie updates.
    async fn debug_state_root_with_updates(
        &self,
        hashed_state: HashedPostState,
        block_id: Option<BlockId>,
    ) -> Result<(B256, TrieUpdates), Eth::Error> {
        self.inner
            .eth_api
            .spawn_blocking_io(move |this| {
                let state = this
                    .provider()
                    .state_by_block_id(block_id.unwrap_or_default())
                    .map_err(Eth::Error::from_eth_err)?;
                state.state_root_with_updates(hashed_state).map_err(Eth::Error::from_eth_err)
            })
            .await
    }

    /// Executes a block and returns the state root after each transaction.
    ///
    /// The block is either canonical or a bad block the engine rejected. Execution stops at the
    /// first transaction that is invalid on top of its predecessors, so the result for a bad block
    /// holds the roots before that transaction.
    pub async fn intermediate_roots(&self, block_hash: B256) -> Result<Vec<B256>, Eth::Error> {
        let block = match self.eth_api().recovered_block(block_hash.into()).await? {
            Some(block) => block,
            None => {
                self.inner
                    .bad_block_store
                    .get(block_hash)
                    .ok_or(EthApiError::HeaderNotFound(block_hash.into()))?
                    .block
            }
        };
        let evm_env = self.eth_api().evm_env_for_header(block.sealed_block().sealed_header())?;

        self.eth_api()
            .spawn_with_state_at_block(block.parent_hash(), move |eth_api, mut db| {
                let tx_count = block.body().transactions().len();
                let mut state = eth_api.apply_pre_execution_changes(&block, &mut db)?;

                // the executor borrows the database, so collect each transaction's changes first
                // and compute the roots once it is dropped
                let mut transaction_states = Vec::with_capacity(tx_count);
                let mut executor = eth_api.block_executor(&block, &mut db, evm_env)?;
                for tx in block.transactions_recovered() {
                    if is_cancelled() {
                        return Err(EthApiError::InternalEthError.into())
                    }
                    let output = match executor.execute_transaction_without_commit(tx) {
                        Ok(output) => output,
                        Err(BlockExecutionError::Validation(_)) => break,
                        Err(err) => return Err(Eth::Error::from_evm_err(err)),
                    };
                    let mut changes = reth_execution_types::TransactionChanges::default();
                    let Ok(()) = output.result().pending_state.visit(&mut changes);
                    let mut transaction_state = reth_execution_types::BlockState::default();
                    transaction_state.commit(&changes);
                    transaction_states.push(transaction_state.into_bundle());
                    executor.commit_transaction(output).map_err(Eth::Error::from_evm_err)?;
                }
                drop(executor);

                let mut roots = Vec::with_capacity(tx_count);
                for transaction_state in transaction_states {
                    if is_cancelled() {
                        return Err(EthApiError::InternalEthError.into())
                    }
                    state.extend(transaction_state);
                    let hashed_state = db
                        .db
                        .inner()
                        .hashed_post_state(&state)
                        .map_err(Eth::Error::from_eth_err)?;
                    let root =
                        db.db.inner().state_root(hashed_state).map_err(Eth::Error::from_eth_err)?;
                    roots.push(root);
                }

                Ok(roots)
            })
            .await
    }
}

impl<Eth> DebugApiServer<RpcTxReq<Eth::NetworkTypes>> for DebugApi<Eth>
where
    Eth: EthTransactions + TraceExt,
{
    /// Handler for `debug_getRawHeader`
    async fn raw_header(&self, block_id: BlockId) -> RpcResult<Bytes> {
        let header = match block_id {
            BlockId::Hash(hash) => self.provider().header(hash.into()).to_rpc_result()?,
            BlockId::Number(number_or_tag) => {
                let number = self
                    .provider()
                    .convert_block_number(number_or_tag)
                    .to_rpc_result()?
                    .ok_or(EthApiError::HeaderNotFound(block_id))?;
                self.provider().header_by_number(number).to_rpc_result()?
            }
        }
        .ok_or(EthApiError::HeaderNotFound(block_id))?;

        let mut res = Vec::new();
        header.encode(&mut res);
        Ok(res.into())
    }

    /// Handler for `debug_getRawBlock`
    async fn raw_block(&self, block_id: BlockId) -> RpcResult<Bytes> {
        let block = self
            .provider()
            .block_by_id(block_id)
            .to_rpc_result()?
            .ok_or(EthApiError::HeaderNotFound(block_id))?;
        let mut res = Vec::new();
        block.encode(&mut res);
        Ok(res.into())
    }

    /// Handler for `debug_getRawBlockAccessList`
    async fn raw_block_access_list(&self, block_id: BlockId) -> RpcResult<Bytes> {
        self.eth_api()
            .get_raw_block_access_list(block_id)
            .await
            .map_err(Into::into)?
            .ok_or_else(|| EthApiError::HeaderNotFound(block_id).into())
    }

    /// Handler for `debug_getRawTransaction`
    ///
    /// If this is a pooled EIP-4844 transaction, the blob sidecar is included.
    ///
    /// Returns the bytes of the transaction for the given hash.
    async fn raw_transaction(&self, hash: B256) -> RpcResult<Option<Bytes>> {
        self.eth_api().raw_transaction_by_hash(hash).await.map_err(Into::into)
    }

    /// Handler for `debug_getRawTransactions`
    /// Returns the bytes of the transaction for the given hash.
    async fn raw_transactions(&self, block_id: BlockId) -> RpcResult<Vec<Bytes>> {
        let block: RecoveredBlock<BlockTy<Eth::Primitives>> = self
            .provider()
            .block_with_senders_by_id(block_id, TransactionVariant::NoHash)
            .to_rpc_result()?
            .unwrap_or_default();
        Ok(block.into_transactions_recovered().map(|tx| tx.encoded_2718().into()).collect())
    }

    /// Handler for `debug_getRawReceipts`
    async fn raw_receipts(&self, block_id: BlockId) -> RpcResult<Vec<Bytes>> {
        Ok(self
            .provider()
            .receipts_by_block_id(block_id)
            .to_rpc_result()?
            .ok_or(EthApiError::HeaderNotFound(block_id))?
            .into_iter()
            .map(|receipt| ReceiptWithBloom::from(receipt).encoded_2718().into())
            .collect())
    }

    /// Handler for `debug_getBadBlocks`
    async fn bad_blocks(&self) -> RpcResult<Vec<serde_json::Value>> {
        let entries = self.inner.bad_block_store.all();
        let mut bad_blocks = Vec::with_capacity(entries.len());

        #[derive(Serialize, Deserialize)]
        struct BadBlockSerde<T> {
            block: T,
            hash: B256,
            rlp: Bytes,
            reason: String,
        }

        for entry in entries {
            let rlp = Bytes::from(alloy_rlp::encode(entry.block.sealed_block()));
            let hash = entry.block.hash();

            let block = entry
                .block
                .clone_into_rpc_block(
                    BlockTransactionsKind::Full,
                    |tx, tx_info| self.eth_api().converter().fill(tx, tx_info),
                    |header, block_size| {
                        self.eth_api().converter().convert_header(header, Some(block_size))
                    },
                )
                .map_err(|err| Eth::Error::from(err).into())?;

            let bad_block =
                serde_json::to_value(BadBlockSerde { block, hash, rlp, reason: entry.reason })
                    .map_err(|err| EthApiError::other(internal_rpc_err(err.to_string())))?;

            bad_blocks.push(bad_block);
        }

        Ok(bad_blocks)
    }

    /// Handler for `debug_clearTxpool`
    async fn debug_clear_txpool(&self) -> RpcResult<()> {
        let pool = self.eth_api().pool();
        let all_hashes = pool.all_transaction_hashes();
        let _ = pool.remove_transactions(all_hashes);
        Ok(())
    }

    /// Handler for `debug_subscribe("traceChain", ...)`.
    async fn debug_subscribe(
        &self,
        pending: PendingSubscriptionSink,
        subscription: String,
        start_exclusive: BlockNumberOrTag,
        end_inclusive: BlockNumberOrTag,
        opts: Option<GethDebugTracingOptions>,
    ) -> jsonrpsee::core::SubscriptionResult {
        if subscription != "traceChain" {
            pending
                .reject(EthApiError::InvalidParams(format!(
                    "unsupported debug subscription: {subscription}"
                )))
                .await;
            return Ok(())
        }

        let start_id = BlockId::Number(start_exclusive);
        let start = match self.eth_api().recovered_block(start_id).await {
            Ok(Some(block)) => block,
            Ok(None) => {
                pending.reject(EthApiError::TracingBlockNotFound(start_id)).await;
                return Ok(())
            }
            Err(err) => {
                pending.reject(err).await;
                return Ok(())
            }
        };
        let end_id = BlockId::Number(end_inclusive);
        let end = match self.eth_api().recovered_block(end_id).await {
            Ok(Some(block)) => block,
            Ok(None) => {
                pending.reject(EthApiError::TracingBlockNotFound(end_id)).await;
                return Ok(())
            }
            Err(err) => {
                pending.reject(err).await;
                return Ok(())
            }
        };

        if start.number() >= end.number() {
            pending
                .reject(EthApiError::InvalidParams(format!(
                    "end block (#{}) needs to come after start block (#{})",
                    end.number(),
                    start.number()
                )))
                .await;
            return Ok(())
        }

        let sink = pending.accept().await?;
        let this = self.clone();
        let task_spawner = self.inner.task_spawner.clone();
        task_spawner.spawn_task(async move {
            let end_number = end.number();
            let opts = opts.unwrap_or_default();

            for number in (start.number() + 1)..=end_number {
                if sink.is_closed() {
                    break
                }

                let block_id = BlockId::number(number);
                let block = match this.eth_api().recovered_block(block_id).await {
                    Ok(Some(block)) => block,
                    Ok(None) => {
                        tracing::warn!(target: "rpc::debug", %number, "Chain tracing block not found");
                        break
                    }
                    Err(err) => {
                        tracing::warn!(target: "rpc::debug", %number, %err, "Failed to load chain tracing block");
                        break
                    }
                };
                let permit = tokio::select! {
                    _ = sink.closed() => break,
                    permit = this.acquire_trace_permit() => match permit {
                        Ok(permit) => permit,
                        Err(err) => {
                            tracing::debug!(target: "rpc::debug", %err, "Failed to acquire trace permit");
                            break
                        }
                    }
                };
                if sink.is_closed() {
                    break
                }
                let traces = match this.trace_chain_block(block.clone(), opts.clone()).await {
                    Ok(traces) => traces,
                    Err(err) => {
                        tracing::warn!(target: "rpc::debug", %number, %err, "Failed to trace chain block");
                        break
                    }
                };
                drop(permit);

                if traces.is_empty() && number != end_number {
                    continue
                }
                let result = ChainBlockTraceResult {
                    block: U256::from(number),
                    hash: block.hash(),
                    traces,
                };
                let message = match SubscriptionMessage::new(
                    sink.method_name(),
                    sink.subscription_id(),
                    &result,
                ) {
                    Ok(message) => message,
                    Err(err) => {
                        tracing::warn!(target: "rpc::debug", %number, %err, "Failed to serialize chain trace");
                        break
                    }
                };
                if sink.send(message).await.is_err() {
                    break
                }
            }

            // Dropping jsonrpsee's final sink unregisters the subscription without notifying the
            // client that this finite stream completed. Retain it so the subscription remains
            // explicitly unsubscribable until the client unsubscribes or disconnects.
            sink.closed().await;
        });

        Ok(())
    }

    /// Handler for `debug_traceBlock`
    async fn debug_trace_block(
        &self,
        rlp_block: Bytes,
        opts: Option<GethDebugTracingOptions>,
    ) -> RpcResult<Vec<TraceResult>> {
        Self::debug_trace_raw_block(self, rlp_block, opts.unwrap_or_default())
            .await
            .map_err(Into::into)
    }

    /// Handler for `debug_traceBlockByHash`
    async fn debug_trace_block_by_hash(
        &self,
        block: B256,
        opts: Option<GethDebugTracingOptions>,
    ) -> RpcResult<Vec<TraceResult>> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_trace_block(self, block.into(), opts.unwrap_or_default())
            .await
            .map_err(Into::into)
    }

    /// Handler for `debug_traceBlockByNumber`
    async fn debug_trace_block_by_number(
        &self,
        block: BlockNumberOrTag,
        opts: Option<GethDebugTracingOptions>,
    ) -> RpcResult<Vec<TraceResult>> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_trace_block(self, block.into(), opts.unwrap_or_default())
            .await
            .map_err(Into::into)
    }

    /// Handler for `debug_traceTransaction`
    async fn debug_trace_transaction(
        &self,
        tx_hash: B256,
        opts: Option<GethDebugTracingOptions>,
    ) -> RpcResult<GethTrace> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_trace_transaction(self, tx_hash, opts.unwrap_or_default())
            .await
            .map_err(Into::into)
    }

    /// Handler for `debug_traceCall`
    async fn debug_trace_call(
        &self,
        request: RpcTxReq<Eth::NetworkTypes>,
        block_id: Option<BlockId>,
        opts: Option<GethDebugTracingCallOptions>,
    ) -> RpcResult<GethTrace> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_trace_call(self, request, block_id, opts.unwrap_or_default())
            .await
            .map_err(Into::into)
    }

    async fn debug_trace_call_many(
        &self,
        bundles: Vec<Bundle<RpcTxReq<Eth::NetworkTypes>>>,
        state_context: Option<StateContext>,
        opts: Option<GethDebugTracingCallOptions>,
    ) -> RpcResult<Vec<Vec<GethTrace>>> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_trace_call_many(self, bundles, state_context, opts).await.map_err(Into::into)
    }

    /// Handler for `debug_executionWitness`
    async fn debug_execution_witness(
        &self,
        block: BlockId,
        mode: Option<ExecutionWitnessMode>,
    ) -> RpcResult<ExecutionWitness> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_execution_witness(self, block, mode).await.map_err(Into::into)
    }

    /// Handler for `debug_executionWitnessByBlockHash`
    async fn debug_execution_witness_by_block_hash(
        &self,
        hash: B256,
        mode: Option<ExecutionWitnessMode>,
    ) -> RpcResult<ExecutionWitness> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_execution_witness_by_block_hash(self, hash, mode).await.map_err(Into::into)
    }

    /// Handler for `debug_accountAt`
    async fn debug_account_at(
        &self,
        block_id: BlockId,
        tx_index: Index,
        address: Address,
    ) -> RpcResult<Option<Account>> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_account_at(self, block_id, tx_index, address).await.map_err(Into::into)
    }

    /// Handler for `debug_accountInfoAt`
    async fn debug_account_info_at(
        &self,
        block_id: BlockId,
        tx_index: Index,
        address: Address,
    ) -> RpcResult<Option<AccountInfo>> {
        let _permit = self.acquire_trace_permit().await;
        Self::debug_account_info_at(self, block_id, tx_index, address).await.map_err(Into::into)
    }

    async fn debug_account_range(
        &self,
        _block_number: BlockNumberOrTag,
        _start: Bytes,
        _max_results: u64,
        _nocode: bool,
        _nostorage: bool,
        _incompletes: bool,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_chaindb_compact(&self) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_chain_config(&self) -> RpcResult<ChainConfig> {
        Ok(crate::chain_config::chain_config(&*self.provider().chain_spec()))
    }

    async fn debug_chaindb_property(&self, _property: String) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_code_by_hash(
        &self,
        hash: B256,
        block_id: Option<BlockId>,
    ) -> RpcResult<Option<Bytes>> {
        Self::debug_code_by_hash(self, hash, block_id).await.map_err(Into::into)
    }

    async fn debug_db_ancient(&self, _kind: String, _number: u64) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_db_ancients(&self) -> RpcResult<()> {
        Ok(())
    }

    /// `debug_db_get` - database key lookup
    ///
    /// Currently supported:
    /// * Contract bytecode associated with a code hash. The key format is: `<0x63><code_hash>`
    ///     * Prefix byte: 0x63 (required)
    ///     * Code hash: 32 bytes
    ///
    ///   Must be provided as either:
    ///     * Hex string: "0x63..." (66 hex characters after 0x)
    ///     * Raw byte string: raw byte string (33 bytes)
    ///
    ///   See Geth impl: <https://github.com/ethereum/go-ethereum/blob/737ffd1bf0cbee378d0111a5b17ae4724fb2216c/core/rawdb/schema.go#L120>
    async fn debug_db_get(&self, key: String) -> RpcResult<Option<Bytes>> {
        let key_bytes = if key.starts_with("0x") {
            decode(&key).map_err(|_| EthApiError::InvalidParams("Invalid hex key".to_string()))?
        } else {
            key.into_bytes()
        };

        if key_bytes.len() != 33 {
            return Err(EthApiError::InvalidParams(format!(
                "Key must be 33 bytes, got {}",
                key_bytes.len()
            ))
            .into());
        }
        if key_bytes[0] != 0x63 {
            return Err(EthApiError::InvalidParams("Key prefix must be 0x63".to_string()).into());
        }

        let code_hash = B256::from_slice(&key_bytes[1..33]);

        // No block ID is provided, so it defaults to the latest block
        self.debug_code_by_hash(code_hash, None).await.map_err(Into::into)
    }

    async fn debug_dump_block(&self, _number: BlockId) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_free_os_memory(&self) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_gc_stats(&self) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_get_accessible_state(
        &self,
        _from: BlockNumberOrTag,
        _to: BlockNumberOrTag,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_get_modified_accounts_by_hash(
        &self,
        _start_hash: B256,
        _end_hash: B256,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_get_modified_accounts_by_number(
        &self,
        _start_number: u64,
        _end_number: u64,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_intermediate_roots(
        &self,
        block_hash: B256,
        _opts: Option<GethDebugTracingCallOptions>,
    ) -> RpcResult<Vec<B256>> {
        let _permit = self.acquire_trace_permit().await;
        self.intermediate_roots(block_hash).await.map_err(Into::into)
    }

    async fn debug_mem_stats(&self) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_preimage(&self, _hash: B256) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_print_block(&self, _number: u64) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_seed_hash(&self, _number: u64) -> RpcResult<B256> {
        Ok(Default::default())
    }

    async fn debug_set_gc_percent(&self, _v: i32) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_set_head(&self, _number: U64) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_set_trie_flush_interval(&self, _interval: String) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_standard_trace_bad_block_to_file(
        &self,
        _block: BlockNumberOrTag,
        _opts: Option<GethDebugTracingCallOptions>,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_standard_trace_block_to_file(
        &self,
        _block: BlockNumberOrTag,
        _opts: Option<GethDebugTracingCallOptions>,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_state_root_with_updates(
        &self,
        hashed_state: HashedPostState,
        block_id: Option<BlockId>,
    ) -> RpcResult<(B256, TrieUpdates)> {
        Self::debug_state_root_with_updates(self, hashed_state, block_id).await.map_err(Into::into)
    }

    async fn debug_storage_range_at(
        &self,
        _block_hash: B256,
        _tx_idx: usize,
        _contract_address: Address,
        _key_start: B256,
        _max_result: u64,
    ) -> RpcResult<()> {
        Ok(())
    }

    async fn debug_trace_bad_block(
        &self,
        block_hash: B256,
        opts: Option<GethDebugTracingCallOptions>,
    ) -> RpcResult<Vec<TraceResult>> {
        let _permit = self.acquire_trace_permit().await;
        let entry = self
            .inner
            .bad_block_store
            .get(block_hash)
            .ok_or_else(|| internal_rpc_err("bad block not found in cache"))?;

        let evm_env = self
            .eth_api()
            .evm_config()
            .evm_env(entry.block.header())
            .map_err(RethError::other)
            .to_rpc_result()?;

        let opts = opts.map(|o| o.tracing_options).unwrap_or_default();
        self.trace_block(entry.block.clone(), evm_env, opts, None).await.map_err(Into::into)
    }
}

impl<Eth: RpcNodeCore> std::fmt::Debug for DebugApi<Eth> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DebugApi").finish_non_exhaustive()
    }
}

impl<Eth: RpcNodeCore> Clone for DebugApi<Eth> {
    fn clone(&self) -> Self {
        Self { inner: Arc::clone(&self.inner) }
    }
}

struct DebugApiInner<Eth: RpcNodeCore> {
    /// The implementation of `eth` API
    eth_api: Eth,
    // restrict the number of concurrent calls to blocking calls
    blocking_task_guard: BlockingTaskGuard,
    /// Spawns long-running subscription tasks.
    task_spawner: Runtime,
    /// Cache for bad blocks.
    bad_block_store: BadBlockStore<BlockTy<Eth::Primitives>>,
}

/// A bounded, deduplicating store of recently observed bad blocks.
#[derive(Clone, Debug)]
struct BadBlockStore<B: BlockTrait> {
    inner: Arc<RwLock<VecDeque<BadBlockEntry<B>>>>,
    limit: usize,
}

/// A cached bad block paired with the reason it was rejected.
#[derive(Clone, Debug)]
struct BadBlockEntry<B: BlockTrait> {
    block: Arc<RecoveredBlock<B>>,
    reason: String,
}

impl<B: BlockTrait> BadBlockStore<B> {
    /// Creates a new store with the given capacity.
    fn new(limit: usize) -> Self {
        Self { inner: Arc::new(RwLock::new(VecDeque::with_capacity(limit))), limit }
    }

    /// Inserts a recovered block with its rejection reason, keeping only the most recent `limit`
    /// entries and deduplicating by block hash.
    fn insert(&self, block: RecoveredBlock<B>, reason: String) {
        let hash = block.hash();
        let mut guard = self.inner.write();

        // skip if we already recorded this bad block , and keep original ordering
        if guard.iter().any(|entry| entry.block.hash() == hash) {
            return;
        }
        guard.push_back(BadBlockEntry { block: Arc::new(block), reason });

        while guard.len() > self.limit {
            guard.pop_front();
        }
    }

    /// Returns all cached bad block entries ordered from newest to oldest.
    fn all(&self) -> Vec<BadBlockEntry<B>> {
        let guard = self.inner.read();
        guard.iter().rev().cloned().collect()
    }

    /// Returns the bad block entry with the given hash, if cached.
    fn get(&self, hash: B256) -> Option<BadBlockEntry<B>> {
        let guard = self.inner.read();
        guard.iter().find(|entry| entry.block.hash() == hash).cloned()
    }
}

impl<B: BlockTrait> Default for BadBlockStore<B> {
    fn default() -> Self {
        Self::new(64)
    }
}

/// Timeout of tracing a block supplied by the caller, whose work no consensus rule bounds.
const RAW_BLOCK_TRACE_TIMEOUT: Duration = Duration::from_secs(30);

/// Default timeout of a JS tracer, matching geth.
#[cfg(feature = "js-tracer")]
const DEFAULT_JS_TRACER_TIMEOUT: Duration = Duration::from_secs(5);

/// Returns whether `tracer` is a JS tracer or a mux tracer that holds one, at any depth.
#[cfg(feature = "js-tracer")]
fn runs_js_tracer(tracer: &GethDebugTracerType, config: &GethDebugTracerConfig) -> bool {
    match tracer {
        GethDebugTracerType::JsTracer(_) => true,
        GethDebugTracerType::BuiltInTracer(GethDebugBuiltInTracerType::MuxTracer) => {
            config.clone().into_mux_config().is_ok_and(|mux| {
                mux.0.iter().any(|(tracer, config)| {
                    runs_js_tracer(tracer, &config.clone().unwrap_or_default())
                })
            })
        }
        GethDebugTracerType::BuiltInTracer(_) => false,
    }
}

/// Interrupts a JS tracer once its request is dropped or its timeout expires.
///
/// Built-in tracers can't be interrupted, so for them this does nothing.
#[derive(Clone, Debug, Default)]
struct JsTraceInterrupt {
    #[cfg(feature = "js-tracer")]
    js: Option<(JsInspectorInterrupt, Duration)>,
}

impl JsTraceInterrupt {
    /// Creates an interrupt if `opts` selects a JS tracer, using its `timeout` or geth's default.
    #[cfg_attr(
        not(feature = "js-tracer"),
        expect(unused_variables, reason = "only JS tracers can be interrupted")
    )]
    fn new(opts: &GethDebugTracingOptions) -> Result<Self, EthApiError> {
        #[cfg(feature = "js-tracer")]
        if let Some(tracer) = &opts.tracer &&
            runs_js_tracer(tracer, &opts.tracer_config)
        {
            let timeout = match &opts.timeout {
                Some(timeout) => humantime::parse_duration(timeout).map_err(|err| {
                    EthApiError::InvalidParams(format!("invalid tracer timeout {timeout:?}: {err}"))
                })?,
                None => DEFAULT_JS_TRACER_TIMEOUT,
            };
            return Ok(Self { js: Some((JsInspectorInterrupt::new(), timeout)) })
        }
        Ok(Self::default())
    }

    /// Lets this interrupt stop the JS tracers of `inspector`, including those in a mux tracer.
    #[cfg_attr(
        not(feature = "js-tracer"),
        expect(clippy::missing_const_for_fn, reason = "only JS tracers can be interrupted")
    )]
    fn attach(&self, mut inspector: DebugInspector) -> DebugInspector {
        #[cfg(feature = "js-tracer")]
        if let Some((interrupt, _)) = &self.js {
            inspector.set_js_interrupt(interrupt);
        }
        inspector
    }

    /// Awaits `trace`, interrupting the tracer if the returned future is dropped or the timeout
    /// expires.
    ///
    /// The interrupt is armed before the future is first polled, so dropping it unpolled still
    /// stops a trace that is already running on a blocking task.
    #[cfg_attr(
        not(feature = "js-tracer"),
        expect(
            clippy::manual_async_fn,
            clippy::redundant_async_block,
            reason = "only JS tracers can be interrupted"
        )
    )]
    fn run<T, E: FromEthApiError>(
        &self,
        trace: impl Future<Output = Result<T, E>>,
    ) -> impl Future<Output = Result<T, E>> {
        #[cfg(feature = "js-tracer")]
        let js = self.js.as_ref().map(|(interrupt, timeout)| (interrupt.drop_guard(), *timeout));
        async move {
            #[cfg(feature = "js-tracer")]
            if let Some((_interrupt_on_drop, timeout)) = js {
                return tokio::time::timeout(timeout, trace)
                    .await
                    .map_err(|_| E::from_eth_err(EthApiError::ExecutionTimedOut(timeout)))?
            }
            trace.await
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{eth::helpers::types::EthRpcConverter, EthApi};
    use alloy_genesis::{Genesis, GenesisAccount};
    use alloy_primitives::{keccak256, U256};
    use evm2::{bytecode::Bytecode, evm::AccountInfo};
    use reth_chainspec::{ChainSpec, ChainSpecBuilder};
    use reth_db_api::{tables, transaction::DbTxMut};
    use reth_db_common::init::init_genesis;
    use reth_evm_ethereum::EthEvmConfig;
    use reth_network_api::noop::NoopNetwork;
    use reth_primitives_traits::StorageEntry;
    use reth_provider::test_utils::{
        create_test_provider_factory, create_test_provider_factory_with_chain_spec, NoopProvider,
    };
    use reth_rpc_eth_api::EthApiServer;
    use reth_transaction_pool::test_utils::testing_pool;
    use reth_trie_common::bundle_state::{AccountStatus, BundleAccount, BundleState, StorageSlot};

    #[tokio::test]
    async fn trace_call_out_of_range_block_error() {
        let eth_api = EthApi::<_, EthRpcConverter<ChainSpec>>::builder(
            NoopProvider::default(),
            testing_pool(),
            NoopNetwork::default(),
            EthEvmConfig::mainnet(),
        )
        .build();
        let debug_api = DebugApi::new(
            eth_api.clone(),
            BlockingTaskGuard::new(1),
            &Runtime::test(),
            futures::stream::empty(),
        );
        let block_id = BlockId::number(0xfffffffff);
        for tx_index in [None, Some(0)] {
            let mut opts: GethDebugTracingCallOptions =
                serde_json::from_value(serde_json::json!({ "tracer": "callTracer" })).unwrap();
            opts.tx_index = tx_index;
            let err = DebugApiServer::debug_trace_call(
                &debug_api,
                Default::default(),
                Some(block_id),
                Some(opts),
            )
            .await
            .unwrap_err();
            assert_eq!(err.code(), -32000);
            assert!(err.message().contains("not found"));
        }

        let err = EthApiServer::call(&eth_api, Default::default(), Some(block_id), None, None)
            .await
            .unwrap_err();
        assert_eq!(err.code(), -32001);
    }

    #[cfg(feature = "js-tracer")]
    fn js_tracer_opts(timeout: Option<&str>) -> GethDebugTracingOptions {
        GethDebugTracingOptions {
            tracer: Some(GethDebugTracerType::JsTracer(
                "{ fault: function() {}, result: function() { return null; } }".into(),
            )),
            timeout: timeout.map(Into::into),
            ..Default::default()
        }
    }

    #[cfg(feature = "js-tracer")]
    #[test]
    fn js_trace_interrupt_stops_attached_js_tracer() {
        let opts = js_tracer_opts(None);
        let interrupt = JsTraceInterrupt::new(&opts).unwrap();
        let inspector = interrupt.attach(DebugInspector::new(opts).unwrap());
        interrupt.js.as_ref().unwrap().0.interrupt();

        let config = EthEvmConfig::mainnet();
        let mut evm = config.evm_with_env(
            evm2::evm::CacheDB::<evm2::evm::EmptyDB>::default(),
            reth_evm_ethereum::EthEvmEnv::new(evm2::SpecId::CANCUN, Default::default(), 1),
        );
        let tx = alloy_consensus::transaction::Recovered::new_unchecked(
            evm2::ethereum::TxEnvelope::Legacy(alloy_consensus::TxLegacy {
                gas_limit: 1_000_000,
                to: alloy_primitives::TxKind::Call(Address::with_last_byte(1)),
                ..Default::default()
            }),
            Address::ZERO,
        );
        let err = Evm::transact_with_inspector(&mut evm, &tx, inspector).unwrap_err();
        assert_eq!(err.to_string(), "fatal error: JavaScript tracing interrupted");
    }

    #[cfg(feature = "js-tracer")]
    #[test]
    fn js_trace_interrupt_covers_mux_tracers_with_js_children() {
        let mux = |config: serde_json::Value| GethDebugTracingOptions {
            tracer: Some(GethDebugTracerType::BuiltInTracer(GethDebugBuiltInTracerType::MuxTracer)),
            tracer_config: serde_json::from_value(config).unwrap(),
            ..Default::default()
        };
        let js = "{ fault: function() {}, result: function() { return null; } }";
        let nested = mux(serde_json::json!({ "muxTracer": { (js): {} } }));
        let nested = serde_json::json!({ "muxTracer": nested.tracer_config });
        for (config, interrupted) in [
            (serde_json::json!({ (js): {} }), true),
            (nested, true),
            (serde_json::json!({ "callTracer": {} }), false),
        ] {
            let opts = mux(config);
            let interrupt = JsTraceInterrupt::new(&opts).unwrap();
            assert_eq!(interrupt.js.is_some(), interrupted);
        }
    }

    #[cfg(feature = "js-tracer")]
    #[tokio::test]
    async fn js_trace_interrupt_times_out_and_interrupts_tracer() {
        let interrupt = JsTraceInterrupt::new(&js_tracer_opts(Some("10ms"))).unwrap();
        let err =
            interrupt.run(std::future::pending::<Result<(), EthApiError>>()).await.unwrap_err();
        assert!(
            matches!(err, EthApiError::ExecutionTimedOut(timeout) if timeout == Duration::from_millis(10))
        );
        assert!(interrupt.js.unwrap().0.is_interrupted());
    }

    #[cfg(feature = "js-tracer")]
    #[test]
    fn js_trace_interrupt_interrupts_tracer_when_dropped_unpolled() {
        let interrupt = JsTraceInterrupt::new(&js_tracer_opts(None)).unwrap();
        let (tracer_interrupt, _) = interrupt.js.clone().unwrap();
        let trace = interrupt.run(std::future::pending::<Result<(), EthApiError>>());
        assert!(!tracer_interrupt.is_interrupted());
        drop(trace);
        assert!(tracer_interrupt.is_interrupted());
    }

    #[test]
    fn witness_preserves_parent_storage_and_code_reads() {
        let address = Address::with_last_byte(1);
        let slots = [B256::with_last_byte(1), B256::with_last_byte(2)];
        let genesis = Genesis::default().extend_accounts([(
            address,
            GenesisAccount::default().with_nonce(Some(1)).with_storage(Some(
                slots.into_iter().map(|slot| (slot, B256::with_last_byte(10))).collect(),
            )),
        )]);
        let factory = create_test_provider_factory_with_chain_spec(Arc::new(
            ChainSpecBuilder::mainnet().genesis(genesis).build(),
        ));
        init_genesis(&factory).unwrap();
        let provider = factory.latest().unwrap();
        let proof = provider.proof(Default::default(), address, &slots).unwrap();
        let mut state = evm2::evm::CacheDB::<evm2::evm::EmptyDB>::default();
        state.cache.accounts.insert(address, None);
        state.cache.storage.entry(address).or_default().wipe();

        let read_code = Bytecode::new_raw(alloy_primitives::bytes!("600100"));
        let created_code = Bytecode::new_raw(alloy_primitives::bytes!("600200"));
        state.cache.contracts.insert(read_code.hash_slow(), read_code.clone());
        let mut output = BundleState::default();
        // Recreating code already read from parent state must not remove it from the witness.
        output.contracts.insert(read_code.hash_slow(), read_code.clone());
        output.contracts.insert(created_code.hash_slow(), created_code.clone());
        let codes = ExecutionWitnessRecord::commit_execution(&mut state, &output);

        for mode in [ExecutionWitnessMode::Legacy, ExecutionWitnessMode::Canonical] {
            let witness = ExecutionWitnessRecord::new(&state)
                .with_canonical_codes(codes.clone())
                .into_execution_witness_without_headers(&provider, mode)
                .unwrap();
            // Canonical mode needs only parent code; legacy also includes execution output.
            let mut expected_codes = vec![read_code.original_bytes()];
            if !mode.is_canonical() {
                expected_codes.push(created_code.original_bytes());
                expected_codes.push(Bytes::new());
                expected_codes.sort_unstable();
            }
            let mut actual_codes = witness.codes;
            actual_codes.sort_unstable();
            assert_eq!(actual_codes, expected_codes);
            for storage in &proof.storage_proofs {
                assert!(!storage.proof.is_empty());
                assert!(storage.proof.iter().all(|node| witness.state.contains(node)));
            }
        }
    }

    #[test]
    fn hashed_post_state_zeroes_destroyed_account_parent_storage() {
        let factory = create_test_provider_factory();
        let address = Address::with_last_byte(1);
        let old_slot = U256::ONE;
        let new_slot = U256::from(2);
        let old_value = U256::from(10);
        let new_value = U256::from(20);
        let hashed_address = keccak256(address);
        let hashed_old_slot = keccak256(B256::from(old_slot));
        let hashed_new_slot = keccak256(B256::from(new_slot));

        let provider_rw = factory.provider_rw().unwrap();
        provider_rw
            .tx_ref()
            .put::<tables::HashedStorages>(
                hashed_address,
                StorageEntry { key: hashed_old_slot, value: old_value },
            )
            .unwrap();
        provider_rw.commit().unwrap();

        let mut bundle_state = BundleState::default();
        bundle_state.state.insert(
            address,
            BundleAccount::new(
                Some(AccountInfo::default()),
                Some(AccountInfo::default()),
                std::iter::once((new_slot, StorageSlot::new_changed(U256::ZERO, new_value)))
                    .collect(),
                AccountStatus::DestroyedChanged,
            ),
        );

        let provider = factory.latest().unwrap();
        let hashed_state = provider.hashed_post_state(&bundle_state).unwrap();
        let storage = &hashed_state.storages[&hashed_address];

        assert_eq!(storage.storage[&hashed_old_slot], U256::ZERO);
        assert_eq!(storage.storage[&hashed_new_slot], new_value);
    }
}