reth_engine_tree/tree/payload_processor/bal/
worker.rs

1use super::BalExecutionError;
2use alloy_consensus::Transaction;
3use alloy_eip7928::BlockAccessIndex;
4use alloy_primitives::{Address, B256, U256};
5use crossbeam_channel::{Receiver, Sender};
6use evm2::{bytecode::Bytecode, evm::AccountInfo, AnyError};
7use reth_evm::{
8    BlockExecutionError, BlockExecutor, BlockExecutorFactory, BlockExecutorFor,
9    BlockValidationError, ConfigureEvm, Database, EvmEnvFor, ExecutableTxFor, ExecutionCtxFor,
10};
11use std::{cell::RefCell, rc::Rc, sync::Arc};
12
13#[derive(Debug, thiserror::Error)]
14pub(super) enum BalWorkerError {
15    /// Worker state or provider setup failed.
16    #[error("BAL worker setup failed: {0}")]
17    Setup(#[source] BalExecutionError),
18    /// Transaction recovery or conversion failed before EVM execution.
19    #[error("BAL worker transaction conversion failed for transaction {tx_index}: {source}")]
20    Transaction {
21        /// Index of the transaction that failed.
22        tx_index: usize,
23        /// The underlying recovery or conversion error.
24        #[source]
25        source: AnyError,
26    },
27    /// EVM transaction execution failed.
28    #[error("BAL worker EVM execution failed for transaction {tx_index}: {source}")]
29    Execution {
30        /// Index of the transaction that failed.
31        tx_index: usize,
32        /// Gas limit of the transaction that failed.
33        tx_gas_limit: u64,
34        /// The underlying execution error.
35        #[source]
36        source: BlockExecutionError,
37    },
38}
39
40impl From<BalWorkerError> for BalExecutionError {
41    fn from(err: BalWorkerError) -> Self {
42        match err {
43            BalWorkerError::Setup(err) => err,
44            BalWorkerError::Transaction { source, .. } => {
45                Self::Execution(BlockValidationError::Other(source).into())
46            }
47            BalWorkerError::Execution { source, .. } => Self::Execution(source),
48        }
49    }
50}
51
52pub(super) struct BalWorkerOutput<R> {
53    pub(super) index: usize,
54    pub(super) signer: Address,
55    pub(super) tx_gas_limit: u64,
56    pub(super) result: R,
57}
58
59type WorkerExecutorResult<'a, Cfg> =
60    <BlockExecutorFor<'a, Cfg> as BlockExecutor>::TransactionResultWithState;
61
62type WorkerResultSender<'a, Cfg> =
63    Sender<Result<BalWorkerOutput<WorkerExecutorResult<'a, Cfg>>, BalWorkerError>>;
64
65#[expect(clippy::too_many_arguments)]
66pub(super) fn spawn_worker<'scope, Evm, Tx, Err, DB, MakeDb>(
67    scope: &rayon::Scope<'scope>,
68    tx_rx: Receiver<(usize, Result<Tx, Err>)>,
69    abort_rx: Receiver<()>,
70    result_tx: WorkerResultSender<'scope, Evm>,
71    evm_config: &'scope Evm,
72    make_db: &'scope MakeDb,
73    received_bal: Arc<<BlockExecutorFor<'scope, Evm> as BlockExecutor>::BlockAccessList>,
74    evm_env: EvmEnvFor<Evm>,
75    ctx: ExecutionCtxFor<'scope, Evm>,
76) where
77    Evm: ConfigureEvm + 'scope,
78    Tx: ExecutableTxFor<Evm> + Send + 'scope,
79    Err: core::error::Error + Send + Sync + 'static,
80    DB: Database + Send + 'scope,
81    MakeDb: Fn(bool) -> Result<DB, BalExecutionError> + Sync + 'scope,
82{
83    scope.spawn(move |_| {
84        let worker_result = (|| -> Result<(), BalWorkerError> {
85            // Keep the cache-filling database across executor resets so a speculative failure
86            // cannot introduce an unindexed provider setup error ahead of its ordered verdict.
87            let database = Rc::new(RefCell::new(make_db(true).map_err(BalWorkerError::Setup)?));
88            'worker: loop {
89                let evm =
90                    evm_config.evm_with_env(WorkerDatabase(database.clone()), evm_env.clone());
91                let mut executor =
92                    evm_config.block_executor_factory().create_executor(evm, ctx.clone());
93                executor.set_block_access_list(Arc::clone(&received_bal));
94
95                loop {
96                    let (tx_index, tx) = crossbeam_channel::select_biased! {
97                        recv(abort_rx) -> _ => break 'worker,
98                        recv(tx_rx) -> msg => match msg {
99                            Ok(ix_tx) => ix_tx,
100                            Err(_) => break 'worker,
101                        },
102                    };
103                    let tx = match tx {
104                        Ok(tx) => tx,
105                        Err(source) => {
106                            let error = BalWorkerError::Transaction {
107                                tx_index,
108                                source: AnyError::new(source),
109                            };
110                            if result_tx.send(Err(error)).is_err() {
111                                break 'worker;
112                            }
113                            continue;
114                        }
115                    };
116                    let signer = *tx.signer();
117                    let tx_gas_limit = tx.tx().gas_limit();
118
119                    executor
120                        .set_block_access_index(BlockAccessIndex::from_tx_index(tx_index as u64));
121                    let message = match executor.execute_transaction_without_commit(tx) {
122                        Ok(result) => {
123                            Ok(BalWorkerOutput { index: tx_index, signer, tx_gas_limit, result })
124                        }
125                        Err(source) => {
126                            Err(BalWorkerError::Execution { tx_index, tx_gas_limit, source })
127                        }
128                    };
129                    let failed = message.is_err();
130                    if result_tx.send(message).is_err() {
131                        break 'worker;
132                    }
133                    if failed {
134                        // The executor trait does not guarantee reuse after an error. Rebuild
135                        // its EVM and state before serving more work: the queue can still contain
136                        // earlier transactions whose verdict must precede this failure.
137                        break;
138                    }
139                }
140            }
141
142            Ok(())
143        })();
144
145        if let Err(err) = worker_result {
146            let _ = result_tx.send(Err(err));
147        }
148    });
149}
150
151// Executor resets retain the same provider. The wrapper is created and used entirely on one
152// worker thread; its shared ownership decouples the database lifetime from each executor.
153struct WorkerDatabase<DB>(Rc<RefCell<DB>>);
154
155impl<DB: Database> Database for WorkerDatabase<DB> {
156    type Error = DB::Error;
157
158    fn is_fatal(error: &Self::Error) -> bool {
159        DB::is_fatal(error)
160    }
161
162    fn get_account(&mut self, address: &Address) -> Result<Option<AccountInfo>, Self::Error> {
163        self.0.borrow_mut().get_account(address)
164    }
165
166    fn get_code_by_hash(&mut self, hash: &B256) -> Result<Bytecode, Self::Error> {
167        self.0.borrow_mut().get_code_by_hash(hash)
168    }
169
170    fn get_storage(&mut self, address: &Address, key: &U256) -> Result<U256, Self::Error> {
171        self.0.borrow_mut().get_storage(address, key)
172    }
173
174    fn get_block_hash(&mut self, number: &U256) -> Result<B256, Self::Error> {
175        self.0.borrow_mut().get_block_hash(number)
176    }
177}
reth_engine_tree/tree/payload_processor/bal/worker.rs

reth_engine_tree/tree/payload_processor/bal/
worker.rs
