1use 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
36pub(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 let changesets = cache.get_or_compute(provider, block_number)?;
101
102 let reverts = cache.get_or_compute_range(provider, (block_number + 1)..=db_tip_block)?;
104
105 let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(tx);
108 let cursor_factory = InMemoryTrieCursorFactory::new(db_cursor_factory, &reverts);
109
110 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 for (nibbles, _old_node) in account_nodes_ref {
117 let node_value = account_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
119 account_nodes.push((*nibbles, node_value));
120 }
121
122 let mut storage_tries = B256Map::default();
124
125 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 for (nibbles, _old_node) in storage_nodes_ref {
133 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#[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 pub(crate) fn new() -> Self {
168 Self { inner: Arc::new(RwLock::new(ChangesetCacheInner::new())) }
169 }
170
171 pub(crate) fn evict(&self, up_to_block: BlockNumber) {
181 self.inner.write().evict(up_to_block)
182 }
183
184 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 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 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 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 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#[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#[derive(Debug)]
443struct ChangesetCacheInner {
444 entries: HashMap<ChangesetRangeKey, Arc<TrieUpdatesSorted>>,
446
447 range_starts: BTreeMap<BlockNumber, Vec<ChangesetRangeKey>>,
449
450 metrics: ChangesetCacheMetrics,
452}
453
454#[derive(Metrics, Clone)]
459#[metrics(scope = "trie.changeset_cache")]
460struct ChangesetCacheMetrics {
461 hits: Counter,
463
464 misses: Counter,
466
467 evictions: Counter,
469
470 size: Gauge,
472}
473
474impl Default for ChangesetCacheInner {
475 fn default() -> Self {
476 Self::new()
477 }
478}
479
480impl ChangesetCacheInner {
481 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 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 let range_starts_to_evict: Vec<u64> =
537 self.range_starts.range(..up_to_block).map(|(num, _)| *num).collect();
538
539 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 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 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 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 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 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 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 assert_eq!(cache.entries.len(), 15);
984
985 cache.evict(4);
987
988 assert_eq!(cache.entries.len(), 11); 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 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 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 assert_eq!(cache.entries.len(), 66);
1024
1025 cache.evict(100);
1028
1029 assert_eq!(cache.entries.len(), 66);
1031
1032 cache.evict(101);
1035
1036 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 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 assert_eq!(cache.entries.len(), 4);
1061
1062 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 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 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 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 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}