reth_engine_tree/tree/txpool_prewarm/
mod.rs

1//! Txpool-driven state prewarming and immutable snapshot publication.
2
3mod control;
4mod worker;
5
6use self::control::Control;
7use crate::tree::{StateProviderBuilder, TxPoolPrewarmCacheSnapshot};
8use alloy_consensus::transaction::Recovered;
9use alloy_primitives::{Address, B256};
10use reth_evm::{ConfigureEvm, EvmEnvFor};
11use reth_primitives_traits::{NodePrimitives, TxTy};
12use reth_provider::{
13    BlockNumReader, DatabaseProviderFactory, PruneCheckpointReader, StageCheckpointReader,
14    StorageSettingsCache, TryIntoHistoricalStateProvider,
15};
16use std::{fmt::Debug, sync::Arc};
17
18/// Coordinates a long-lived worker and the latest completed immutable snapshot.
19pub(crate) struct Handle<N, P, Evm>
20where
21    N: NodePrimitives,
22    Evm: ConfigureEvm<Primitives = N>,
23{
24    control: Arc<Control<Job<N, P, Evm>>>,
25}
26
27impl<N, P, Evm> Debug for Handle<N, P, Evm>
28where
29    N: NodePrimitives,
30    Evm: ConfigureEvm<Primitives = N>,
31{
32    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33        f.debug_struct("Handle").field("control", &self.control).finish()
34    }
35}
36
37impl<N, P, Evm> Handle<N, P, Evm>
38where
39    N: NodePrimitives,
40    P: DatabaseProviderFactory + 'static,
41    P::Provider: BlockNumReader
42        + PruneCheckpointReader
43        + StageCheckpointReader
44        + StorageSettingsCache
45        + TryIntoHistoricalStateProvider
46        + 'static,
47    Evm: ConfigureEvm<Primitives = N> + 'static,
48{
49    /// Spawns the long-lived worker, which owns its mutable read cache and starts a fresh one for
50    /// each new head.
51    pub(crate) fn spawn(
52        runtime: &reth_tasks::Runtime,
53        source: Arc<dyn Source<N>>,
54        evm_config: Evm,
55    ) -> Self {
56        let (control, commands) = Control::new();
57        let publication = control.publication();
58        runtime.spawn_critical_os_thread("txpool-prewarm", "txpool prewarm worker", async move {
59            worker::Worker::new(commands, publication, source, evm_config).run()
60        });
61        Self { control }
62    }
63
64    /// Pauses speculative work.
65    ///
66    /// Returns a guard that will resume the worker when dropped. There could be multiple
67    /// outstanding guards, in which case the worker will not resume until all guards are dropped.
68    ///
69    /// Pausing is asynchronous and never blocks the caller: the worker observes it between
70    /// transactions, so speculative work may overlap the guard's scope by at most one
71    /// transaction.
72    pub(crate) fn pause(&self) -> impl Drop + Send + 'static {
73        self.control.pause()
74    }
75
76    /// Returns the latest fully published snapshot for `parent_hash`, or `None` if no snapshot is
77    /// available for that hash.
78    pub(crate) fn snapshot(&self, parent_hash: B256) -> Option<TxPoolPrewarmCacheSnapshot> {
79        self.control.snapshot(parent_hash)
80    }
81
82    /// Starts continuous warming for the latest canonical head.
83    pub(crate) fn start(
84        &self,
85        parent_hash: B256,
86        evm_env: EvmEnvFor<Evm>,
87        provider_builder: StateProviderBuilder<N, P>,
88    ) {
89        self.control.start(parent_hash, Job { evm_env, provider_builder });
90    }
91}
92
93/// A live, forward-only view of the pool's best transactions for one canonical parent.
94///
95/// Returning [`None`](Iterator::next) only means no transaction is currently ready. The same
96/// iterator can yield transactions that become pending later.
97pub type Transactions<N> = Box<dyn Iterator<Item = Transaction<N>> + Send>;
98
99/// A transaction selected from the txpool for cache-only prewarming.
100#[derive(Debug, Clone)]
101pub struct Transaction<N: NodePrimitives> {
102    /// Transaction hash.
103    pub hash: B256,
104    /// Recovered sender.
105    pub sender: Address,
106    /// Recovered consensus transaction.
107    pub transaction: Recovered<TxTy<N>>,
108}
109
110/// Source of txpool transactions for best-effort cache prewarming.
111pub trait Source<N: NodePrimitives>: Send + Sync + Debug {
112    /// Opens a live best-transactions iterator for `parent_hash`.
113    ///
114    /// The worker opens this once per canonical parent and retains it across empty polls, snapshot
115    /// publications, and validation pauses. Sources should return [`None`] if they are not yet
116    /// tracking `parent_hash`.
117    fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<N>>;
118}
119
120/// A request to warm txpool transactions against one fully validated parent state.
121struct Job<N: NodePrimitives, P, Evm: ConfigureEvm<Primitives = N>> {
122    evm_env: EvmEnvFor<Evm>,
123    provider_builder: StateProviderBuilder<N, P>,
124}
reth_engine_tree/tree/txpool_prewarm/mod.rs

reth_engine_tree/tree/txpool_prewarm/
mod.rs
