reth_engine_tree/tree/payload_processor/

prewarm.rs

//! Caching and prewarming related functionality.
//!
//! Prewarming executes transactions in parallel before the actual block execution
//! to populate the execution cache with state that will likely be accessed during
//! block processing.
//!
//! ## How Prewarming Works
//!
//! 1. Incoming transactions are split into two streams: one for prewarming (executed in parallel)
//!    and one for actual execution (executed sequentially)
//! 2. Prewarming tasks execute transactions in parallel using shared caches
//! 3. When actual block execution happens, it benefits from the warmed cache

use crate::tree::{
    cached_state::{CachedStateProvider, SavedCache},
    payload_processor::{
        bal::{self, total_slots, BALSlotIter},
        multiproof::{MultiProofMessage, VersionedMultiProofTargets},
        PayloadExecutionCache,
    },
    precompile_cache::{CachedPrecompile, PrecompileCacheMap},
    ExecutionEnv, StateProviderBuilder,
};
use alloy_consensus::transaction::TxHashRef;
use alloy_eip7928::BlockAccessList;
use alloy_eips::eip4895::Withdrawal;
use alloy_evm::Database;
use alloy_primitives::{keccak256, map::B256Set, B256};
use crossbeam_channel::{Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
use metrics::{Counter, Gauge, Histogram};
use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, RecoveredTx, SpecFor};
use reth_metrics::Metrics;
use reth_primitives_traits::NodePrimitives;
use reth_provider::{
    AccountReader, BlockExecutionOutput, BlockReader, StateProvider, StateProviderFactory,
    StateReader,
};
use reth_revm::{database::StateProviderDatabase, state::EvmState};
use reth_tasks::Runtime;
use reth_trie::MultiProofTargets;
use std::{
    ops::Range,
    sync::{
        atomic::{AtomicBool, Ordering},
        mpsc::{self, channel, Receiver, Sender},
        Arc,
    },
    time::Instant,
};
use tracing::{debug, debug_span, instrument, trace, warn, Span};

/// Determines the prewarming mode: transaction-based, BAL-based, or skipped.
#[derive(Debug)]
pub enum PrewarmMode<Tx> {
    /// Prewarm by executing transactions from a stream.
    Transactions(Receiver<Tx>),
    /// Prewarm by prefetching slots from a Block Access List.
    BlockAccessList(Arc<BlockAccessList>),
    /// Transaction prewarming is skipped (e.g. small blocks where the overhead exceeds the
    /// benefit). No workers are spawned.
    Skipped,
}

/// A wrapper for transactions that includes their index in the block.
#[derive(Clone)]
struct IndexedTransaction<Tx> {
    /// The transaction index in the block.
    index: usize,
    /// The wrapped transaction.
    tx: Tx,
}

/// A task that is responsible for caching and prewarming the cache by executing transactions
/// individually in parallel.
///
/// Note: This task runs until cancelled externally.
#[derive(Debug)]
pub struct PrewarmCacheTask<N, P, Evm>
where
    N: NodePrimitives,
    Evm: ConfigureEvm<Primitives = N>,
{
    /// The executor used to spawn execution tasks.
    executor: Runtime,
    /// Shared execution cache.
    execution_cache: PayloadExecutionCache,
    /// Context provided to execution tasks
    ctx: PrewarmContext<N, P, Evm>,
    /// How many transactions should be executed in parallel
    max_concurrency: usize,
    /// Sender to emit evm state outcome messages, if any.
    to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
    /// Receiver for events produced by tx execution
    actions_rx: Receiver<PrewarmTaskEvent<N::Receipt>>,
    /// Parent span for tracing
    parent_span: Span,
}

