reth_engine_tree/tree/txpool_prewarm/
mod.rs

1//! Txpool-driven state prewarming and immutable snapshot publication.
2//!
3//! If block execution reuses prewarm results, the worker executes pool transactions in a predicted
4//! environment of the next block and records their results, which prewarming of that block may
5//! reuse, see [`PoolResults`].
6
7mod control;
8mod worker;
9
10use self::control::Control;
11use crate::tree::{
12    payload_processor::reuse::{
13        cfg_reuses_prewarm_results, BlockEnvFields, PoolResults, PoolSeeds, POOL_BENEFICIARY,
14    },
15    TxPoolPrewarmCacheSnapshot,
16};
17use alloy_consensus::transaction::Recovered;
18use alloy_primitives::{Address, B256};
19use parking_lot::RwLock;
20use reth_evm::{ConfigureEvm, EvmEnv, EvmEnvFor, HaltReasonFor};
21use reth_primitives_traits::{NodePrimitives, SealedHeader, TxTy};
22use reth_provider::{
23    BlockNumReader, ChangeSetReader, DatabaseProviderFactory, DatabaseProviderROFactory,
24    PruneCheckpointReader, StageCheckpointReader, StateProvider, StorageChangeSetReader,
25    StorageSettingsCache,
26};
27use reth_storage_overlay::OverlayStateProviderFactory;
28use revm::context::Block;
29use std::{fmt::Debug, sync::Arc};
30
31/// Coordinates a long-lived worker and the latest completed immutable snapshot.
32pub(crate) struct Handle<N, P, Evm>
33where
34    N: NodePrimitives,
35    Evm: ConfigureEvm<Primitives = N>,
36{
37    control: Arc<Control<Job<N, P, Evm>>>,
38    /// Predicts the environment of the next block.
39    evm_config: Evm,
40    /// Whether to record results of pool transactions in the predicted environment of the next
41    /// block.
42    record_results: bool,
43    /// The parent, predicted environment and results of the recorded speculation, if any.
44    speculation: RwLock<Option<Speculation<Evm>>>,
45}
46
47impl<N, P, Evm> Debug for Handle<N, P, Evm>
48where
49    N: NodePrimitives,
50    Evm: ConfigureEvm<Primitives = N>,
51{
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        f.debug_struct("Handle").field("control", &self.control).finish()
54    }
55}
56
57impl<N, P, Evm> Handle<N, P, Evm>
58where
59    N: NodePrimitives,
60    P: DatabaseProviderFactory + 'static,
61    P::Provider: BlockNumReader
62        + PruneCheckpointReader
63        + StageCheckpointReader
64        + ChangeSetReader
65        + StorageChangeSetReader
66        + StorageSettingsCache
67        + 'static,
68    OverlayStateProviderFactory<P, N>: DatabaseProviderROFactory<Provider: StateProvider> + Send,
69    Evm: ConfigureEvm<Primitives = N> + 'static,
70{
71    /// Spawns the long-lived worker, which owns its mutable read cache and starts a fresh one for
72    /// each new head.
73    ///
74    /// If `record_results` is set, the worker records results of pool transactions that
75    /// prewarming may reuse.
76    pub(crate) fn spawn(
77        runtime: &reth_tasks::Runtime,
78        source: Arc<dyn Source<N>>,
79        evm_config: Evm,
80        record_results: bool,
81    ) -> Self {
82        let (control, commands) = Control::new();
83        let publication = control.publication();
84        let worker_evm_config = evm_config.clone();
85        runtime.spawn_critical_os_thread("txpool-prewarm", "txpool prewarm worker", async move {
86            worker::Worker::new(commands, publication, source, worker_evm_config).run()
87        });
88        Self { control, evm_config, record_results, speculation: RwLock::new(None) }
89    }
90
91    /// Pauses speculative work.
92    ///
93    /// Returns a guard that will resume the worker when dropped. There could be multiple
94    /// outstanding guards, in which case the worker will not resume until all guards are dropped.
95    ///
96    /// Pausing is asynchronous and never blocks the caller: the worker observes it between
97    /// transactions, so speculative work may overlap the guard's scope by at most one
98    /// transaction.
99    pub(crate) fn pause(&self) -> impl Drop + Send + 'static {
100        self.control.pause()
101    }
102
103    /// Returns the latest fully published snapshot for `parent_hash`, or `None` if no snapshot is
104    /// available for that hash.
105    pub(crate) fn snapshot(&self, parent_hash: B256) -> Option<TxPoolPrewarmCacheSnapshot> {
106        self.control.snapshot(parent_hash)
107    }
108
109    /// Starts continuous warming for the latest canonical head `parent`.
110    ///
111    /// Results are recorded in the predicted environment of the next block if it is available and
112    /// the block may reuse them, otherwise warming uses the head block's own environment
113    /// `evm_env`.
114    pub(crate) fn start(
115        &self,
116        parent: &SealedHeader<N::BlockHeader>,
117        evm_env: EvmEnvFor<Evm>,
118        state_provider_factory: OverlayStateProviderFactory<P, N>,
119    ) {
120        let mut speculation = self.speculation.write();
121        // The worker keeps executed transactions of the same head, so it would not record them
122        // again.
123        if speculation.as_ref().is_none_or(|s| s.parent_hash != parent.hash()) {
124            *speculation = self
125                .record_results
126                .then(|| self.evm_config.speculative_next_evm_env(parent, POOL_BENEFICIARY))
127                .flatten()
128                .filter(|evm_env| cfg_reuses_prewarm_results(&evm_env.cfg_env))
129                .map(|evm_env| Speculation {
130                    parent_hash: parent.hash(),
131                    evm_env,
132                    results: Arc::default(),
133                });
134        }
135        let job = Job {
136            evm_env: speculation.as_ref().map_or(evm_env, |s| s.evm_env.clone()),
137            state_provider_factory,
138            results: speculation.as_ref().map(|s| Arc::clone(&s.results)),
139        };
140        drop(speculation);
141        self.control.start(parent.hash(), job);
142    }
143
144    /// Returns the recorded results of pool transactions on top of `parent_hash` that prewarming
145    /// of a block with `evm_env` may reuse, `None` if there are none.
146    pub(crate) fn seeds(
147        &self,
148        parent_hash: B256,
149        evm_env: &EvmEnvFor<Evm>,
150    ) -> Option<PoolSeeds<HaltReasonFor<Evm>>> {
151        let speculation = self.speculation.read();
152        let speculation = speculation
153            .as_ref()
154            .filter(|s| s.parent_hash == parent_hash && !s.results.is_empty())?;
155        let differs = env_differences(&speculation.evm_env, evm_env)?;
156        Some(PoolSeeds::new(
157            Arc::clone(&speculation.results),
158            evm_env.block_env.beneficiary(),
159            differs,
160        ))
161    }
162}
163
164/// Results of pool transactions recorded on top of a parent in a predicted block environment.
165struct Speculation<Evm: ConfigureEvm> {
166    parent_hash: B256,
167    evm_env: EvmEnvFor<Evm>,
168    results: Arc<PoolResults<HaltReasonFor<Evm>>>,
169}
170
171/// Returns the block environment fields that differ between the `predicted` environment and the
172/// `actual` one, or `None` if they differ in a way that every execution may depend on.
173fn env_differences<Spec: Clone + PartialEq, B: Block>(
174    predicted: &EvmEnv<Spec, B>,
175    actual: &EvmEnv<Spec, B>,
176) -> Option<BlockEnvFields> {
177    let mut cfg_env = predicted.cfg_env.clone();
178    // Results are recorded with these checks disabled and show that the checks would pass.
179    cfg_env.disable_nonce_check = actual.cfg_env.disable_nonce_check;
180    cfg_env.disable_balance_check = actual.cfg_env.disable_balance_check;
181    if cfg_env != actual.cfg_env {
182        return None
183    }
184    let (predicted, actual) = (&predicted.block_env, &actual.block_env);
185    (predicted.number() == actual.number() &&
186        predicted.basefee() == actual.basefee() &&
187        predicted.difficulty() == actual.difficulty() &&
188        predicted.blob_excess_gas_and_price() == actual.blob_excess_gas_and_price() &&
189        predicted.slot_num() == actual.slot_num())
190    .then(|| BlockEnvFields {
191        // Recorded results never access the beneficiary.
192        beneficiary: false,
193        timestamp: predicted.timestamp() != actual.timestamp(),
194        prevrandao: predicted.prevrandao() != actual.prevrandao(),
195        gas_limit: predicted.gas_limit() != actual.gas_limit(),
196    })
197}
198
199/// A live, forward-only view of the pool's best transactions for one canonical parent.
200///
201/// Returning [`None`](Iterator::next) only means no transaction is currently ready. The same
202/// iterator can yield transactions that become pending later.
203pub type Transactions<N> = Box<dyn Iterator<Item = Transaction<N>> + Send>;
204
205/// A transaction selected from the txpool for prewarming.
206#[derive(Debug, Clone)]
207pub struct Transaction<N: NodePrimitives> {
208    /// Transaction hash.
209    pub hash: B256,
210    /// Recovered sender.
211    pub sender: Address,
212    /// Recovered consensus transaction.
213    pub transaction: Recovered<TxTy<N>>,
214}
215
216/// Source of txpool transactions for best-effort prewarming.
217pub trait Source<N: NodePrimitives>: Send + Sync + Debug {
218    /// Opens a live best-transactions iterator for `parent_hash`.
219    ///
220    /// The worker opens this once per canonical parent and retains it across empty polls, snapshot
221    /// publications, and validation pauses. Sources should return [`None`] if they are not yet
222    /// tracking `parent_hash`.
223    fn best_transactions(&self, parent_hash: B256) -> Option<Transactions<N>>;
224}
225
226/// A request to warm txpool transactions against one fully validated parent state.
227struct Job<N: NodePrimitives, P, Evm: ConfigureEvm<Primitives = N>> {
228    evm_env: EvmEnvFor<Evm>,
229    state_provider_factory: OverlayStateProviderFactory<P, N>,
230    /// Where to record results of pool transactions, if `evm_env` is the predicted environment
231    /// of the next block.
232    results: Option<Arc<PoolResults<HaltReasonFor<Evm>>>>,
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use alloy_consensus::Header;
239    use alloy_primitives::U256;
240    use reth_chainspec::ChainSpecBuilder;
241    use reth_ethereum_primitives::EthPrimitives;
242    use reth_evm_ethereum::EthEvmConfig;
243    use reth_provider::test_utils::MockEthProvider;
244    use reth_storage_overlay::OverlayManager;
245    use reth_tasks::Runtime;
246    use revm::context::result::HaltReason;
247
248    type TestHandle = Handle<EthPrimitives, MockEthProvider, EthEvmConfig>;
249
250    /// A pool that never tracks the head.
251    #[derive(Debug)]
252    struct NoPool;
253
254    impl Source<EthPrimitives> for NoPool {
255        fn best_transactions(&self, _parent_hash: B256) -> Option<Transactions<EthPrimitives>> {
256            None
257        }
258    }
259
260    fn start(handle: &TestHandle, parent: &SealedHeader) {
261        let state_provider_factory = OverlayStateProviderFactory::new(
262            MockEthProvider::default(),
263            OverlayManager::default().overlay_builder(parent.hash()),
264        );
265        handle.start(parent, Default::default(), state_provider_factory);
266    }
267
268    fn recorded(handle: &TestHandle) -> Option<Arc<PoolResults<HaltReason>>> {
269        handle.speculation.read().as_ref().map(|s| Arc::clone(&s.results))
270    }
271
272    #[test]
273    fn keeps_results_of_the_same_head() {
274        let handle =
275            TestHandle::spawn(&Runtime::test(), Arc::new(NoPool), EthEvmConfig::mainnet(), true);
276        let parent = SealedHeader::seal_slow(Header::default());
277        start(&handle, &parent);
278        let results = recorded(&handle).unwrap();
279        start(&handle, &parent);
280        assert!(Arc::ptr_eq(&results, &recorded(&handle).unwrap()));
281
282        start(&handle, &SealedHeader::seal_slow(Header { number: 1, ..Default::default() }));
283        assert!(!Arc::ptr_eq(&results, &recorded(&handle).unwrap()));
284    }
285
286    #[test]
287    fn offers_no_seeds_without_results() {
288        let handle =
289            TestHandle::spawn(&Runtime::test(), Arc::new(NoPool), EthEvmConfig::mainnet(), true);
290        let parent = SealedHeader::seal_slow(Header::default());
291        start(&handle, &parent);
292        let evm_env = handle.speculation.read().as_ref().unwrap().evm_env.clone();
293        assert!(handle.seeds(parent.hash(), &evm_env).is_none());
294    }
295
296    #[test]
297    fn records_no_results_with_eip8037() {
298        let chain_spec = ChainSpecBuilder::mainnet().amsterdam_activated().build();
299        let evm_config = EthEvmConfig::new(Arc::new(chain_spec));
300        let handle = TestHandle::spawn(&Runtime::test(), Arc::new(NoPool), evm_config, true);
301        start(&handle, &SealedHeader::seal_slow(Header::default()));
302        assert!(recorded(&handle).is_none());
303    }
304
305    #[test]
306    fn predicted_env_differs_in_unobserved_fields() {
307        let config = EthEvmConfig::mainnet();
308        // A half full parent without blobs keeps the base fee and excess blob gas.
309        let parent = Header {
310            number: 25_000_000,
311            timestamp: 1_800_000_000,
312            gas_limit: 60_000_000,
313            gas_used: 30_000_000,
314            base_fee_per_gas: Some(1_000_000_000),
315            excess_blob_gas: Some(0),
316            blob_gas_used: Some(0),
317            ..Default::default()
318        };
319        let block = Header {
320            number: parent.number + 1,
321            timestamp: parent.timestamp + 12,
322            beneficiary: Address::repeat_byte(1),
323            mix_hash: B256::repeat_byte(2),
324            gas_limit: parent.gas_limit + 1,
325            ..parent.clone()
326        };
327        let mut predicted = config
328            .speculative_next_evm_env(&SealedHeader::seal_slow(parent), POOL_BENEFICIARY)
329            .unwrap();
330        // Recorded results disable the nonce check.
331        predicted.cfg_env.disable_nonce_check = true;
332        let mut actual = config.evm_env(&block).unwrap();
333        assert_eq!(
334            env_differences(&predicted, &actual),
335            Some(BlockEnvFields { prevrandao: true, gas_limit: true, ..Default::default() })
336        );
337
338        actual.block_env.timestamp += U256::from(12);
339        assert_eq!(
340            env_differences(&predicted, &actual),
341            Some(BlockEnvFields {
342                timestamp: true,
343                prevrandao: true,
344                gas_limit: true,
345                ..Default::default()
346            })
347        );
348
349        let mut other_chain = actual.clone();
350        other_chain.cfg_env.chain_id += 1;
351        assert_eq!(env_differences(&predicted, &other_chain), None);
352
353        actual.block_env.basefee += 1;
354        assert_eq!(env_differences(&predicted, &actual), None);
355    }
356}
reth_engine_tree/tree/txpool_prewarm/mod.rs

reth_engine_tree/tree/txpool_prewarm/
mod.rs
