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

reth_engine_tree/tree/txpool_prewarm/
mod.rs