impl<N, P, Evm> PrewarmCacheTask<N, P, Evm>
where
    N: NodePrimitives,
    P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
    Evm: ConfigureEvm<Primitives = N> + 'static,
{
    /// Initializes the task with the given transactions pending execution
    pub fn new(
        executor: Runtime,
        execution_cache: PayloadExecutionCache,
        ctx: PrewarmContext<N, P, Evm>,
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
        max_concurrency: usize,
    ) -> (Self, Sender<PrewarmTaskEvent<N::Receipt>>) {
        let (actions_tx, actions_rx) = channel();

        trace!(
            target: "engine::tree::payload_processor::prewarm",
            max_concurrency,
            transaction_count = ctx.env.transaction_count,
            "Initialized prewarm task"
        );

        (
            Self {
                executor,
                execution_cache,
                ctx,
                max_concurrency,
                to_multi_proof,
                actions_rx,
                parent_span: Span::current(),
            },
            actions_tx,
        )
    }

    /// Spawns all pending transactions as blocking tasks by first chunking them.
    ///
    /// For Optimism chains, special handling is applied to the first transaction if it's a
    /// deposit transaction (type 0x7E/126) which sets critical metadata that affects all
    /// subsequent transactions in the block.
    fn spawn_all<Tx>(
        &self,
        pending: mpsc::Receiver<Tx>,
        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
    ) where
        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
    {
        let executor = self.executor.clone();
        let ctx = self.ctx.clone();
        let max_concurrency = self.max_concurrency;
        let span = Span::current();

        self.executor.spawn_blocking(move || {
            let _enter = debug_span!(target: "engine::tree::payload_processor::prewarm", parent: span, "spawn_all").entered();

            let (done_tx, done_rx) = mpsc::channel();

            // When transaction_count is 0, it means the count is unknown. In this case, spawn
            // max workers to handle potentially many transactions in parallel rather
            // than bottlenecking on a single worker.
            let transaction_count = ctx.env.transaction_count;
            let workers_needed = if transaction_count == 0 {
                max_concurrency
            } else {
                transaction_count.min(max_concurrency)
            };

            // Spawn workers
            let tx_sender = ctx.clone().spawn_workers(workers_needed, &executor,  to_multi_proof.clone(), done_tx.clone());

            // Distribute transactions to workers
            let mut tx_index = 0usize;
            while let Ok(tx) = pending.recv() {
                // Stop distributing if termination was requested
                if ctx.terminate_execution.load(Ordering::Relaxed) {
                    trace!(
                        target: "engine::tree::payload_processor::prewarm",
                        "Termination requested, stopping transaction distribution"
                    );
                    break;
                }

                let indexed_tx = IndexedTransaction { index: tx_index, tx };

                // Send transaction to the workers
                // Ignore send errors: workers listen to terminate_execution and may
                // exit early when signaled.
                let _ = tx_sender.send(indexed_tx);

                tx_index += 1;
            }

            // Send withdrawal prefetch targets after all transactions have been distributed
            if let Some(to_multi_proof) = to_multi_proof
                && let Some(withdrawals) = &ctx.env.withdrawals
                && !withdrawals.is_empty()
            {
                let targets =
                    multiproof_targets_from_withdrawals(withdrawals, ctx.v2_proofs_enabled);
                let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
            }

            // drop sender and wait for all tasks to finish
            drop(done_tx);
            drop(tx_sender);
            while done_rx.recv().is_ok() {}

            let _ = actions_tx
                .send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: tx_index });
        });
    }

    /// This method calls `ExecutionCache::update_with_guard` which requires exclusive access.
    /// It should only be called after ensuring that:
    /// 1. All prewarming tasks have completed execution
    /// 2. No other concurrent operations are accessing the cache
    ///
    /// Saves the warmed caches back into the shared slot after prewarming completes.
    ///
    /// This consumes the `SavedCache` held by the task, which releases its usage guard and allows
    /// the new, warmed cache to be inserted.
    ///
    /// This method is called from `run()` only after all execution tasks are complete.
    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
    fn save_cache(
        self,
        execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
        valid_block_rx: mpsc::Receiver<()>,
    ) {
        let start = Instant::now();

        let Self { execution_cache, ctx: PrewarmContext { env, metrics, saved_cache, .. }, .. } =
            self;
        let hash = env.hash;

        if let Some(saved_cache) = saved_cache {
            debug!(target: "engine::caching", parent_hash=?hash, "Updating execution cache");
            // Perform all cache operations atomically under the lock
            execution_cache.update_with_guard(|cached| {
                // consumes the `SavedCache` held by the prewarming task, which releases its usage
                // guard
                let (caches, cache_metrics, disable_cache_metrics) = saved_cache.split();
                let new_cache = SavedCache::new(hash, caches, cache_metrics)
                    .with_disable_cache_metrics(disable_cache_metrics);

                // Insert state into cache while holding the lock
                // Access the BundleState through the shared ExecutionOutcome
                if new_cache.cache().insert_state(&execution_outcome.state).is_err() {
                    // Clear the cache on error to prevent having a polluted cache
                    *cached = None;
                    debug!(target: "engine::caching", "cleared execution cache on update error");
                    return;
                }

                new_cache.update_metrics();

                if valid_block_rx.recv().is_ok() {
                    // Replace the shared cache with the new one; the previous cache (if any) is
                    // dropped.
                    *cached = Some(new_cache);
                } else {
                    // Block was invalid; caches were already mutated by insert_state above,
                    // so we must clear to prevent using polluted state
                    *cached = None;
                    debug!(target: "engine::caching", "cleared execution cache on invalid block");
                }
            });

            let elapsed = start.elapsed();
            debug!(target: "engine::caching", parent_hash=?hash, elapsed=?elapsed, "Updated execution cache");

            metrics.cache_saving_duration.set(elapsed.as_secs_f64());
        }
    }

    /// Runs BAL-based prewarming by spawning workers to prefetch storage slots.
    ///
    /// Divides the total slots across `max_concurrency` workers, each responsible for
    /// prefetching a range of slots from the BAL.
    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
    fn run_bal_prewarm(
        &self,
        bal: Arc<BlockAccessList>,
        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
    ) {
        // Only prefetch if we have a cache to populate
        if self.ctx.saved_cache.is_none() {
            trace!(
                target: "engine::tree::payload_processor::prewarm",
                "Skipping BAL prewarm - no cache available"
            );
            self.send_bal_hashed_state(&bal);
            let _ =
                actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
            return;
        }

        let total_slots = total_slots(&bal);

        trace!(
            target: "engine::tree::payload_processor::prewarm",
            total_slots,
            max_concurrency = self.max_concurrency,
            "Starting BAL prewarm"
        );

        if total_slots == 0 {
            self.send_bal_hashed_state(&bal);
            let _ =
                actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
            return;
        }

        let (done_tx, done_rx) = mpsc::channel();

        // Calculate number of workers needed (at most max_concurrency)
        let workers_needed = total_slots.min(self.max_concurrency);

        // Calculate slots per worker
        let slots_per_worker = total_slots / workers_needed;
        let remainder = total_slots % workers_needed;

        // Spawn workers with their assigned ranges
        for i in 0..workers_needed {
            let start = i * slots_per_worker + i.min(remainder);
            let extra = if i < remainder { 1 } else { 0 };
            let end = start + slots_per_worker + extra;

            self.ctx.spawn_bal_worker(
                i,
                &self.executor,
                Arc::clone(&bal),
                start..end,
                done_tx.clone(),
            );
        }

        // Drop our handle to done_tx so we can detect completion
        drop(done_tx);

        // Wait for all workers to complete
        let mut completed_workers = 0;
        while done_rx.recv().is_ok() {
            completed_workers += 1;
        }

        trace!(
            target: "engine::tree::payload_processor::prewarm",
            completed_workers,
            "All BAL prewarm workers completed"
        );

        // Convert BAL to HashedPostState and send to multiproof task
        self.send_bal_hashed_state(&bal);

        // Signal that execution has finished
        let _ = actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
    }

    /// Converts the BAL to [`HashedPostState`](reth_trie::HashedPostState) and sends it to the
    /// multiproof task.
    fn send_bal_hashed_state(&self, bal: &BlockAccessList) {
        let Some(to_multi_proof) = &self.to_multi_proof else { return };

        let provider = match self.ctx.provider.build() {
            Ok(provider) => provider,
            Err(err) => {
                warn!(
                    target: "engine::tree::payload_processor::prewarm",
                    ?err,
                    "Failed to build provider for BAL hashed state conversion"
                );
                return;
            }
        };

        match bal::bal_to_hashed_post_state(bal, &provider) {
            Ok(hashed_state) => {
                debug!(
                    target: "engine::tree::payload_processor::prewarm",
                    accounts = hashed_state.accounts.len(),
                    storages = hashed_state.storages.len(),
                    "Converted BAL to hashed post state"
                );
                let _ = to_multi_proof.send(MultiProofMessage::HashedStateUpdate(hashed_state));
                let _ = to_multi_proof.send(MultiProofMessage::FinishedStateUpdates);
            }
            Err(err) => {
                warn!(
                    target: "engine::tree::payload_processor::prewarm",
                    ?err,
                    "Failed to convert BAL to hashed state"
                );
            }
        }
    }

    /// Executes the task.
    ///
    /// This will execute the transactions until all transactions have been processed or the task
    /// was cancelled.
    #[instrument(
        parent = &self.parent_span,
        level = "debug",
        target = "engine::tree::payload_processor::prewarm",
        name = "prewarm and caching",
        skip_all
    )]
    pub fn run<Tx>(self, mode: PrewarmMode<Tx>, actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>)
    where
        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
    {
        // Spawn execution tasks based on mode
        match mode {
            PrewarmMode::Transactions(pending) => {
                self.spawn_all(pending, actions_tx, self.to_multi_proof.clone());
            }
            PrewarmMode::BlockAccessList(bal) => {
                self.run_bal_prewarm(bal, actions_tx);
            }
            PrewarmMode::Skipped => {
                let _ = actions_tx
                    .send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
            }
        }

        let mut final_execution_outcome = None;
        let mut finished_execution = false;
        while let Ok(event) = self.actions_rx.recv() {
            match event {
                PrewarmTaskEvent::TerminateTransactionExecution => {
                    // stop tx processing
                    debug!(target: "engine::tree::prewarm", "Terminating prewarm execution");
                    self.ctx.terminate_execution.store(true, Ordering::Relaxed);
                }
                PrewarmTaskEvent::Terminate { execution_outcome, valid_block_rx } => {
                    trace!(target: "engine::tree::payload_processor::prewarm", "Received termination signal");
                    final_execution_outcome =
                        Some(execution_outcome.map(|outcome| (outcome, valid_block_rx)));

                    if finished_execution {
                        // all tasks are done, we can exit, which will save caches and exit
                        break
                    }
                }
                PrewarmTaskEvent::FinishedTxExecution { executed_transactions } => {
                    trace!(target: "engine::tree::payload_processor::prewarm", "Finished prewarm execution signal");
                    self.ctx.metrics.transactions.set(executed_transactions as f64);
                    self.ctx.metrics.transactions_histogram.record(executed_transactions as f64);

                    finished_execution = true;

                    if final_execution_outcome.is_some() {
                        // all tasks are done, we can exit, which will save caches and exit
                        break
                    }
                }
            }
        }

        debug!(target: "engine::tree::payload_processor::prewarm", "Completed prewarm execution");

        // save caches and finish using the shared ExecutionOutcome
        if let Some(Some((execution_outcome, valid_block_rx))) = final_execution_outcome {
            self.save_cache(execution_outcome, valid_block_rx);
        }
    }
}

