reth_trie/verify/
storage.rs

1//! Storage verification using one worker's long-lived transaction and reusable cursors.
2
3use super::{compare::SingleVerifier, pipeline::WorkerContext};
4use crate::{
5    hashed_cursor::HashedStorageCursor,
6    trie::{StorageRootCalculation, DEFAULT_INTERMEDIATE_THRESHOLD},
7    trie_cursor::{depth_first, noop::NoopStorageTrieCursor, TrieStorageCursor},
8    StorageRootProgress,
9};
10use alloy_consensus::EMPTY_ROOT_HASH;
11use alloy_primitives::{B256, U256};
12use crossbeam_channel::{Receiver, Sender};
13use reth_execution_errors::StateRootError;
14
15#[cfg(feature = "metrics")]
16use crate::{metrics::TrieRootMetrics, trie::TrieType};
17
18/// Cursors for one parallel storage verification thread of a [`Verifier`](super::Verifier).
19///
20/// Both cursors may start at any hashed address. `tx` is the transaction backing both cursors; it
21/// is held without being read and dropped after them.
22#[derive(Debug)]
23pub struct StorageWorker<T, H, TX> {
24    /// Cursor over the storage trie table.
25    pub trie: T,
26    /// Cursor over the hashed storage table.
27    pub hashed: H,
28    /// Transaction backing `trie` and `hashed`.
29    pub tx: TX,
30}
31
32impl<T, H, TX> StorageWorker<T, H, TX>
33where
34    T: TrieStorageCursor,
35    H: HashedStorageCursor<Value = U256>,
36{
37    /// Verifies the storage of each address received from `jobs`, completing requested roots
38    /// and reporting findings through `context`. Returns once `jobs` is exhausted or
39    /// verification is cancelled.
40    pub(super) fn run(
41        mut self,
42        jobs: Receiver<(B256, Option<Sender<B256>>)>,
43        context: &WorkerContext,
44    ) -> Result<(), StateRootError> {
45        #[cfg(feature = "metrics")]
46        let metrics = TrieRootMetrics::new(TrieType::Storage);
47
48        loop {
49            let Some((address, root_sender)) = context.receive(&jobs) else { return Ok(()) };
50            self.hashed.set_hashed_address(address);
51            self.trie.set_hashed_address(address);
52            let root = if self.hashed.is_storage_empty()? {
53                let mut verifier = SingleVerifier::new(Some(address), &mut self.trie)?;
54                if !verifier.finalize(&mut |output| context.report_one(output))? {
55                    return Ok(());
56                }
57                EMPTY_ROOT_HASH
58            } else {
59                let Some(root) = self.verify_storage(
60                    address,
61                    context,
62                    #[cfg(feature = "metrics")]
63                    &metrics,
64                )?
65                else {
66                    return Ok(())
67                };
68                root
69            };
70            if let Some(sender) = root_sender &&
71                !context.send(&sender, root)
72            {
73                return Ok(());
74            }
75        }
76    }
77
78    /// Recomputes the storage trie of `address` from hashed state alone and compares it against
79    /// the storage trie table. Work is done in bounded rounds so memory stays bounded for large
80    /// storages and cancellation is observed between rounds.
81    ///
82    /// Returns `None` if verification was cancelled.
83    fn verify_storage(
84        &mut self,
85        address: B256,
86        context: &WorkerContext,
87        #[cfg(feature = "metrics")] metrics: &TrieRootMetrics,
88    ) -> Result<Option<B256>, StateRootError> {
89        let mut verifier = SingleVerifier::new(Some(address), &mut self.trie)?;
90        let mut state = None;
91        loop {
92            if context.is_cancelled() {
93                return Ok(None);
94            }
95            let result = StorageRootCalculation {
96                hashed_address: address,
97                prefix_set: Default::default(),
98                previous_state: state.take(),
99                walk_all_changed_branch_children: false,
100                threshold: DEFAULT_INTERMEDIATE_THRESHOLD,
101                retain_updates: true,
102            }
103            .calculate(
104                &mut NoopStorageTrieCursor::default(),
105                &mut self.hashed,
106                #[cfg(feature = "metrics")]
107                metrics,
108            )?;
109            let (root, updates) = match result {
110                StorageRootProgress::Complete(root, _, updates) => (Some(root), updates),
111                StorageRootProgress::Progress(next, _, updates) => {
112                    state = Some(*next);
113                    (None, updates)
114                }
115            };
116            let mut nodes = updates.storage_nodes.into_iter().collect::<Vec<_>>();
117            nodes.sort_unstable_by(|a, b| depth_first::cmp(&a.0, &b.0));
118            let mut emit = |output| context.report_one(output);
119            for (path, node) in nodes {
120                if context.is_cancelled() {
121                    return Ok(None);
122                }
123                if !verifier.next(&mut emit, path, node)? {
124                    return Ok(None);
125                }
126            }
127            if let Some(root) = root {
128                if !verifier.finalize(&mut emit)? {
129                    return Ok(None);
130                }
131                return Ok(Some(root));
132            }
133        }
134    }
135}
136
137/// Storage to verify, including whether the account walk needs its root.
138#[derive(Debug)]
139pub struct StorageJob {
140    /// Hashed address of the storage to verify.
141    pub address: B256,
142    /// Whether to return the computed root to the account walk. False still verifies the storage
143    /// trie, but discards its root and does not reserve space in the root backlog. Hashed storage
144    /// needs a root even without a hashed account row; only stale trie-only storage may omit it.
145    pub root_required: bool,
146}
reth_trie/verify/storage.rs

reth_trie/verify/
storage.rs
