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

reth_execution_cache/
lib.rs
