Skip to main content

reth_trie_db/
changesets.rs

1//! Trie changeset computation and caching utilities.
2//!
3//! This module provides functionality to compute trie changesets for a given block,
4//! which represent the old trie node values before the block was processed.
5//!
6//! It also provides an efficient in-memory cache for these changesets, which is essential for:
7//! - **Reorg support**: Quickly access changesets to revert blocks during chain reorganizations
8//! - **Memory efficiency**: Automatic eviction ensures bounded memory usage
9
10use crate::{
11    DatabaseHashedCursorFactory, DatabaseStateRoot, DatabaseTrieCursorFactory, TrieTableAdapter,
12};
13use alloy_primitives::{map::B256Map, BlockNumber, B256};
14use parking_lot::RwLock;
15use reth_primitives_traits::FastInstant as Instant;
16use reth_storage_api::{
17    BlockNumReader, ChangeSetReader, DBProvider, StorageChangeSetReader, StorageSettingsCache,
18};
19use reth_storage_errors::provider::{ProviderError, ProviderResult};
20use reth_trie::{
21    trie_cursor::{InMemoryTrieCursorFactory, TrieCursor, TrieCursorFactory},
22    TrieInputSorted,
23};
24use reth_trie_common::updates::{StorageTrieUpdatesSorted, TrieUpdatesSorted};
25use std::{
26    collections::{BTreeMap, HashMap},
27    ops::RangeInclusive,
28    sync::Arc,
29};
30use tracing::{debug, warn};
31
32#[cfg(test)]
33use reth_trie::changesets::compute_trie_changesets;
34
35#[cfg(feature = "metrics")]
36use reth_metrics::{
37    metrics::{Counter, Gauge},
38    Metrics,
39};
40
41/// Computes trie changesets for a block.
42///
43/// # Algorithm
44///
45/// For block N:
46/// 1. Query cumulative `HashedPostState` revert from the database tip to after N.
47/// 2. Calculate cumulative `TrieUpdates` revert from the database tip to after N.
48/// 3. Query per-block `HashedPostState` revert for block N.
49/// 4. Overlay the post-N trie and calculate trie updates to the pre-N state.
50///
51/// # Arguments
52///
53/// * `provider` - Database provider with changeset access
54/// * `block_number` - Block number to compute changesets for
55///
56/// # Returns
57///
58/// Changesets (old trie node values) for the specified block
59///
60/// # Errors
61///
62/// Returns error if:
63/// - Block number exceeds database tip
64/// - Database access fails
65/// - State root computation fails
66pub fn compute_block_trie_changesets<Provider>(
67    provider: &Provider,
68    block_number: BlockNumber,
69) -> Result<TrieUpdatesSorted, ProviderError>
70where
71    Provider: DBProvider
72        + ChangeSetReader
73        + StorageChangeSetReader
74        + BlockNumReader
75        + StorageSettingsCache,
76{
77    let db_tip_block = provider.best_block_number()?;
78    crate::with_adapter!(provider, |A| {
79        compute_range_trie_changesets_inner::<_, A>(
80            provider,
81            block_number..=block_number,
82            db_tip_block,
83        )
84    })
85}
86
87fn compute_range_trie_changesets<Provider>(
88    provider: &Provider,
89    range: RangeInclusive<BlockNumber>,
90    db_tip_block: BlockNumber,
91) -> Result<TrieUpdatesSorted, ProviderError>
92where
93    Provider: DBProvider
94        + ChangeSetReader
95        + StorageChangeSetReader
96        + BlockNumReader
97        + StorageSettingsCache,
98{
99    crate::with_adapter!(provider, |A| {
100        compute_range_trie_changesets_inner::<_, A>(provider, range, db_tip_block)
101    })
102}
103
104fn compute_range_trie_changesets_inner<Provider, A>(
105    provider: &Provider,
106    range: RangeInclusive<BlockNumber>,
107    db_tip_block: BlockNumber,
108) -> Result<TrieUpdatesSorted, ProviderError>
109where
110    Provider: DBProvider
111        + ChangeSetReader
112        + StorageChangeSetReader
113        + BlockNumReader
114        + StorageSettingsCache,
115    A: TrieTableAdapter,
116{
117    let start_block = *range.start();
118    let end_block = *range.end();
119
120    if start_block > end_block {
121        return Ok(TrieUpdatesSorted::default())
122    }
123
124    if end_block > db_tip_block {
125        return Err(ProviderError::InsufficientChangesets {
126            requested: end_block,
127            available: 0..=db_tip_block,
128        })
129    }
130
131    debug!(
132        target: "trie::changeset_cache",
133        start_block,
134        end_block,
135        db_tip_block,
136        "Computing range trie changesets from database state"
137    );
138
139    // Step 1: collect the state revert for the requested range.
140    let range_state_revert = crate::state::from_reverts_auto(provider, range)?;
141    let range_prefix_sets = range_state_revert.construct_prefix_sets();
142
143    type DbStateRoot<'a, TX, A> = reth_trie::StateRoot<
144        DatabaseTrieCursorFactory<&'a TX, A>,
145        DatabaseHashedCursorFactory<&'a TX>,
146    >;
147
148    let (range_nodes, range_state) = if end_block == db_tip_block {
149        debug!(
150            target: "trie::changeset_cache",
151            start_block,
152            end_block,
153            db_tip_block,
154            "Skipping tail trie revert computation for tip-ended range"
155        );
156
157        (Arc::default(), Arc::new(range_state_revert))
158    } else {
159        // Step 2: collect the state revert from the database tip to just after the range.
160        let tail_state_revert = end_block
161            .checked_add(1)
162            .map(|next_block| crate::state::from_reverts_auto(provider, next_block..))
163            .transpose()?
164            .unwrap_or_default();
165
166        // Step 3: compute trie reverts from the database tip to just after the range.
167        let tail_input = TrieInputSorted::new(
168            Arc::default(),
169            Arc::new(tail_state_revert.clone()),
170            tail_state_revert.construct_prefix_sets(),
171        );
172        let tail_trie_revert = DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(
173            provider.tx_ref(),
174            tail_input,
175        )
176        .map_err(ProviderError::other)?
177        .1
178        .into_sorted();
179
180        // Step 4: overlay the post-range trie and compute the trie revert to the pre-range state.
181        let mut pre_range_state_revert = tail_state_revert;
182        pre_range_state_revert.extend_ref_and_sort(&range_state_revert);
183
184        (Arc::new(tail_trie_revert), Arc::new(pre_range_state_revert))
185    };
186
187    let range_input = TrieInputSorted::new(range_nodes, range_state, range_prefix_sets);
188    let range_trie_revert =
189        DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(provider.tx_ref(), range_input)
190            .map_err(ProviderError::other)?
191            .1
192            .into_sorted();
193
194    debug!(
195        target: "trie::changeset_cache",
196        start_block,
197        end_block,
198        num_account_nodes = range_trie_revert.account_nodes_ref().len(),
199        num_storage_tries = range_trie_revert.storage_tries_ref().len(),
200        "Computed range trie changesets successfully"
201    );
202
203    Ok(range_trie_revert)
204}
205
206/// Computes block trie updates using the changeset cache.
207///
208/// # Algorithm
209///
210/// For block N:
211/// 1. Get cumulative trie reverts from block N+1 to db tip using the cache
212/// 2. Create an overlay cursor factory with these reverts (representing trie state after block N)
213/// 3. Walk through account trie changesets for block N
214/// 4. For each changed path, look up the current value using the overlay cursor
215/// 5. Walk through storage trie changesets for block N
216/// 6. For each changed path, look up the current value using the overlay cursor
217/// 7. Return the collected trie updates
218///
219/// # Arguments
220///
221/// * `cache` - Handle to the changeset cache for retrieving trie reverts
222/// * `provider` - Database provider for accessing changesets and block data
223/// * `block_number` - Block number to compute trie updates for
224///
225/// # Returns
226///
227/// Trie updates representing the state of trie nodes after the block was processed
228///
229/// # Errors
230///
231/// Returns error if:
232/// - Block number exceeds database tip
233/// - Database access fails
234/// - Cache retrieval fails
235pub fn compute_block_trie_updates<Provider>(
236    cache: &ChangesetCache,
237    provider: &Provider,
238    block_number: BlockNumber,
239) -> ProviderResult<TrieUpdatesSorted>
240where
241    Provider: DBProvider
242        + ChangeSetReader
243        + StorageChangeSetReader
244        + BlockNumReader
245        + StorageSettingsCache,
246{
247    crate::with_adapter!(provider, |A| {
248        compute_block_trie_updates_inner::<_, A>(cache, provider, block_number)
249    })
250}
251
252fn compute_block_trie_updates_inner<Provider, A>(
253    cache: &ChangesetCache,
254    provider: &Provider,
255    block_number: BlockNumber,
256) -> ProviderResult<TrieUpdatesSorted>
257where
258    Provider: DBProvider
259        + ChangeSetReader
260        + StorageChangeSetReader
261        + BlockNumReader
262        + StorageSettingsCache,
263    A: TrieTableAdapter,
264{
265    let tx = provider.tx_ref();
266
267    let db_tip_block = provider.best_block_number()?;
268
269    // Step 1: Get the trie changesets for the target block from cache
270    let changesets = cache.get_or_compute(provider, block_number)?;
271
272    // Step 2: Get the trie reverts for the state after the target block using the cache
273    let reverts = cache.get_or_compute_range(provider, (block_number + 1)..=db_tip_block)?;
274
275    // Step 3: Create an InMemoryTrieCursorFactory with the reverts
276    // This gives us the trie state as it was after the target block was processed
277    let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(tx);
278    let cursor_factory = InMemoryTrieCursorFactory::new(db_cursor_factory, &reverts);
279
280    // Step 4: Collect all account trie nodes that changed in the target block
281    let account_nodes_ref = changesets.account_nodes_ref();
282    let mut account_nodes = Vec::with_capacity(account_nodes_ref.len());
283    let mut account_cursor = cursor_factory.account_trie_cursor()?;
284
285    // Iterate over the account nodes from the changesets
286    for (nibbles, _old_node) in account_nodes_ref {
287        // Look up the current value of this trie node using the overlay cursor
288        let node_value = account_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
289        account_nodes.push((*nibbles, node_value));
290    }
291
292    // Step 5: Collect all storage trie nodes that changed in the target block
293    let mut storage_tries = B256Map::default();
294
295    // Iterate over the storage tries from the changesets
296    for (hashed_address, storage_changeset) in changesets.storage_tries_ref() {
297        let mut storage_cursor = cursor_factory.storage_trie_cursor(*hashed_address)?;
298        let storage_nodes_ref = storage_changeset.storage_nodes_ref();
299        let mut storage_nodes = Vec::with_capacity(storage_nodes_ref.len());
300
301        // Iterate over the storage nodes for this account
302        for (nibbles, _old_node) in storage_nodes_ref {
303            // Look up the current value of this storage trie node
304            let node_value = storage_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
305            storage_nodes.push((*nibbles, node_value));
306        }
307
308        storage_tries.insert(
309            *hashed_address,
310            StorageTrieUpdatesSorted { storage_nodes, is_deleted: storage_changeset.is_deleted },
311        );
312    }
313
314    Ok(TrieUpdatesSorted::new(account_nodes, storage_tries))
315}
316
317/// Thread-safe changeset cache.
318///
319/// This type wraps a shared, mutable reference to the cache inner.
320/// The `RwLock` enables concurrent reads while ensuring exclusive access for writes.
321#[derive(Debug, Clone)]
322pub struct ChangesetCache {
323    inner: Arc<RwLock<ChangesetCacheInner>>,
324}
325
326impl Default for ChangesetCache {
327    fn default() -> Self {
328        Self::new()
329    }
330}
331
332impl ChangesetCache {
333    /// Creates a new cache.
334    ///
335    /// The cache has no capacity limit and relies on explicit eviction
336    /// via the `evict()` method to manage memory usage.
337    pub fn new() -> Self {
338        Self { inner: Arc::new(RwLock::new(ChangesetCacheInner::new())) }
339    }
340
341    /// Retrieves changesets for a block.
342    ///
343    /// Returns `None` if the block is not in the cache (either evicted or never computed).
344    /// Updates hit/miss metrics accordingly.
345    pub fn get(
346        &self,
347        block_hash: B256,
348        block_number: BlockNumber,
349    ) -> Option<Arc<TrieUpdatesSorted>> {
350        self.inner.read().get(&ChangesetRangeKey::single(block_number, block_hash))
351    }
352
353    /// Evicts changesets for blocks below the given block number.
354    ///
355    /// This should be called after blocks are persisted to the database to free
356    /// memory for changesets that are no longer needed in the cache.
357    ///
358    /// # Arguments
359    ///
360    /// * `up_to_block` - Evict blocks with number < this value. Blocks with number >= this value
361    ///   are retained.
362    pub fn evict(&self, up_to_block: BlockNumber) {
363        self.inner.write().evict(up_to_block)
364    }
365
366    /// Gets changesets from cache, or computes them on-the-fly if missing.
367    ///
368    /// This is the primary API for retrieving changesets. It checks the cache first, then falls
369    /// back to computing from database state if missing.
370    ///
371    /// # Arguments
372    ///
373    /// * `block_number` - Block number (for cache insertion and logging)
374    /// * `provider` - Database provider for DB access
375    ///
376    /// # Returns
377    ///
378    /// Changesets for the block, either from cache or computed on-the-fly.
379    pub fn get_or_compute<P>(
380        &self,
381        provider: &P,
382        block_number: BlockNumber,
383    ) -> ProviderResult<Arc<TrieUpdatesSorted>>
384    where
385        P: DBProvider
386            + ChangeSetReader
387            + StorageChangeSetReader
388            + BlockNumReader
389            + StorageSettingsCache,
390    {
391        self.get_or_compute_range(provider, block_number..=block_number)
392    }
393
394    /// Gets or computes trie reverts for a range of blocks.
395    ///
396    /// If all blocks in the range are cached, this method retrieves and accumulates those
397    /// per-block trie changesets (reverts) in reverse order (newest to oldest), so that older
398    /// values take precedence when there are conflicts.
399    ///
400    /// If any block is missing from cache, this falls back to one aggregate database computation
401    /// for the whole range. The aggregate result restores the trie to the state before the range
402    /// and is inserted into the range cache.
403    ///
404    /// # Arguments
405    ///
406    /// * `provider` - Database provider for DB access and block lookups
407    /// * `range` - Block range to accumulate reverts for (inclusive)
408    ///
409    /// # Returns
410    ///
411    /// Accumulated trie reverts for all blocks in the specified range
412    ///
413    /// # Errors
414    ///
415    /// Returns error if:
416    /// - Any block in the range is beyond the database tip
417    /// - Database access fails
418    /// - Block hash lookup fails
419    /// - Changeset computation fails
420    pub fn get_or_compute_range<P>(
421        &self,
422        provider: &P,
423        range: RangeInclusive<BlockNumber>,
424    ) -> ProviderResult<Arc<TrieUpdatesSorted>>
425    where
426        P: DBProvider
427            + ChangeSetReader
428            + StorageChangeSetReader
429            + BlockNumReader
430            + StorageSettingsCache,
431    {
432        let db_tip_block = provider.best_block_number()?;
433
434        let start_block = *range.start();
435        let end_block = *range.end();
436
437        // If range end is beyond the tip, return an error
438        if end_block > db_tip_block {
439            return Err(ProviderError::InsufficientChangesets {
440                requested: end_block,
441                available: 0..=db_tip_block,
442            });
443        }
444
445        let timer = Instant::now();
446
447        debug!(
448            target: "trie::changeset_cache",
449            start_block,
450            end_block,
451            db_tip_block,
452            "Starting get_or_compute_range"
453        );
454
455        if start_block > end_block {
456            debug!(
457                target: "trie::changeset_cache",
458                start_block,
459                end_block,
460                "Empty changeset range requested"
461            );
462            return Ok(Arc::new(TrieUpdatesSorted::default()))
463        }
464
465        let end_block_hash = provider.block_hash(end_block)?.ok_or_else(|| {
466            ProviderError::other(std::io::Error::new(
467                std::io::ErrorKind::NotFound,
468                format!("block hash not found for block number {}", end_block),
469            ))
470        })?;
471        let range_key = ChangesetRangeKey::new(start_block, end_block, end_block_hash);
472
473        if let Some(accumulated_reverts) = self.inner.read().get(&range_key) {
474            let elapsed = timer.elapsed();
475
476            debug!(
477                target: "trie::changeset_cache",
478                ?elapsed,
479                start_block,
480                end_block,
481                ?end_block_hash,
482                num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
483                "Changeset cache HIT for block range"
484            );
485
486            return Ok(accumulated_reverts)
487        }
488
489        let mut cached_reverts =
490            Vec::with_capacity(end_block.saturating_sub(start_block).saturating_add(1) as usize);
491        let mut all_cached = true;
492
493        for block_number in range.rev() {
494            // Get the block hash for this block number
495            let block_hash = if block_number == end_block {
496                end_block_hash
497            } else {
498                provider.block_hash(block_number)?.ok_or_else(|| {
499                    ProviderError::other(std::io::Error::new(
500                        std::io::ErrorKind::NotFound,
501                        format!("block hash not found for block number {}", block_number),
502                    ))
503                })?
504            };
505
506            debug!(
507                target: "trie::changeset_cache",
508                block_number,
509                ?block_hash,
510                "Looked up block hash for block number in range"
511            );
512
513            let block_key = ChangesetRangeKey::single(block_number, block_hash);
514            if let Some(changesets) = self.inner.read().get(&block_key) {
515                cached_reverts.push(changesets);
516            } else {
517                all_cached = false;
518                break
519            }
520        }
521
522        if all_cached {
523            // `merge_slice` gives precedence to earlier items, so pass reverts oldest-to-newest.
524            cached_reverts.reverse();
525            let accumulated_reverts = Arc::new(TrieUpdatesSorted::merge_slice(&cached_reverts));
526            let elapsed = timer.elapsed();
527
528            let num_account_nodes = accumulated_reverts.account_nodes_ref().len();
529            let num_storage_tries = accumulated_reverts.storage_tries_ref().len();
530
531            debug!(
532                target: "trie::changeset_cache",
533                ?elapsed,
534                start_block,
535                end_block,
536                num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
537                num_account_nodes,
538                num_storage_tries,
539                "Finished accumulating cached trie reverts for block range"
540            );
541
542            self.inner.write().insert(range_key, Arc::clone(&accumulated_reverts));
543            return Ok(accumulated_reverts)
544        }
545
546        warn!(
547            target: "trie::changeset_cache",
548            start_block,
549            end_block,
550            "Changeset cache MISS in range, falling back to aggregate DB-based computation"
551        );
552
553        let accumulated_reverts = Arc::new(compute_range_trie_changesets(
554            provider,
555            start_block..=end_block,
556            db_tip_block,
557        )?);
558
559        let elapsed = timer.elapsed();
560
561        let num_account_nodes = accumulated_reverts.account_nodes_ref().len();
562        let num_storage_tries = accumulated_reverts.storage_tries_ref().len();
563
564        debug!(
565            target: "trie::changeset_cache",
566            ?elapsed,
567            start_block,
568            end_block,
569            ?end_block_hash,
570            num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
571            num_account_nodes,
572            num_storage_tries,
573            "Finished accumulating trie reverts for block range"
574        );
575
576        self.inner.write().insert(range_key, Arc::clone(&accumulated_reverts));
577
578        Ok(accumulated_reverts)
579    }
580}
581
582/// Cache key for one contiguous range of canonical trie changesets.
583///
584/// The end block hash disambiguates canonical rewrites where the same block numbers later refer to
585/// a different chain. For a single block, `start_block == end_block` and `end_block_hash` is that
586/// block's hash.
587#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
588struct ChangesetRangeKey {
589    start_block: BlockNumber,
590    end_block: BlockNumber,
591    end_block_hash: B256,
592}
593
594impl ChangesetRangeKey {
595    const fn new(start_block: BlockNumber, end_block: BlockNumber, end_block_hash: B256) -> Self {
596        Self { start_block, end_block, end_block_hash }
597    }
598
599    const fn single(block_number: BlockNumber, block_hash: B256) -> Self {
600        Self::new(block_number, block_number, block_hash)
601    }
602}
603
604/// In-memory cache for trie changesets with explicit eviction policy.
605///
606/// Holds changesets for blocks or block ranges that have been validated but not yet persisted.
607/// Keyed by canonical block range. Eviction is controlled
608/// explicitly by the engine API tree handler when persistence completes.
609///
610/// ## Eviction Policy
611///
612/// Unlike traditional caches with automatic eviction, this cache requires explicit
613/// eviction calls. The engine API tree handler calls `evict(block_number)` after
614/// blocks are persisted to the database, ensuring changesets remain available
615/// until their corresponding blocks are safely on disk.
616///
617/// ## Metrics
618///
619/// The cache maintains several metrics for observability:
620/// - `hits`: Number of successful cache lookups
621/// - `misses`: Number of failed cache lookups
622/// - `evictions`: Number of blocks evicted
623/// - `size`: Current number of cached blocks
624#[derive(Debug)]
625struct ChangesetCacheInner {
626    /// Cache entries keyed by inclusive block range plus the range's canonical end hash.
627    entries: HashMap<ChangesetRangeKey, Arc<TrieUpdatesSorted>>,
628
629    /// Range start block to cache keys mapping for eviction.
630    range_starts: BTreeMap<BlockNumber, Vec<ChangesetRangeKey>>,
631
632    /// Metrics for monitoring cache behavior
633    #[cfg(feature = "metrics")]
634    metrics: ChangesetCacheMetrics,
635}
636
637#[cfg(feature = "metrics")]
638/// Metrics for the changeset cache.
639///
640/// These metrics provide visibility into cache performance and help identify
641/// potential issues like high miss rates.
642#[derive(Metrics, Clone)]
643#[metrics(scope = "trie.changeset_cache")]
644struct ChangesetCacheMetrics {
645    /// Cache hit counter
646    hits: Counter,
647
648    /// Cache miss counter
649    misses: Counter,
650
651    /// Eviction counter
652    evictions: Counter,
653
654    /// Current cache size (number of entries)
655    size: Gauge,
656}
657
658impl Default for ChangesetCacheInner {
659    fn default() -> Self {
660        Self::new()
661    }
662}
663
664impl ChangesetCacheInner {
665    /// Creates a new empty changeset cache.
666    ///
667    /// The cache has no capacity limit and relies on explicit eviction
668    /// via the `evict()` method to manage memory usage.
669    fn new() -> Self {
670        Self {
671            entries: HashMap::new(),
672            range_starts: BTreeMap::new(),
673            #[cfg(feature = "metrics")]
674            metrics: Default::default(),
675        }
676    }
677
678    fn get(&self, key: &ChangesetRangeKey) -> Option<Arc<TrieUpdatesSorted>> {
679        match self.entries.get(key) {
680            Some(changesets) => {
681                #[cfg(feature = "metrics")]
682                self.metrics.hits.increment(1);
683                Some(Arc::clone(changesets))
684            }
685            None => {
686                #[cfg(feature = "metrics")]
687                self.metrics.misses.increment(1);
688                None
689            }
690        }
691    }
692
693    fn insert(&mut self, key: ChangesetRangeKey, changesets: Arc<TrieUpdatesSorted>) {
694        debug!(
695            target: "trie::changeset_cache",
696            ?key,
697            cache_size_before = self.entries.len(),
698            "Inserting changeset into cache"
699        );
700
701        let is_new_entry = self.entries.insert(key, changesets).is_none();
702
703        if is_new_entry {
704            self.range_starts.entry(key.start_block).or_default().push(key);
705        }
706
707        // Update size metric
708        #[cfg(feature = "metrics")]
709        self.metrics.size.set(self.entries.len() as f64);
710
711        debug!(
712            target: "trie::changeset_cache",
713            ?key,
714            cache_size_after = self.entries.len(),
715            "Changeset inserted into cache"
716        );
717    }
718
719    fn evict(&mut self, up_to_block: BlockNumber) {
720        debug!(
721            target: "trie::changeset_cache",
722            up_to_block,
723            cache_size_before = self.entries.len(),
724            "Starting cache eviction"
725        );
726
727        // Find all block numbers that should be evicted (< up_to_block)
728        let range_starts_to_evict: Vec<u64> =
729            self.range_starts.range(..up_to_block).map(|(num, _)| *num).collect();
730
731        // Remove entries for each block number below threshold
732        #[cfg(feature = "metrics")]
733        let mut evicted_count = 0;
734        #[cfg(not(feature = "metrics"))]
735        let mut evicted_count = 0;
736
737        for start_block in &range_starts_to_evict {
738            if let Some(keys) = self.range_starts.remove(start_block) {
739                debug!(
740                    target: "trie::changeset_cache",
741                    start_block,
742                    num_ranges = keys.len(),
743                    "Evicting ranges from cache"
744                );
745                for key in keys {
746                    if self.entries.remove(&key).is_some() {
747                        evicted_count += 1;
748                    }
749                }
750            }
751        }
752
753        debug!(
754            target: "trie::changeset_cache",
755            up_to_block,
756            evicted_count,
757            cache_size_after = self.entries.len(),
758            "Finished cache eviction"
759        );
760
761        // Update metrics if we evicted anything
762        #[cfg(feature = "metrics")]
763        if evicted_count > 0 {
764            self.metrics.evictions.increment(evicted_count as u64);
765            self.metrics.size.set(self.entries.len() as f64);
766        }
767    }
768}
769
770#[cfg(test)]
771mod tests {
772    use super::*;
773    use alloy_consensus::Header;
774    use alloy_primitives::{
775        keccak256,
776        map::{B256Map, HashMap},
777        Address, U256,
778    };
779    use reth_db::tables;
780    use reth_db_api::{
781        models::{AccountBeforeTx, BlockNumberAddress},
782        transaction::DbTxMut,
783    };
784    use reth_primitives_traits::{Account, StorageEntry};
785    use reth_provider::{
786        test_utils::create_test_provider_factory, StaticFileProviderFactory, StaticFileSegment,
787        StaticFileWriter,
788    };
789    use reth_stages_types::{StageCheckpoint, StageId};
790    use reth_storage_api::{StageCheckpointWriter, TrieWriter};
791    use reth_trie::{BranchNodeCompact, Nibbles, StateRoot};
792
793    // Helper function to create empty TrieUpdatesSorted for testing
794    fn create_test_changesets() -> Arc<TrieUpdatesSorted> {
795        Arc::new(TrieUpdatesSorted::new(vec![], B256Map::default()))
796    }
797
798    fn insert_test_changesets(
799        cache: &mut ChangesetCacheInner,
800        block_hash: B256,
801        block_number: BlockNumber,
802        changesets: Arc<TrieUpdatesSorted>,
803    ) {
804        cache.insert(ChangesetRangeKey::single(block_number, block_hash), changesets);
805    }
806
807    fn get_test_changesets(
808        cache: &ChangesetCacheInner,
809        block_hash: B256,
810        block_number: BlockNumber,
811    ) -> Option<Arc<TrieUpdatesSorted>> {
812        cache.get(&ChangesetRangeKey::single(block_number, block_hash))
813    }
814
815    fn test_account(balance: u64) -> Account {
816        Account { balance: U256::from(balance), ..Default::default() }
817    }
818
819    fn test_storage(slot: u64, value: u64) -> StorageEntry {
820        StorageEntry { key: B256::from(U256::from(slot)), value: U256::from(value) }
821    }
822
823    fn seed_headers(
824        factory: &impl StaticFileProviderFactory<
825            Primitives: reth_primitives_traits::NodePrimitives<BlockHeader = Header>,
826        >,
827        end_block: BlockNumber,
828    ) {
829        let static_file_provider = factory.static_file_provider();
830        let mut header_writer =
831            static_file_provider.latest_writer(StaticFileSegment::Headers).unwrap();
832        for block_number in 0..=end_block {
833            let header = Header { number: block_number, ..Default::default() };
834            header_writer
835                .append_header(&header, &B256::with_last_byte(block_number as u8))
836                .unwrap();
837        }
838        header_writer.commit().unwrap();
839    }
840
841    fn legacy_compute_range_trie_changesets<Provider>(
842        provider: &Provider,
843        range: RangeInclusive<BlockNumber>,
844    ) -> TrieUpdatesSorted
845    where
846        Provider: DBProvider
847            + ChangeSetReader
848            + StorageChangeSetReader
849            + BlockNumReader
850            + StorageSettingsCache,
851    {
852        let mut accumulated_reverts = TrieUpdatesSorted::default();
853        for block_number in range.rev() {
854            let changesets = legacy_compute_block_trie_changesets(provider, block_number);
855            accumulated_reverts.extend_ref_and_sort(&changesets);
856        }
857        accumulated_reverts
858    }
859
860    fn legacy_compute_block_trie_changesets<Provider>(
861        provider: &Provider,
862        block_number: BlockNumber,
863    ) -> TrieUpdatesSorted
864    where
865        Provider: DBProvider
866            + ChangeSetReader
867            + StorageChangeSetReader
868            + BlockNumReader
869            + StorageSettingsCache,
870    {
871        crate::with_adapter!(provider, |A| {
872            legacy_compute_block_trie_changesets_inner::<_, A>(provider, block_number)
873        })
874    }
875
876    fn legacy_compute_block_trie_changesets_inner<Provider, A>(
877        provider: &Provider,
878        block_number: BlockNumber,
879    ) -> TrieUpdatesSorted
880    where
881        Provider: DBProvider
882            + ChangeSetReader
883            + StorageChangeSetReader
884            + BlockNumReader
885            + StorageSettingsCache,
886        A: TrieTableAdapter,
887    {
888        let individual_state_revert =
889            crate::state::from_reverts_auto(provider, block_number..=block_number).unwrap();
890        let cumulative_state_revert =
891            crate::state::from_reverts_auto(provider, (block_number + 1)..).unwrap();
892
893        let mut cumulative_state_revert_prev = cumulative_state_revert.clone();
894        cumulative_state_revert_prev.extend_ref_and_sort(&individual_state_revert);
895
896        type DbStateRoot<'a, TX, A> =
897            StateRoot<DatabaseTrieCursorFactory<&'a TX, A>, DatabaseHashedCursorFactory<&'a TX>>;
898
899        let input_prev = TrieInputSorted::new(
900            Arc::default(),
901            Arc::new(cumulative_state_revert_prev.clone()),
902            cumulative_state_revert_prev.construct_prefix_sets(),
903        );
904        let cumulative_trie_updates_prev =
905            DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(
906                provider.tx_ref(),
907                input_prev,
908            )
909            .unwrap()
910            .1
911            .into_sorted();
912
913        let input = TrieInputSorted::new(
914            Arc::new(cumulative_trie_updates_prev.clone()),
915            Arc::new(cumulative_state_revert),
916            individual_state_revert.construct_prefix_sets(),
917        );
918        let trie_updates =
919            DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(provider.tx_ref(), input)
920                .unwrap()
921                .1
922                .into_sorted();
923
924        let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(provider.tx_ref());
925        let overlay_factory =
926            InMemoryTrieCursorFactory::new(db_cursor_factory, &cumulative_trie_updates_prev);
927
928        compute_trie_changesets(&overlay_factory, &trie_updates).unwrap()
929    }
930
931    fn seed_tip_trie_tables<Provider, A>(provider: &Provider)
932    where
933        Provider: DBProvider + TrieWriter,
934        A: TrieTableAdapter,
935    {
936        type DbStateRoot<'a, TX, A> =
937            StateRoot<DatabaseTrieCursorFactory<&'a TX, A>, DatabaseHashedCursorFactory<&'a TX>>;
938
939        let (_, trie_updates) =
940            DbStateRoot::<_, A>::from_tx(provider.tx_ref()).root_with_updates().unwrap();
941        provider.write_trie_updates(trie_updates).unwrap();
942    }
943
944    #[test]
945    fn cached_range_merge_keeps_oldest_revert_values() {
946        let factory = create_test_provider_factory();
947        seed_headers(&factory, 2);
948
949        let provider = factory.provider_rw().unwrap();
950        provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(2)).unwrap();
951
952        let cache = ChangesetCache::new();
953        let path = Nibbles::from_nibbles([0x1, 0x2]);
954        let older_node = BranchNodeCompact::new(0b0001, 0, 0, vec![], None);
955        let newer_node = BranchNodeCompact::new(0b0010, 0, 0, vec![], None);
956
957        {
958            let mut cache = cache.inner.write();
959            insert_test_changesets(
960                &mut cache,
961                B256::with_last_byte(1),
962                1,
963                Arc::new(TrieUpdatesSorted::new(
964                    vec![(path, Some(older_node.clone()))],
965                    B256Map::default(),
966                )),
967            );
968            insert_test_changesets(
969                &mut cache,
970                B256::with_last_byte(2),
971                2,
972                Arc::new(TrieUpdatesSorted::new(
973                    vec![(path, Some(newer_node))],
974                    B256Map::default(),
975                )),
976            );
977        }
978
979        let accumulated = cache.get_or_compute_range(&*provider, 1..=2).unwrap();
980        assert_eq!(accumulated.account_nodes_ref(), &[(path, Some(older_node))]);
981    }
982
983    #[test]
984    fn aggregate_range_reverts_to_pre_range_state() {
985        let factory = create_test_provider_factory();
986        seed_headers(&factory, 3);
987
988        let provider = factory.provider_rw().unwrap();
989        let address = Address::with_last_byte(1);
990        let hashed_address = keccak256(address);
991        let slot1 = B256::from(U256::from(1));
992        let slot2 = B256::from(U256::from(2));
993        let account1 = test_account(10);
994        let account2 = test_account(20);
995        let account3 = test_account(30);
996
997        provider.tx_ref().put::<tables::HashedAccounts>(hashed_address, account3).unwrap();
998        provider
999            .tx_ref()
1000            .put::<tables::HashedStorages>(
1001                hashed_address,
1002                StorageEntry { key: keccak256(slot1), value: U256::from(25) },
1003            )
1004            .unwrap();
1005        provider
1006            .tx_ref()
1007            .put::<tables::HashedStorages>(
1008                hashed_address,
1009                StorageEntry { key: keccak256(slot2), value: U256::from(20) },
1010            )
1011            .unwrap();
1012
1013        provider
1014            .tx_ref()
1015            .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
1016            .unwrap();
1017        provider
1018            .tx_ref()
1019            .put::<tables::AccountChangeSets>(2, AccountBeforeTx { address, info: Some(account1) })
1020            .unwrap();
1021        provider
1022            .tx_ref()
1023            .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: Some(account2) })
1024            .unwrap();
1025
1026        provider
1027            .tx_ref()
1028            .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(1, 0))
1029            .unwrap();
1030        provider
1031            .tx_ref()
1032            .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(2, 0))
1033            .unwrap();
1034        provider
1035            .tx_ref()
1036            .put::<tables::StorageChangeSets>(
1037                BlockNumberAddress((2, address)),
1038                StorageEntry { key: slot1, value: U256::from(10) },
1039            )
1040            .unwrap();
1041        provider
1042            .tx_ref()
1043            .put::<tables::StorageChangeSets>(
1044                BlockNumberAddress((3, address)),
1045                StorageEntry { key: slot1, value: U256::from(15) },
1046            )
1047            .unwrap();
1048
1049        provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
1050        crate::with_adapter!(provider, |A| seed_tip_trie_tables::<_, A>(&*provider));
1051
1052        let actual = compute_range_trie_changesets(&*provider, 1..=3, 3).unwrap();
1053        let storage_revert = actual
1054            .storage_tries_ref()
1055            .get(&hashed_address)
1056            .expect("created account storage trie should be deleted by range revert");
1057        assert!(storage_revert.is_deleted());
1058        assert!(storage_revert.storage_nodes_ref().is_empty());
1059
1060        let cache = ChangesetCache::new();
1061        let from_cache_api = cache.get_or_compute_range(&*provider, 1..=3).unwrap();
1062        assert_eq!(*from_cache_api, actual);
1063        assert_eq!(cache.inner.read().entries.len(), 1);
1064
1065        let block_changesets = cache.get_or_compute(&*provider, 2).unwrap();
1066        assert_eq!(*block_changesets, legacy_compute_block_trie_changesets(&*provider, 2));
1067        assert_eq!(cache.inner.read().entries.len(), 2);
1068    }
1069
1070    #[test]
1071    fn aggregate_range_matches_legacy_per_block_merge_with_storage_wipe() {
1072        let factory = create_test_provider_factory();
1073        seed_headers(&factory, 3);
1074
1075        let provider = factory.provider_rw().unwrap();
1076        let address = Address::with_last_byte(1);
1077        let slot1 = B256::from(U256::from(1));
1078        let slot2 = B256::from(U256::from(2));
1079        let account1 = test_account(10);
1080        let account2 = test_account(20);
1081
1082        provider
1083            .tx_ref()
1084            .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
1085            .unwrap();
1086        provider
1087            .tx_ref()
1088            .put::<tables::AccountChangeSets>(2, AccountBeforeTx { address, info: Some(account1) })
1089            .unwrap();
1090        provider
1091            .tx_ref()
1092            .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: Some(account2) })
1093            .unwrap();
1094
1095        provider
1096            .tx_ref()
1097            .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(1, 0))
1098            .unwrap();
1099        provider
1100            .tx_ref()
1101            .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(2, 0))
1102            .unwrap();
1103        provider
1104            .tx_ref()
1105            .put::<tables::StorageChangeSets>(
1106                BlockNumberAddress((2, address)),
1107                StorageEntry { key: slot1, value: U256::from(10) },
1108            )
1109            .unwrap();
1110        provider
1111            .tx_ref()
1112            .put::<tables::StorageChangeSets>(
1113                BlockNumberAddress((3, address)),
1114                StorageEntry { key: slot1, value: U256::from(15) },
1115            )
1116            .unwrap();
1117        provider
1118            .tx_ref()
1119            .put::<tables::StorageChangeSets>(
1120                BlockNumberAddress((3, address)),
1121                StorageEntry { key: slot2, value: U256::from(20) },
1122            )
1123            .unwrap();
1124
1125        provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
1126        crate::with_adapter!(provider, |A| seed_tip_trie_tables::<_, A>(&*provider));
1127
1128        let expected = legacy_compute_range_trie_changesets(&*provider, 2..=3);
1129        let actual = compute_range_trie_changesets(&*provider, 2..=3, 3).unwrap();
1130        assert_eq!(actual, expected);
1131    }
1132
1133    #[test]
1134    fn test_insert_and_retrieve_single_entry() {
1135        let mut cache = ChangesetCacheInner::new();
1136        let hash = B256::random();
1137        let changesets = create_test_changesets();
1138
1139        insert_test_changesets(&mut cache, hash, 100, Arc::clone(&changesets));
1140
1141        // Should be able to retrieve it
1142        let retrieved = get_test_changesets(&cache, hash, 100);
1143        assert!(retrieved.is_some());
1144        assert_eq!(cache.entries.len(), 1);
1145    }
1146
1147    #[test]
1148    fn test_insert_multiple_entries() {
1149        let mut cache = ChangesetCacheInner::new();
1150
1151        // Insert 10 blocks
1152        let mut hashes = Vec::new();
1153        for i in 0..10 {
1154            let hash = B256::random();
1155            insert_test_changesets(&mut cache, hash, 100 + i, create_test_changesets());
1156            hashes.push((100 + i, hash));
1157        }
1158
1159        // Should be able to retrieve all
1160        assert_eq!(cache.entries.len(), 10);
1161        for (block_number, hash) in hashes {
1162            assert!(get_test_changesets(&cache, hash, block_number).is_some());
1163        }
1164    }
1165
1166    #[test]
1167    fn test_eviction_when_explicitly_called() {
1168        let mut cache = ChangesetCacheInner::new();
1169
1170        // Insert 15 blocks (0-14)
1171        let mut hashes = Vec::new();
1172        for i in 0..15 {
1173            let hash = B256::random();
1174            insert_test_changesets(&mut cache, hash, i, create_test_changesets());
1175            hashes.push((i, hash));
1176        }
1177
1178        // All blocks should be present (no automatic eviction)
1179        assert_eq!(cache.entries.len(), 15);
1180
1181        // Explicitly evict blocks < 4
1182        cache.evict(4);
1183
1184        // Blocks 0-3 should be evicted
1185        assert_eq!(cache.entries.len(), 11); // blocks 4-14 = 11 blocks
1186
1187        // Verify blocks 0-3 are evicted
1188        for i in 0..4 {
1189            assert!(
1190                get_test_changesets(&cache, hashes[i as usize].1, i).is_none(),
1191                "Block {} should be evicted",
1192                i
1193            );
1194        }
1195
1196        // Verify blocks 4-14 are still present
1197        for i in 4..15 {
1198            assert!(
1199                get_test_changesets(&cache, hashes[i as usize].1, i).is_some(),
1200                "Block {} should be present",
1201                i
1202            );
1203        }
1204    }
1205
1206    #[test]
1207    fn test_eviction_with_persistence_watermark() {
1208        let mut cache = ChangesetCacheInner::new();
1209
1210        // Insert blocks 100-165
1211        let mut hashes = HashMap::new();
1212        for i in 100..=165 {
1213            let hash = B256::random();
1214            insert_test_changesets(&mut cache, hash, i, create_test_changesets());
1215            hashes.insert(i, hash);
1216        }
1217
1218        // All blocks should be present (no automatic eviction)
1219        assert_eq!(cache.entries.len(), 66);
1220
1221        // Simulate persistence up to block 164, with 64-block retention window
1222        // Eviction threshold = 164 - 64 = 100
1223        cache.evict(100);
1224
1225        // Blocks 100-165 should remain (66 blocks)
1226        assert_eq!(cache.entries.len(), 66);
1227
1228        // Simulate persistence up to block 165
1229        // Eviction threshold = 165 - 64 = 101
1230        cache.evict(101);
1231
1232        // Blocks 101-165 should remain (65 blocks)
1233        assert_eq!(cache.entries.len(), 65);
1234        assert!(get_test_changesets(&cache, hashes[&100], 100).is_none());
1235        assert!(get_test_changesets(&cache, hashes[&101], 101).is_some());
1236    }
1237
1238    #[test]
1239    fn test_out_of_order_inserts_with_explicit_eviction() {
1240        let mut cache = ChangesetCacheInner::new();
1241
1242        // Insert blocks in random order
1243        let hash_10 = B256::random();
1244        insert_test_changesets(&mut cache, hash_10, 10, create_test_changesets());
1245
1246        let hash_5 = B256::random();
1247        insert_test_changesets(&mut cache, hash_5, 5, create_test_changesets());
1248
1249        let hash_15 = B256::random();
1250        insert_test_changesets(&mut cache, hash_15, 15, create_test_changesets());
1251
1252        let hash_3 = B256::random();
1253        insert_test_changesets(&mut cache, hash_3, 3, create_test_changesets());
1254
1255        // All blocks should be present (no automatic eviction)
1256        assert_eq!(cache.entries.len(), 4);
1257
1258        // Explicitly evict blocks < 5
1259        cache.evict(5);
1260
1261        assert!(get_test_changesets(&cache, hash_3, 3).is_none(), "Block 3 should be evicted");
1262        assert!(get_test_changesets(&cache, hash_5, 5).is_some(), "Block 5 should be present");
1263        assert!(get_test_changesets(&cache, hash_10, 10).is_some(), "Block 10 should be present");
1264        assert!(get_test_changesets(&cache, hash_15, 15).is_some(), "Block 15 should be present");
1265    }
1266
1267    #[test]
1268    fn test_multiple_blocks_same_number() {
1269        let mut cache = ChangesetCacheInner::new();
1270
1271        // Insert multiple blocks with same number (side chains)
1272        let hash_1a = B256::random();
1273        let hash_1b = B256::random();
1274        insert_test_changesets(&mut cache, hash_1a, 100, create_test_changesets());
1275        insert_test_changesets(&mut cache, hash_1b, 100, create_test_changesets());
1276
1277        // Both should be retrievable
1278        assert!(get_test_changesets(&cache, hash_1a, 100).is_some());
1279        assert!(get_test_changesets(&cache, hash_1b, 100).is_some());
1280        assert_eq!(cache.entries.len(), 2);
1281    }
1282
1283    #[test]
1284    fn test_ranges_with_same_numbers_and_different_end_hashes_are_distinct() {
1285        let mut cache = ChangesetCacheInner::new();
1286        let path = Nibbles::from_nibbles_unchecked([0x01]);
1287        let hash_a = B256::with_last_byte(1);
1288        let hash_b = B256::with_last_byte(2);
1289        let key_a = ChangesetRangeKey::new(10, 20, hash_a);
1290        let key_b = ChangesetRangeKey::new(10, 20, hash_b);
1291        let changesets_a = Arc::new(TrieUpdatesSorted::new(
1292            vec![(path, Some(BranchNodeCompact::new(0b0001, 0, 0, vec![], None)))],
1293            B256Map::default(),
1294        ));
1295        let changesets_b = Arc::new(TrieUpdatesSorted::new(
1296            vec![(path, Some(BranchNodeCompact::new(0b0010, 0, 0, vec![], None)))],
1297            B256Map::default(),
1298        ));
1299
1300        cache.insert(key_a, Arc::clone(&changesets_a));
1301        cache.insert(key_b, Arc::clone(&changesets_b));
1302
1303        assert_eq!(cache.entries.len(), 2);
1304        assert_eq!(
1305            cache.get(&key_a).unwrap().account_nodes_ref(),
1306            changesets_a.account_nodes_ref()
1307        );
1308        assert_eq!(
1309            cache.get(&key_b).unwrap().account_nodes_ref(),
1310            changesets_b.account_nodes_ref()
1311        );
1312
1313        cache.evict(11);
1314        assert!(cache.get(&key_a).is_none());
1315        assert!(cache.get(&key_b).is_none());
1316    }
1317
1318    #[test]
1319    fn test_eviction_removes_all_side_chains() {
1320        let mut cache = ChangesetCacheInner::new();
1321
1322        // Insert multiple blocks at same height (side chains)
1323        let hash_10a = B256::random();
1324        let hash_10b = B256::random();
1325        let hash_10c = B256::random();
1326        insert_test_changesets(&mut cache, hash_10a, 10, create_test_changesets());
1327        insert_test_changesets(&mut cache, hash_10b, 10, create_test_changesets());
1328        insert_test_changesets(&mut cache, hash_10c, 10, create_test_changesets());
1329
1330        let hash_20 = B256::random();
1331        insert_test_changesets(&mut cache, hash_20, 20, create_test_changesets());
1332
1333        assert_eq!(cache.entries.len(), 4);
1334
1335        // Evict blocks < 15 - should remove all three side chains at height 10
1336        cache.evict(15);
1337
1338        assert_eq!(cache.entries.len(), 1);
1339        assert!(get_test_changesets(&cache, hash_10a, 10).is_none());
1340        assert!(get_test_changesets(&cache, hash_10b, 10).is_none());
1341        assert!(get_test_changesets(&cache, hash_10c, 10).is_none());
1342        assert!(get_test_changesets(&cache, hash_20, 20).is_some());
1343    }
1344}