1use crate::{
11 DatabaseHashedCursorFactory, DatabaseStateRoot, DatabaseTrieCursorFactory, TrieTableAdapter,
12};
13use alloy_primitives::{map::B256Map, BlockNumber, B256};
14use parking_lot::RwLock;
15use reth_primitives_traits::FastInstant as Instant;
16use reth_storage_api::{
17 BlockNumReader, ChangeSetReader, DBProvider, StorageChangeSetReader, StorageSettingsCache,
18};
19use reth_storage_errors::provider::{ProviderError, ProviderResult};
20use reth_trie::{
21 trie_cursor::{InMemoryTrieCursorFactory, TrieCursor, TrieCursorFactory},
22 TrieInputSorted,
23};
24use reth_trie_common::updates::{StorageTrieUpdatesSorted, TrieUpdatesSorted};
25use std::{
26 collections::{BTreeMap, HashMap},
27 ops::RangeInclusive,
28 sync::Arc,
29};
30use tracing::{debug, warn};
31
32#[cfg(test)]
33use reth_trie::changesets::compute_trie_changesets;
34
35#[cfg(feature = "metrics")]
36use reth_metrics::{
37 metrics::{Counter, Gauge},
38 Metrics,
39};
40
41pub fn compute_block_trie_changesets<Provider>(
67 provider: &Provider,
68 block_number: BlockNumber,
69) -> Result<TrieUpdatesSorted, ProviderError>
70where
71 Provider: DBProvider
72 + ChangeSetReader
73 + StorageChangeSetReader
74 + BlockNumReader
75 + StorageSettingsCache,
76{
77 let db_tip_block = provider.best_block_number()?;
78 crate::with_adapter!(provider, |A| {
79 compute_range_trie_changesets_inner::<_, A>(
80 provider,
81 block_number..=block_number,
82 db_tip_block,
83 )
84 })
85}
86
87fn compute_range_trie_changesets<Provider>(
88 provider: &Provider,
89 range: RangeInclusive<BlockNumber>,
90 db_tip_block: BlockNumber,
91) -> Result<TrieUpdatesSorted, ProviderError>
92where
93 Provider: DBProvider
94 + ChangeSetReader
95 + StorageChangeSetReader
96 + BlockNumReader
97 + StorageSettingsCache,
98{
99 crate::with_adapter!(provider, |A| {
100 compute_range_trie_changesets_inner::<_, A>(provider, range, db_tip_block)
101 })
102}
103
104fn compute_range_trie_changesets_inner<Provider, A>(
105 provider: &Provider,
106 range: RangeInclusive<BlockNumber>,
107 db_tip_block: BlockNumber,
108) -> Result<TrieUpdatesSorted, ProviderError>
109where
110 Provider: DBProvider
111 + ChangeSetReader
112 + StorageChangeSetReader
113 + BlockNumReader
114 + StorageSettingsCache,
115 A: TrieTableAdapter,
116{
117 let start_block = *range.start();
118 let end_block = *range.end();
119
120 if start_block > end_block {
121 return Ok(TrieUpdatesSorted::default())
122 }
123
124 if end_block > db_tip_block {
125 return Err(ProviderError::InsufficientChangesets {
126 requested: end_block,
127 available: 0..=db_tip_block,
128 })
129 }
130
131 debug!(
132 target: "trie::changeset_cache",
133 start_block,
134 end_block,
135 db_tip_block,
136 "Computing range trie changesets from database state"
137 );
138
139 let range_state_revert = crate::state::from_reverts_auto(provider, range)?;
141 let range_prefix_sets = range_state_revert.construct_prefix_sets();
142
143 type DbStateRoot<'a, TX, A> = reth_trie::StateRoot<
144 DatabaseTrieCursorFactory<&'a TX, A>,
145 DatabaseHashedCursorFactory<&'a TX>,
146 >;
147
148 let (range_nodes, range_state) = if end_block == db_tip_block {
149 debug!(
150 target: "trie::changeset_cache",
151 start_block,
152 end_block,
153 db_tip_block,
154 "Skipping tail trie revert computation for tip-ended range"
155 );
156
157 (Arc::default(), Arc::new(range_state_revert))
158 } else {
159 let tail_state_revert = end_block
161 .checked_add(1)
162 .map(|next_block| crate::state::from_reverts_auto(provider, next_block..))
163 .transpose()?
164 .unwrap_or_default();
165
166 let tail_input = TrieInputSorted::new(
168 Arc::default(),
169 Arc::new(tail_state_revert.clone()),
170 tail_state_revert.construct_prefix_sets(),
171 );
172 let tail_trie_revert = DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(
173 provider.tx_ref(),
174 tail_input,
175 )
176 .map_err(ProviderError::other)?
177 .1
178 .into_sorted();
179
180 let mut pre_range_state_revert = tail_state_revert;
182 pre_range_state_revert.extend_ref_and_sort(&range_state_revert);
183
184 (Arc::new(tail_trie_revert), Arc::new(pre_range_state_revert))
185 };
186
187 let range_input = TrieInputSorted::new(range_nodes, range_state, range_prefix_sets);
188 let range_trie_revert =
189 DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(provider.tx_ref(), range_input)
190 .map_err(ProviderError::other)?
191 .1
192 .into_sorted();
193
194 debug!(
195 target: "trie::changeset_cache",
196 start_block,
197 end_block,
198 num_account_nodes = range_trie_revert.account_nodes_ref().len(),
199 num_storage_tries = range_trie_revert.storage_tries_ref().len(),
200 "Computed range trie changesets successfully"
201 );
202
203 Ok(range_trie_revert)
204}
205
206pub fn compute_block_trie_updates<Provider>(
236 cache: &ChangesetCache,
237 provider: &Provider,
238 block_number: BlockNumber,
239) -> ProviderResult<TrieUpdatesSorted>
240where
241 Provider: DBProvider
242 + ChangeSetReader
243 + StorageChangeSetReader
244 + BlockNumReader
245 + StorageSettingsCache,
246{
247 crate::with_adapter!(provider, |A| {
248 compute_block_trie_updates_inner::<_, A>(cache, provider, block_number)
249 })
250}
251
252fn compute_block_trie_updates_inner<Provider, A>(
253 cache: &ChangesetCache,
254 provider: &Provider,
255 block_number: BlockNumber,
256) -> ProviderResult<TrieUpdatesSorted>
257where
258 Provider: DBProvider
259 + ChangeSetReader
260 + StorageChangeSetReader
261 + BlockNumReader
262 + StorageSettingsCache,
263 A: TrieTableAdapter,
264{
265 let tx = provider.tx_ref();
266
267 let db_tip_block = provider.best_block_number()?;
268
269 let changesets = cache.get_or_compute(provider, block_number)?;
271
272 let reverts = cache.get_or_compute_range(provider, (block_number + 1)..=db_tip_block)?;
274
275 let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(tx);
278 let cursor_factory = InMemoryTrieCursorFactory::new(db_cursor_factory, &reverts);
279
280 let account_nodes_ref = changesets.account_nodes_ref();
282 let mut account_nodes = Vec::with_capacity(account_nodes_ref.len());
283 let mut account_cursor = cursor_factory.account_trie_cursor()?;
284
285 for (nibbles, _old_node) in account_nodes_ref {
287 let node_value = account_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
289 account_nodes.push((*nibbles, node_value));
290 }
291
292 let mut storage_tries = B256Map::default();
294
295 for (hashed_address, storage_changeset) in changesets.storage_tries_ref() {
297 let mut storage_cursor = cursor_factory.storage_trie_cursor(*hashed_address)?;
298 let storage_nodes_ref = storage_changeset.storage_nodes_ref();
299 let mut storage_nodes = Vec::with_capacity(storage_nodes_ref.len());
300
301 for (nibbles, _old_node) in storage_nodes_ref {
303 let node_value = storage_cursor.seek_exact(*nibbles)?.map(|(_, node)| node);
305 storage_nodes.push((*nibbles, node_value));
306 }
307
308 storage_tries.insert(
309 *hashed_address,
310 StorageTrieUpdatesSorted { storage_nodes, is_deleted: storage_changeset.is_deleted },
311 );
312 }
313
314 Ok(TrieUpdatesSorted::new(account_nodes, storage_tries))
315}
316
317#[derive(Debug, Clone)]
322pub struct ChangesetCache {
323 inner: Arc<RwLock<ChangesetCacheInner>>,
324}
325
326impl Default for ChangesetCache {
327 fn default() -> Self {
328 Self::new()
329 }
330}
331
332impl ChangesetCache {
333 pub fn new() -> Self {
338 Self { inner: Arc::new(RwLock::new(ChangesetCacheInner::new())) }
339 }
340
341 pub fn get(
346 &self,
347 block_hash: B256,
348 block_number: BlockNumber,
349 ) -> Option<Arc<TrieUpdatesSorted>> {
350 self.inner.read().get(&ChangesetRangeKey::single(block_number, block_hash))
351 }
352
353 pub fn evict(&self, up_to_block: BlockNumber) {
363 self.inner.write().evict(up_to_block)
364 }
365
366 pub fn get_or_compute<P>(
380 &self,
381 provider: &P,
382 block_number: BlockNumber,
383 ) -> ProviderResult<Arc<TrieUpdatesSorted>>
384 where
385 P: DBProvider
386 + ChangeSetReader
387 + StorageChangeSetReader
388 + BlockNumReader
389 + StorageSettingsCache,
390 {
391 self.get_or_compute_range(provider, block_number..=block_number)
392 }
393
394 pub fn get_or_compute_range<P>(
421 &self,
422 provider: &P,
423 range: RangeInclusive<BlockNumber>,
424 ) -> ProviderResult<Arc<TrieUpdatesSorted>>
425 where
426 P: DBProvider
427 + ChangeSetReader
428 + StorageChangeSetReader
429 + BlockNumReader
430 + StorageSettingsCache,
431 {
432 let db_tip_block = provider.best_block_number()?;
433
434 let start_block = *range.start();
435 let end_block = *range.end();
436
437 if end_block > db_tip_block {
439 return Err(ProviderError::InsufficientChangesets {
440 requested: end_block,
441 available: 0..=db_tip_block,
442 });
443 }
444
445 let timer = Instant::now();
446
447 debug!(
448 target: "trie::changeset_cache",
449 start_block,
450 end_block,
451 db_tip_block,
452 "Starting get_or_compute_range"
453 );
454
455 if start_block > end_block {
456 debug!(
457 target: "trie::changeset_cache",
458 start_block,
459 end_block,
460 "Empty changeset range requested"
461 );
462 return Ok(Arc::new(TrieUpdatesSorted::default()))
463 }
464
465 let end_block_hash = provider.block_hash(end_block)?.ok_or_else(|| {
466 ProviderError::other(std::io::Error::new(
467 std::io::ErrorKind::NotFound,
468 format!("block hash not found for block number {}", end_block),
469 ))
470 })?;
471 let range_key = ChangesetRangeKey::new(start_block, end_block, end_block_hash);
472
473 if let Some(accumulated_reverts) = self.inner.read().get(&range_key) {
474 let elapsed = timer.elapsed();
475
476 debug!(
477 target: "trie::changeset_cache",
478 ?elapsed,
479 start_block,
480 end_block,
481 ?end_block_hash,
482 num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
483 "Changeset cache HIT for block range"
484 );
485
486 return Ok(accumulated_reverts)
487 }
488
489 let mut cached_reverts =
490 Vec::with_capacity(end_block.saturating_sub(start_block).saturating_add(1) as usize);
491 let mut all_cached = true;
492
493 for block_number in range.rev() {
494 let block_hash = if block_number == end_block {
496 end_block_hash
497 } else {
498 provider.block_hash(block_number)?.ok_or_else(|| {
499 ProviderError::other(std::io::Error::new(
500 std::io::ErrorKind::NotFound,
501 format!("block hash not found for block number {}", block_number),
502 ))
503 })?
504 };
505
506 debug!(
507 target: "trie::changeset_cache",
508 block_number,
509 ?block_hash,
510 "Looked up block hash for block number in range"
511 );
512
513 let block_key = ChangesetRangeKey::single(block_number, block_hash);
514 if let Some(changesets) = self.inner.read().get(&block_key) {
515 cached_reverts.push(changesets);
516 } else {
517 all_cached = false;
518 break
519 }
520 }
521
522 if all_cached {
523 cached_reverts.reverse();
525 let accumulated_reverts = Arc::new(TrieUpdatesSorted::merge_slice(&cached_reverts));
526 let elapsed = timer.elapsed();
527
528 let num_account_nodes = accumulated_reverts.account_nodes_ref().len();
529 let num_storage_tries = accumulated_reverts.storage_tries_ref().len();
530
531 debug!(
532 target: "trie::changeset_cache",
533 ?elapsed,
534 start_block,
535 end_block,
536 num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
537 num_account_nodes,
538 num_storage_tries,
539 "Finished accumulating cached trie reverts for block range"
540 );
541
542 self.inner.write().insert(range_key, Arc::clone(&accumulated_reverts));
543 return Ok(accumulated_reverts)
544 }
545
546 warn!(
547 target: "trie::changeset_cache",
548 start_block,
549 end_block,
550 "Changeset cache MISS in range, falling back to aggregate DB-based computation"
551 );
552
553 let accumulated_reverts = Arc::new(compute_range_trie_changesets(
554 provider,
555 start_block..=end_block,
556 db_tip_block,
557 )?);
558
559 let elapsed = timer.elapsed();
560
561 let num_account_nodes = accumulated_reverts.account_nodes_ref().len();
562 let num_storage_tries = accumulated_reverts.storage_tries_ref().len();
563
564 debug!(
565 target: "trie::changeset_cache",
566 ?elapsed,
567 start_block,
568 end_block,
569 ?end_block_hash,
570 num_blocks = end_block.saturating_sub(start_block).saturating_add(1),
571 num_account_nodes,
572 num_storage_tries,
573 "Finished accumulating trie reverts for block range"
574 );
575
576 self.inner.write().insert(range_key, Arc::clone(&accumulated_reverts));
577
578 Ok(accumulated_reverts)
579 }
580}
581
582#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
588struct ChangesetRangeKey {
589 start_block: BlockNumber,
590 end_block: BlockNumber,
591 end_block_hash: B256,
592}
593
594impl ChangesetRangeKey {
595 const fn new(start_block: BlockNumber, end_block: BlockNumber, end_block_hash: B256) -> Self {
596 Self { start_block, end_block, end_block_hash }
597 }
598
599 const fn single(block_number: BlockNumber, block_hash: B256) -> Self {
600 Self::new(block_number, block_number, block_hash)
601 }
602}
603
604#[derive(Debug)]
625struct ChangesetCacheInner {
626 entries: HashMap<ChangesetRangeKey, Arc<TrieUpdatesSorted>>,
628
629 range_starts: BTreeMap<BlockNumber, Vec<ChangesetRangeKey>>,
631
632 #[cfg(feature = "metrics")]
634 metrics: ChangesetCacheMetrics,
635}
636
637#[cfg(feature = "metrics")]
638#[derive(Metrics, Clone)]
643#[metrics(scope = "trie.changeset_cache")]
644struct ChangesetCacheMetrics {
645 hits: Counter,
647
648 misses: Counter,
650
651 evictions: Counter,
653
654 size: Gauge,
656}
657
658impl Default for ChangesetCacheInner {
659 fn default() -> Self {
660 Self::new()
661 }
662}
663
664impl ChangesetCacheInner {
665 fn new() -> Self {
670 Self {
671 entries: HashMap::new(),
672 range_starts: BTreeMap::new(),
673 #[cfg(feature = "metrics")]
674 metrics: Default::default(),
675 }
676 }
677
678 fn get(&self, key: &ChangesetRangeKey) -> Option<Arc<TrieUpdatesSorted>> {
679 match self.entries.get(key) {
680 Some(changesets) => {
681 #[cfg(feature = "metrics")]
682 self.metrics.hits.increment(1);
683 Some(Arc::clone(changesets))
684 }
685 None => {
686 #[cfg(feature = "metrics")]
687 self.metrics.misses.increment(1);
688 None
689 }
690 }
691 }
692
693 fn insert(&mut self, key: ChangesetRangeKey, changesets: Arc<TrieUpdatesSorted>) {
694 debug!(
695 target: "trie::changeset_cache",
696 ?key,
697 cache_size_before = self.entries.len(),
698 "Inserting changeset into cache"
699 );
700
701 let is_new_entry = self.entries.insert(key, changesets).is_none();
702
703 if is_new_entry {
704 self.range_starts.entry(key.start_block).or_default().push(key);
705 }
706
707 #[cfg(feature = "metrics")]
709 self.metrics.size.set(self.entries.len() as f64);
710
711 debug!(
712 target: "trie::changeset_cache",
713 ?key,
714 cache_size_after = self.entries.len(),
715 "Changeset inserted into cache"
716 );
717 }
718
719 fn evict(&mut self, up_to_block: BlockNumber) {
720 debug!(
721 target: "trie::changeset_cache",
722 up_to_block,
723 cache_size_before = self.entries.len(),
724 "Starting cache eviction"
725 );
726
727 let range_starts_to_evict: Vec<u64> =
729 self.range_starts.range(..up_to_block).map(|(num, _)| *num).collect();
730
731 #[cfg(feature = "metrics")]
733 let mut evicted_count = 0;
734 #[cfg(not(feature = "metrics"))]
735 let mut evicted_count = 0;
736
737 for start_block in &range_starts_to_evict {
738 if let Some(keys) = self.range_starts.remove(start_block) {
739 debug!(
740 target: "trie::changeset_cache",
741 start_block,
742 num_ranges = keys.len(),
743 "Evicting ranges from cache"
744 );
745 for key in keys {
746 if self.entries.remove(&key).is_some() {
747 evicted_count += 1;
748 }
749 }
750 }
751 }
752
753 debug!(
754 target: "trie::changeset_cache",
755 up_to_block,
756 evicted_count,
757 cache_size_after = self.entries.len(),
758 "Finished cache eviction"
759 );
760
761 #[cfg(feature = "metrics")]
763 if evicted_count > 0 {
764 self.metrics.evictions.increment(evicted_count as u64);
765 self.metrics.size.set(self.entries.len() as f64);
766 }
767 }
768}
769
770#[cfg(test)]
771mod tests {
772 use super::*;
773 use alloy_consensus::Header;
774 use alloy_primitives::{
775 keccak256,
776 map::{B256Map, HashMap},
777 Address, U256,
778 };
779 use reth_db::tables;
780 use reth_db_api::{
781 models::{AccountBeforeTx, BlockNumberAddress},
782 transaction::DbTxMut,
783 };
784 use reth_primitives_traits::{Account, StorageEntry};
785 use reth_provider::{
786 test_utils::create_test_provider_factory, StaticFileProviderFactory, StaticFileSegment,
787 StaticFileWriter,
788 };
789 use reth_stages_types::{StageCheckpoint, StageId};
790 use reth_storage_api::{StageCheckpointWriter, TrieWriter};
791 use reth_trie::{BranchNodeCompact, Nibbles, StateRoot};
792
793 fn create_test_changesets() -> Arc<TrieUpdatesSorted> {
795 Arc::new(TrieUpdatesSorted::new(vec![], B256Map::default()))
796 }
797
798 fn insert_test_changesets(
799 cache: &mut ChangesetCacheInner,
800 block_hash: B256,
801 block_number: BlockNumber,
802 changesets: Arc<TrieUpdatesSorted>,
803 ) {
804 cache.insert(ChangesetRangeKey::single(block_number, block_hash), changesets);
805 }
806
807 fn get_test_changesets(
808 cache: &ChangesetCacheInner,
809 block_hash: B256,
810 block_number: BlockNumber,
811 ) -> Option<Arc<TrieUpdatesSorted>> {
812 cache.get(&ChangesetRangeKey::single(block_number, block_hash))
813 }
814
815 fn test_account(balance: u64) -> Account {
816 Account { balance: U256::from(balance), ..Default::default() }
817 }
818
819 fn test_storage(slot: u64, value: u64) -> StorageEntry {
820 StorageEntry { key: B256::from(U256::from(slot)), value: U256::from(value) }
821 }
822
823 fn seed_headers(
824 factory: &impl StaticFileProviderFactory<
825 Primitives: reth_primitives_traits::NodePrimitives<BlockHeader = Header>,
826 >,
827 end_block: BlockNumber,
828 ) {
829 let static_file_provider = factory.static_file_provider();
830 let mut header_writer =
831 static_file_provider.latest_writer(StaticFileSegment::Headers).unwrap();
832 for block_number in 0..=end_block {
833 let header = Header { number: block_number, ..Default::default() };
834 header_writer
835 .append_header(&header, &B256::with_last_byte(block_number as u8))
836 .unwrap();
837 }
838 header_writer.commit().unwrap();
839 }
840
841 fn legacy_compute_range_trie_changesets<Provider>(
842 provider: &Provider,
843 range: RangeInclusive<BlockNumber>,
844 ) -> TrieUpdatesSorted
845 where
846 Provider: DBProvider
847 + ChangeSetReader
848 + StorageChangeSetReader
849 + BlockNumReader
850 + StorageSettingsCache,
851 {
852 let mut accumulated_reverts = TrieUpdatesSorted::default();
853 for block_number in range.rev() {
854 let changesets = legacy_compute_block_trie_changesets(provider, block_number);
855 accumulated_reverts.extend_ref_and_sort(&changesets);
856 }
857 accumulated_reverts
858 }
859
860 fn legacy_compute_block_trie_changesets<Provider>(
861 provider: &Provider,
862 block_number: BlockNumber,
863 ) -> TrieUpdatesSorted
864 where
865 Provider: DBProvider
866 + ChangeSetReader
867 + StorageChangeSetReader
868 + BlockNumReader
869 + StorageSettingsCache,
870 {
871 crate::with_adapter!(provider, |A| {
872 legacy_compute_block_trie_changesets_inner::<_, A>(provider, block_number)
873 })
874 }
875
876 fn legacy_compute_block_trie_changesets_inner<Provider, A>(
877 provider: &Provider,
878 block_number: BlockNumber,
879 ) -> TrieUpdatesSorted
880 where
881 Provider: DBProvider
882 + ChangeSetReader
883 + StorageChangeSetReader
884 + BlockNumReader
885 + StorageSettingsCache,
886 A: TrieTableAdapter,
887 {
888 let individual_state_revert =
889 crate::state::from_reverts_auto(provider, block_number..=block_number).unwrap();
890 let cumulative_state_revert =
891 crate::state::from_reverts_auto(provider, (block_number + 1)..).unwrap();
892
893 let mut cumulative_state_revert_prev = cumulative_state_revert.clone();
894 cumulative_state_revert_prev.extend_ref_and_sort(&individual_state_revert);
895
896 type DbStateRoot<'a, TX, A> =
897 StateRoot<DatabaseTrieCursorFactory<&'a TX, A>, DatabaseHashedCursorFactory<&'a TX>>;
898
899 let input_prev = TrieInputSorted::new(
900 Arc::default(),
901 Arc::new(cumulative_state_revert_prev.clone()),
902 cumulative_state_revert_prev.construct_prefix_sets(),
903 );
904 let cumulative_trie_updates_prev =
905 DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(
906 provider.tx_ref(),
907 input_prev,
908 )
909 .unwrap()
910 .1
911 .into_sorted();
912
913 let input = TrieInputSorted::new(
914 Arc::new(cumulative_trie_updates_prev.clone()),
915 Arc::new(cumulative_state_revert),
916 individual_state_revert.construct_prefix_sets(),
917 );
918 let trie_updates =
919 DbStateRoot::<_, A>::overlay_root_from_nodes_with_updates(provider.tx_ref(), input)
920 .unwrap()
921 .1
922 .into_sorted();
923
924 let db_cursor_factory = DatabaseTrieCursorFactory::<_, A>::new(provider.tx_ref());
925 let overlay_factory =
926 InMemoryTrieCursorFactory::new(db_cursor_factory, &cumulative_trie_updates_prev);
927
928 compute_trie_changesets(&overlay_factory, &trie_updates).unwrap()
929 }
930
931 fn seed_tip_trie_tables<Provider, A>(provider: &Provider)
932 where
933 Provider: DBProvider + TrieWriter,
934 A: TrieTableAdapter,
935 {
936 type DbStateRoot<'a, TX, A> =
937 StateRoot<DatabaseTrieCursorFactory<&'a TX, A>, DatabaseHashedCursorFactory<&'a TX>>;
938
939 let (_, trie_updates) =
940 DbStateRoot::<_, A>::from_tx(provider.tx_ref()).root_with_updates().unwrap();
941 provider.write_trie_updates(trie_updates).unwrap();
942 }
943
944 #[test]
945 fn cached_range_merge_keeps_oldest_revert_values() {
946 let factory = create_test_provider_factory();
947 seed_headers(&factory, 2);
948
949 let provider = factory.provider_rw().unwrap();
950 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(2)).unwrap();
951
952 let cache = ChangesetCache::new();
953 let path = Nibbles::from_nibbles([0x1, 0x2]);
954 let older_node = BranchNodeCompact::new(0b0001, 0, 0, vec![], None);
955 let newer_node = BranchNodeCompact::new(0b0010, 0, 0, vec![], None);
956
957 {
958 let mut cache = cache.inner.write();
959 insert_test_changesets(
960 &mut cache,
961 B256::with_last_byte(1),
962 1,
963 Arc::new(TrieUpdatesSorted::new(
964 vec![(path, Some(older_node.clone()))],
965 B256Map::default(),
966 )),
967 );
968 insert_test_changesets(
969 &mut cache,
970 B256::with_last_byte(2),
971 2,
972 Arc::new(TrieUpdatesSorted::new(
973 vec![(path, Some(newer_node))],
974 B256Map::default(),
975 )),
976 );
977 }
978
979 let accumulated = cache.get_or_compute_range(&*provider, 1..=2).unwrap();
980 assert_eq!(accumulated.account_nodes_ref(), &[(path, Some(older_node))]);
981 }
982
983 #[test]
984 fn aggregate_range_reverts_to_pre_range_state() {
985 let factory = create_test_provider_factory();
986 seed_headers(&factory, 3);
987
988 let provider = factory.provider_rw().unwrap();
989 let address = Address::with_last_byte(1);
990 let hashed_address = keccak256(address);
991 let slot1 = B256::from(U256::from(1));
992 let slot2 = B256::from(U256::from(2));
993 let account1 = test_account(10);
994 let account2 = test_account(20);
995 let account3 = test_account(30);
996
997 provider.tx_ref().put::<tables::HashedAccounts>(hashed_address, account3).unwrap();
998 provider
999 .tx_ref()
1000 .put::<tables::HashedStorages>(
1001 hashed_address,
1002 StorageEntry { key: keccak256(slot1), value: U256::from(25) },
1003 )
1004 .unwrap();
1005 provider
1006 .tx_ref()
1007 .put::<tables::HashedStorages>(
1008 hashed_address,
1009 StorageEntry { key: keccak256(slot2), value: U256::from(20) },
1010 )
1011 .unwrap();
1012
1013 provider
1014 .tx_ref()
1015 .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
1016 .unwrap();
1017 provider
1018 .tx_ref()
1019 .put::<tables::AccountChangeSets>(2, AccountBeforeTx { address, info: Some(account1) })
1020 .unwrap();
1021 provider
1022 .tx_ref()
1023 .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: Some(account2) })
1024 .unwrap();
1025
1026 provider
1027 .tx_ref()
1028 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(1, 0))
1029 .unwrap();
1030 provider
1031 .tx_ref()
1032 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(2, 0))
1033 .unwrap();
1034 provider
1035 .tx_ref()
1036 .put::<tables::StorageChangeSets>(
1037 BlockNumberAddress((2, address)),
1038 StorageEntry { key: slot1, value: U256::from(10) },
1039 )
1040 .unwrap();
1041 provider
1042 .tx_ref()
1043 .put::<tables::StorageChangeSets>(
1044 BlockNumberAddress((3, address)),
1045 StorageEntry { key: slot1, value: U256::from(15) },
1046 )
1047 .unwrap();
1048
1049 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
1050 crate::with_adapter!(provider, |A| seed_tip_trie_tables::<_, A>(&*provider));
1051
1052 let actual = compute_range_trie_changesets(&*provider, 1..=3, 3).unwrap();
1053 let storage_revert = actual
1054 .storage_tries_ref()
1055 .get(&hashed_address)
1056 .expect("created account storage trie should be deleted by range revert");
1057 assert!(storage_revert.is_deleted());
1058 assert!(storage_revert.storage_nodes_ref().is_empty());
1059
1060 let cache = ChangesetCache::new();
1061 let from_cache_api = cache.get_or_compute_range(&*provider, 1..=3).unwrap();
1062 assert_eq!(*from_cache_api, actual);
1063 assert_eq!(cache.inner.read().entries.len(), 1);
1064
1065 let block_changesets = cache.get_or_compute(&*provider, 2).unwrap();
1066 assert_eq!(*block_changesets, legacy_compute_block_trie_changesets(&*provider, 2));
1067 assert_eq!(cache.inner.read().entries.len(), 2);
1068 }
1069
1070 #[test]
1071 fn aggregate_range_matches_legacy_per_block_merge_with_storage_wipe() {
1072 let factory = create_test_provider_factory();
1073 seed_headers(&factory, 3);
1074
1075 let provider = factory.provider_rw().unwrap();
1076 let address = Address::with_last_byte(1);
1077 let slot1 = B256::from(U256::from(1));
1078 let slot2 = B256::from(U256::from(2));
1079 let account1 = test_account(10);
1080 let account2 = test_account(20);
1081
1082 provider
1083 .tx_ref()
1084 .put::<tables::AccountChangeSets>(1, AccountBeforeTx { address, info: None })
1085 .unwrap();
1086 provider
1087 .tx_ref()
1088 .put::<tables::AccountChangeSets>(2, AccountBeforeTx { address, info: Some(account1) })
1089 .unwrap();
1090 provider
1091 .tx_ref()
1092 .put::<tables::AccountChangeSets>(3, AccountBeforeTx { address, info: Some(account2) })
1093 .unwrap();
1094
1095 provider
1096 .tx_ref()
1097 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(1, 0))
1098 .unwrap();
1099 provider
1100 .tx_ref()
1101 .put::<tables::StorageChangeSets>(BlockNumberAddress((1, address)), test_storage(2, 0))
1102 .unwrap();
1103 provider
1104 .tx_ref()
1105 .put::<tables::StorageChangeSets>(
1106 BlockNumberAddress((2, address)),
1107 StorageEntry { key: slot1, value: U256::from(10) },
1108 )
1109 .unwrap();
1110 provider
1111 .tx_ref()
1112 .put::<tables::StorageChangeSets>(
1113 BlockNumberAddress((3, address)),
1114 StorageEntry { key: slot1, value: U256::from(15) },
1115 )
1116 .unwrap();
1117 provider
1118 .tx_ref()
1119 .put::<tables::StorageChangeSets>(
1120 BlockNumberAddress((3, address)),
1121 StorageEntry { key: slot2, value: U256::from(20) },
1122 )
1123 .unwrap();
1124
1125 provider.save_stage_checkpoint(StageId::Finish, StageCheckpoint::new(3)).unwrap();
1126 crate::with_adapter!(provider, |A| seed_tip_trie_tables::<_, A>(&*provider));
1127
1128 let expected = legacy_compute_range_trie_changesets(&*provider, 2..=3);
1129 let actual = compute_range_trie_changesets(&*provider, 2..=3, 3).unwrap();
1130 assert_eq!(actual, expected);
1131 }
1132
1133 #[test]
1134 fn test_insert_and_retrieve_single_entry() {
1135 let mut cache = ChangesetCacheInner::new();
1136 let hash = B256::random();
1137 let changesets = create_test_changesets();
1138
1139 insert_test_changesets(&mut cache, hash, 100, Arc::clone(&changesets));
1140
1141 let retrieved = get_test_changesets(&cache, hash, 100);
1143 assert!(retrieved.is_some());
1144 assert_eq!(cache.entries.len(), 1);
1145 }
1146
1147 #[test]
1148 fn test_insert_multiple_entries() {
1149 let mut cache = ChangesetCacheInner::new();
1150
1151 let mut hashes = Vec::new();
1153 for i in 0..10 {
1154 let hash = B256::random();
1155 insert_test_changesets(&mut cache, hash, 100 + i, create_test_changesets());
1156 hashes.push((100 + i, hash));
1157 }
1158
1159 assert_eq!(cache.entries.len(), 10);
1161 for (block_number, hash) in hashes {
1162 assert!(get_test_changesets(&cache, hash, block_number).is_some());
1163 }
1164 }
1165
1166 #[test]
1167 fn test_eviction_when_explicitly_called() {
1168 let mut cache = ChangesetCacheInner::new();
1169
1170 let mut hashes = Vec::new();
1172 for i in 0..15 {
1173 let hash = B256::random();
1174 insert_test_changesets(&mut cache, hash, i, create_test_changesets());
1175 hashes.push((i, hash));
1176 }
1177
1178 assert_eq!(cache.entries.len(), 15);
1180
1181 cache.evict(4);
1183
1184 assert_eq!(cache.entries.len(), 11); for i in 0..4 {
1189 assert!(
1190 get_test_changesets(&cache, hashes[i as usize].1, i).is_none(),
1191 "Block {} should be evicted",
1192 i
1193 );
1194 }
1195
1196 for i in 4..15 {
1198 assert!(
1199 get_test_changesets(&cache, hashes[i as usize].1, i).is_some(),
1200 "Block {} should be present",
1201 i
1202 );
1203 }
1204 }
1205
1206 #[test]
1207 fn test_eviction_with_persistence_watermark() {
1208 let mut cache = ChangesetCacheInner::new();
1209
1210 let mut hashes = HashMap::new();
1212 for i in 100..=165 {
1213 let hash = B256::random();
1214 insert_test_changesets(&mut cache, hash, i, create_test_changesets());
1215 hashes.insert(i, hash);
1216 }
1217
1218 assert_eq!(cache.entries.len(), 66);
1220
1221 cache.evict(100);
1224
1225 assert_eq!(cache.entries.len(), 66);
1227
1228 cache.evict(101);
1231
1232 assert_eq!(cache.entries.len(), 65);
1234 assert!(get_test_changesets(&cache, hashes[&100], 100).is_none());
1235 assert!(get_test_changesets(&cache, hashes[&101], 101).is_some());
1236 }
1237
1238 #[test]
1239 fn test_out_of_order_inserts_with_explicit_eviction() {
1240 let mut cache = ChangesetCacheInner::new();
1241
1242 let hash_10 = B256::random();
1244 insert_test_changesets(&mut cache, hash_10, 10, create_test_changesets());
1245
1246 let hash_5 = B256::random();
1247 insert_test_changesets(&mut cache, hash_5, 5, create_test_changesets());
1248
1249 let hash_15 = B256::random();
1250 insert_test_changesets(&mut cache, hash_15, 15, create_test_changesets());
1251
1252 let hash_3 = B256::random();
1253 insert_test_changesets(&mut cache, hash_3, 3, create_test_changesets());
1254
1255 assert_eq!(cache.entries.len(), 4);
1257
1258 cache.evict(5);
1260
1261 assert!(get_test_changesets(&cache, hash_3, 3).is_none(), "Block 3 should be evicted");
1262 assert!(get_test_changesets(&cache, hash_5, 5).is_some(), "Block 5 should be present");
1263 assert!(get_test_changesets(&cache, hash_10, 10).is_some(), "Block 10 should be present");
1264 assert!(get_test_changesets(&cache, hash_15, 15).is_some(), "Block 15 should be present");
1265 }
1266
1267 #[test]
1268 fn test_multiple_blocks_same_number() {
1269 let mut cache = ChangesetCacheInner::new();
1270
1271 let hash_1a = B256::random();
1273 let hash_1b = B256::random();
1274 insert_test_changesets(&mut cache, hash_1a, 100, create_test_changesets());
1275 insert_test_changesets(&mut cache, hash_1b, 100, create_test_changesets());
1276
1277 assert!(get_test_changesets(&cache, hash_1a, 100).is_some());
1279 assert!(get_test_changesets(&cache, hash_1b, 100).is_some());
1280 assert_eq!(cache.entries.len(), 2);
1281 }
1282
1283 #[test]
1284 fn test_ranges_with_same_numbers_and_different_end_hashes_are_distinct() {
1285 let mut cache = ChangesetCacheInner::new();
1286 let path = Nibbles::from_nibbles_unchecked([0x01]);
1287 let hash_a = B256::with_last_byte(1);
1288 let hash_b = B256::with_last_byte(2);
1289 let key_a = ChangesetRangeKey::new(10, 20, hash_a);
1290 let key_b = ChangesetRangeKey::new(10, 20, hash_b);
1291 let changesets_a = Arc::new(TrieUpdatesSorted::new(
1292 vec![(path, Some(BranchNodeCompact::new(0b0001, 0, 0, vec![], None)))],
1293 B256Map::default(),
1294 ));
1295 let changesets_b = Arc::new(TrieUpdatesSorted::new(
1296 vec![(path, Some(BranchNodeCompact::new(0b0010, 0, 0, vec![], None)))],
1297 B256Map::default(),
1298 ));
1299
1300 cache.insert(key_a, Arc::clone(&changesets_a));
1301 cache.insert(key_b, Arc::clone(&changesets_b));
1302
1303 assert_eq!(cache.entries.len(), 2);
1304 assert_eq!(
1305 cache.get(&key_a).unwrap().account_nodes_ref(),
1306 changesets_a.account_nodes_ref()
1307 );
1308 assert_eq!(
1309 cache.get(&key_b).unwrap().account_nodes_ref(),
1310 changesets_b.account_nodes_ref()
1311 );
1312
1313 cache.evict(11);
1314 assert!(cache.get(&key_a).is_none());
1315 assert!(cache.get(&key_b).is_none());
1316 }
1317
1318 #[test]
1319 fn test_eviction_removes_all_side_chains() {
1320 let mut cache = ChangesetCacheInner::new();
1321
1322 let hash_10a = B256::random();
1324 let hash_10b = B256::random();
1325 let hash_10c = B256::random();
1326 insert_test_changesets(&mut cache, hash_10a, 10, create_test_changesets());
1327 insert_test_changesets(&mut cache, hash_10b, 10, create_test_changesets());
1328 insert_test_changesets(&mut cache, hash_10c, 10, create_test_changesets());
1329
1330 let hash_20 = B256::random();
1331 insert_test_changesets(&mut cache, hash_20, 20, create_test_changesets());
1332
1333 assert_eq!(cache.entries.len(), 4);
1334
1335 cache.evict(15);
1337
1338 assert_eq!(cache.entries.len(), 1);
1339 assert!(get_test_changesets(&cache, hash_10a, 10).is_none());
1340 assert!(get_test_changesets(&cache, hash_10b, 10).is_none());
1341 assert!(get_test_changesets(&cache, hash_10c, 10).is_none());
1342 assert!(get_test_changesets(&cache, hash_20, 20).is_some());
1343 }
1344}