/// Context required by tx execution tasks.
#[derive(Debug, Clone)]
pub struct PrewarmContext<N, P, Evm>
where
    N: NodePrimitives,
    Evm: ConfigureEvm<Primitives = N>,
{
    /// The execution environment.
    pub env: ExecutionEnv<Evm>,
    /// The EVM configuration.
    pub evm_config: Evm,
    /// The saved cache.
    pub saved_cache: Option<SavedCache>,
    /// Provider to obtain the state
    pub provider: StateProviderBuilder<N, P>,
    /// The metrics for the prewarm task.
    pub metrics: PrewarmMetrics,
    /// An atomic bool that tells prewarm tasks to not start any more execution.
    pub terminate_execution: Arc<AtomicBool>,
    /// Whether the precompile cache is disabled.
    pub precompile_cache_disabled: bool,
    /// The precompile cache map.
    pub precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
    /// Whether V2 proof calculation is enabled.
    pub v2_proofs_enabled: bool,
}

impl<N, P, Evm> PrewarmContext<N, P, Evm>
where
    N: NodePrimitives,
    P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
    Evm: ConfigureEvm<Primitives = N> + 'static,
{
    /// Splits this context into an evm, an evm config, metrics, the atomic bool for terminating
    /// execution, and whether V2 proofs are enabled.
    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
    fn evm_for_ctx(
        self,
    ) -> Option<(EvmFor<Evm, impl Database>, PrewarmMetrics, Arc<AtomicBool>, bool)> {
        let Self {
            env,
            evm_config,
            saved_cache,
            provider,
            metrics,
            terminate_execution,
            precompile_cache_disabled,
            precompile_cache_map,
            v2_proofs_enabled,
        } = self;

        let mut state_provider = match provider.build() {
            Ok(provider) => provider,
            Err(err) => {
                trace!(
                    target: "engine::tree::payload_processor::prewarm",
                    %err,
                    "Failed to build state provider in prewarm thread"
                );
                return None
            }
        };

        // Use the caches to create a new provider with caching
        if let Some(saved_cache) = saved_cache {
            let caches = saved_cache.cache().clone();
            let cache_metrics = saved_cache.metrics().clone();
            state_provider =
                Box::new(CachedStateProvider::new_prewarm(state_provider, caches, cache_metrics));
        }

        let state_provider = StateProviderDatabase::new(state_provider);

        let mut evm_env = env.evm_env;

        // we must disable the nonce check so that we can execute the transaction even if the nonce
        // doesn't match what's on chain.
        evm_env.cfg_env.disable_nonce_check = true;

        // disable the balance check so that transactions from senders who were funded by earlier
        // transactions in the block can still be prewarmed
        evm_env.cfg_env.disable_balance_check = true;

        // create a new executor and disable nonce checks in the env
        let spec_id = *evm_env.spec_id();
        let mut evm = evm_config.evm_with_env(state_provider, evm_env);

        if !precompile_cache_disabled {
            // Only cache pure precompiles to avoid issues with stateful precompiles
            evm.precompiles_mut().map_pure_precompiles(|address, precompile| {
                CachedPrecompile::wrap(
                    precompile,
                    precompile_cache_map.cache_for_address(*address),
                    spec_id,
                    None, // No metrics for prewarm
                )
            });
        }

        Some((evm, metrics, terminate_execution, v2_proofs_enabled))
    }

    /// Accepts a [`CrossbeamReceiver`] of transactions and a handle to prewarm task. Executes
    /// transactions and streams [`MultiProofMessage::PrefetchProofs`] messages for each
    /// transaction.
    ///
    /// This function processes transactions sequentially from the receiver and emits outcome events
    /// via the provided sender. Execution errors are logged and tracked but do not stop the batch
    /// processing unless the task is explicitly cancelled.
    ///
    /// Note: There are no ordering guarantees; this does not reflect the state produced by
    /// sequential execution.
    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
    fn transact_batch<Tx>(
        self,
        txs: CrossbeamReceiver<IndexedTransaction<Tx>>,
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
        done_tx: Sender<()>,
    ) where
        Tx: ExecutableTxFor<Evm>,
    {
        let Some((mut evm, metrics, terminate_execution, v2_proofs_enabled)) = self.evm_for_ctx()
        else {
            return
        };

        while let Ok(IndexedTransaction { index, tx }) = txs.recv() {
            let _enter = debug_span!(
                target: "engine::tree::payload_processor::prewarm",
                "prewarm tx",
                index,
            )
            .entered();

            // create the tx env
            let start = Instant::now();

            // If the task was cancelled, stop execution, and exit.
            if terminate_execution.load(Ordering::Relaxed) {
                break
            }

            let (tx_env, tx) = tx.into_parts();
            let res = match evm.transact(tx_env) {
                Ok(res) => res,
                Err(err) => {
                    trace!(
                        target: "engine::tree::payload_processor::prewarm",
                        %err,
                        tx_hash=%tx.tx().tx_hash(),
                        sender=%tx.signer(),
                        "Error when executing prewarm transaction",
                    );
                    // Track transaction execution errors
                    metrics.transaction_errors.increment(1);
                    // skip error because we can ignore these errors and continue with the next tx
                    continue
                }
            };
            metrics.execution_duration.record(start.elapsed());

            // If the task was cancelled, stop execution, and exit.
            if terminate_execution.load(Ordering::Relaxed) {
                break
            }

            // Only send outcome for transactions after the first txn
            // as the main execution will be just as fast
            if index > 0 {
                let (targets, storage_targets) =
                    multiproof_targets_from_state(res.state, v2_proofs_enabled);
                metrics.prefetch_storage_targets.record(storage_targets as f64);
                if let Some(to_multi_proof) = &to_multi_proof {
                    let _ = to_multi_proof.send(MultiProofMessage::PrefetchProofs(targets));
                }
            }

            metrics.total_runtime.record(start.elapsed());
        }

        // send a message to the main task to flag that we're done
        let _ = done_tx.send(());
    }

    /// Spawns worker tasks that pull transactions from a shared channel.
    ///
    /// Returns the sender for distributing transactions to workers.
    fn spawn_workers<Tx>(
        self,
        workers_needed: usize,
        task_executor: &Runtime,
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
        done_tx: Sender<()>,
    ) -> CrossbeamSender<IndexedTransaction<Tx>>
    where
        Tx: ExecutableTxFor<Evm> + Send + 'static,
    {
        let (tx_sender, tx_receiver) = crossbeam_channel::unbounded();

        // Spawn workers that all pull from the shared receiver
        let executor = task_executor.clone();
        let span = Span::current();
        task_executor.spawn_blocking(move || {
            let _enter = span.entered();
            for idx in 0..workers_needed {
                let ctx = self.clone();
                let to_multi_proof = to_multi_proof.clone();
                let done_tx = done_tx.clone();
                let rx = tx_receiver.clone();
                let span = debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);
                executor.spawn_blocking(move || {
                    let _enter = span.entered();
                    ctx.transact_batch(rx, to_multi_proof, done_tx);
                });
            }
        });

        tx_sender
    }

    /// Spawns a worker task for BAL slot prefetching.
    ///
    /// The worker iterates over the specified range of slots in the BAL and ensures
    /// each slot is loaded into the cache by accessing it through the state provider.
    fn spawn_bal_worker(
        &self,
        idx: usize,
        executor: &Runtime,
        bal: Arc<BlockAccessList>,
        range: Range<usize>,
        done_tx: Sender<()>,
    ) {
        let ctx = self.clone();
        let span = debug_span!(
            target: "engine::tree::payload_processor::prewarm",
            "bal prewarm worker",
            idx,
            range_start = range.start,
            range_end = range.end
        );

        executor.spawn_blocking(move || {
            let _enter = span.entered();
            ctx.prefetch_bal_slots(bal, range, done_tx);
        });
    }

    /// Prefetches storage slots from a BAL range into the cache.
    ///
    /// This iterates through the specified range of slots and accesses them via the state
    /// provider to populate the cache.
    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
    fn prefetch_bal_slots(
        self,
        bal: Arc<BlockAccessList>,
        range: Range<usize>,
        done_tx: Sender<()>,
    ) {
        let Self { saved_cache, provider, metrics, .. } = self;

        // Build state provider
        let state_provider = match provider.build() {
            Ok(provider) => provider,
            Err(err) => {
                trace!(
                    target: "engine::tree::payload_processor::prewarm",
                    %err,
                    "Failed to build state provider in BAL prewarm thread"
                );
                let _ = done_tx.send(());
                return;
            }
        };

        // Wrap with cache (guaranteed to be Some since run_bal_prewarm checks)
        let saved_cache = saved_cache.expect("BAL prewarm should only run with cache");
        let caches = saved_cache.cache().clone();
        let cache_metrics = saved_cache.metrics().clone();
        let state_provider = CachedStateProvider::new(state_provider, caches, cache_metrics);

        let start = Instant::now();

        // Track last seen address to avoid fetching the same account multiple times.
        let mut last_address = None;

        // Iterate through the assigned range of slots
        for (address, slot) in BALSlotIter::new(&bal, range.clone()) {
            // Fetch the account if this is a different address than the last one
            if last_address != Some(address) {
                let _ = state_provider.basic_account(&address);
                last_address = Some(address);
            }

            // Access the slot to populate the cache
            let _ = state_provider.storage(address, slot);
        }

        let elapsed = start.elapsed();

        trace!(
            target: "engine::tree::payload_processor::prewarm",
            ?range,
            elapsed_ms = elapsed.as_millis(),
            "BAL prewarm worker completed"
        );

        // Signal completion
        let _ = done_tx.send(());
        metrics.bal_slot_iteration_duration.record(elapsed.as_secs_f64());
    }
}

