reth_engine_tree/tree/txpool_prewarm/
worker.rs

1use super::{
2    control::{Command, Publication},
3    Job, Source, Transactions,
4};
5use crate::tree::{StateProviderDatabase, TxPoolPrewarmCacheSnapshot as Snapshot};
6use alloy_evm::Evm;
7use alloy_primitives::B256;
8use crossbeam_channel::{Receiver, RecvTimeoutError, TryRecvError};
9use reth_evm::ConfigureEvm;
10use reth_primitives_traits::NodePrimitives;
11use reth_provider::{BlockReader, StateProviderFactory, StateReader};
12use reth_revm::{cached::CachedReads, db::State};
13use std::{
14    sync::Arc,
15    time::{Duration, Instant},
16};
17use tracing::{debug, trace};
18
19/// Maximum interval between snapshot publications and delay when no transaction is ready.
20const REFRESH_INTERVAL: Duration = Duration::from_millis(100);
21
22/// Delay while waiting for pool maintenance to advance to the state being warmed.
23const HEAD_POLL_INTERVAL: Duration = Duration::from_millis(10);
24
25/// The txpool prewarming worker.
26///
27/// A long-lived loop that speculatively executes the pool's best transactions on top of the
28/// current canonical state, recording every state read in a [`CachedReads`]. Roughly every
29/// [`REFRESH_INTERVAL`] it publishes an immutable snapshot of that cache, which block validation
30/// and payload building use to seed their own caches.
31///
32/// The worker is driven by [`Command`]s: `Start` points it at a new parent state, and
33/// `Pause`/`Resume` bracket cache-sensitive work elsewhere. Commands are only applied between
34/// batches, never while an EVM or state provider is alive.
35pub(super) struct Worker<N, P, Evm>
36where
37    N: NodePrimitives,
38    Evm: ConfigureEvm<Primitives = N>,
39{
40    /// Control commands from the [`Handle`](super::Handle).
41    commands: Receiver<Command<Job<N, P, Evm>>>,
42    /// Shared slot the latest snapshot is published into.
43    publication: Publication,
44    /// The txpool view transactions are drawn from.
45    source: Arc<dyn Source<N>>,
46    /// Configures the EVM used for speculative execution.
47    evm_config: Evm,
48    /// The parent state to warm, from the most recent `Start` command.
49    job: Option<(B256, Job<N, P, Evm>)>,
50    /// Outstanding pauses; the worker only warms while this is zero.
51    pauses: u64,
52    /// Read-through cache filled by execution; replaced whenever the warmed parent changes.
53    cache: CachedReads,
54    /// Parent whose state the `cache` reads were collected against.
55    cache_parent: Option<B256>,
56    /// Cache entry counts as of the last publication. The cache only ever grows, so a change
57    /// means it holds unpublished reads.
58    published_entries: (usize, usize, usize),
59    /// Live best-transactions iterator, tagged with the parent it was opened for.
60    transactions: Option<(B256, Transactions<N>)>,
61}
62
63impl<N, P, Evm> Worker<N, P, Evm>
64where
65    N: NodePrimitives,
66    P: BlockReader + StateProviderFactory + StateReader + Clone,
67    Evm: ConfigureEvm<Primitives = N>,
68{
69    pub(super) fn new(
70        commands: Receiver<Command<Job<N, P, Evm>>>,
71        publication: Publication,
72        source: Arc<dyn Source<N>>,
73        evm_config: Evm,
74    ) -> Self {
75        Self {
76            commands,
77            publication,
78            source,
79            evm_config,
80            job: None,
81            pauses: 0,
82            cache: CachedReads::default(),
83            cache_parent: None,
84            published_entries: (0, 0, 0),
85            transactions: None,
86        }
87    }
88
89    /// Runs until the control side is dropped, which is the worker's shutdown signal.
90    pub(super) fn run(mut self) {
91        let _ = self.run_until_disconnected();
92    }
93
94    fn run_until_disconnected(&mut self) -> Result<(), ChannelDisconnected> {
95        loop {
96            let parent_hash = self.wait_until_runnable()?;
97
98            // The pool tracks canonical heads on its own schedule; check back shortly if it is
99            // not tracking this parent yet.
100            if !self.open_transactions(parent_hash) {
101                self.idle(HEAD_POLL_INTERVAL)?;
102                continue
103            }
104
105            if self.cache_parent != Some(parent_hash) {
106                self.cache = CachedReads::default();
107                self.cache_parent = Some(parent_hash);
108                self.published_entries = (0, 0, 0);
109                debug!(
110                    target: "engine::tree::txpool_prewarm",
111                    ?parent_hash,
112                    "started txpool prewarming"
113                );
114            }
115
116            let batch = self.warm_one_batch();
117
118            // A pending command may pause the worker or point it at a new parent: apply it (at
119            // the top of the loop) before spending time on publication.
120            if !self.commands.is_empty() {
121                continue
122            }
123            self.publish_snapshot_if_dirty();
124            if batch == BatchEnd::Rest {
125                self.idle(REFRESH_INTERVAL)?;
126            }
127        }
128    }
129
130    /// Blocks until the worker holds a job and no pauses are outstanding, applying every command
131    /// that arrives in the meantime. Returns the parent hash to warm.
132    fn wait_until_runnable(&mut self) -> Result<B256, ChannelDisconnected> {
133        loop {
134            self.apply_pending_commands()?;
135
136            if self.pauses == 0 &&
137                let Some((parent_hash, _)) = self.job.as_ref()
138            {
139                return Ok(*parent_hash)
140            }
141
142            let command = self.commands.recv().map_err(|_| ChannelDisconnected)?;
143            self.apply(command);
144        }
145    }
146
147    /// Ensures the transaction iterator matches `parent_hash`, opening a fresh one when the head
148    /// changed. Returns `false` while the pool is not yet tracking that parent.
149    fn open_transactions(&mut self, parent_hash: B256) -> bool {
150        if self.transactions.as_ref().is_none_or(|(parent, _)| *parent != parent_hash) {
151            // Release the stale iterator before asking the pool for a new one.
152            self.transactions = None;
153            self.transactions = self
154                .source
155                .best_transactions(parent_hash)
156                .map(|transactions| (parent_hash, transactions));
157        }
158        self.transactions.is_some()
159    }
160
161    /// Speculatively executes pool transactions against the parent state for at most
162    /// [`REFRESH_INTERVAL`], filling the cache with every state read.
163    ///
164    /// Stops early once the pool has no transaction ready or a command arrives. Commands are
165    /// never consumed here: a pending command merely ends the batch and is applied by the main
166    /// loop after the EVM and state provider built here are dropped.
167    fn warm_one_batch(&mut self) -> BatchEnd {
168        let (_, job) = self.job.as_ref().expect("wait_until_runnable installed a job");
169        let (parent_hash, transactions) =
170            self.transactions.as_mut().expect("open_transactions installed an iterator");
171
172        // Building a state provider opens a database transaction; don't bother under a pending
173        // command.
174        if !self.commands.is_empty() {
175            return BatchEnd::GoAgain
176        }
177
178        let state_provider = match job.provider_builder.build() {
179            Ok(provider) => provider,
180            Err(err) => {
181                trace!(
182                    target: "engine::tree::txpool_prewarm",
183                    %err,
184                    ?parent_hash,
185                    "failed to build txpool prewarming state provider"
186                );
187                return BatchEnd::Rest
188            }
189        };
190        let mut state = State::builder()
191            .with_database(self.cache.as_db_mut(StateProviderDatabase::new(state_provider)))
192            .build();
193        // The environment is the head block's own, not a predicted next-block one, and execution
194        // is out of context by design: transaction viability is the pool's business, so nonce,
195        // balance and (one-block-stale) basefee checks must not gate which state gets warmed.
196        let mut evm_env = job.evm_env.clone();
197        evm_env.cfg_env.disable_nonce_check = true;
198        evm_env.cfg_env.disable_balance_check = true;
199        evm_env.cfg_env.disable_base_fee = true;
200        let mut evm = self.evm_config.evm_with_env(&mut state, evm_env);
201
202        let deadline = Instant::now() + REFRESH_INTERVAL;
203        while self.commands.is_empty() && Instant::now() < deadline {
204            let Some(transaction) = transactions.next() else { return BatchEnd::Rest };
205            if let Err(err) = evm.transact(transaction.transaction) {
206                trace!(
207                    target: "engine::tree::txpool_prewarm",
208                    %err,
209                    tx_hash = ?transaction.hash,
210                    sender = %transaction.sender,
211                    "speculative txpool transaction execution failed"
212                );
213            }
214        }
215        BatchEnd::GoAgain
216    }
217
218    /// Publishes a fresh snapshot if the cache gained reads since the last publication.
219    fn publish_snapshot_if_dirty(&mut self) {
220        let entries = entry_counts(&self.cache);
221        if entries == self.published_entries {
222            return
223        }
224
225        let parent_hash = self.cache_parent.expect("reads only accumulate after a cache reset");
226        *self.publication.write() = Some(Snapshot::new(parent_hash, Arc::new(self.cache.clone())));
227        self.published_entries = entries;
228        let (accounts, storage, bytecodes) = entries;
229        debug!(
230            target: "engine::tree::txpool_prewarm",
231            ?parent_hash,
232            accounts,
233            storage,
234            bytecodes,
235            "published txpool prewarming snapshot"
236        );
237    }
238
239    /// Rests until a command arrives (applying it) or `timeout` elapses, whichever comes first.
240    fn idle(&mut self, timeout: Duration) -> Result<(), ChannelDisconnected> {
241        match self.commands.recv_timeout(timeout) {
242            Ok(command) => {
243                self.apply(command);
244                Ok(())
245            }
246            Err(RecvTimeoutError::Timeout) => Ok(()),
247            Err(RecvTimeoutError::Disconnected) => Err(ChannelDisconnected),
248        }
249    }
250
251    /// Applies every command already sitting in the channel, without blocking.
252    fn apply_pending_commands(&mut self) -> Result<(), ChannelDisconnected> {
253        loop {
254            match self.commands.try_recv() {
255                Ok(command) => self.apply(command),
256                Err(TryRecvError::Empty) => return Ok(()),
257                Err(TryRecvError::Disconnected) => return Err(ChannelDisconnected),
258            }
259        }
260    }
261
262    /// Applies a single command.
263    ///
264    /// Only called while no EVM or state provider is alive, so a paused worker holds no
265    /// execution resources.
266    fn apply(&mut self, command: Command<Job<N, P, Evm>>) {
267        match command {
268            Command::Start { parent_hash, job } => self.job = Some((parent_hash, job)),
269            Command::Pause => {
270                self.pauses =
271                    self.pauses.checked_add(1).expect("txpool prewarm pause count overflow");
272            }
273            Command::Resume => {
274                self.pauses = self
275                    .pauses
276                    .checked_sub(1)
277                    .expect("txpool prewarm resumed without a matching pause");
278            }
279        }
280    }
281}
282
283/// What to do after a warming batch.
284#[derive(Debug, Clone, Copy, PartialEq, Eq)]
285enum BatchEnd {
286    /// Batch again right away: the refresh deadline passed or a command interrupted the batch
287    /// while transactions may still be ready.
288    GoAgain,
289    /// Idle until something changes: the pool had no transaction ready, or no state provider
290    /// could be built.
291    Rest,
292}
293
294/// The control channel closed: every sender is dropped and the worker shuts down.
295#[derive(Debug, Clone, Copy, PartialEq, Eq)]
296struct ChannelDisconnected;
297
298/// Returns `(accounts, storage slots, bytecodes)` cached in `reads`.
299fn entry_counts(reads: &CachedReads) -> (usize, usize, usize) {
300    (
301        reads.accounts.len(),
302        reads.accounts.values().map(|account| account.storage.len()).sum(),
303        reads.contracts.len(),
304    )
305}
306
307#[cfg(test)]
308mod tests {
309    use super::{super::Transaction as PoolTransaction, *};
310    use crate::tree::StateProviderBuilder;
311    use alloy_consensus::{transaction::Recovered, Signed, TxLegacy};
312    use alloy_primitives::{Address, Signature, TxKind, U256};
313    use crossbeam_channel::{unbounded, Sender};
314    use parking_lot::{Mutex, RwLock};
315    use reth_ethereum_primitives::{EthPrimitives, TransactionSigned};
316    use reth_evm_ethereum::EthEvmConfig;
317    use reth_provider::test_utils::MockEthProvider;
318    use std::{
319        collections::{HashMap, VecDeque},
320        sync::atomic::{AtomicUsize, Ordering},
321        thread::{self, JoinHandle},
322    };
323
324    /// Upper bound on any single wait; failures surface as panics well before CI timeouts.
325    const WAIT_LIMIT: Duration = Duration::from_secs(5);
326    const POLL_INTERVAL: Duration = Duration::from_millis(2);
327
328    type TestJob = Job<EthPrimitives, MockEthProvider, EthEvmConfig>;
329
330    /// Drives a live worker thread through its public seams only: commands in, the publication
331    /// slot and the scripted pool out.
332    struct Harness {
333        commands: Sender<Command<TestJob>>,
334        publication: Publication,
335        pool: Arc<ScriptedPool>,
336        worker: Option<JoinHandle<()>>,
337    }
338
339    impl Harness {
340        fn spawn() -> Self {
341            let (commands, receiver) = unbounded();
342            let publication: Publication = Arc::new(RwLock::new(None));
343            let pool = Arc::new(ScriptedPool::default());
344            let worker = thread::spawn({
345                let publication = Arc::clone(&publication);
346                let source: Arc<dyn Source<EthPrimitives>> = pool.clone();
347                move || Worker::new(receiver, publication, source, EthEvmConfig::mainnet()).run()
348            });
349            Self { commands, publication, pool, worker: Some(worker) }
350        }
351
352        /// Drops the control channel and waits for the worker thread to exit.
353        fn shutdown(mut self) {
354            let (disconnected, _) = unbounded();
355            self.commands = disconnected;
356            let worker = self.worker.take().expect("worker already joined");
357            wait_until("the worker thread exits", || worker.is_finished());
358            worker.join().unwrap();
359        }
360
361        /// Points the worker at `parent_hash`, as [`Handle::start`](super::super::Handle) does.
362        fn start(&self, parent_hash: B256) {
363            let job = Job {
364                evm_env: Default::default(),
365                provider_builder: StateProviderBuilder::new(
366                    MockEthProvider::default(),
367                    parent_hash,
368                    None,
369                ),
370            };
371            self.commands.send(Command::Start { parent_hash, job }).unwrap();
372        }
373
374        /// Pauses the worker, as [`Handle::pause`](super::super::Handle) does. Fire-and-forget
375        /// is still deterministic: the worker never publishes work it performs after the pause
376        /// is queued until [`Self::resume`], because publication is skipped while a command is
377        /// pending and a consumed pause blocks the loop.
378        fn pause(&self) {
379            self.commands.send(Command::Pause).unwrap();
380        }
381
382        fn resume(&self) {
383            self.commands.send(Command::Resume).unwrap();
384        }
385
386        /// Blocks until a published snapshot satisfies `accept`.
387        fn published(&self, accept: impl Fn(&Snapshot) -> bool) -> Snapshot {
388            let deadline = Instant::now() + WAIT_LIMIT;
389            loop {
390                let snapshot = self.publication.read().as_ref().cloned();
391                if let Some(snapshot) = snapshot &&
392                    accept(&snapshot)
393                {
394                    return snapshot
395                }
396                assert!(Instant::now() < deadline, "timed out waiting for a matching snapshot");
397                thread::sleep(POLL_INTERVAL);
398            }
399        }
400
401        fn published_for(&self, parent_hash: B256) -> Snapshot {
402            self.published(|snapshot| snapshot.parent_hash() == parent_hash)
403        }
404
405        fn published_entry_counts(&self) -> Option<(usize, usize, usize)> {
406            self.publication.read().as_ref().map(|snapshot| snapshot.entry_counts())
407        }
408    }
409
410    impl Drop for Harness {
411        fn drop(&mut self) {
412            // Disconnect and reap the worker thread so it cannot outlive the test.
413            let (disconnected, _) = unbounded();
414            self.commands = disconnected;
415            if let Some(worker) = self.worker.take() {
416                let _ = worker.join();
417            }
418        }
419    }
420
421    /// A pool the tests script per parent: [`Self::push`] hands a transaction to the iterator
422    /// opened for that parent, and unknown parents read as "not tracking this head yet".
423    #[derive(Debug, Default)]
424    struct ScriptedPool {
425        queues: Arc<Mutex<HashMap<B256, VecDeque<PoolTransaction<EthPrimitives>>>>>,
426        /// Iterators handed out; the worker is expected to open exactly one per parent.
427        opened: AtomicUsize,
428        /// Polls answered with "not tracking this head yet".
429        not_ready: AtomicUsize,
430    }
431
432    impl ScriptedPool {
433        fn push(&self, parent_hash: B256, transaction: PoolTransaction<EthPrimitives>) {
434            self.queues.lock().entry(parent_hash).or_default().push_back(transaction);
435        }
436    }
437
438    impl Source<EthPrimitives> for ScriptedPool {
439        fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<EthPrimitives>> {
440            if !self.queues.lock().contains_key(&parent_hash) {
441                self.not_ready.fetch_add(1, Ordering::Relaxed);
442                return None
443            }
444            self.opened.fetch_add(1, Ordering::Relaxed);
445            let queues = Arc::clone(&self.queues);
446            Some(Box::new(std::iter::from_fn(move || {
447                queues.lock().get_mut(&parent_hash)?.pop_front()
448            })))
449        }
450    }
451
452    /// A signed transfer to `recipient`. The signature is a dummy: the worker executes with the
453    /// attached sender and disabled nonce/balance checks, so it is never recovered or validated.
454    fn transfer(recipient: u8) -> PoolTransaction<EthPrimitives> {
455        let transaction = TxLegacy {
456            gas_limit: 21_000,
457            to: TxKind::Call(Address::repeat_byte(recipient)),
458            value: U256::from(1),
459            ..Default::default()
460        };
461        let hash = B256::repeat_byte(recipient);
462        let signed = TransactionSigned::Legacy(Signed::new_unchecked(
463            transaction,
464            Signature::test_signature(),
465            hash,
466        ));
467        let sender = Address::repeat_byte(0xAA);
468        PoolTransaction { hash, sender, transaction: Recovered::new_unchecked(signed, sender) }
469    }
470
471    fn wait_until(what: &str, condition: impl Fn() -> bool) {
472        let deadline = Instant::now() + WAIT_LIMIT;
473        while !condition() {
474            assert!(Instant::now() < deadline, "timed out waiting until {what}");
475            thread::sleep(POLL_INTERVAL);
476        }
477    }
478
479    #[test]
480    fn warms_pool_transactions_into_a_published_snapshot() {
481        let harness = Harness::spawn();
482        let parent_hash = B256::repeat_byte(0x01);
483
484        // Start before the pool tracks the head, covering the poll-and-retry path.
485        harness.start(parent_hash);
486        wait_until("the untracked head is polled", || {
487            harness.pool.not_ready.load(Ordering::Relaxed) >= 1
488        });
489        harness.pool.push(parent_hash, transfer(0xB0));
490
491        let snapshot = harness.published_for(parent_hash);
492        let (accounts, _, _) = snapshot.entry_counts();
493        assert!(accounts >= 1, "speculative execution should cache account reads");
494    }
495
496    #[test]
497    fn pause_quiesces_the_worker_until_resume() {
498        let harness = Harness::spawn();
499        let parent_hash = B256::repeat_byte(0x01);
500        harness.start(parent_hash);
501        harness.pool.push(parent_hash, transfer(0xB0));
502        let before = harness.published_for(parent_hash).entry_counts();
503
504        harness.pause();
505        harness.pool.push(parent_hash, transfer(0xB1));
506        thread::sleep(REFRESH_INTERVAL * 2);
507        assert_eq!(
508            harness.published_entry_counts(),
509            Some(before),
510            "a paused worker must not publish"
511        );
512
513        harness.resume();
514        harness.published(|snapshot| snapshot.entry_counts() != before);
515    }
516
517    #[test]
518    fn overlapping_pauses_require_matching_resumes() {
519        let harness = Harness::spawn();
520        let parent_hash = B256::repeat_byte(0x01);
521        harness.start(parent_hash);
522        harness.pool.push(parent_hash, transfer(0xB0));
523        let before = harness.published_for(parent_hash).entry_counts();
524
525        harness.pause();
526        harness.pause();
527        harness.pool.push(parent_hash, transfer(0xB1));
528        harness.resume();
529        thread::sleep(REFRESH_INTERVAL * 2);
530        assert_eq!(
531            harness.published_entry_counts(),
532            Some(before),
533            "one resume must not release two pauses"
534        );
535
536        harness.resume();
537        harness.published(|snapshot| snapshot.entry_counts() != before);
538    }
539
540    #[test]
541    fn reuses_the_iterator_per_head_and_reopens_on_switch() {
542        let harness = Harness::spawn();
543        let first = B256::repeat_byte(0x01);
544        let second = B256::repeat_byte(0x02);
545
546        harness.start(first);
547        harness.pool.push(first, transfer(0xB0));
548        let before = harness.published_for(first).entry_counts();
549        harness.pool.push(first, transfer(0xB1));
550        harness.published(|snapshot| snapshot.entry_counts() != before);
551        assert_eq!(harness.pool.opened.load(Ordering::Relaxed), 1, "one iterator per head");
552
553        harness.start(second);
554        harness.pool.push(second, transfer(0xB2));
555        harness.published_for(second);
556        assert_eq!(harness.pool.opened.load(Ordering::Relaxed), 2);
557    }
558
559    #[test]
560    fn newest_start_wins() {
561        let harness = Harness::spawn();
562        let stale = B256::repeat_byte(0x01);
563        let newest = B256::repeat_byte(0x02);
564
565        harness.start(stale);
566        harness.start(newest);
567        harness.pool.push(newest, transfer(0xB0));
568
569        harness.published_for(newest);
570    }
571
572    #[test]
573    fn shuts_down_when_control_is_dropped() {
574        let harness = Harness::spawn();
575        let parent_hash = B256::repeat_byte(0x01);
576        harness.start(parent_hash);
577        harness.pool.push(parent_hash, transfer(0xB0));
578        harness.published_for(parent_hash);
579
580        harness.shutdown();
581    }
582}
reth_engine_tree/tree/txpool_prewarm/worker.rs

reth_engine_tree/tree/txpool_prewarm/
worker.rs
