Skip to main content

reth_storage_overlay/
changeset_cache.rs

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