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

reth_engine_tree/tree/txpool_prewarm/
worker.rs