/// Returns a set of [`VersionedMultiProofTargets`] and the total amount of storage targets, based
/// on the given state.
fn multiproof_targets_from_state(
    state: EvmState,
    v2_enabled: bool,
) -> (VersionedMultiProofTargets, usize) {
    if v2_enabled {
        multiproof_targets_v2_from_state(state)
    } else {
        multiproof_targets_legacy_from_state(state)
    }
}

/// Returns legacy [`MultiProofTargets`] and the total amount of storage targets, based on the
/// given state.
fn multiproof_targets_legacy_from_state(state: EvmState) -> (VersionedMultiProofTargets, usize) {
    let mut targets = MultiProofTargets::with_capacity(state.len());
    let mut storage_targets = 0;
    for (addr, account) in state {
        // if the account was not touched, or if the account was selfdestructed, do not
        // fetch proofs for it
        //
        // Since selfdestruct can only happen in the same transaction, we can skip
        // prefetching proofs for selfdestructed accounts
        //
        // See: https://eips.ethereum.org/EIPS/eip-6780
        if !account.is_touched() || account.is_selfdestructed() {
            continue
        }

        let mut storage_set =
            B256Set::with_capacity_and_hasher(account.storage.len(), Default::default());
        for (key, slot) in account.storage {
            // do nothing if unchanged
            if !slot.is_changed() {
                continue
            }

            storage_set.insert(keccak256(B256::new(key.to_be_bytes())));
        }

        storage_targets += storage_set.len();
        targets.insert(keccak256(addr), storage_set);
    }

    (VersionedMultiProofTargets::Legacy(targets), storage_targets)
}

