1use crate::{database_state_frontiers, OverlayManager, OverlayStateProvider};
11use alloy_eips::BlockNumHash;
12use alloy_primitives::{map::B256Map, BlockNumber, B256};
13use parking_lot::RwLock;
14use reth_metrics::{
15 metrics::{Counter, Gauge},
16 Metrics,
17};
18use reth_primitives_traits::{AlloyBlockHeader, FastInstant as Instant, NodePrimitives};
19use reth_storage_api::{
20 BlockNumReader, ChangeSetReader, DBProvider, PruneCheckpointReader, StageCheckpointReader,
21 StorageChangeSetReader, StorageSettingsCache,
22};
23use reth_storage_errors::provider::{ProviderError, ProviderResult};
24use reth_trie::trie_cursor::{InMemoryTrieCursorFactory, TrieCursor, TrieCursorFactory};
25use reth_trie_common::updates::{StorageTrieUpdatesSorted, TrieUpdatesSorted};
26use reth_trie_db::{DatabaseTrieCursorFactory, TrieTableAdapter};
27use std::{
28 collections::{BTreeMap, HashMap},
29 ops::RangeInclusive,
30 sync::Arc,
31};
32use tracing::{debug, warn};
33
34#[cfg(test)]
35use reth_trie::{changesets::compute_trie_changesets, HashedPostStateSorted, TrieInputSorted};
36#[cfg(test)]
37use reth_trie_db::{DatabaseHashedCursorFactory, DatabaseHashedPostState, DatabaseStateRoot};
38
39pub(crate) fn compute_block_trie_updates<N, Provider>(
68 overlay_manager: &OverlayManager<N>,
69 provider: &Provider,
70 block_number: BlockNumber,
71) -> ProviderResult<TrieUpdatesSorted>
72where
73 N: NodePrimitives,
74 Provider: DBProvider
75 + ChangeSetReader
76 + StorageChangeSetReader
77 + PruneCheckpointReader
78 + StageCheckpointReader
79 + BlockNumReader
80 + StorageSettingsCache,
81{
82 reth_trie_db::with_adapter!(provider, |A| {
83 compute_block_trie_updates_inner::<_, _, A>(overlay_manager, provider, block_number)
84 })
85}
86
87fn compute_block_trie_updates_inner<N, Provider, A>(
88 overlay_manager: &OverlayManager<N>,
89 provider: &Provider,
90 block_number: BlockNumber,
91) -> ProviderResult<TrieUpdatesSorted>
92where
93 N: NodePrimitives,
94 Provider: DBProvider
95 + ChangeSetReader
96 + StorageChangeSetReader
97 + PruneCheckpointReader
98 + StageCheckpointReader
99 + BlockNumReader
100 + StorageSettingsCache,
101 A: TrieTableAdapter,
102{
103 let tx = provider.tx_ref();
104 let cache = overlay_manager.changeset_cache();
105 let (partial_state_trie, finish) = database_state_frontiers(provider)?;
106
107 let changesets = cache.get_or_compute(
109 overlay_manager,
110 provider,
111 block_number,
112 partial_state_trie,
113 finish,
114 )?;
115
116 let reverts = cache.get_or_compute_range(
118 overlay_manager,
119 provider,
120 (block_number + 1)..=finish.number,
121 partial_state_trie,
122 finish,
123 )?;
124
125 let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(tx);
128 let cursor_factory = InMemoryTrieCursorFactory::new(db_cursor_factory, &reverts);
129
130 let account_nodes_ref = changesets.account_nodes_ref();
132 let mut account_nodes = Vec::with_capacity(account_nodes_ref.len());
133 let mut account_cursor = cursor_factory.account_trie_cursor()?;
134
135 for (nibbles, _old_node) in account_nodes_ref {
137 let node_value = account_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
139 account_nodes.push((*nibbles, node_value));
140 }
141
142 let mut storage_tries = B256Map::default();
144
145 for (hashed_address, storage_changeset) in changesets.storage_tries_ref() {
147 let mut storage_cursor = cursor_factory.storage_trie_cursor(*hashed_address)?;
148 let storage_nodes_ref = storage_changeset.storage_nodes_ref();
149 let mut storage_nodes = Vec::with_capacity(storage_nodes_ref.len());
150
151 for (nibbles, _old_node) in storage_nodes_ref {
153 let node_value = storage_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
155 storage_nodes.push((*nibbles, node_value));
156 }
157
158 storage_tries.insert(*hashed_address, StorageTrieUpdatesSorted { storage_nodes });
159 }
160
161 Ok(TrieUpdatesSorted::new(account_nodes, storage_tries))
162}
163
164#[derive(Debug, Clone)]
169pub(crate) struct ChangesetCache {
170 inner: Arc<RwLock<ChangesetCacheInner>>,
171}
172
173impl Default for ChangesetCache {
174 fn default() -> Self {
175 Self::new()
176 }
177}
178
179impl ChangesetCache {
180 pub(crate) fn new() -> Self {
185 Self { inner: Arc::new(RwLock::new(ChangesetCacheInner::new())) }
186 }
187
188 pub(crate) fn evict(&self, up_to_block: BlockNumber) {
198 self.inner.write().evict(up_to_block)
199 }
200
201 pub(crate) fn get_or_compute<N, P>(
215 &self,
216 overlay_manager: &OverlayManager<N>,
217 provider: &P,
218 block_number: BlockNumber,
219 partial_state_trie: BlockNumHash,
220 finish: BlockNumHash,
221 ) -> ProviderResult<Arc<TrieUpdatesSorted>>
222 where
223 N: NodePrimitives,
224 P: DBProvider
225 + ChangeSetReader
226 + StorageChangeSetReader
227 + StageCheckpointReader
228 + PruneCheckpointReader
229 + BlockNumReader
230 + StorageSettingsCache,
231 {
232 self.get_or_compute_range(
233 overlay_manager,
234 provider,
235 block_number..=block_number,
236 partial_state_trie,
237 finish,
238 )
239 }
240
241 pub(crate) fn get_or_compute_range<N, P>(
268 &self,
269 overlay_manager: &OverlayManager<N>,
270 provider: &P,
271 range: RangeInclusive<BlockNumber>,
272 partial_state_trie: BlockNumHash,
273 finish: BlockNumHash,
274 ) -> ProviderResult<Arc<TrieUpdatesSorted>>
275 where
276 N: NodePrimitives,
277 P: DBProvider
278 + ChangeSetReader
279 + StorageChangeSetReader
280 + StageCheckpointReader
281 + PruneCheckpointReader
282 + BlockNumReader
283 + StorageSettingsCache,
284 {
285 let start_block = *range.start();
286 let end_block = *range.end();
287 let timer = Instant::now();
288
289 if end_block > finish.number {
290 return Err(ProviderError::InsufficientChangesets {
291 requested: end_block,
292 available: 0..=finish.number,
293 });
294 }
295
296 debug!(
297 target: "trie::changeset_cache",
298 start_block,
299 end_block,
300 ?partial_state_trie,
301 ?finish,
302 "Starting get_or_compute_range"
303 );
304
305 if start_block > end_block {
306 debug!(
307 target: "trie::changeset_cache",
308 start_block,
309 end_block,
310 "Empty changeset range requested"
311 );
312 return Ok(Arc::new(TrieUpdatesSorted::default()))
313 }
314
315 let end_block_hash = provider.block_hash(end_block)?.ok_or_else(|| {
316 ProviderError::other(std::io::Error::new(
317 std::io::ErrorKind::NotFound,
318 format!("block hash not found for block number {}", end_block),
319 ))
320 })?;
321 let range_key = ChangesetRangeKey::new(start_block, end_block, end_block_hash);
322
323 if let Some(accumulated_reverts) = self.inner.read().get(&range_key) {
324 let elapsed = timer.elapsed();
325
326 debug!(
327 target: "trie::changeset_cache",
328 ?elapsed,
329 start_block,
330 end_block,
331 ?end_block_hash,
332 num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
333 "Changeset cache HIT for block range"
334 );
335
336 return Ok(accumulated_reverts)
337 }
338
339 let mut cached_reverts =
340 Vec::with_capacity(end_block.saturating_sub(start_block).saturating_add(1) as usize);
341 let mut all_cached = true;
342
343 for block_number in range.rev() {
344 let block_hash = if block_number == end_block {
346 end_block_hash
347 } else {
348 provider.block_hash(block_number)?.ok_or_else(|| {
349 ProviderError::other(std::io::Error::new(
350 std::io::ErrorKind::NotFound,
351 format!("block hash not found for block number {}", block_number),
352 ))
353 })?
354 };
355
356 debug!(
357 target: "trie::changeset_cache",
358 block_number,
359 ?block_hash,
360 "Looked up block hash for block number in range"
361 );
362
363 let block_key = ChangesetRangeKey::single(block_number, block_hash);
364 if let Some(changesets) = self.inner.read().get(&block_key) {
365 cached_reverts.push(changesets);
366 } else {
367 all_cached = false;
368 break
369 }
370 }
371
372 if all_cached {
373 cached_reverts.reverse();
375 let accumulated_reverts = Arc::new(TrieUpdatesSorted::merge_slice(&cached_reverts));
376 let elapsed = timer.elapsed();
377
378 let num_account_nodes = accumulated_reverts.account_nodes_ref().len();
379 let num_storage_tries = accumulated_reverts.storage_tries_ref().len();
380
381 debug!(
382 target: "trie::changeset_cache",
383 ?elapsed,
384 start_block,
385 end_block,
386 num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
387 num_account_nodes,
388 num_storage_tries,
389 "Finished accumulating cached trie reverts for block range"
390 );
391
392 self.inner.write().insert(range_key, Arc::clone(&accumulated_reverts));
393 return Ok(accumulated_reverts)
394 }
395
396 warn!(
397 target: "trie::changeset_cache",
398 start_block,
399 end_block,
400 "Changeset cache MISS in range, falling back to aggregate DB-based computation"
401 );
402
403 let overlay = overlay_manager
404 .overlay_builder(finish.hash)
405 .with_no_reverts()
406 .build_state_trie_overlay_at_frontiers(provider, partial_state_trie, finish, true)?;
407
408 let mut forward_updates = overlay_manager
411 .parent_chain(finish.hash)
412 .take_while(|block| block.recovered_block().number() >= start_block)
413 .map(|block| (block.recovered_block().number(), block.trie_updates()))
414 .collect::<Vec<_>>();
415 forward_updates.reverse();
416
417 let state_trie_provider = OverlayStateProvider::<&P, N>::new_with_state_trie(
418 provider,
419 overlay,
420 provider.cached_storage_settings().is_v2(),
421 );
422
423 let accumulated_reverts = Arc::new(reth_trie_db::compute_range_trie_changesets(
424 provider,
425 &state_trie_provider,
426 &forward_updates,
427 start_block..=end_block,
428 finish.number,
429 )?);
430
431 let elapsed = timer.elapsed();
432
433 let num_account_nodes = accumulated_reverts.account_nodes_ref().len();
434 let num_storage_tries = accumulated_reverts.storage_tries_ref().len();
435
436 debug!(
437 target: "trie::changeset_cache",
438 ?elapsed,
439 start_block,
440 end_block,
441 ?end_block_hash,
442 num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
443 num_account_nodes,
444 num_storage_tries,
445 "Finished accumulating trie reverts for block range"
446 );
447
448 self.inner.write().insert(range_key, Arc::clone(&accumulated_reverts));
449
450 Ok(accumulated_reverts)
451 }
452}
453
454#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
460struct ChangesetRangeKey {
461 start_block: BlockNumber,
462 end_block: BlockNumber,
463 end_block_hash: B256,
464}
465
466impl ChangesetRangeKey {
467 const fn new(start_block: BlockNumber, end_block: BlockNumber, end_block_hash: B256) -> Self {
468 Self { start_block, end_block, end_block_hash }
469 }
470
471 const fn single(block_number: BlockNumber, block_hash: B256) -> Self {
472 Self::new(block_number, block_number, block_hash)
473 }
474}
475
476#[derive(Debug)]
497struct ChangesetCacheInner {
498 entries: HashMap<ChangesetRangeKey, Arc<TrieUpdatesSorted>>,
500
501 range_starts: BTreeMap<BlockNumber, Vec<ChangesetRangeKey>>,
503
504 metrics: ChangesetCacheMetrics,
506}
507
508#[derive(Metrics, Clone)]
513#[metrics(scope = "trie.changeset_cache")]
514struct ChangesetCacheMetrics {
515 hits: Counter,
517
518 misses: Counter,
520
521 evictions: Counter,
523
524 size: Gauge,
526}
527
528impl Default for ChangesetCacheInner {
529 fn default() -> Self {
530 Self::new()
531 }
532}
533
534impl ChangesetCacheInner {
535 fn new() -> Self {
540 Self { entries: HashMap::new(), range_starts: BTreeMap::new(), metrics: Default::default() }
541 }
542
543 fn get(&self, key: &ChangesetRangeKey) -> Option<Arc<TrieUpdatesSorted>> {
544 match self.entries.get(key) {
545 Some(changesets) => {
546 self.metrics.hits.increment(1);
547 Some(Arc::clone(changesets))
548 }
549 None => {
550 self.metrics.misses.increment(1);
551 None
552 }
553 }
554 }
555
556 fn insert(&mut self, key: ChangesetRangeKey, changesets: Arc<TrieUpdatesSorted>) {
557 debug!(
558 target: "trie::changeset_cache",
559 ?key,
560 cache_size_before = self.entries.len(),
561 "Inserting changeset into cache"
562 );
563
564 let is_new_entry = self.entries.insert(key, changesets).is_none();
565
566 if is_new_entry {
567 self.range_starts.entry(key.start_block).or_default().push(key);
568 }
569
570 self.metrics.size.set(self.entries.len() as f64);
572
573 debug!(
574 target: "trie::changeset_cache",
575 ?key,
576 cache_size_after = self.entries.len(),
577 "Changeset inserted into cache"
578 );
579 }
580
581 fn evict(&mut self, up_to_block: BlockNumber) {
582 debug!(
583 target: "trie::changeset_cache",
584 up_to_block,
585 cache_size_before = self.entries.len(),
586 "Starting cache eviction"
587 );
588
589 let range_starts_to_evict: Vec<u64> =
591 self.range_starts.range(..up_to_block).map(|(num, _)| *num).collect();
592
593 let mut evicted_count = 0;
595
596 for start_block in &range_starts_to_evict {
597 if let Some(keys) = self.range_starts.remove(start_block) {
598 debug!(
599 target: "trie::changeset_cache",
600 start_block,
601 num_ranges = keys.len(),
602 "Evicting ranges from cache"
603 );
604 for key in keys {
605 if self.entries.remove(&key).is_some() {
606 evicted_count += 1;
607 }
608 }
609 }
610 }
611
612 debug!(
613 target: "trie::changeset_cache",
614 up_to_block,
615 evicted_count,
616 cache_size_after = self.entries.len(),
617 "Finished cache eviction"
618 );
619
620 if evicted_count > 0 {
622 self.metrics.evictions.increment(evicted_count as u64);
623 self.metrics.size.set(self.entries.len() as f64);
624 }
625 }
626}
627
628#[cfg(test)]
629mod tests {
630 use super::*;
631 use crate::StateTrieOverlay;
632 use alloy_consensus::Header;
633 use alloy_primitives::{
634 keccak256,
635 map::{B256Map, HashMap},
636 Address, U256,
637 };
638 use reth_chain_state::{test_utils::TestBlockBuilder, ExecutedBlock};
639 use reth_db::{
640 models::{AccountBeforeTx, BlockNumberAddress},
641 tables,
642 transaction::DbTxMut,
643 };
644 use reth_primitives_traits::{Account, StorageEntry};
645 use reth_provider::{
646 test_utils::create_test_provider_factory, BlockWriter, StaticFileProviderFactory,
647 StaticFileSegment, StaticFileWriter,
648 };
649 use reth_stages_types::{FinishCheckpoint, StageCheckpoint, StageId};
650 use reth_storage_api::{StageCheckpointWriter, TrieWriter};
651 use reth_trie::{BranchNodeCompact, Nibbles, SortedTrieData, StateRoot};
652
653 fn create_test_changesets() -> Arc<TrieUpdatesSorted> {
655 Arc::new(TrieUpdatesSorted::new(vec![], B256Map::default()))
656 }
657
658 fn empty_overlay() -> StateTrieOverlay {
659 StateTrieOverlay::new(TrieInputSorted::default())
660 }
661
662 fn insert_test_changesets(
663 cache: &mut ChangesetCacheInner,
664 block_hash: B256,
665 block_number: BlockNumber,
666 changesets: Arc<TrieUpdatesSorted>,
667 ) {
668 cache.insert(ChangesetRangeKey::single(block_number, block_hash), changesets);
669 }
670
671 fn get_test_changesets(
672 cache: &ChangesetCacheInner,
673 block_hash: B256,
674 block_number: BlockNumber,
675 ) -> Option<Arc<TrieUpdatesSorted>> {
676 cache.get(&ChangesetRangeKey::single(block_number, block_hash))
677 }
678
679 fn test_account(balance: u64) -> Account {
680 Account { balance: U256::from(balance), ..Default::default() }
681 }
682
683 fn test_storage(slot: u64, value: u64) -> StorageEntry {
684 StorageEntry { key: B256::from(U256::from(slot)), value: U256::from(value) }
685 }
686
687 fn seed_headers(
688 factory: &impl StaticFileProviderFactory<
689 Primitives: reth_primitives_traits::NodePrimitives<BlockHeader = Header>,
690 >,
691 end_block: BlockNumber,
692 ) {
693 let static_file_provider = factory.static_file_provider();
694 let mut header_writer =
695 static_file_provider.latest_writer(StaticFileSegment::Headers).unwrap();
696 for block_number in 0..=end_block {
697 let header = Header { number: block_number, ..Default::default() };
698 header_writer
699 .append_header(&header, &B256::with_last_byte(block_number as u8))
700 .unwrap();
701 }
702 header_writer.commit().unwrap();
703 }
704
705 fn legacy_compute_range_trie_changesets<Provider>(
706 provider: &Provider,
707 range: RangeInclusive<BlockNumber>,
708 ) -> TrieUpdatesSorted
709 where
710 Provider: DBProvider
711 + ChangeSetReader
712 + StorageChangeSetReader
713 + BlockNumReader
714 + StorageSettingsCache,
715 {
716 let mut accumulated_reverts = TrieUpdatesSorted::default();
717 for block_number in range.rev() {
718 let changesets = legacy_compute_block_trie_changesets(provider, block_number);
719 accumulated_reverts.extend_ref_and_sort(&changesets);
720 }
721 accumulated_reverts
722 }
723
724 fn legacy_compute_block_trie_changesets<Provider>(
725 provider: &Provider,
726 block_number: BlockNumber,
727 ) -> TrieUpdatesSorted
728 where
729 Provider: DBProvider
730 + ChangeSetReader
731 + StorageChangeSetReader
732 + BlockNumReader
733 + StorageSettingsCache,
734 {
735 reth_trie_db::with_adapter!(provider, |A| {
736 legacy_compute_block_trie_changesets_inner::<_, A>(provider, block_number)
737 })
738 }
739
740 fn legacy_compute_block_trie_changesets_inner<Provider, A>(
741 provider: &Provider,
742 block_number: BlockNumber,
743 ) -> TrieUpdatesSorted
744 where
745 Provider: DBProvider
746 + ChangeSetReader
747 + StorageChangeSetReader
748 + BlockNumReader
749 + StorageSettingsCache,
750 A: TrieTableAdapter,
751 {
752 let individual_state_revert =
753 HashedPostStateSorted::from_reverts(provider, block_number..=block_number).unwrap();
754 let cumulative_state_revert =
755 HashedPostStateSorted::from_reverts(provider, (block_number + 1)..).unwrap();
756
757 let mut cumulative_state_revert_prev = cumulative_state_revert.clone();
758 cumulative_state_revert_prev.extend_ref_and_sort(&individual_state_revert);
759
760 type DbStateRoot<'a, TX, A> =
761 StateRoot<DatabaseTrieCursorFactory<&'a TX, A>, DatabaseHashedCursorFactory<&'a TX>>;
762
763 let input_prev = TrieInputSorted::new(
764 Arc::default(),
765 Arc::new(cumulative_state_revert_prev.clone()),
766 cumulative_state_revert_prev.construct_prefix_sets(),
767 );
768 let cumulative_trie_updates_prev =
769 DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(
770 provider.tx_ref(),
771 input_prev,
772 )
773 .unwrap()
774 .1
775 .into_sorted();
776
777 let input = TrieInputSorted::new(
778 Arc::new(cumulative_trie_updates_prev.clone()),
779 Arc::new(cumulative_state_revert),
780 individual_state_revert.construct_prefix_sets(),
781 );
782 let trie_updates =
783 DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(provider.tx_ref(), input)
784 .unwrap()
785 .1
786 .into_sorted();
787
788 let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(provider.tx_ref());
789 let state_provider_factory =
790 InMemoryTrieCursorFactory::new(db_cursor_factory, &cumulative_trie_updates_prev);
791
792 compute_trie_changesets(&state_provider_factory, &trie_updates).unwrap()
793 }
794
795 fn seed_tip_trie_tables<Provider, A>(provider: &Provider)
796 where
797 Provider: DBProvider + TrieWriter,
798 A: TrieTableAdapter,
799 {
800 type DbStateRoot<'a, TX, A> =
801 StateRoot<DatabaseTrieCursorFactory<&'a TX, A>, DatabaseHashedCursorFactory<&'a TX>>;
802
803 let (_, trie_updates) =
804 DbStateRoot::<_, A>::from_tx(provider.tx_ref()).root_with_updates().unwrap();
805 provider.write_trie_updates(trie_updates).unwrap();
806 }
807
808 #[test]
809 fn consumers_restore_trie_values_masked_by_deferred_blocks() {
810 let manager = OverlayManager::<reth_ethereum_primitives::EthPrimitives>::default();
811 let factory = create_test_provider_factory();
812 let blocks = TestBlockBuilder::eth().get_executed_blocks(0..11).collect::<Vec<_>>();
813 let provider = factory.provider_rw().unwrap();
814 for block in &blocks {
815 provider.insert_block(block.recovered_block()).unwrap();
816 }
817 provider
818 .save_stage_checkpoint(
819 StageId::Finish,
820 StageCheckpoint::new(10)
821 .with_finish_stage_checkpoint(FinishCheckpoint { partial_state_trie: Some(5) }),
822 )
823 .unwrap();
824 let path = Nibbles::from_nibbles([1]);
825 let updates = |mask| {
826 Arc::new(TrieUpdatesSorted::new(
827 vec![(path, Some(BranchNodeCompact::new(mask, 0, 0, vec![], None)))],
828 B256Map::default(),
829 ))
830 };
831 let block4 = updates(1);
832 let block9 = updates(2);
833 let masked = TrieUpdatesSorted::disjointed_merge_batch(&[&block4], &[&block9]);
835 assert!(masked.is_empty());
836 provider.write_trie_updates_sorted(&masked).unwrap();
837 provider.commit().unwrap();
838 let provider = factory.provider_rw().unwrap();
839 for (start, end, before) in [
840 (4, 4, Arc::new(TrieUpdatesSorted::new(vec![(path, None)], B256Map::default()))),
841 (5, 10, Arc::clone(&block4)),
842 (5, 5, Arc::default()),
843 ] {
844 manager.changeset_cache().inner.write().insert(
845 ChangesetRangeKey::new(start, end, blocks[end as usize].recovered_block().hash()),
846 before,
847 );
848 }
849 let (frontier, finish) = database_state_frontiers(&*provider).unwrap();
850 let overlay = manager
851 .overlay_builder(blocks[4].recovered_block().hash())
852 .build_state_trie_overlay_at_frontiers(&*provider, frontier, finish, true)
853 .unwrap();
854 assert_eq!(overlay.input().nodes, block4);
855 assert_eq!(manager.compute_block_trie_updates(&*provider, 4).unwrap(), *block4);
856 }
857
858 #[test]
859 fn calculated_reverts_preserve_deletions_across_frontiers() {
860 let factory = create_test_provider_factory();
861 let blocks = TestBlockBuilder::eth().get_executed_blocks(0..4).collect::<Vec<_>>();
862 let address = Address::with_last_byte(1);
863 let hashed_address = keccak256(address);
864 let removed = Nibbles::from_nibbles([1]);
865 let replaced = Nibbles::from_nibbles([2]);
866 let node = BranchNodeCompact::new(0b0001, 0, 0, vec![], None);
867 let storage_updates = |storage_nodes| {
868 TrieUpdatesSorted::new(
869 vec![],
870 B256Map::from_iter([(hashed_address, StorageTrieUpdatesSorted { storage_nodes })]),
871 )
872 };
873 let finish_updates =
874 Arc::new(storage_updates(vec![(removed, None), (replaced, Some(node.clone()))]));
875 let provider_rw = factory.provider_rw().unwrap();
876 for block in &blocks {
877 provider_rw.insert_block(block.recovered_block()).unwrap();
878 }
879 provider_rw
880 .write_trie_updates_sorted(&storage_updates(vec![(removed, Some(node.clone()))]))
881 .unwrap();
882 provider_rw
884 .tx_ref()
885 .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
886 .unwrap();
887 provider_rw
888 .save_stage_checkpoint(
889 StageId::Finish,
890 StageCheckpoint::new(2)
891 .with_finish_stage_checkpoint(FinishCheckpoint { partial_state_trie: Some(1) }),
892 )
893 .unwrap();
894 provider_rw.commit().unwrap();
895
896 let manager = OverlayManager::<reth_ethereum_primitives::EthPrimitives>::default();
897 for (block, updates) in [
898 (&blocks[1], Arc::new(storage_updates(vec![(removed, Some(node))]))),
899 (&blocks[2], Arc::clone(&finish_updates)),
900 (&blocks[3], Arc::default()),
901 ] {
902 manager.insert_block(ExecutedBlock::new(
903 Arc::clone(&block.recovered_block),
904 Arc::clone(&block.execution_output),
905 SortedTrieData::new(Arc::default(), updates),
906 ));
907 }
908 let before = factory.provider().unwrap();
909 let cache = manager.changeset_cache();
910 let (frontier, finish) = database_state_frontiers(&before).unwrap();
911 let result =
912 cache.get_or_compute_range(&manager, &before, 1..=2, frontier, finish).unwrap();
913 assert_eq!(
914 result.storage_tries_ref()[&hashed_address].storage_nodes_ref(),
915 &[(removed, None), (replaced, None)],
916 );
917
918 let provider_rw = factory.provider_rw().unwrap();
919 provider_rw.write_trie_updates_sorted(&finish_updates).unwrap();
920 provider_rw.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
921 provider_rw.commit().unwrap();
922 let after = factory.provider().unwrap();
923 let (frontier, finish) = database_state_frontiers(&after).unwrap();
924 let advanced =
925 cache.get_or_compute_range(&manager, &after, 1..=2, frontier, finish).unwrap();
926 assert!(Arc::ptr_eq(&advanced, &result));
927
928 cache.evict(2);
930 let advanced =
931 cache.get_or_compute_range(&manager, &after, 1..=2, frontier, finish).unwrap();
932 assert_eq!(advanced, result);
933
934 let cached = manager.get_or_compute_cached_changesets_range(&before, 1..=2).unwrap();
936 assert!(Arc::ptr_eq(&cached, &advanced));
937 }
938
939 #[test]
940 fn calculated_historical_range_excludes_tail_reverts() {
941 let factory = create_test_provider_factory();
942 let provider = factory.provider_rw().unwrap();
943 let address = Address::with_last_byte(1);
944 let hashed_address = keccak256(address);
945 let path = Nibbles::from_nibbles([1]);
946 let base_nodes = Arc::new(TrieUpdatesSorted::new(
947 vec![],
948 B256Map::from_iter([(
949 hashed_address,
950 StorageTrieUpdatesSorted {
951 storage_nodes: vec![(
952 path,
953 Some(BranchNodeCompact::new(0b0001, 0, 0, vec![], None)),
954 )],
955 },
956 )]),
957 ));
958 provider
960 .tx_ref()
961 .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: None })
962 .unwrap();
963 let state_trie_provider =
964 OverlayStateProvider::<&_, reth_ethereum_primitives::EthPrimitives>::new_with_state_trie(
965 &*provider,
966 StateTrieOverlay::new(TrieInputSorted::new(
967 Arc::clone(&base_nodes),
968 Arc::default(),
969 Default::default(),
970 )),
971 provider.cached_storage_settings().is_v2(),
972 );
973
974 let end = 2;
975 for range in [1..=end, end + 1..=end] {
976 let result = reth_trie_db::compute_range_trie_changesets(
977 &*provider,
978 &state_trie_provider,
979 &[(3, Arc::clone(&base_nodes))],
980 range,
981 3,
982 )
983 .unwrap();
984 assert!(result.is_empty());
985 }
986 }
987
988 #[test]
989 fn persisted_range_retains_transient_nodes_without_executed_blocks() {
990 let factory = create_test_provider_factory();
991 seed_headers(&factory, 2);
992 let provider = factory.provider_rw().unwrap();
993 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(2)).unwrap();
994 let address = Address::with_last_byte(1);
995 let hashed_address = keccak256(address);
996 provider
997 .tx_ref()
998 .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
999 .unwrap();
1000 provider
1001 .tx_ref()
1002 .put::<tables::AccountChangeSets>(
1003 2,
1004 AccountBeforeTx { address, info: Some(test_account(1)) },
1005 )
1006 .unwrap();
1007 for slot in 0..257 {
1010 provider
1011 .tx_ref()
1012 .put::<tables::StorageChangeSets>(
1013 BlockNumberAddress((1, address)),
1014 test_storage(slot, 0),
1015 )
1016 .unwrap();
1017 provider
1018 .tx_ref()
1019 .put::<tables::StorageChangeSets>(
1020 BlockNumberAddress((2, address)),
1021 test_storage(slot, 1),
1022 )
1023 .unwrap();
1024 }
1025 let state_trie_provider =
1026 OverlayStateProvider::<&_, reth_ethereum_primitives::EthPrimitives>::new_with_state_trie(
1027 &*provider,
1028 empty_overlay(),
1029 provider.cached_storage_settings().is_v2(),
1030 );
1031 let result = reth_trie_db::compute_range_trie_changesets(
1032 &*provider,
1033 &state_trie_provider,
1034 &[],
1035 1..=2,
1036 2,
1037 )
1038 .unwrap();
1039 let nodes = result.storage_tries_ref()[&hashed_address].storage_nodes_ref();
1040 assert!(!nodes.is_empty());
1041 assert!(nodes.iter().all(|(_, node)| node.is_none()));
1042
1043 let first = reth_trie_db::compute_range_trie_changesets(
1044 &*provider,
1045 &state_trie_provider,
1046 &[],
1047 1..=1,
1048 2,
1049 )
1050 .unwrap();
1051 let second = reth_trie_db::compute_range_trie_changesets(
1052 &*provider,
1053 &state_trie_provider,
1054 &[],
1055 2..=2,
1056 2,
1057 )
1058 .unwrap();
1059 let mut merged = second.clone();
1060 merged.extend_ref_and_sort(&first);
1061 assert_eq!(result, merged);
1062 let forward = [(1, Arc::new(second.clone())), (2, Arc::new(first.clone()))];
1063 assert_eq!(
1064 reth_trie_db::compute_range_trie_changesets(
1065 &*provider,
1066 &state_trie_provider,
1067 &forward,
1068 1..=2,
1069 2
1070 )
1071 .unwrap(),
1072 result,
1073 );
1074 let manager = OverlayManager::<reth_ethereum_primitives::EthPrimitives>::default();
1075 assert_eq!(manager.compute_block_trie_updates(&*provider, 1).unwrap(), second);
1076 assert_eq!(manager.compute_block_trie_updates(&*provider, 2).unwrap(), first);
1077
1078 let at_first = OverlayStateProvider::<&_, reth_ethereum_primitives::EthPrimitives>::new_with_state_trie(
1080 &*provider,
1081 StateTrieOverlay::new(TrieInputSorted::new(
1082 Arc::new(second),
1083 Arc::new(HashedPostStateSorted::from_reverts(&*provider, 2..=2).unwrap()),
1084 Default::default(),
1085 )),
1086 provider.cached_storage_settings().is_v2(),
1087 );
1088 let at_first =
1089 reth_trie_db::compute_range_trie_changesets(&*provider, &at_first, &[], 1..=1, 1)
1090 .unwrap();
1091 assert_eq!(first, at_first);
1092 }
1093
1094 #[test]
1095 fn cached_range_merge_keeps_oldest_revert_values() {
1096 let factory = create_test_provider_factory();
1097 seed_headers(&factory, 2);
1098
1099 let provider = factory.provider_rw().unwrap();
1100 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(2)).unwrap();
1101
1102 let cache = ChangesetCache::new();
1103 let path = Nibbles::from_nibbles([0x1, 0x2]);
1104 let older_node = BranchNodeCompact::new(0b0001, 0, 0, vec![], None);
1105 let newer_node = BranchNodeCompact::new(0b0010, 0, 0, vec![], None);
1106
1107 {
1108 let mut cache = cache.inner.write();
1109 insert_test_changesets(
1110 &mut cache,
1111 B256::with_last_byte(1),
1112 1,
1113 Arc::new(TrieUpdatesSorted::new(
1114 vec![(path, Some(older_node.clone()))],
1115 B256Map::default(),
1116 )),
1117 );
1118 insert_test_changesets(
1119 &mut cache,
1120 B256::with_last_byte(2),
1121 2,
1122 Arc::new(TrieUpdatesSorted::new(
1123 vec![(path, Some(newer_node))],
1124 B256Map::default(),
1125 )),
1126 );
1127 }
1128
1129 let overlay_manager = OverlayManager::<reth_ethereum_primitives::EthPrimitives>::default();
1130 let (partial_state_trie, finish) = database_state_frontiers(&*provider).unwrap();
1131 let accumulated = cache
1132 .get_or_compute_range(&overlay_manager, &*provider, 1..=2, partial_state_trie, finish)
1133 .unwrap();
1134 assert_eq!(accumulated.account_nodes_ref(), &[(path, Some(older_node))]);
1135 }
1136
1137 #[test]
1138 fn aggregate_range_reverts_to_pre_range_state() {
1139 let factory = create_test_provider_factory();
1140 seed_headers(&factory, 3);
1141
1142 let provider = factory.provider_rw().unwrap();
1143 let address = Address::with_last_byte(1);
1144 let hashed_address = keccak256(address);
1145 let slot1 = B256::with_last_byte(1);
1146 let slot2 = B256::with_last_byte(2);
1147 let account1 = test_account(10);
1148 let account2 = test_account(20);
1149 let account3 = test_account(30);
1150
1151 provider.tx_ref().put::<tables::HashedAccounts>(hashed_address, account3).unwrap();
1152 provider
1153 .tx_ref()
1154 .put::<tables::HashedStorages>(
1155 hashed_address,
1156 StorageEntry { key: keccak256(slot1), value: U256::from(25) },
1157 )
1158 .unwrap();
1159 provider
1160 .tx_ref()
1161 .put::<tables::HashedStorages>(
1162 hashed_address,
1163 StorageEntry { key: keccak256(slot2), value: U256::from(20) },
1164 )
1165 .unwrap();
1166
1167 provider
1168 .tx_ref()
1169 .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
1170 .unwrap();
1171 provider
1172 .tx_ref()
1173 .put::<tables::AccountChangeSets>(2, AccountBeforeTx { address, info: Some(account1) })
1174 .unwrap();
1175 provider
1176 .tx_ref()
1177 .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: Some(account2) })
1178 .unwrap();
1179
1180 provider
1181 .tx_ref()
1182 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(1, 0))
1183 .unwrap();
1184 provider
1185 .tx_ref()
1186 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(2, 0))
1187 .unwrap();
1188 provider
1189 .tx_ref()
1190 .put::<tables::StorageChangeSets>(
1191 BlockNumberAddress((2, address)),
1192 StorageEntry { key: slot1, value: U256::from(10) },
1193 )
1194 .unwrap();
1195 provider
1196 .tx_ref()
1197 .put::<tables::StorageChangeSets>(
1198 BlockNumberAddress((3, address)),
1199 StorageEntry { key: slot1, value: U256::from(15) },
1200 )
1201 .unwrap();
1202
1203 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
1204 reth_trie_db::with_adapter!(provider, |A| seed_tip_trie_tables::<_, A>(&*provider));
1205
1206 let overlay = empty_overlay();
1207 let state_trie_provider =
1208 OverlayStateProvider::<&_, reth_ethereum_primitives::EthPrimitives>::new_with_state_trie(
1209 &*provider,
1210 overlay,
1211 provider.cached_storage_settings().is_v2(),
1212 );
1213 let actual = reth_trie_db::compute_range_trie_changesets(
1214 &*provider,
1215 &state_trie_provider,
1216 &[],
1217 1..=3,
1218 3,
1219 )
1220 .unwrap();
1221 assert!(actual.storage_tries_ref().get(&hashed_address).is_none());
1222
1223 let cache = ChangesetCache::new();
1224 let overlay_manager = OverlayManager::<reth_ethereum_primitives::EthPrimitives>::default();
1225 let (partial_state_trie, finish) = database_state_frontiers(&*provider).unwrap();
1226 let from_cache_api = cache
1227 .get_or_compute_range(&overlay_manager, &*provider, 1..=3, partial_state_trie, finish)
1228 .unwrap();
1229 assert_eq!(*from_cache_api, actual);
1230 assert_eq!(cache.inner.read().entries.len(), 1);
1231
1232 let block_changesets = cache
1233 .get_or_compute(&overlay_manager, &*provider, 2, partial_state_trie, finish)
1234 .unwrap();
1235 assert_eq!(*block_changesets, legacy_compute_block_trie_changesets(&*provider, 2));
1236 assert_eq!(cache.inner.read().entries.len(), 2);
1237 }
1238
1239 #[test]
1240 fn aggregate_range_matches_legacy_per_block_merge_with_storage_wipe() {
1241 let factory = create_test_provider_factory();
1242 seed_headers(&factory, 3);
1243
1244 let provider = factory.provider_rw().unwrap();
1245 let address = Address::with_last_byte(1);
1246 let slot1 = B256::with_last_byte(1);
1247 let slot2 = B256::with_last_byte(2);
1248 let account1 = test_account(10);
1249 let account2 = test_account(20);
1250
1251 provider
1252 .tx_ref()
1253 .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
1254 .unwrap();
1255 provider
1256 .tx_ref()
1257 .put::<tables::AccountChangeSets>(2, AccountBeforeTx { address, info: Some(account1) })
1258 .unwrap();
1259 provider
1260 .tx_ref()
1261 .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: Some(account2) })
1262 .unwrap();
1263
1264 provider
1265 .tx_ref()
1266 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(1, 0))
1267 .unwrap();
1268 provider
1269 .tx_ref()
1270 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(2, 0))
1271 .unwrap();
1272 provider
1273 .tx_ref()
1274 .put::<tables::StorageChangeSets>(
1275 BlockNumberAddress((2, address)),
1276 StorageEntry { key: slot1, value: U256::from(10) },
1277 )
1278 .unwrap();
1279 provider
1280 .tx_ref()
1281 .put::<tables::StorageChangeSets>(
1282 BlockNumberAddress((3, address)),
1283 StorageEntry { key: slot1, value: U256::from(15) },
1284 )
1285 .unwrap();
1286 provider
1287 .tx_ref()
1288 .put::<tables::StorageChangeSets>(
1289 BlockNumberAddress((3, address)),
1290 StorageEntry { key: slot2, value: U256::from(20) },
1291 )
1292 .unwrap();
1293
1294 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
1295 reth_trie_db::with_adapter!(provider, |A| seed_tip_trie_tables::<_, A>(&*provider));
1296
1297 let expected = legacy_compute_range_trie_changesets(&*provider, 2..=3);
1298 let overlay = empty_overlay();
1299 let state_trie_provider =
1300 OverlayStateProvider::<&_, reth_ethereum_primitives::EthPrimitives>::new_with_state_trie(
1301 &*provider,
1302 overlay,
1303 provider.cached_storage_settings().is_v2(),
1304 );
1305 let actual = reth_trie_db::compute_range_trie_changesets(
1306 &*provider,
1307 &state_trie_provider,
1308 &[],
1309 2..=3,
1310 3,
1311 )
1312 .unwrap();
1313 assert_eq!(actual, expected);
1314 }
1315
1316 #[test]
1317 fn test_eviction_when_explicitly_called() {
1318 let mut cache = ChangesetCacheInner::new();
1319
1320 let mut hashes = Vec::new();
1322 for i in 0..15 {
1323 let hash = B256::random();
1324 insert_test_changesets(&mut cache, hash, i, create_test_changesets());
1325 hashes.push((i, hash));
1326 }
1327
1328 assert_eq!(cache.entries.len(), 15);
1330
1331 cache.evict(4);
1333
1334 assert_eq!(cache.entries.len(), 11); for i in 0..4 {
1339 assert!(
1340 get_test_changesets(&cache, hashes[i as usize].1, i).is_none(),
1341 "Block {} should be evicted",
1342 i
1343 );
1344 }
1345
1346 for i in 4..15 {
1348 assert!(
1349 get_test_changesets(&cache, hashes[i as usize].1, i).is_some(),
1350 "Block {} should be present",
1351 i
1352 );
1353 }
1354 }
1355
1356 #[test]
1357 fn test_eviction_with_persistence_watermark() {
1358 let mut cache = ChangesetCacheInner::new();
1359
1360 let mut hashes = HashMap::new();
1362 for i in 100..=165 {
1363 let hash = B256::random();
1364 insert_test_changesets(&mut cache, hash, i, create_test_changesets());
1365 hashes.insert(i, hash);
1366 }
1367
1368 assert_eq!(cache.entries.len(), 66);
1370
1371 cache.evict(100);
1374
1375 assert_eq!(cache.entries.len(), 66);
1377
1378 cache.evict(101);
1381
1382 assert_eq!(cache.entries.len(), 65);
1384 assert!(get_test_changesets(&cache, hashes[&100], 100).is_none());
1385 assert!(get_test_changesets(&cache, hashes[&101], 101).is_some());
1386 }
1387
1388 #[test]
1389 fn test_out_of_order_inserts_with_explicit_eviction() {
1390 let mut cache = ChangesetCacheInner::new();
1391
1392 let hash_10 = B256::random();
1394 insert_test_changesets(&mut cache, hash_10, 10, create_test_changesets());
1395
1396 let hash_5 = B256::random();
1397 insert_test_changesets(&mut cache, hash_5, 5, create_test_changesets());
1398
1399 let hash_15 = B256::random();
1400 insert_test_changesets(&mut cache, hash_15, 15, create_test_changesets());
1401
1402 let hash_3 = B256::random();
1403 insert_test_changesets(&mut cache, hash_3, 3, create_test_changesets());
1404
1405 assert_eq!(cache.entries.len(), 4);
1407
1408 cache.evict(5);
1410
1411 assert!(get_test_changesets(&cache, hash_3, 3).is_none(), "Block 3 should be evicted");
1412 assert!(get_test_changesets(&cache, hash_5, 5).is_some(), "Block 5 should be present");
1413 assert!(get_test_changesets(&cache, hash_10, 10).is_some(), "Block 10 should be present");
1414 assert!(get_test_changesets(&cache, hash_15, 15).is_some(), "Block 15 should be present");
1415 }
1416
1417 #[test]
1418 fn test_ranges_with_same_numbers_and_different_end_hashes_are_distinct() {
1419 let mut cache = ChangesetCacheInner::new();
1420 let path = Nibbles::from_nibbles_unchecked([0x01]);
1421 let hash_a = B256::with_last_byte(1);
1422 let hash_b = B256::with_last_byte(2);
1423 let key_a = ChangesetRangeKey::new(10, 20, hash_a);
1424 let key_b = ChangesetRangeKey::new(10, 20, hash_b);
1425 let changesets_a = Arc::new(TrieUpdatesSorted::new(
1426 vec![(path, Some(BranchNodeCompact::new(0b0001, 0, 0, vec![], None)))],
1427 B256Map::default(),
1428 ));
1429 let changesets_b = Arc::new(TrieUpdatesSorted::new(
1430 vec![(path, Some(BranchNodeCompact::new(0b0010, 0, 0, vec![], None)))],
1431 B256Map::default(),
1432 ));
1433
1434 cache.insert(key_a, Arc::clone(&changesets_a));
1435 cache.insert(key_b, Arc::clone(&changesets_b));
1436
1437 assert_eq!(cache.entries.len(), 2);
1438 assert_eq!(
1439 cache.get(&key_a).unwrap().account_nodes_ref(),
1440 changesets_a.account_nodes_ref()
1441 );
1442 assert_eq!(
1443 cache.get(&key_b).unwrap().account_nodes_ref(),
1444 changesets_b.account_nodes_ref()
1445 );
1446
1447 cache.evict(11);
1448 assert!(cache.get(&key_a).is_none());
1449 assert!(cache.get(&key_b).is_none());
1450 }
1451
1452 #[test]
1453 fn test_eviction_removes_all_side_chains() {
1454 let mut cache = ChangesetCacheInner::new();
1455
1456 let hash_10a = B256::random();
1458 let hash_10b = B256::random();
1459 let hash_10c = B256::random();
1460 insert_test_changesets(&mut cache, hash_10a, 10, create_test_changesets());
1461 insert_test_changesets(&mut cache, hash_10b, 10, create_test_changesets());
1462 insert_test_changesets(&mut cache, hash_10c, 10, create_test_changesets());
1463
1464 let hash_20 = B256::random();
1465 insert_test_changesets(&mut cache, hash_20, 20, create_test_changesets());
1466
1467 assert_eq!(cache.entries.len(), 4);
1468
1469 cache.evict(15);
1471
1472 assert_eq!(cache.entries.len(), 1);
1473 assert!(get_test_changesets(&cache, hash_10a, 10).is_none());
1474 assert!(get_test_changesets(&cache, hash_10b, 10).is_none());
1475 assert!(get_test_changesets(&cache, hash_10c, 10).is_none());
1476 assert!(get_test_changesets(&cache, hash_20, 20).is_some());
1477 }
1478}