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

reth_engine_tree/tree/txpool_prewarm/
mod.rs