/// Returns V2 [`reth_trie_parallel::targets_v2::MultiProofTargetsV2`] and the total amount of
/// storage targets, based on the given state.
fn multiproof_targets_v2_from_state(state: EvmState) -> (VersionedMultiProofTargets, usize) {
    use reth_trie::proof_v2;
    use reth_trie_parallel::targets_v2::MultiProofTargetsV2;

    let mut targets = MultiProofTargetsV2::default();
    let mut storage_target_count = 0;
    for (addr, account) in state {
        // if the account was not touched, or if the account was selfdestructed, do not
        // fetch proofs for it
        //
        // Since selfdestruct can only happen in the same transaction, we can skip
        // prefetching proofs for selfdestructed accounts
        //
        // See: https://eips.ethereum.org/EIPS/eip-6780
        if !account.is_touched() || account.is_selfdestructed() {
            continue
        }

        let hashed_address = keccak256(addr);
        targets.account_targets.push(hashed_address.into());

        let mut storage_slots = Vec::with_capacity(account.storage.len());
        for (key, slot) in account.storage {
            // do nothing if unchanged
            if !slot.is_changed() {
                continue
            }

            let hashed_slot = keccak256(B256::new(key.to_be_bytes()));
            storage_slots.push(proof_v2::Target::from(hashed_slot));
        }

        storage_target_count += storage_slots.len();
        if !storage_slots.is_empty() {
            targets.storage_targets.insert(hashed_address, storage_slots);
        }
    }

    (VersionedMultiProofTargets::V2(targets), storage_target_count)
}

