reth_execution_cache/
lib.rs

1//! Cross-block execution cache for payload processing.
2//!
3//! This crate provides the core caching infrastructure used during block execution:
4//! - [`ExecutionCache`]: Fixed-size concurrent caches for accounts, storage, and bytecode
5//! - [`SavedCache`]: An execution cache snapshot associated with a specific block hash
6//! - [`PayloadExecutionCache`]: Thread-safe wrapper for sharing cached state across payload
7//!   processing tasks
8
9#![doc(
10    html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
11    html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
12    issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
13)]
14#![cfg_attr(docsrs, feature(doc_cfg))]
15#![cfg_attr(not(test), warn(unused_crate_dependencies))]
16
17mod cached_state;
18pub use cached_state::*;
19
20mod txpool;
21pub use txpool::*;
22
23use alloy_primitives::B256;
24use metrics::{Counter, Histogram};
25use parking_lot::Mutex;
26use reth_metrics::Metrics;
27use reth_primitives_traits::FastInstant as Instant;
28use std::{sync::Arc, time::Duration};
29use tracing::{debug, instrument, warn};
30
31/// A guarded, thread-safe cache of execution state that tracks the most recent block's caches.
32///
33/// This is the cross-block cache used to accelerate sequential payload processing.
34/// When a new block arrives, its parent's cached state can be reused to avoid
35/// redundant database lookups.
36///
37/// This process assumes that payloads are received sequentially.
38///
39/// ## Cache Safety
40///
41/// **CRITICAL**: Cache update operations require exclusive access. All concurrent cache users
42/// (such as prewarming tasks) must be terminated before calling
43/// [`PayloadExecutionCache::update_with_guard`], otherwise the cache may be corrupted or cleared.
44#[derive(Clone, Debug, Default)]
45pub struct PayloadExecutionCache {
46    /// Guarded cloneable cache identified by a block hash.
47    inner: Arc<Mutex<Option<SavedCache>>>,
48    /// Metrics for cache operations.
49    metrics: PayloadExecutionCacheMetrics,
50}
51
52impl PayloadExecutionCache {
53    /// Returns the cache for `parent_hash` if it's available for use.
54    ///
55    /// See [`Self::checkout`] for when a cache is available.
56    pub fn get_cache_for(&self, parent_hash: B256) -> Option<SavedCache> {
57        match self.checkout(parent_hash) {
58            CacheCheckout::Available(cache) => Some(cache),
59            CacheCheckout::Empty | CacheCheckout::InUse { .. } => None,
60        }
61    }
62
63    /// Checks out the stored cache for `parent_hash`.
64    ///
65    /// The stored cache is available when no other task holds a reference to it (checked via the
66    /// Arc reference count). If it was saved for another block, it is cleared and re-keyed to
67    /// `parent_hash` before it is returned.
68    #[instrument(level = "debug", target = "engine::tree::payload_processor", skip(self))]
69    pub fn checkout(&self, parent_hash: B256) -> CacheCheckout {
70        let start = Instant::now();
71        let mut cache = self.inner.lock();
72
73        let elapsed = start.elapsed();
74        self.metrics.execution_cache_wait_duration.record(elapsed.as_secs_f64());
75        if elapsed.as_millis() > 5 {
76            warn!(blocked_for=?elapsed, "Blocked waiting for execution cache mutex");
77        }
78
79        if let Some(c) = cache.as_mut() {
80            let cached_hash = c.executed_block_hash();
81            // Check that the cache hash matches the parent hash of the current block. It won't
82            // match in case it's a fork block.
83            let hash_matches = cached_hash == parent_hash;
84            // Check `is_available()` to ensure no other tasks (e.g., prewarming) currently hold
85            // a reference to this cache. We can only reuse it when we have exclusive access.
86            let available = c.is_available();
87            let usage_count = c.usage_count();
88
89            debug!(
90                target: "engine::caching",
91                %cached_hash,
92                %parent_hash,
93                hash_matches,
94                available,
95                usage_count,
96                "Existing cache found"
97            );
98
99            if available {
100                if !hash_matches {
101                    // Fork block: clear and update the hash on the ORIGINAL before cloning.
102                    // This prevents the canonical chain from matching on the stale hash
103                    // and picking up polluted data if the fork block fails.
104                    c.clear_with_hash(parent_hash);
105                }
106                return CacheCheckout::Available(c.clone())
107            }
108            self.metrics.execution_cache_in_use.increment(1);
109            return CacheCheckout::InUse { cached_hash }
110        }
111
112        debug!(target: "engine::caching", %parent_hash, "No cache found");
113        CacheCheckout::Empty
114    }
115
116    /// Waits for the mutex protecting the stored `Option<SavedCache>` to be released.
117    ///
118    /// This does not wait for other users to drop their [`ExecutionCache`] clones or for removed
119    /// caches to finish dropping after unlocking. A subsequent [`Self::get_cache_for`] can still
120    /// return `None`, causing its caller to allocate a new cache while those drops run.
121    ///
122    /// Returns only the time spent waiting for the mutex, excluding post-unlock cleanup.
123    pub fn wait_for_availability(&self) -> Duration {
124        let start = Instant::now();
125        // Acquire lock to wait for any current holders to finish
126        let _guard = self.inner.lock();
127        let elapsed = start.elapsed();
128        if elapsed.as_millis() > 5 {
129            debug!(
130                target: "engine::tree::payload_processor",
131                blocked_for=?elapsed,
132                "Waited for execution cache to become available"
133            );
134        }
135        elapsed
136    }
137
138    /// Runs `update_fn` with mutable access to the stored `Option<SavedCache>` under the mutex.
139    /// Returns the closure's result after releasing the mutex, allowing removed caches to be
140    /// dropped outside the lock.
141    ///
142    /// Drop extra [`SavedCache`] or [`ExecutionCache`] clones of the stored cache before the
143    /// closure returns: [`Self::get_cache_for`] requires that cache's Arc strong reference
144    /// count to be one.
145    ///
146    /// ## CRITICAL SAFETY REQUIREMENT
147    ///
148    /// **Before calling this method, you MUST ensure there are no other active cache users.**
149    /// This includes:
150    /// - No running prewarming task instances that could write to the cache
151    /// - No concurrent transactions that might access the cached state
152    /// - All prewarming operations must be completed or cancelled
153    ///
154    /// Violating this requirement can result in cache corruption, incorrect state data,
155    /// and potential consensus failures.
156    pub fn update_with_guard<F, R>(&self, update_fn: F) -> R
157    where
158        F: FnOnce(&mut Option<SavedCache>) -> R,
159    {
160        let mut guard = self.inner.lock();
161        update_fn(&mut guard)
162    }
163}
164
165/// Outcome of [`PayloadExecutionCache::checkout`].
166#[derive(Debug)]
167pub enum CacheCheckout {
168    /// The stored cache, keyed to the requested parent hash.
169    Available(SavedCache),
170    /// No cache is stored yet.
171    Empty,
172    /// The stored cache is still held by another task, such as a prewarm task that has not saved
173    /// its block yet.
174    InUse {
175        /// Hash of the block whose post-state the stored cache holds.
176        cached_hash: B256,
177    },
178}
179
180/// Metrics for [`PayloadExecutionCache`] operations.
181#[derive(Metrics, Clone)]
182#[metrics(scope = "consensus.engine.beacon")]
183struct PayloadExecutionCacheMetrics {
184    /// Counter for when the execution cache was unavailable because other threads
185    /// (e.g., prewarming) are still using it.
186    execution_cache_in_use: Counter,
187    /// Time spent waiting for execution cache mutex to become available.
188    execution_cache_wait_duration: Histogram,
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194
195    #[test]
196    fn single_checkout_blocks_second() {
197        let cache = PayloadExecutionCache::default();
198        let hash = B256::repeat_byte(1u8);
199
200        cache.update_with_guard(|slot| {
201            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
202        });
203
204        let first = cache.get_cache_for(hash);
205        assert!(first.is_some());
206
207        let second = cache.get_cache_for(hash);
208        assert!(second.is_none());
209    }
210
211    #[test]
212    fn checkout_available_after_drop() {
213        let cache = PayloadExecutionCache::default();
214        let hash = B256::repeat_byte(2u8);
215
216        cache.update_with_guard(|slot| {
217            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
218        });
219
220        let checked_out = cache.get_cache_for(hash);
221        assert!(checked_out.is_some());
222        drop(checked_out);
223
224        let second = cache.get_cache_for(hash);
225        assert!(second.is_some());
226    }
227
228    #[test]
229    fn raw_cache_handle_blocks_checkout_until_drop() {
230        let cache = PayloadExecutionCache::default();
231        let hash = B256::repeat_byte(3u8);
232
233        cache.update_with_guard(|slot| {
234            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
235        });
236
237        let checked_out = cache.get_cache_for(hash).expect("checkout should succeed");
238        let cache_handle = checked_out.cache().clone();
239        drop(checked_out);
240
241        let blocked = cache.get_cache_for(hash);
242        assert!(blocked.is_none(), "raw ExecutionCache handle should keep slot in use");
243
244        drop(cache_handle);
245
246        let available = cache.get_cache_for(hash);
247        assert!(available.is_some(), "checkout should succeed after raw handle is dropped");
248    }
249
250    #[test]
251    fn hash_mismatch_clears_and_retags() {
252        let cache = PayloadExecutionCache::default();
253        let hash_a = B256::repeat_byte(0xAA);
254        let hash_b = B256::repeat_byte(0xBB);
255
256        cache.update_with_guard(|slot| {
257            *slot = Some(SavedCache::new(hash_a, ExecutionCache::new(1_000)))
258        });
259
260        let checked_out = cache.get_cache_for(hash_b);
261        assert!(checked_out.is_some());
262        assert_eq!(checked_out.unwrap().executed_block_hash(), hash_b);
263    }
264
265    #[test]
266    fn empty_cache_returns_none() {
267        let cache = PayloadExecutionCache::default();
268        assert!(cache.get_cache_for(B256::ZERO).is_none());
269    }
270
271    #[test]
272    fn checkout_distinguishes_empty_from_in_use() {
273        let cache = PayloadExecutionCache::default();
274        let hash = B256::repeat_byte(7u8);
275        assert!(matches!(cache.checkout(hash), CacheCheckout::Empty));
276
277        cache.update_with_guard(|slot| {
278            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
279        });
280        let held = cache.get_cache_for(hash).expect("checkout should succeed");
281
282        assert!(matches!(
283            cache.checkout(B256::repeat_byte(8u8)),
284            CacheCheckout::InUse { cached_hash } if cached_hash == hash
285        ));
286        drop(held);
287    }
288}
reth_execution_cache/lib.rs

reth_execution_cache/
lib.rs
