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 = State::builder()
204            .with_database(self.cache.as_db_mut(StateProviderDatabase::new(state_provider)))
205            .build();
206        // The environment is the head block's own, not a predicted next-block one, and execution
207        // is out of context by design: transaction viability is the pool's business, so nonce,
208        // balance and (one-block-stale) basefee checks must not gate which state gets warmed.
209        let mut evm_env = job.evm_env.clone();
210        evm_env.cfg_env.disable_nonce_check = true;
211        evm_env.cfg_env.disable_balance_check = true;
212        evm_env.cfg_env.disable_base_fee = true;
213        let mut evm = self.evm_config.evm_with_env(&mut state, evm_env);
214
215        let deadline = Instant::now() + REFRESH_INTERVAL;
216        while self.commands.is_empty() && Instant::now() < deadline {
217            let Some(transaction) = transactions.next() else { return BatchEnd::Rest };
218            if let Err(err) = evm.transact(transaction.transaction) {
219                trace!(
220                    target: "engine::tree::txpool_prewarm",
221                    %err,
222                    tx_hash = ?transaction.hash,
223                    sender = %transaction.sender,
224                    "speculative txpool transaction execution failed"
225                );
226            }
227        }
228        BatchEnd::GoAgain
229    }
230
231    /// Publishes a fresh snapshot if the cache gained reads since the last publication.
232    fn publish_snapshot_if_dirty(&mut self) {
233        let entries = entry_counts(&self.cache);
234        if entries == self.published_entries {
235            return
236        }
237
238        let parent_hash = self.cache_parent.expect("reads only accumulate after a cache reset");
239        *self.publication.write() = Some(Snapshot::new(parent_hash, Arc::new(self.cache.clone())));
240        self.published_entries = entries;
241        let (accounts, storage, bytecodes) = entries;
242        debug!(
243            target: "engine::tree::txpool_prewarm",
244            ?parent_hash,
245            accounts,
246            storage,
247            bytecodes,
248            "published txpool prewarming snapshot"
249        );
250    }
251
252    /// Rests until a command arrives (applying it) or `timeout` elapses, whichever comes first.
253    fn idle(&mut self, timeout: Duration) -> Result<(), ChannelDisconnected> {
254        match self.commands.recv_timeout(timeout) {
255            Ok(command) => {
256                self.apply(command);
257                Ok(())
258            }
259            Err(RecvTimeoutError::Timeout) => Ok(()),
260            Err(RecvTimeoutError::Disconnected) => Err(ChannelDisconnected),
261        }
262    }
263
264    /// Applies every command already sitting in the channel, without blocking.
265    fn apply_pending_commands(&mut self) -> Result<(), ChannelDisconnected> {
266        loop {
267            match self.commands.try_recv() {
268                Ok(command) => self.apply(command),
269                Err(TryRecvError::Empty) => return Ok(()),
270                Err(TryRecvError::Disconnected) => return Err(ChannelDisconnected),
271            }
272        }
273    }
274
275    /// Applies a single command.
276    ///
277    /// Only called while no EVM or state provider is alive, so a paused worker holds no
278    /// execution resources.
279    fn apply(&mut self, command: Command<Job<N, P, Evm>>) {
280        match command {
281            Command::Start { parent_hash, job } => self.job = Some((parent_hash, job)),
282            Command::Pause => {
283                self.pauses =
284                    self.pauses.checked_add(1).expect("txpool prewarm pause count overflow");
285            }
286            Command::Resume => {
287                self.pauses = self
288                    .pauses
289                    .checked_sub(1)
290                    .expect("txpool prewarm resumed without a matching pause");
291            }
292        }
293    }
294}
295
296/// What to do after a warming batch.
297#[derive(Debug, Clone, Copy, PartialEq, Eq)]
298enum BatchEnd {
299    /// Batch again right away: the refresh deadline passed or a command interrupted the batch
300    /// while transactions may still be ready.
301    GoAgain,
302    /// Idle until something changes: the pool had no transaction ready, or no state provider
303    /// could be built.
304    Rest,
305}
306
307/// The control channel closed: every sender is dropped and the worker shuts down.
308#[derive(Debug, Clone, Copy, PartialEq, Eq)]
309struct ChannelDisconnected;
310
311/// Returns `(accounts, storage slots, bytecodes)` cached in `reads`.
312fn entry_counts(reads: &CachedReads) -> (usize, usize, usize) {
313    (
314        reads.accounts.len(),
315        reads.accounts.values().map(|account| account.storage.len()).sum(),
316        reads.contracts.len(),
317    )
318}
319
320#[cfg(test)]
321mod tests {
322    use super::{super::Transaction as PoolTransaction, *};
323    use alloy_consensus::{transaction::Recovered, Signed, TxLegacy};
324    use alloy_primitives::{Address, Signature, TxKind, U256};
325    use crossbeam_channel::{unbounded, Sender};
326    use parking_lot::{Mutex, RwLock};
327    use reth_ethereum_primitives::{EthPrimitives, TransactionSigned};
328    use reth_evm_ethereum::EthEvmConfig;
329    use reth_provider::test_utils::MockEthProvider;
330    use reth_stages_api::{StageCheckpoint, StageId};
331    use std::{
332        collections::{HashMap, VecDeque},
333        sync::atomic::{AtomicUsize, Ordering},
334        thread::{self, JoinHandle},
335    };
336
337    /// Upper bound on any single wait; failures surface as panics well before CI timeouts.
338    const WAIT_LIMIT: Duration = Duration::from_secs(5);
339    const POLL_INTERVAL: Duration = Duration::from_millis(2);
340
341    type TestJob = Job<EthPrimitives, MockEthProvider, EthEvmConfig>;
342
343    /// Drives a live worker thread through its public seams only: commands in, the publication
344    /// slot and the scripted pool out.
345    struct Harness {
346        commands: Sender<Command<TestJob>>,
347        publication: Publication,
348        pool: Arc<ScriptedPool>,
349        worker: Option<JoinHandle<()>>,
350    }
351
352    impl Harness {
353        fn spawn() -> Self {
354            let (commands, receiver) = unbounded();
355            let publication: Publication = Arc::new(RwLock::new(None));
356            let pool = Arc::new(ScriptedPool::default());
357            let worker = thread::spawn({
358                let publication = Arc::clone(&publication);
359                let source: Arc<dyn Source<EthPrimitives>> = pool.clone();
360                move || Worker::new(receiver, publication, source, EthEvmConfig::mainnet()).run()
361            });
362            Self { commands, publication, pool, worker: Some(worker) }
363        }
364
365        /// Drops the control channel and waits for the worker thread to exit.
366        fn shutdown(mut self) {
367            let (disconnected, _) = unbounded();
368            self.commands = disconnected;
369            let worker = self.worker.take().expect("worker already joined");
370            wait_until("the worker thread exits", || worker.is_finished());
371            worker.join().unwrap();
372        }
373
374        /// Points the worker at `parent_hash`, as [`Handle::start`](super::super::Handle) does.
375        fn start(&self, parent_hash: B256) {
376            let provider = MockEthProvider::default();
377            provider.enable_database_provider();
378            provider.add_header(parent_hash, Default::default());
379            provider.add_stage_checkpoint(StageId::Finish, StageCheckpoint::new(0));
380            let job = Job {
381                evm_env: Default::default(),
382                state_provider_factory: OverlayStateProviderFactory::new(
383                    provider,
384                    reth_storage_overlay::OverlayManager::default().overlay_builder(parent_hash),
385                ),
386            };
387            self.commands.send(Command::Start { parent_hash, job }).unwrap();
388        }
389
390        /// Pauses the worker, as [`Handle::pause`](super::super::Handle) does. Fire-and-forget
391        /// is still deterministic: the worker never publishes work it performs after the pause
392        /// is queued until [`Self::resume`], because publication is skipped while a command is
393        /// pending and a consumed pause blocks the loop.
394        fn pause(&self) {
395            self.commands.send(Command::Pause).unwrap();
396        }
397
398        fn resume(&self) {
399            self.commands.send(Command::Resume).unwrap();
400        }
401
402        /// Blocks until a published snapshot satisfies `accept`.
403        fn published(&self, accept: impl Fn(&Snapshot) -> bool) -> Snapshot {
404            let deadline = Instant::now() + WAIT_LIMIT;
405            loop {
406                let snapshot = self.publication.read().as_ref().cloned();
407                if let Some(snapshot) = snapshot &&
408                    accept(&snapshot)
409                {
410                    return snapshot
411                }
412                assert!(Instant::now() < deadline, "timed out waiting for a matching snapshot");
413                thread::sleep(POLL_INTERVAL);
414            }
415        }
416
417        fn published_for(&self, parent_hash: B256) -> Snapshot {
418            self.published(|snapshot| snapshot.parent_hash() == parent_hash)
419        }
420
421        fn published_entry_counts(&self) -> Option<(usize, usize, usize)> {
422            self.publication.read().as_ref().map(|snapshot| snapshot.entry_counts())
423        }
424    }
425
426    impl Drop for Harness {
427        fn drop(&mut self) {
428            // Disconnect and reap the worker thread so it cannot outlive the test.
429            let (disconnected, _) = unbounded();
430            self.commands = disconnected;
431            if let Some(worker) = self.worker.take() {
432                let _ = worker.join();
433            }
434        }
435    }
436
437    /// A pool the tests script per parent: [`Self::push`] hands a transaction to the iterator
438    /// opened for that parent, and unknown parents read as "not tracking this head yet".
439    #[derive(Debug, Default)]
440    struct ScriptedPool {
441        queues: Arc<Mutex<HashMap<B256, VecDeque<PoolTransaction<EthPrimitives>>>>>,
442        /// Iterators handed out; the worker is expected to open exactly one per parent.
443        opened: AtomicUsize,
444        /// Polls answered with "not tracking this head yet".
445        not_ready: AtomicUsize,
446    }
447
448    impl ScriptedPool {
449        fn push(&self, parent_hash: B256, transaction: PoolTransaction<EthPrimitives>) {
450            self.queues.lock().entry(parent_hash).or_default().push_back(transaction);
451        }
452    }
453
454    impl Source<EthPrimitives> for ScriptedPool {
455        fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<EthPrimitives>> {
456            if !self.queues.lock().contains_key(&parent_hash) {
457                self.not_ready.fetch_add(1, Ordering::Relaxed);
458                return None
459            }
460            self.opened.fetch_add(1, Ordering::Relaxed);
461            let queues = Arc::clone(&self.queues);
462            Some(Box::new(std::iter::from_fn(move || {
463                queues.lock().get_mut(&parent_hash)?.pop_front()
464            })))
465        }
466    }
467
468    /// A signed transfer to `recipient`. The signature is a dummy: the worker executes with the
469    /// attached sender and disabled nonce/balance checks, so it is never recovered or validated.
470    fn transfer(recipient: u8) -> PoolTransaction<EthPrimitives> {
471        let transaction = TxLegacy {
472            gas_limit: 21_000,
473            to: TxKind::Call(Address::repeat_byte(recipient)),
474            value: U256::from(1),
475            ..Default::default()
476        };
477        let hash = B256::repeat_byte(recipient);
478        let signed = TransactionSigned::Legacy(Signed::new_unchecked(
479            transaction,
480            Signature::test_signature(),
481            hash,
482        ));
483        let sender = Address::repeat_byte(0xAA);
484        PoolTransaction { hash, sender, transaction: Recovered::new_unchecked(signed, sender) }
485    }
486
487    fn wait_until(what: &str, condition: impl Fn() -> bool) {
488        let deadline = Instant::now() + WAIT_LIMIT;
489        while !condition() {
490            assert!(Instant::now() < deadline, "timed out waiting until {what}");
491            thread::sleep(POLL_INTERVAL);
492        }
493    }
494
495    #[test]
496    fn warms_pool_transactions_into_a_published_snapshot() {
497        let harness = Harness::spawn();
498        let parent_hash = B256::repeat_byte(0x01);
499
500        // Start before the pool tracks the head, covering the poll-and-retry path.
501        harness.start(parent_hash);
502        wait_until("the untracked head is polled", || {
503            harness.pool.not_ready.load(Ordering::Relaxed) >= 1
504        });
505        harness.pool.push(parent_hash, transfer(0xB0));
506
507        let snapshot = harness.published_for(parent_hash);
508        let (accounts, _, _) = snapshot.entry_counts();
509        assert!(accounts >= 1, "speculative execution should cache account reads");
510    }
511
512    #[test]
513    fn pause_quiesces_the_worker_until_resume() {
514        let harness = Harness::spawn();
515        let parent_hash = B256::repeat_byte(0x01);
516        harness.start(parent_hash);
517        harness.pool.push(parent_hash, transfer(0xB0));
518        let before = harness.published_for(parent_hash).entry_counts();
519
520        harness.pause();
521        harness.pool.push(parent_hash, transfer(0xB1));
522        thread::sleep(REFRESH_INTERVAL * 2);
523        assert_eq!(
524            harness.published_entry_counts(),
525            Some(before),
526            "a paused worker must not publish"
527        );
528
529        harness.resume();
530        harness.published(|snapshot| snapshot.entry_counts() != before);
531    }
532
533    #[test]
534    fn overlapping_pauses_require_matching_resumes() {
535        let harness = Harness::spawn();
536        let parent_hash = B256::repeat_byte(0x01);
537        harness.start(parent_hash);
538        harness.pool.push(parent_hash, transfer(0xB0));
539        let before = harness.published_for(parent_hash).entry_counts();
540
541        harness.pause();
542        harness.pause();
543        harness.pool.push(parent_hash, transfer(0xB1));
544        harness.resume();
545        thread::sleep(REFRESH_INTERVAL * 2);
546        assert_eq!(
547            harness.published_entry_counts(),
548            Some(before),
549            "one resume must not release two pauses"
550        );
551
552        harness.resume();
553        harness.published(|snapshot| snapshot.entry_counts() != before);
554    }
555
556    #[test]
557    fn reuses_the_iterator_per_head_and_reopens_on_switch() {
558        let harness = Harness::spawn();
559        let first = B256::repeat_byte(0x01);
560        let second = B256::repeat_byte(0x02);
561
562        harness.start(first);
563        harness.pool.push(first, transfer(0xB0));
564        let before = harness.published_for(first).entry_counts();
565        harness.pool.push(first, transfer(0xB1));
566        harness.published(|snapshot| snapshot.entry_counts() != before);
567        assert_eq!(harness.pool.opened.load(Ordering::Relaxed), 1, "one iterator per head");
568
569        harness.start(second);
570        harness.pool.push(second, transfer(0xB2));
571        harness.published_for(second);
572        assert_eq!(harness.pool.opened.load(Ordering::Relaxed), 2);
573    }
574
575    #[test]
576    fn newest_start_wins() {
577        let harness = Harness::spawn();
578        let stale = B256::repeat_byte(0x01);
579        let newest = B256::repeat_byte(0x02);
580
581        harness.start(stale);
582        harness.start(newest);
583        harness.pool.push(newest, transfer(0xB0));
584
585        harness.published_for(newest);
586    }
587
588    #[test]
589    fn shuts_down_when_control_is_dropped() {
590        let harness = Harness::spawn();
591        let parent_hash = B256::repeat_byte(0x01);
592        harness.start(parent_hash);
593        harness.pool.push(parent_hash, transfer(0xB0));
594        harness.published_for(parent_hash);
595
596        harness.shutdown();
597    }
598}
reth_engine_tree/tree/txpool_prewarm/worker.rs

reth_engine_tree/tree/txpool_prewarm/
worker.rs