/// Returns [`VersionedMultiProofTargets`] for withdrawal addresses.
///
/// Withdrawals only modify account balances (no storage), so the targets contain
/// only account-level entries with empty storage sets.
fn multiproof_targets_from_withdrawals(
    withdrawals: &[Withdrawal],
    v2_enabled: bool,
) -> VersionedMultiProofTargets {
    use reth_trie_parallel::targets_v2::MultiProofTargetsV2;
    if v2_enabled {
        VersionedMultiProofTargets::V2(MultiProofTargetsV2 {
            account_targets: withdrawals.iter().map(|w| keccak256(w.address).into()).collect(),
            ..Default::default()
        })
    } else {
        VersionedMultiProofTargets::Legacy(
            withdrawals.iter().map(|w| (keccak256(w.address), Default::default())).collect(),
        )
    }
}

/// The events the pre-warm task can handle.
///
/// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the main
/// execution path without cloning the expensive `BundleState`.
#[derive(Debug)]
pub enum PrewarmTaskEvent<R> {
    /// Forcefully terminate all remaining transaction execution.
    TerminateTransactionExecution,
    /// Forcefully terminate the task on demand and update the shared cache with the given output
    /// before exiting.
    Terminate {
        /// The final execution outcome. Using `Arc` allows sharing with the main execution
        /// path without cloning the expensive `BundleState`.
        execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
        /// Receiver for the block validation result.
        ///
        /// Cache saving is racing the state root validation. We optimistically construct the
        /// updated cache but only save it once we know the block is valid.
        valid_block_rx: mpsc::Receiver<()>,
    },
    /// Finished executing all transactions
    FinishedTxExecution {
        /// Number of transactions executed
        executed_transactions: usize,
    },
}

/// Metrics for transactions prewarming.
#[derive(Metrics, Clone)]
#[metrics(scope = "sync.prewarm")]
pub struct PrewarmMetrics {
    /// The number of transactions to prewarm
    pub(crate) transactions: Gauge,
    /// A histogram of the number of transactions to prewarm
    pub(crate) transactions_histogram: Histogram,
    /// A histogram of duration per transaction prewarming
    pub(crate) total_runtime: Histogram,
    /// A histogram of EVM execution duration per transaction prewarming
    pub(crate) execution_duration: Histogram,
    /// A histogram for prefetch targets per transaction prewarming
    pub(crate) prefetch_storage_targets: Histogram,
    /// A histogram of duration for cache saving
    pub(crate) cache_saving_duration: Gauge,
    /// Counter for transaction execution errors during prewarming
    pub(crate) transaction_errors: Counter,
    /// A histogram of BAL slot iteration duration during prefetching
    pub(crate) bal_slot_iteration_duration: Histogram,
}