reth_snap_sync/account/
store.rs

1//! Persists account ranges, their storage and code, and the coverage cursor in one transaction.
2//!
3//! Each range replaces its key interval, and ranges commit in key order. Accounts scheduled for
4//! repair are fetched again on their own and replace what the state holds for them.
5
6use crate::{
7    common::SnapRecord, repair::StoredRepairs, storage::persisted_storage_root, SnapAttemptStore,
8    SnapCatchUpStore, SnapStorageStore, SnapSyncError, SnapWrite, StateRepairs, StorageProgress,
9};
10use alloy_primitives::{
11    map::{B256Map, B256Set},
12    B256, KECCAK256_EMPTY, U256,
13};
14use reth_db_api::{
15    tables,
16    transaction::{DbTx, DbTxMut},
17    RawKey, RawTable,
18};
19use reth_downloaders::snap::VerifiedAccountRange;
20use reth_primitives_traits::Account;
21use reth_storage_api::{
22    BlockHashReader, DBProvider, MetadataProvider, MetadataWriter, SnapAttemptId, StateWriter,
23};
24use reth_trie_common::{
25    root::storage_root, HashedPostState, HashedPostStateSorted, HashedStorage, TrieAccount,
26    EMPTY_ROOT_HASH,
27};
28use revm::{bytecode::Bytecode, database::states::StateChangeset};
29use serde::{Deserialize, Serialize};
30use std::ops::Bound;
31
32/// Persistence for the account ranges an attempt downloads.
33///
34/// Blanket-implemented over the node's writers, so accounts, their dependencies and the coverage
35/// join the caller's transaction and commit together or not at all.
36pub trait SnapAccountStore {
37    /// Returns the coverage recorded for the attempt `write` belongs to, recording that no
38    /// account has been downloaded yet when there is none.
39    fn start_account_coverage(&self, write: SnapWrite) -> Result<AccountCoverage, SnapSyncError>
40    where
41        Self: MetadataWriter;
42
43    /// Returns the coverage recorded for the attempt `write` belongs to, if any.
44    fn account_coverage(&self, write: SnapWrite) -> Result<Option<AccountCoverage>, SnapSyncError>;
45
46    /// Persists `range` with its storage and code, replacing its key interval.
47    ///
48    /// Storage must match each account's root, supplied or persisted ahead of the range by
49    /// [`SnapStorageStore`], and code must be supplied or already stored. Persisted storage only
50    /// counts once catch-up has carried it to the pivot.
51    fn commit_account_range(
52        &self,
53        write: SnapWrite,
54        range: &VerifiedAccountRange,
55        storages: B256Map<HashedStorage>,
56        bytecodes: Vec<(B256, Bytecode)>,
57    ) -> Result<AccountCoverage, SnapSyncError>
58    where
59        Self: MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>;
60
61    /// Removes storage in `interval` except the contracts in `kept`, which are sorted and inside
62    /// it.
63    fn remove_storages_except(
64        &self,
65        interval: (Bound<B256>, Bound<B256>),
66        kept: &[B256],
67    ) -> Result<(), SnapSyncError>
68    where
69        Self: DBProvider<Tx: DbTxMut>;
70
71    /// Returns the repairs scheduled for the attempt `write` belongs to.
72    fn snap_repairs(&self, write: SnapWrite) -> Result<StateRepairs, SnapSyncError>;
73
74    /// Adds `repairs` to those scheduled for the attempt `write` belongs to.
75    fn schedule_snap_repairs(
76        &self,
77        write: SnapWrite,
78        repairs: StateRepairs,
79    ) -> Result<(), SnapSyncError>
80    where
81        Self: MetadataWriter;
82
83    /// Writes the pivot's account at `range`'s origin and its `slots`, returning how many accounts
84    /// remain scheduled. Refused until catch-up reaches the pivot.
85    fn commit_account_repair(
86        &self,
87        write: SnapWrite,
88        range: &VerifiedAccountRange,
89        slots: Vec<(B256, U256)>,
90    ) -> Result<usize, SnapSyncError>
91    where
92        Self: BlockHashReader + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>;
93}
94
95/// How far the account key space has been downloaded.
96#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
97pub struct AccountCoverage {
98    // Next key to request, or none once the trie is exhausted.
99    next: Option<B256>,
100}
101
102impl AccountCoverage {
103    /// Nothing downloaded yet.
104    pub const START: Self = Self { next: Some(B256::ZERO) };
105
106    /// Every account downloaded.
107    pub const COMPLETE: Self = Self { next: None };
108
109    /// Key the next range is requested from, or `None` once every account is downloaded.
110    pub const fn next(&self) -> Option<B256> {
111        self.next
112    }
113
114    /// Returns whether every account is downloaded.
115    pub const fn is_complete(&self) -> bool {
116        self.next.is_none()
117    }
118
119    // The coverage after `range`, which must start at this cursor and end past it.
120    fn advance(&self, range: &VerifiedAccountRange) -> Result<Self, SnapSyncError> {
121        let origin = range.origin();
122        if self.next != Some(origin) {
123            return Err(SnapSyncError::OutOfOrderRange { expected: self.next, got: origin })
124        }
125        if range.next().is_some_and(|next| next <= origin) {
126            return Err(SnapSyncError::NoProgress { origin })
127        }
128        Ok(Self { next: range.next() })
129    }
130
131    // Keys from this cursor up to `to`, which a range proved hold only its own accounts.
132    fn interval_to(&self, to: &Self) -> (Bound<B256>, Bound<B256>) {
133        (
134            self.next.map_or(Bound::Unbounded, Bound::Included),
135            to.next.map_or(Bound::Unbounded, Bound::Excluded),
136        )
137    }
138}
139
140// The coverage record as persisted, tied to the attempt that recorded it.
141#[derive(Serialize, Deserialize)]
142pub(crate) struct StoredCoverage {
143    // Encoding version, checked before the rest is decoded.
144    version: u32,
145    // Attempt the coverage belongs to.
146    attempt: SnapAttemptId,
147    // How far that attempt has downloaded.
148    coverage: AccountCoverage,
149}
150
151impl SnapRecord for StoredCoverage {
152    const KEY: &'static str = "snap_account_coverage";
153    const VERSION: u32 = 1;
154}
155
156impl StoredCoverage {
157    // `coverage` for `attempt` at this build's version.
158    const fn new(attempt: SnapAttemptId, coverage: AccountCoverage) -> Self {
159        Self { version: Self::VERSION, attempt, coverage }
160    }
161}
162
163impl<T: MetadataProvider> SnapAccountStore for T {
164    // Resumes the recorded coverage, so restarting an attempt keeps its progress.
165    fn start_account_coverage(&self, write: SnapWrite) -> Result<AccountCoverage, SnapSyncError>
166    where
167        Self: MetadataWriter,
168    {
169        if let Some(coverage) = self.account_coverage(write)? {
170            return Ok(coverage)
171        }
172        let start = AccountCoverage::START;
173        StoredCoverage::new(write.attempt(), start).write(self)?;
174        Ok(start)
175    }
176
177    // A record left by another attempt reads as no coverage.
178    fn account_coverage(&self, write: SnapWrite) -> Result<Option<AccountCoverage>, SnapSyncError> {
179        self.authorize_snap_write(write)?;
180        let Some(stored) = StoredCoverage::read(self)? else { return Ok(None) };
181        Ok((stored.attempt == write.attempt()).then_some(stored.coverage))
182    }
183
184    // Every check runs before the first write, so a refused range changes nothing.
185    fn commit_account_range(
186        &self,
187        write: SnapWrite,
188        range: &VerifiedAccountRange,
189        storages: B256Map<HashedStorage>,
190        bytecodes: Vec<(B256, Bytecode)>,
191    ) -> Result<AccountCoverage, SnapSyncError>
192    where
193        Self: MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
194    {
195        let attempt = self.authorize_snap_write(write)?;
196        if range.state_root() != attempt.state_root() {
197            return Err(SnapSyncError::RootMismatch {
198                expected: attempt.state_root(),
199                got: range.state_root(),
200            })
201        }
202        let coverage = self.account_coverage(write)?.ok_or(SnapSyncError::NoCoverage)?;
203        let advanced = coverage.advance(range)?;
204        let progress = self.storage_progress(write, range.origin())?;
205        // Slots persisted before the pivot moved hold its old values until the lists reach it.
206        if progress != StorageProgress::START &&
207            let Some(catch_up) = self.catch_up_progress(write)? &&
208            catch_up.applied().number < attempt.pivot().number
209        {
210            return Err(SnapSyncError::CatchUpBehindPivot {
211                applied: catch_up.applied().number,
212                pivot: attempt.pivot().number,
213            })
214        }
215        let dependencies = RangeDependencies::new(range.accounts(), storages, bytecodes);
216        let persisted = dependencies.verify(range.accounts(), &progress, self.tx_ref())?;
217
218        let interval = coverage.interval_to(&advanced);
219        self.remove::<tables::HashedAccounts>(interval)?;
220        self.remove_storages_except(interval, &persisted)?;
221        dependencies.write(self)?;
222        StoredCoverage::new(write.attempt(), advanced).write(self)?;
223        // A page ending before contracts with persisted storage leaves them to the next range.
224        if let Some(next) = advanced.next() {
225            progress.carry_to(self, write, next)?;
226        }
227        Ok(advanced)
228    }
229
230    fn remove_storages_except(
231        &self,
232        (mut start, end): (Bound<B256>, Bound<B256>),
233        kept: &[B256],
234    ) -> Result<(), SnapSyncError>
235    where
236        Self: DBProvider<Tx: DbTxMut>,
237    {
238        for account in kept {
239            self.remove::<tables::HashedStorages>((start, Bound::Excluded(*account)))?;
240            // Walks cannot start at an excluded key, so the next piece starts one key past it.
241            let Some(after) = U256::from_be_bytes(account.0).checked_add(U256::from(1)) else {
242                return Ok(())
243            };
244            start = Bound::Included(after.into());
245        }
246        self.remove::<tables::HashedStorages>((start, end))?;
247        Ok(())
248    }
249
250    // A record left by another attempt reads as nothing scheduled.
251    fn snap_repairs(&self, write: SnapWrite) -> Result<StateRepairs, SnapSyncError> {
252        self.authorize_snap_write(write)?;
253        let Some(stored) = StoredRepairs::read(self)? else { return Ok(StateRepairs::default()) };
254        Ok(if stored.attempt == write.attempt() { stored.repairs } else { StateRepairs::default() })
255    }
256
257    fn schedule_snap_repairs(
258        &self,
259        write: SnapWrite,
260        repairs: StateRepairs,
261    ) -> Result<(), SnapSyncError>
262    where
263        Self: MetadataWriter,
264    {
265        let mut scheduled = self.snap_repairs(write)?;
266        scheduled.extend(repairs);
267        StoredRepairs::store(self, write.attempt(), scheduled)
268    }
269
270    // Every check runs before the first write, so a refused repair changes nothing.
271    #[allow(clippy::clone_on_copy)]
272    fn commit_account_repair(
273        &self,
274        write: SnapWrite,
275        range: &VerifiedAccountRange,
276        slots: Vec<(B256, U256)>,
277    ) -> Result<usize, SnapSyncError>
278    where
279        Self: BlockHashReader + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
280    {
281        // The repaired values belong to the pivot, so it must still be canonical.
282        let attempt = self.authorize_canonical_snap_write(write)?;
283        if range.state_root() != attempt.state_root() {
284            return Err(SnapSyncError::RootMismatch {
285                expected: attempt.state_root(),
286                got: range.state_root(),
287            })
288        }
289        let applied =
290            self.catch_up_progress(write)?.ok_or(SnapSyncError::NoCatchUpProgress)?.applied();
291        if applied != attempt.pivot() {
292            return Err(SnapSyncError::CatchUpBehindPivot {
293                applied: applied.number,
294                pivot: attempt.pivot().number,
295            })
296        }
297        let hashed_address = range.origin();
298        // The proof starts at the origin, so an account keyed past it means none is there.
299        let account = range
300            .accounts()
301            .first()
302            .filter(|(hash, _)| *hash == hashed_address)
303            .map(|(_, account)| account);
304        if let Some(hash) = account.map(|account| account.code_hash) &&
305            hash != KECCAK256_EMPTY &&
306            self.tx_ref().get::<RawTable<tables::Bytecodes>>(RawKey::new(hash))?.is_none()
307        {
308            return Err(SnapSyncError::MissingCode { hash })
309        }
310
311        let has_storage = account.is_some_and(|account| account.storage_root != EMPTY_ROOT_HASH);
312        let mut state = HashedPostState::default().with_accounts([(
313            hashed_address,
314            account.map(|account| Account::from(account.clone())),
315        )]);
316        if has_storage {
317            // Zero values remove their slots.
318            state = state
319                .with_storages([(hashed_address, HashedStorage::from_iter(slots.iter().copied()))]);
320        } else {
321            self.remove::<tables::HashedStorages>(hashed_address..=hashed_address)?;
322        }
323        self.write_hashed_state(&state.into_sorted())?;
324
325        let mut remaining = self.snap_repairs(write)?;
326        remaining.resolve(hashed_address, has_storage.then_some(slots.as_slice()));
327        let accounts = remaining.len();
328        StoredRepairs::store(self, write.attempt(), remaining)?;
329        Ok(accounts)
330    }
331}
332
333// Accounts of a range with the storage and code supplied for them.
334struct RangeDependencies {
335    // Accounts and storage, sorted once for both the root checks and the write.
336    state: HashedPostStateSorted,
337    // Code keyed by the hash it was supplied under.
338    bytecodes: Vec<(B256, Bytecode)>,
339}
340
341impl RangeDependencies {
342    #[allow(clippy::clone_on_copy)]
343    fn new(
344        accounts: &[(B256, TrieAccount)],
345        storages: B256Map<HashedStorage>,
346        bytecodes: Vec<(B256, Bytecode)>,
347    ) -> Self {
348        let state = HashedPostState::default()
349            .with_accounts(
350                accounts
351                    .iter()
352                    .map(|(hash, account)| (*hash, Some(Account::from(account.clone())))),
353            )
354            .with_storages(storages)
355            .into_sorted();
356        Self { state, bytecodes }
357    }
358
359    // Checks storage against each account's root, and code against its hash or the table.
360    // Returns the contracts whose storage was persisted ahead of the range, in key order.
361    fn verify(
362        &self,
363        accounts: &[(B256, TrieAccount)],
364        progress: &StorageProgress,
365        tx: &impl DbTx,
366    ) -> Result<Vec<B256>, SnapSyncError> {
367        let mut available = self.supplied_code()?;
368        let mut contracts = B256Set::default();
369        let mut persisted = Vec::new();
370        for (hash, account) in accounts {
371            if account.storage_root != EMPTY_ROOT_HASH {
372                contracts.insert(*hash);
373                if self.verify_storage(*hash, account, progress, tx)? {
374                    persisted.push(*hash);
375                }
376            }
377            // Only presence matters, so stored code is not decoded.
378            if account.code_hash != KECCAK256_EMPTY &&
379                available.insert(account.code_hash) &&
380                tx.get::<RawTable<tables::Bytecodes>>(RawKey::new(account.code_hash))?.is_none()
381            {
382                return Err(SnapSyncError::MissingCode { hash: account.code_hash })
383            }
384        }
385        if let Some(account) =
386            self.state.account_storages().keys().find(|hash| !contracts.contains(*hash))
387        {
388            return Err(SnapSyncError::UnexpectedStorage { account: *account })
389        }
390        Ok(persisted)
391    }
392
393    // Hashes of the supplied code, refusing code filed under a hash it does not hash to.
394    fn supplied_code(&self) -> Result<B256Set, SnapSyncError> {
395        let mut hashes = B256Set::default();
396        for (hash, code) in &self.bytecodes {
397            let got = code.hash_slow();
398            if got != *hash {
399                return Err(SnapSyncError::CodeMismatch { expected: *hash, got })
400            }
401            hashes.insert(*hash);
402        }
403        Ok(hashes)
404    }
405
406    // Checks the contract's storage, supplied or persisted, against its root. Returns whether it
407    // was persisted.
408    fn verify_storage(
409        &self,
410        hash: B256,
411        account: &TrieAccount,
412        progress: &StorageProgress,
413        tx: &impl DbTx,
414    ) -> Result<bool, SnapSyncError> {
415        let (got, persisted) = match self.state.account_storages().get(&hash) {
416            // Zero slots are deletions, which the trie does not hold.
417            Some(storage) => (
418                storage_root(
419                    storage
420                        .storage_slots_ref()
421                        .iter()
422                        .filter(|(_, value)| !value.is_zero())
423                        .copied(),
424                ),
425                false,
426            ),
427            None if progress.is_complete(hash) => (persisted_storage_root(tx, hash)?, true),
428            None => return Err(SnapSyncError::MissingStorage { account: hash }),
429        };
430        if got != account.storage_root {
431            return Err(SnapSyncError::StorageRootMismatch {
432                account: hash,
433                expected: account.storage_root,
434                got,
435            })
436        }
437        Ok(persisted)
438    }
439
440    // Writes the accounts with their storage and code.
441    fn write(self, writer: &impl StateWriter) -> Result<(), SnapSyncError> {
442        writer.write_hashed_state(&self.state)?;
443        writer.write_state_changes(StateChangeset {
444            contracts: self.bytecodes,
445            ..Default::default()
446        })?;
447        Ok(())
448    }
449}
450
451#[cfg(test)]
452mod tests {
453    use super::*;
454    use crate::{
455        test_utils::{
456            account, generation, hashed_factory, insert_generation_headers, key, state_root,
457            stored_slots, verified_range, verified_repair,
458        },
459        SnapGeneration, StorageChunk,
460    };
461    use alloy_primitives::{Bytes, U256};
462    use reth_db_api::cursor::DbCursorRO;
463    use reth_provider::{
464        test_utils::MockNodeTypesWithDB, DatabaseProviderFactory, ProviderFactory,
465    };
466    use reth_trie_common::root::storage_root_unsorted;
467
468    const FAR: B256 = B256::repeat_byte(0xaa);
469    const SLOT: B256 = B256::repeat_byte(0x55);
470    // A slot the stale state holds and the pivot does not.
471    const OTHER: B256 = B256::repeat_byte(0x66);
472    // Sorts between the contract and the far account, holding nothing at the pivot.
473    const ABSENT: B256 = B256::repeat_byte(0x33);
474
475    fn code() -> Bytecode {
476        Bytecode::new_raw(Bytes::from_static(&[0x60, 0x00]))
477    }
478
479    // Storage holding one slot, with the root the contract fixture commits to.
480    fn storage() -> (B256, HashedStorage) {
481        let storage = HashedStorage::from_iter([(SLOT, U256::from(7))]);
482        (
483            storage_root_unsorted(storage.storage.iter().map(|(slot, value)| (*slot, *value))),
484            storage,
485        )
486    }
487
488    // A plain account, a contract with storage and code, and a far account.
489    fn accounts() -> Vec<(B256, TrieAccount)> {
490        let mut contract = account(2);
491        contract.storage_root = storage().0;
492        contract.code_hash = code().hash_slow();
493        vec![(key(1), account(1)), (key(2), contract), (FAR, account(3))]
494    }
495
496    // Storage and code that satisfy the contract in `accounts`.
497    fn dependencies() -> (B256Map<HashedStorage>, Vec<(B256, Bytecode)>) {
498        (B256Map::from_iter([(key(2), storage().1)]), vec![(code().hash_slow(), code())])
499    }
500
501    // An attempt anchored to the trie's root, with its coverage recorded.
502    fn started(
503        accounts: &[(B256, TrieAccount)],
504    ) -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite, SnapGeneration) {
505        let factory = hashed_factory();
506        let generation = generation(1, state_root(accounts));
507        let provider = factory.database_provider_rw().unwrap();
508        let write = provider.start_snap_attempt(generation).unwrap();
509        provider.start_account_coverage(write).unwrap();
510        provider.commit().unwrap();
511        (factory, write, generation)
512    }
513
514    fn stored_accounts(provider: &impl DBProvider) -> Vec<B256> {
515        let mut cursor = provider.tx_ref().cursor_read::<tables::HashedAccounts>().unwrap();
516        cursor.walk(None).unwrap().map(|entry| entry.unwrap().0).collect()
517    }
518
519    fn stored(provider: &impl DBProvider) -> (Vec<B256>, bool, bool) {
520        let tx = provider.tx_ref();
521        (
522            stored_accounts(provider),
523            tx.get::<tables::HashedStorages>(key(2)).unwrap().is_some(),
524            tx.get::<tables::Bytecodes>(code().hash_slow()).unwrap().is_some(),
525        )
526    }
527
528    #[test]
529    fn accounts_dependencies_and_coverage_commit_together() {
530        let accounts = accounts();
531        let (factory, write, _) = started(&accounts);
532        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
533        let (storages, bytecodes) = dependencies();
534
535        let provider = factory.database_provider_rw().unwrap();
536        let coverage = provider.commit_account_range(write, &range, storages, bytecodes).unwrap();
537        provider.commit().unwrap();
538
539        assert!(coverage.is_complete());
540        let provider = factory.database_provider_rw().unwrap();
541        assert_eq!(stored(&provider), (vec![key(1), key(2), FAR], true, true));
542        assert_eq!(provider.account_coverage(write).unwrap(), Some(coverage));
543    }
544
545    #[test]
546    fn an_interrupted_commit_leaves_nothing_behind() {
547        let accounts = accounts();
548        let (factory, write, _) = started(&accounts);
549        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
550        let (storages, bytecodes) = dependencies();
551
552        let provider = factory.database_provider_rw().unwrap();
553        provider.commit_account_range(write, &range, storages, bytecodes).unwrap();
554        drop(provider);
555
556        let provider = factory.database_provider_rw().unwrap();
557        assert_eq!(stored(&provider), (Vec::new(), false, false));
558        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
559    }
560
561    #[test]
562    fn a_partial_range_moves_the_coverage_to_its_next_key() {
563        let accounts = accounts();
564        let (factory, write, _) = started(&accounts);
565        // Only the first account, with a proof placing the next one at key 2.
566        let range = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
567
568        let provider = factory.database_provider_rw().unwrap();
569        let coverage =
570            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
571
572        assert_eq!(coverage.next(), Some(key(2)));
573        assert_eq!(stored_accounts(&provider), [key(1)]);
574    }
575
576    #[test]
577    fn a_proven_empty_tail_completes_the_coverage() {
578        let accounts = vec![(key(1), account(1)), (key(2), account(2))];
579        let (factory, write, _) = started(&accounts);
580        let tail = verified_range(&accounts, 0..0, key(3), &[key(3)]);
581        assert!(tail.accounts().is_empty());
582        let provider = factory.database_provider_rw().unwrap();
583        let coverage = AccountCoverage { next: Some(key(3)) };
584        StoredCoverage::new(write.attempt(), coverage).write(&provider).unwrap();
585
586        let coverage =
587            provider.commit_account_range(write, &tail, Default::default(), Vec::new()).unwrap();
588
589        assert!(coverage.is_complete());
590        assert!(stored_accounts(&provider).is_empty());
591    }
592
593    #[test]
594    fn a_range_from_another_origin_changes_nothing() {
595        let accounts = accounts();
596        let (factory, write, _) = started(&accounts);
597        let provider = factory.database_provider_rw().unwrap();
598        let first = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
599        provider.commit_account_range(write, &first, Default::default(), Vec::new()).unwrap();
600
601        // The same range again, and one skipping ahead of the cursor.
602        let duplicate =
603            provider.commit_account_range(write, &first, Default::default(), Vec::new());
604        let skipped = verified_range(&accounts, 2..3, FAR, &[FAR]);
605        let ahead = provider.commit_account_range(write, &skipped, Default::default(), Vec::new());
606
607        assert!(matches!(duplicate, Err(SnapSyncError::OutOfOrderRange { .. })));
608        assert!(matches!(ahead, Err(SnapSyncError::OutOfOrderRange { .. })));
609        assert_eq!(provider.account_coverage(write).unwrap().unwrap().next(), Some(key(2)));
610        assert_eq!(stored_accounts(&provider), [key(1)]);
611    }
612
613    #[test]
614    fn a_contract_without_its_storage_or_code_is_refused() {
615        let accounts = accounts();
616        let (factory, write, _) = started(&accounts);
617        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
618        let (storages, bytecodes) = dependencies();
619        let provider = factory.database_provider_rw().unwrap();
620
621        let no_storage =
622            provider.commit_account_range(write, &range, Default::default(), bytecodes.clone());
623        assert!(
624            matches!(no_storage, Err(SnapSyncError::MissingStorage { account }) if account == key(2))
625        );
626
627        let no_code = provider.commit_account_range(write, &range, storages, Vec::new());
628        assert!(matches!(no_code, Err(SnapSyncError::MissingCode { .. })));
629
630        let stray = B256Map::from_iter([(key(1), storage().1), (key(2), storage().1)]);
631        let unexpected = provider.commit_account_range(write, &range, stray, bytecodes);
632        assert!(
633            matches!(unexpected, Err(SnapSyncError::UnexpectedStorage { account }) if account == key(1))
634        );
635
636        assert_eq!(stored(&provider), (Vec::new(), false, false));
637        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
638    }
639
640    #[test]
641    fn storage_and_code_are_checked_against_what_the_account_commits_to() {
642        let accounts = accounts();
643        let (factory, write, _) = started(&accounts);
644        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
645        let (storages, bytecodes) = dependencies();
646        let provider = factory.database_provider_rw().unwrap();
647
648        // A slot short of the storage root, as a partial download would be.
649        let partial = B256Map::from_iter([(key(2), HashedStorage::default())]);
650        let short = provider.commit_account_range(write, &range, partial, bytecodes);
651        assert!(matches!(
652            short,
653            Err(SnapSyncError::StorageRootMismatch { account, expected, .. })
654                if account == key(2) && expected == storage().0
655        ));
656
657        // Code filed under a hash it does not hash to.
658        let other = Bytecode::new_raw(Bytes::from_static(&[0x60, 0x01]));
659        let relabelled = vec![(code().hash_slow(), other)];
660        let wrong = provider.commit_account_range(write, &range, storages, relabelled);
661        assert!(matches!(
662            wrong,
663            Err(SnapSyncError::CodeMismatch { expected, .. }) if expected == code().hash_slow()
664        ));
665
666        assert_eq!(stored(&provider), (Vec::new(), false, false));
667        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
668    }
669
670    #[test]
671    fn code_already_stored_need_not_be_supplied() {
672        let accounts = accounts();
673        let (factory, write, _) = started(&accounts);
674        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
675        let (storages, _) = dependencies();
676        let provider = factory.database_provider_rw().unwrap();
677        provider
678            .write_state_changes(StateChangeset {
679                contracts: vec![(code().hash_slow(), code())],
680                ..Default::default()
681            })
682            .unwrap();
683
684        let coverage = provider.commit_account_range(write, &range, storages, Vec::new()).unwrap();
685
686        assert!(coverage.is_complete());
687    }
688
689    #[test]
690    fn starting_coverage_again_resumes_where_the_attempt_left_off() {
691        let accounts = accounts();
692        let (factory, write, _) = started(&accounts);
693        let range = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
694        let provider = factory.database_provider_rw().unwrap();
695        let committed =
696            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
697
698        assert_eq!(provider.start_account_coverage(write).unwrap(), committed);
699        assert_eq!(provider.account_coverage(write).unwrap(), Some(committed));
700    }
701
702    #[test]
703    fn a_write_from_a_replaced_attempt_changes_nothing() {
704        let accounts = accounts();
705        let (factory, replaced, generation) = started(&accounts);
706        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
707        let (storages, bytecodes) = dependencies();
708        let provider = factory.database_provider_rw().unwrap();
709        let current = provider.start_snap_attempt(generation).unwrap();
710
711        let refused = provider.commit_account_range(replaced, &range, storages, bytecodes);
712
713        assert!(matches!(refused, Err(SnapSyncError::StaleWrite { .. })));
714        assert_eq!(stored(&provider), (Vec::new(), false, false));
715        // The replaced attempt's coverage is not the new attempt's either.
716        assert_eq!(provider.account_coverage(current).unwrap(), None);
717        assert!(matches!(
718            provider.account_coverage(replaced),
719            Err(SnapSyncError::StaleWrite { .. })
720        ));
721    }
722
723    #[test]
724    fn a_new_attempt_replaces_what_an_earlier_one_left_in_the_interval() {
725        let earlier = accounts();
726        let (factory, write, _) = started(&earlier);
727        let (storages, bytecodes) = dependencies();
728        let provider = factory.database_provider_rw().unwrap();
729        let range = verified_range(&earlier, 0..3, B256::ZERO, &[]);
730        provider.commit_account_range(write, &range, storages, bytecodes).unwrap();
731        provider.commit().unwrap();
732
733        // At the new root the contract is gone and the first account changed.
734        let current = vec![(key(1), account(9)), (FAR, account(3))];
735        let provider = factory.database_provider_rw().unwrap();
736        let write = provider.start_snap_attempt(generation(2, state_root(&current))).unwrap();
737        provider.start_account_coverage(write).unwrap();
738        let range = verified_range(&current, 0..2, B256::ZERO, &[]);
739        let coverage =
740            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
741
742        assert!(coverage.is_complete());
743        assert_eq!(stored(&provider), (vec![key(1), FAR], false, true));
744        let first = provider.tx_ref().get::<tables::HashedAccounts>(key(1)).unwrap().unwrap();
745        assert_eq!(first.nonce, 9);
746    }
747
748    #[test]
749    fn a_range_proved_against_another_root_is_refused() {
750        let accounts = accounts();
751        let (factory, write, _) = started(&accounts);
752        let provider = factory.database_provider_rw().unwrap();
753        let other = vec![(key(7), account(7))];
754        let range = verified_range(&other, 0..1, B256::ZERO, &[]);
755
756        let refused = provider.commit_account_range(write, &range, Default::default(), Vec::new());
757
758        assert!(matches!(refused, Err(SnapSyncError::RootMismatch { .. })));
759        assert!(stored_accounts(&provider).is_empty());
760    }
761
762    #[test]
763    fn a_range_without_recorded_coverage_is_refused() {
764        let accounts = accounts();
765        let factory = hashed_factory();
766        let provider = factory.database_provider_rw().unwrap();
767        let write = provider.start_snap_attempt(generation(1, state_root(&accounts))).unwrap();
768        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
769
770        assert!(matches!(
771            provider.commit_account_range(write, &range, Default::default(), Vec::new()),
772            Err(SnapSyncError::NoCoverage)
773        ));
774    }
775
776    #[test]
777    fn coverage_survives_reopening_the_database() {
778        let accounts = accounts();
779        let (factory, write, _) = started(&accounts);
780        let range = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
781        let provider = factory.database_provider_rw().unwrap();
782        let coverage =
783            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
784        provider.commit().unwrap();
785
786        let reopened = factory.database_provider_rw().unwrap();
787
788        assert_eq!(reopened.account_coverage(write).unwrap(), Some(coverage));
789    }
790
791    #[test]
792    fn storage_persisted_ahead_of_the_range_commits_with_it() {
793        let accounts = accounts();
794        let (factory, write, _) = started(&accounts);
795        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
796        let (_, bytecodes) = dependencies();
797        let provider = factory.database_provider_rw().unwrap();
798        // Left behind for an account that has no storage at this root.
799        let leftover = HashedPostState::default().with_storages([(key(1), storage().1)]);
800        provider.write_hashed_state(&leftover.into_sorted()).unwrap();
801        let slots = vec![(SLOT, U256::from(7))];
802        let chunk = StorageChunk::new(key(2), storage().0, B256::ZERO, slots, None);
803        provider.commit_storage_chunk(write, B256::ZERO, chunk).unwrap();
804
805        let coverage =
806            provider.commit_account_range(write, &range, Default::default(), bytecodes).unwrap();
807
808        assert!(coverage.is_complete());
809        assert_eq!(stored(&provider), (vec![key(1), key(2), FAR], true, true));
810        assert!(provider.tx_ref().get::<tables::HashedStorages>(key(1)).unwrap().is_none());
811    }
812
813    #[test]
814    fn persisted_storage_must_be_complete_and_match_the_account_root() {
815        let accounts = accounts();
816        let (factory, write, _) = started(&accounts);
817        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
818        let (_, bytecodes) = dependencies();
819
820        // Part way through the contract.
821        let provider = factory.database_provider_rw().unwrap();
822        let slots = vec![(SLOT, U256::from(7))];
823        let next = Some(B256::repeat_byte(0x66));
824        let partial = StorageChunk::new(key(2), storage().0, B256::ZERO, slots, next);
825        provider.commit_storage_chunk(write, B256::ZERO, partial).unwrap();
826        let incomplete =
827            provider.commit_account_range(write, &range, Default::default(), bytecodes.clone());
828        assert!(
829            matches!(incomplete, Err(SnapSyncError::MissingStorage { account }) if account == key(2))
830        );
831        drop(provider);
832
833        // Complete, but not the storage the account commits to.
834        let provider = factory.database_provider_rw().unwrap();
835        let slots = vec![(SLOT, U256::from(8))];
836        let wrong = StorageChunk::new(key(2), storage().0, B256::ZERO, slots, None);
837        provider.commit_storage_chunk(write, B256::ZERO, wrong).unwrap();
838        let mismatched =
839            provider.commit_account_range(write, &range, Default::default(), bytecodes);
840        assert!(matches!(
841            mismatched,
842            Err(SnapSyncError::StorageRootMismatch { account, .. }) if account == key(2)
843        ));
844        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
845    }
846
847    #[test]
848    fn a_record_this_build_cannot_read_is_reported() {
849        let accounts = accounts();
850        let (factory, write, _) = started(&accounts);
851
852        for record in [br#"{"version":999}"#.to_vec(), b"{}".to_vec()] {
853            let provider = factory.database_provider_rw().unwrap();
854            provider.write_metadata(StoredCoverage::KEY, record).unwrap();
855
856            assert!(matches!(
857                provider.account_coverage(write),
858                Err(SnapSyncError::UnsupportedRecord { .. })
859            ));
860        }
861    }
862
863    // `started`, with headers so the pivot is canonical, and stale values for the contract and
864    // for `ABSENT`, both scheduled for repair.
865    fn repairing(
866        accounts: &[(B256, TrieAccount)],
867    ) -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite) {
868        let (factory, write, _) = started(accounts);
869        insert_generation_headers(&factory);
870        let provider = factory.database_provider_rw().unwrap();
871        let stale = HashedPostState::default()
872            .with_accounts([
873                (key(2), Some(Account::from(account(9)))),
874                (ABSENT, Some(Account::from(account(9)))),
875            ])
876            .with_storages([
877                (key(2), HashedStorage::from_iter([(SLOT, U256::from(1)), (OTHER, U256::from(2))])),
878                (ABSENT, HashedStorage::from_iter([(SLOT, U256::from(1))])),
879            ]);
880        provider.write_hashed_state(&stale.into_sorted()).unwrap();
881        let mut repairs = StateRepairs::default();
882        repairs.insert_slot(key(2), SLOT);
883        repairs.insert_slot(key(2), OTHER);
884        repairs.insert_account(ABSENT);
885        provider.schedule_snap_repairs(write, repairs).unwrap();
886        provider.commit().unwrap();
887        (factory, write)
888    }
889
890    fn repair(
891        factory: &ProviderFactory<MockNodeTypesWithDB>,
892        write: SnapWrite,
893        range: &VerifiedAccountRange,
894        slots: Vec<(B256, U256)>,
895    ) -> Result<usize, SnapSyncError> {
896        let provider = factory.database_provider_rw().unwrap();
897        let remaining = provider.commit_account_repair(write, range, slots)?;
898        provider.commit().unwrap();
899        Ok(remaining)
900    }
901
902    #[test]
903    #[allow(clippy::clone_on_copy)]
904    fn a_repair_takes_the_pivot_account_and_scheduled_slots() {
905        let accounts = accounts();
906        let (factory, write) = repairing(&accounts);
907        let range = verified_repair(&accounts, 1..2, key(2), &[key(2)]);
908        // The pivot holds 7 at `SLOT` and nothing at `OTHER`.
909        let slots = vec![(SLOT, U256::from(7)), (OTHER, U256::ZERO)];
910
911        let refused = repair(&factory, write, &range, slots.clone());
912        assert!(matches!(refused, Err(SnapSyncError::MissingCode { .. })));
913        let provider = factory.database_provider_rw().unwrap();
914        provider
915            .write_state_changes(StateChangeset {
916                contracts: vec![(code().hash_slow(), code())],
917                ..Default::default()
918            })
919            .unwrap();
920        provider.commit().unwrap();
921        let remaining = repair(&factory, write, &range, slots).unwrap();
922
923        assert_eq!(remaining, 1);
924        let provider = factory.database_provider_ro().unwrap();
925        let mut expected = StateRepairs::default();
926        expected.insert_account(ABSENT);
927        assert_eq!(provider.snap_repairs(write).unwrap(), expected);
928        let stored = provider.tx_ref().get::<tables::HashedAccounts>(key(2)).unwrap();
929        assert_eq!(stored, Some(Account::from(accounts[1].1.clone())));
930        assert_eq!(stored_slots(&provider, key(2)), [(SLOT, U256::from(7))]);
931    }
932
933    #[test]
934    fn an_account_the_pivot_lacks_is_removed_with_its_storage() {
935        let accounts = accounts();
936        let (factory, write) = repairing(&accounts);
937        // The far account, the first past `ABSENT`, proves it absent.
938        let range = verified_repair(&accounts, 2..3, ABSENT, &[ABSENT, FAR]);
939
940        repair(&factory, write, &range, Vec::new()).unwrap();
941
942        let provider = factory.database_provider_ro().unwrap();
943        assert_eq!(provider.tx_ref().get::<tables::HashedAccounts>(ABSENT).unwrap(), None);
944        assert!(stored_slots(&provider, ABSENT).is_empty());
945        assert_eq!(provider.snap_repairs(write).unwrap().len(), 1);
946    }
947
948    #[test]
949    fn a_repair_waits_for_catch_up_to_reach_the_pivot() {
950        let accounts = accounts();
951        let (factory, write) = repairing(&accounts);
952        let provider = factory.database_provider_rw().unwrap();
953        let write =
954            provider.advance_snap_pivot(write, generation(2, state_root(&accounts))).unwrap();
955        provider.commit().unwrap();
956        let range = verified_repair(&accounts, 2..3, ABSENT, &[ABSENT, FAR]);
957
958        let refused = repair(&factory, write, &range, Vec::new());
959
960        assert!(matches!(refused, Err(SnapSyncError::CatchUpBehindPivot { applied: 1, pivot: 2 })));
961    }
962
963    #[test]
964    fn a_new_attempt_starts_without_repairs() {
965        let accounts = accounts();
966        let (factory, _) = repairing(&accounts);
967        let provider = factory.database_provider_rw().unwrap();
968
969        let restarted = provider.start_snap_attempt(generation(1, state_root(&accounts))).unwrap();
970
971        assert!(provider.snap_repairs(restarted).unwrap().is_empty());
972        assert!(StoredRepairs::read(&provider).unwrap().is_none());
973    }
974}
reth_snap_sync/account/store.rs

reth_snap_sync/account/
store.rs
