reth_trie/verify/
mod.rs

1//! Parallel verification of persisted account and storage tries.
2
3use self::{
4    compare::SingleVerifier,
5    pipeline::{StoragePipeline, WorkerContext},
6};
7use crate::{
8    hashed_cursor::{HashedCursor, HashedStorageCursor},
9    trie_cursor::{
10        depth_first::{self, DepthFirstTrieIterator},
11        TrieCursor, TrieStorageCursor,
12    },
13    Nibbles,
14};
15use alloy_primitives::{B256, U256};
16use alloy_rlp::Encodable;
17use alloy_trie::BranchNodeCompact;
18use crossbeam_channel::{Receiver, Sender};
19use reth_execution_errors::StateRootError;
20use reth_primitives_traits::Account;
21use reth_storage_errors::db::DatabaseError;
22use std::{
23    sync::Arc,
24    thread::{self, JoinHandle},
25};
26
27mod compare;
28mod pipeline;
29mod storage;
30pub use storage::{StorageJob, StorageWorker};
31
32/// Compares the persisted account and storage trie tables against tries recomputed from the hashed
33/// state tables, yielding an [`Output`] for every inconsistency found.
34///
35/// The hashed tables are the source of truth. Storage tries are recomputed in parallel, so storage
36/// findings are not ordered relative to account findings or to [`Output::Progress`]. Iteration
37/// stops after the first error. Only hashed account rows contribute account leaves; orphan storage
38/// tries are still verified independently.
39///
40/// Storage work ahead of the account walk is bounded by the configured storage lookahead.
41/// Findings use a bounded channel shared by account and storage producers, so a slow consumer
42/// applies backpressure without buffering an entire storage trie while waiting for its root.
43#[derive(Debug)]
44pub struct Verifier<T: TrieCursor, H> {
45    events: Receiver<Result<Output, StateRootError>>,
46    // The sole sender is dropped to broadcast cancellation to every producer.
47    cancel: Option<Sender<()>>,
48    // Retain account cursors until all producers have joined, before backing transactions drop.
49    background: Option<JoinHandle<AccountVerifier<T, H>>>,
50}
51
52const FINDINGS_CAPACITY: usize = 1024;
53
54const ACCOUNT_UPDATE_THRESHOLD: usize = 1024;
55
56/// Default maximum number of storage roots computed or in flight ahead of the account walk.
57pub const DEFAULT_STORAGE_LOOKAHEAD: usize = 1_000_000;
58
59impl<T, H> Verifier<T, H>
60where
61    T: TrieCursor + Send + 'static,
62    H: HashedCursor<Value = Account> + Send + 'static,
63{
64    /// Starts verification and its background storage root computation.
65    ///
66    /// - `trie` reads the account trie table.
67    /// - `accounts` reads the hashed accounts table in ascending address order.
68    /// - `storage_addresses` yields a job for each address present in hashed storage or stored
69    ///   storage tries exactly once, in ascending address order, including addresses without an
70    ///   account row. Jobs with hashed storage must request a root even without an account row. It
71    ///   is walked on a background thread to schedule storage work ahead of `accounts`.
72    /// - `storage_lookahead` bounds storage jobs whose requested roots have not been consumed,
73    ///   including storage without a hashed account row.
74    /// - `storage_workers` sets the storage parallelism, one thread per worker.
75    ///
76    /// Storage cursors may start at any hashed address. All cursors must read the same database
77    /// version, and any transactions backing them must stay open until the verifier is dropped.
78    /// The account walk runs on a producer thread. Account and storage findings share a bounded
79    /// channel, consumed by this iterator independently of storage root waits. Account cursors
80    /// must be transferable to that thread. Dropping the verifier cancels and joins all work.
81    ///
82    /// # Errors
83    ///
84    /// Returns an error if the account trie cannot be read or the account thread cannot be
85    /// spawned. Errors starting storage threads are yielded by the iterator.
86    ///
87    /// # Panics
88    ///
89    /// Panics if `storage_workers` is empty or `storage_lookahead` is zero.
90    pub fn new<L, W, C, TX>(
91        trie: T,
92        accounts: H,
93        storage_addresses: L,
94        storage_workers: Vec<StorageWorker<W, C, TX>>,
95        storage_lookahead: usize,
96    ) -> Result<Self, DatabaseError>
97    where
98        L: Iterator<Item = Result<StorageJob, DatabaseError>> + Send + 'static,
99        W: TrieStorageCursor + Send + 'static,
100        C: HashedStorageCursor<Value = U256> + Send + 'static,
101        TX: Send + 'static,
102    {
103        let account_verifier = SingleVerifier::new(None, trie)?;
104        assert!(!storage_workers.is_empty(), "verification requires storage workers");
105        assert!(storage_lookahead > 0, "storage lookahead must be positive");
106        let (events_tx, events) = crossbeam_channel::bounded(FINDINGS_CAPACITY);
107        let (cancel, cancel_rx) = crossbeam_channel::bounded(0);
108        let context = WorkerContext::new(events_tx, cancel_rx);
109        let mut account = AccountVerifier {
110            accounts,
111            account_verifier,
112            hash_builder: crate::HashBuilder::default().with_updates(true),
113            account_rlp: Vec::with_capacity(crate::TRIE_ACCOUNT_RLP_MAX_SIZE),
114        };
115        let span = tracing::Span::current();
116        let background = thread::Builder::new()
117            .name("trie-accounts".into())
118            .spawn(move || {
119                let _entered = span.enter();
120                match StoragePipeline::spawn(
121                    storage_addresses,
122                    storage_workers,
123                    context.clone(),
124                    storage_lookahead,
125                ) {
126                    Ok(mut pipeline) => {
127                        // Report failures before joining: the consumer cancels all producers on
128                        // error.
129                        context.run(|| account.run(&mut pipeline, &context));
130                        pipeline.finish();
131                    }
132                    Err(error) => context.run(|| Err(error.into())),
133                }
134                account
135            })
136            .map_err(|error| DatabaseError::Custom(Arc::new(error)))?;
137        Ok(Self { events, cancel: Some(cancel), background: Some(background) })
138    }
139}
140
141impl<T: TrieCursor, H> Verifier<T, H> {
142    fn stop(&mut self) {
143        self.cancel.take();
144        if let Some(background) = self.background.take() &&
145            let Err(panic) = background.join() &&
146            !thread::panicking()
147        {
148            std::panic::resume_unwind(panic);
149        }
150    }
151}
152
153impl<T: TrieCursor, H> Iterator for Verifier<T, H> {
154    type Item = Result<Output, StateRootError>;
155
156    fn next(&mut self) -> Option<Self::Item> {
157        self.cancel.as_ref()?;
158        match self.events.recv() {
159            Ok(Ok(output)) => Some(Ok(output)),
160            Ok(Err(error)) => {
161                self.stop();
162                Some(Err(error))
163            }
164            Err(_) => {
165                self.stop();
166                None
167            }
168        }
169    }
170}
171
172impl<T: TrieCursor, H> Drop for Verifier<T, H> {
173    fn drop(&mut self) {
174        self.stop();
175    }
176}
177
178/// Account producer state, returned from its thread so cursor lifetimes remain tied to `Verifier`.
179#[derive(Debug)]
180struct AccountVerifier<T: TrieCursor, H> {
181    accounts: H,
182    account_verifier: SingleVerifier<DepthFirstTrieIterator<T>>,
183    hash_builder: crate::HashBuilder,
184    account_rlp: Vec<u8>,
185}
186
187impl<T, H> AccountVerifier<T, H>
188where
189    T: TrieCursor,
190    H: HashedCursor<Value = Account>,
191{
192    fn run(
193        &mut self,
194        pipeline: &mut StoragePipeline,
195        context: &WorkerContext,
196    ) -> Result<(), StateRootError> {
197        let mut next_account = self.accounts.seek(B256::ZERO)?;
198        loop {
199            if context.is_cancelled() {
200                return Ok(());
201            }
202            // The scheduler announces storage addresses in order before dispatching their jobs.
203            let next_root = pipeline.next_root(next_account.as_ref().map(|(address, _)| *address));
204            if context.is_cancelled() {
205                return Ok(());
206            }
207            let (address, root) = match (next_account.as_ref(), next_root) {
208                (_, Some(storage)) => storage,
209                (Some((address, _)), None) => (*address, alloy_consensus::EMPTY_ROOT_HASH),
210                (None, None) => break,
211            };
212            let account = match next_account.take() {
213                Some((key, account)) if key == address => {
214                    next_account = self.accounts.next()?;
215                    account
216                }
217                next => {
218                    next_account = next;
219                    // Canonical state roots only include leaves present in hashed accounts.
220                    // Orphan storage is still verified, but cannot create an account leaf.
221                    continue;
222                }
223            };
224            self.account_rlp.clear();
225            account.into_trie_account(root).encode(&mut self.account_rlp);
226            self.hash_builder.add_leaf(Nibbles::unpack(address), &self.account_rlp);
227            if self.hash_builder.updates_len() >= ACCOUNT_UPDATE_THRESHOLD &&
228                !self.verify_account_updates(context)?
229            {
230                return Ok(());
231            }
232        }
233        // Computing the root flushes the remaining branch node updates.
234        self.hash_builder.root();
235        if !self.verify_account_updates(context)? {
236            return Ok(());
237        }
238        self.account_verifier.finalize(&mut |output| context.report_one(output))?;
239        Ok(())
240    }
241
242    fn verify_account_updates(&mut self, context: &WorkerContext) -> Result<bool, StateRootError> {
243        let (builder, updates) = std::mem::take(&mut self.hash_builder).split();
244        self.hash_builder = builder.with_updates(true);
245        // The account trie root is calculated but never persisted. Updates are sorted depth-first
246        // because that is the order in which the `HashBuilder` completes branch nodes, so each
247        // batch continues where the previous one ended.
248        let mut nodes =
249            updates.into_iter().filter(|(path, _)| !path.is_empty()).collect::<Vec<_>>();
250        nodes.sort_unstable_by(|a, b| depth_first::cmp(&a.0, &b.0));
251        for (path, node) in nodes {
252            if !self.account_verifier.next(&mut |output| context.report_one(output), path, node)? ||
253                !context.report_one(Output::Progress(path))
254            {
255                return Ok(false);
256            }
257        }
258        Ok(true)
259    }
260}
261
262/// Output describes an inconsistency found when comparing the hashed state tables
263/// ([`HashedCursor`]) with that of the trie tables ([`TrieCursor`]). The hashed
264/// tables are considered the source of truth; outputs are on the part of the trie tables.
265#[derive(Debug, Clone, PartialEq, Eq)]
266pub enum Output {
267    /// An extra account node was found.
268    AccountExtra(Nibbles, BranchNodeCompact),
269    /// An extra storage node was found.
270    StorageExtra(B256, Nibbles, BranchNodeCompact),
271    /// An account node had the wrong value.
272    AccountWrong {
273        /// Path of the node
274        path: Nibbles,
275        /// The node's expected value.
276        expected: BranchNodeCompact,
277        /// The node's found value.
278        found: BranchNodeCompact,
279    },
280    /// A storage node had the wrong value.
281    StorageWrong {
282        /// The account the storage trie belongs to.
283        account: B256,
284        /// Path of the node
285        path: Nibbles,
286        /// The node's expected value.
287        expected: BranchNodeCompact,
288        /// The node's found value.
289        found: BranchNodeCompact,
290    },
291    /// An account node was missing.
292    AccountMissing(Nibbles, BranchNodeCompact),
293    /// A storage node was missing.
294    StorageMissing(B256, Nibbles, BranchNodeCompact),
295    /// Progress indicator with the path of the last verified account trie node.
296    Progress(Nibbles),
297}
298
299/// Returns the default number of [`StorageWorker`]s to pass to [`Verifier::new`]: one per available
300/// CPU, or one if the CPU count cannot be determined.
301pub fn default_storage_workers() -> usize {
302    thread::available_parallelism().map_or(1, usize::from)
303}
reth_trie/verify/mod.rs

reth_trie/verify/
mod.rs
