reth_engine_tree/tree/payload_processor/bal/
worker.rs

1use super::BalExecutionError;
2use alloy_consensus::Transaction;
3use alloy_eip7928::BlockAccessIndex;
4use alloy_evm::{
5    block::{BlockExecutionError, BlockExecutor, BlockExecutorFactory},
6    Evm,
7};
8use alloy_primitives::Address;
9use crossbeam_channel::{Receiver, Sender};
10use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Database, EvmEnvFor, ExecutionCtxFor};
11use revm::{database::State, state::bal::Bal as RevmBal};
12use std::sync::Arc;
13
14#[derive(Debug, thiserror::Error)]
15pub(super) enum BalWorkerError {
16    /// Worker state or provider setup failed.
17    #[error("BAL worker setup failed: {0}")]
18    Setup(#[source] BalExecutionError),
19    /// Transaction recovery or conversion failed before EVM execution.
20    #[error("BAL worker transaction conversion failed: {0}")]
21    Transaction(Box<dyn core::error::Error + Send + Sync + 'static>),
22    /// EVM transaction execution failed.
23    #[error("BAL worker EVM execution failed: {0}")]
24    Execution(BlockExecutionError),
25}
26
27impl From<BalWorkerError> for BalExecutionError {
28    fn from(err: BalWorkerError) -> Self {
29        match err {
30            BalWorkerError::Setup(err) => err,
31            BalWorkerError::Transaction(err) => Self::Other(err),
32            BalWorkerError::Execution(err) => Self::Execution(err),
33        }
34    }
35}
36
37pub(super) struct BalWorkerOutput<R> {
38    pub(super) index: usize,
39    pub(super) signer: Address,
40    pub(super) tx_gas_limit: u64,
41    pub(super) result: R,
42}
43
44type WorkerExecutorResult<Cfg> =
45    <<Cfg as ConfigureEvm>::BlockExecutorFactory as BlockExecutorFactory>::TxExecutionResult;
46
47type WorkerResultSender<Cfg> =
48    Sender<Result<BalWorkerOutput<WorkerExecutorResult<Cfg>>, BalWorkerError>>;
49
50#[expect(clippy::too_many_arguments)]
51pub(super) fn spawn_worker<'scope, Evm, Tx, Err, DB, MakeDb>(
52    scope: &rayon::Scope<'scope>,
53    tx_rx: Receiver<(usize, Result<Tx, Err>)>,
54    abort_rx: Receiver<()>,
55    result_tx: WorkerResultSender<Evm>,
56    evm_config: &'scope Evm,
57    make_db: &'scope MakeDb,
58    received_bal_revm: Arc<RevmBal>,
59    evm_env: EvmEnvFor<Evm>,
60    ctx: ExecutionCtxFor<'scope, Evm>,
61) where
62    Evm: ConfigureEvm + 'scope,
63    Tx: ExecutableTxFor<Evm> + Send + 'scope,
64    Err: core::error::Error + Send + Sync + 'static,
65    DB: Database + Send + 'scope,
66    MakeDb: Fn(bool) -> Result<DB, BalExecutionError> + Sync + 'scope,
67{
68    scope.spawn(move |_| {
69        let worker_result = (|| -> Result<(), BalWorkerError> {
70            // Create a database with fill_on_miss=true ensuring misses
71            // are inserted for the other workers.
72            let database = make_db(true).map_err(BalWorkerError::Setup)?;
73            let mut worker_state = State::builder()
74                .with_database(database)
75                .with_bal(received_bal_revm)
76                .with_bundle_update()
77                .build();
78            let evm = evm_config.evm_with_env(&mut worker_state, evm_env);
79            let mut executor = evm_config.create_executor_with_state(evm, ctx.clone());
80
81            loop {
82                let (index, tx) = crossbeam_channel::select_biased! {
83                    recv(abort_rx) -> _ => break,
84                    recv(tx_rx) -> msg => match msg {
85                        Ok(ix_tx) => ix_tx,
86                        Err(_) => break,
87                    },
88                };
89                let tx = tx.map_err(|e| BalWorkerError::Transaction(Box::new(e)))?;
90                let signer = *tx.signer();
91                let tx_gas_limit = tx.tx().gas_limit();
92
93                executor.evm_mut().db_mut().set_bal_index(BlockAccessIndex::new(index as u64 + 1));
94                let result = executor
95                    .execute_transaction_without_commit(tx)
96                    .map_err(BalWorkerError::Execution)?;
97
98                if result_tx
99                    .send(Ok(BalWorkerOutput { index, signer, tx_gas_limit, result }))
100                    .is_err()
101                {
102                    break;
103                }
104            }
105
106            Ok(())
107        })();
108
109        if let Err(err) = worker_result {
110            let _ = result_tx.send(Err(err));
111        }
112    });
113}
reth_engine_tree/tree/payload_processor/bal/worker.rs

reth_engine_tree/tree/payload_processor/bal/
worker.rs
