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

reth_engine_tree/tree/txpool_prewarm/
mod.rs
