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    /// A cache is considered available when:
56    /// - It exists and matches the requested parent hash
57    /// - No other tasks are currently using it (checked via Arc reference count)
58    #[instrument(level = "debug", target = "engine::tree::payload_processor", skip(self))]
59    pub fn get_cache_for(&self, parent_hash: B256) -> Option<SavedCache> {
60        let start = Instant::now();
61        let mut cache = self.inner.lock();
62
63        let elapsed = start.elapsed();
64        self.metrics.execution_cache_wait_duration.record(elapsed.as_secs_f64());
65        if elapsed.as_millis() > 5 {
66            warn!(blocked_for=?elapsed, "Blocked waiting for execution cache mutex");
67        }
68
69        if let Some(c) = cache.as_mut() {
70            let cached_hash = c.executed_block_hash();
71            // Check that the cache hash matches the parent hash of the current block. It won't
72            // match in case it's a fork block.
73            let hash_matches = cached_hash == parent_hash;
74            // Check `is_available()` to ensure no other tasks (e.g., prewarming) currently hold
75            // a reference to this cache. We can only reuse it when we have exclusive access.
76            let available = c.is_available();
77            let usage_count = c.usage_count();
78
79            debug!(
80                target: "engine::caching",
81                %cached_hash,
82                %parent_hash,
83                hash_matches,
84                available,
85                usage_count,
86                "Existing cache found"
87            );
88
89            if available {
90                if !hash_matches {
91                    // Fork block: clear and update the hash on the ORIGINAL before cloning.
92                    // This prevents the canonical chain from matching on the stale hash
93                    // and picking up polluted data if the fork block fails.
94                    c.clear_with_hash(parent_hash);
95                }
96                return Some(c.clone())
97            } else if hash_matches {
98                self.metrics.execution_cache_in_use.increment(1);
99            }
100        } else {
101            debug!(target: "engine::caching", %parent_hash, "No cache found");
102        }
103
104        None
105    }
106
107    /// Waits until the execution cache becomes available for use.
108    ///
109    /// This acquires a write lock to ensure exclusive access, then immediately releases it.
110    /// This is useful for synchronization before starting payload processing.
111    ///
112    /// Returns the time spent waiting for the lock.
113    pub fn wait_for_availability(&self) -> Duration {
114        let start = Instant::now();
115        // Acquire lock to wait for any current holders to finish
116        let _guard = self.inner.lock();
117        let elapsed = start.elapsed();
118        if elapsed.as_millis() > 5 {
119            debug!(
120                target: "engine::tree::payload_processor",
121                blocked_for=?elapsed,
122                "Waited for execution cache to become available"
123            );
124        }
125        elapsed
126    }
127
128    /// Updates the cache with a closure that has exclusive access to the guard.
129    /// This ensures that all cache operations happen atomically.
130    ///
131    /// ## CRITICAL SAFETY REQUIREMENT
132    ///
133    /// **Before calling this method, you MUST ensure there are no other active cache users.**
134    /// This includes:
135    /// - No running prewarming task instances that could write to the cache
136    /// - No concurrent transactions that might access the cached state
137    /// - All prewarming operations must be completed or cancelled
138    ///
139    /// Violating this requirement can result in cache corruption, incorrect state data,
140    /// and potential consensus failures.
141    pub fn update_with_guard<F>(&self, update_fn: F)
142    where
143        F: FnOnce(&mut Option<SavedCache>),
144    {
145        let mut guard = self.inner.lock();
146        update_fn(&mut guard);
147    }
148}
149
150/// Metrics for [`PayloadExecutionCache`] operations.
151#[derive(Metrics, Clone)]
152#[metrics(scope = "consensus.engine.beacon")]
153struct PayloadExecutionCacheMetrics {
154    /// Counter for when the execution cache was unavailable because other threads
155    /// (e.g., prewarming) are still using it.
156    execution_cache_in_use: Counter,
157    /// Time spent waiting for execution cache mutex to become available.
158    execution_cache_wait_duration: Histogram,
159}
160
161#[cfg(test)]
162mod tests {
163    use super::*;
164
165    #[test]
166    fn single_checkout_blocks_second() {
167        let cache = PayloadExecutionCache::default();
168        let hash = B256::from([1u8; 32]);
169
170        cache.update_with_guard(|slot| {
171            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
172        });
173
174        let first = cache.get_cache_for(hash);
175        assert!(first.is_some());
176
177        let second = cache.get_cache_for(hash);
178        assert!(second.is_none());
179    }
180
181    #[test]
182    fn checkout_available_after_drop() {
183        let cache = PayloadExecutionCache::default();
184        let hash = B256::from([2u8; 32]);
185
186        cache.update_with_guard(|slot| {
187            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
188        });
189
190        let checked_out = cache.get_cache_for(hash);
191        assert!(checked_out.is_some());
192        drop(checked_out);
193
194        let second = cache.get_cache_for(hash);
195        assert!(second.is_some());
196    }
197
198    #[test]
199    fn raw_cache_handle_blocks_checkout_until_drop() {
200        let cache = PayloadExecutionCache::default();
201        let hash = B256::from([3u8; 32]);
202
203        cache.update_with_guard(|slot| {
204            *slot = Some(SavedCache::new(hash, ExecutionCache::new(1_000)))
205        });
206
207        let checked_out = cache.get_cache_for(hash).expect("checkout should succeed");
208        let cache_handle = checked_out.cache().clone();
209        drop(checked_out);
210
211        let blocked = cache.get_cache_for(hash);
212        assert!(blocked.is_none(), "raw ExecutionCache handle should keep slot in use");
213
214        drop(cache_handle);
215
216        let available = cache.get_cache_for(hash);
217        assert!(available.is_some(), "checkout should succeed after raw handle is dropped");
218    }
219
220    #[test]
221    fn hash_mismatch_clears_and_retags() {
222        let cache = PayloadExecutionCache::default();
223        let hash_a = B256::from([0xAA; 32]);
224        let hash_b = B256::from([0xBB; 32]);
225
226        cache.update_with_guard(|slot| {
227            *slot = Some(SavedCache::new(hash_a, ExecutionCache::new(1_000)))
228        });
229
230        let checked_out = cache.get_cache_for(hash_b);
231        assert!(checked_out.is_some());
232        assert_eq!(checked_out.unwrap().executed_block_hash(), hash_b);
233    }
234
235    #[test]
236    fn empty_cache_returns_none() {
237        let cache = PayloadExecutionCache::default();
238        assert!(cache.get_cache_for(B256::ZERO).is_none());
239    }
240}
reth_execution_cache/lib.rs

reth_execution_cache/
lib.rs
