reth_snap_sync/account/

store.rs

//! Persists account ranges, their storage and code, and the coverage cursor in one transaction.
//!
//! Each range replaces its key interval, and ranges commit in key order. Accounts scheduled for
//! repair are fetched again on their own and replace what the state holds for them.

use crate::{
    common::SnapRecord, repair::StoredRepairs, storage::persisted_storage_root, SnapAttemptStore,
    SnapCatchUpStore, SnapStorageStore, SnapSyncError, SnapWrite, StateRepairs, StorageProgress,
};
use alloy_primitives::{
    map::{B256Map, B256Set},
    B256, KECCAK256_EMPTY, U256,
};
use reth_db_api::{
    tables,
    transaction::{DbTx, DbTxMut},
    RawKey, RawTable,
};
use reth_downloaders::snap::VerifiedAccountRange;
use reth_primitives_traits::Account;
use reth_storage_api::{
    BlockHashReader, DBProvider, MetadataProvider, MetadataWriter, SnapAttemptId, StateWriter,
};
use reth_trie_common::{
    root::storage_root, HashedPostState, HashedPostStateSorted, HashedStorage, TrieAccount,
    EMPTY_ROOT_HASH,
};
use revm::{bytecode::Bytecode, database::states::StateChangeset};
use serde::{Deserialize, Serialize};
use std::ops::Bound;

/// Persistence for the account ranges an attempt downloads.
///
/// Blanket-implemented over the node's writers, so accounts, their dependencies and the coverage
/// join the caller's transaction and commit together or not at all.
pub trait SnapAccountStore {
    /// Returns the coverage recorded for the attempt `write` belongs to, recording that no
    /// account has been downloaded yet when there is none.
    fn start_account_coverage(&self, write: SnapWrite) -> Result<AccountCoverage, SnapSyncError>
    where
        Self: MetadataWriter;

    /// Returns the coverage recorded for the attempt `write` belongs to, if any.
    fn account_coverage(&self, write: SnapWrite) -> Result<Option<AccountCoverage>, SnapSyncError>;

    /// Persists `range` with its storage and code, replacing its key interval.
    ///
    /// Storage must match each account's root, supplied or persisted ahead of the range by
    /// [`SnapStorageStore`], and code must be supplied or already stored. Persisted storage only
    /// counts once catch-up has carried it to the pivot.
    fn commit_account_range(
        &self,
        write: SnapWrite,
        range: &VerifiedAccountRange,
        storages: B256Map<HashedStorage>,
        bytecodes: Vec<(B256, Bytecode)>,
    ) -> Result<AccountCoverage, SnapSyncError>
    where
        Self: MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>;

    /// Removes storage in `interval` except the contracts in `kept`, which are sorted and inside
    /// it.
    fn remove_storages_except(
        &self,
        interval: (Bound<B256>, Bound<B256>),
        kept: &[B256],
    ) -> Result<(), SnapSyncError>
    where
        Self: DBProvider<Tx: DbTxMut>;

    /// Returns the repairs scheduled for the attempt `write` belongs to.
    fn snap_repairs(&self, write: SnapWrite) -> Result<StateRepairs, SnapSyncError>;

    /// Adds `repairs` to those scheduled for the attempt `write` belongs to.
    fn schedule_snap_repairs(
        &self,
        write: SnapWrite,
        repairs: StateRepairs,
    ) -> Result<(), SnapSyncError>
    where
        Self: MetadataWriter;

    /// Writes the pivot's account at `range`'s origin and its `slots`, returning how many accounts
    /// remain scheduled. Refused until catch-up reaches the pivot.
    fn commit_account_repair(
        &self,
        write: SnapWrite,
        range: &VerifiedAccountRange,
        slots: Vec<(B256, U256)>,
    ) -> Result<usize, SnapSyncError>
    where
        Self: BlockHashReader + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>;
}

/// How far the account key space has been downloaded.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct AccountCoverage {
    // Next key to request, or none once the trie is exhausted.
    next: Option<B256>,
}

impl AccountCoverage {
    /// Nothing downloaded yet.
    pub const START: Self = Self { next: Some(B256::ZERO) };

    /// Every account downloaded.
    pub const COMPLETE: Self = Self { next: None };

    /// Key the next range is requested from, or `None` once every account is downloaded.
    pub const fn next(&self) -> Option<B256> {
        self.next
    }

    /// Returns whether every account is downloaded.
    pub const fn is_complete(&self) -> bool {
        self.next.is_none()
    }

    // The coverage after `range`, which must start at this cursor and end past it.
    fn advance(&self, range: &VerifiedAccountRange) -> Result<Self, SnapSyncError> {
        let origin = range.origin();
        if self.next != Some(origin) {
            return Err(SnapSyncError::OutOfOrderRange { expected: self.next, got: origin })
        }
        if range.next().is_some_and(|next| next <= origin) {
            return Err(SnapSyncError::NoProgress { origin })
        }
        Ok(Self { next: range.next() })
    }

    // Keys from this cursor up to `to`, which a range proved hold only its own accounts.
    fn interval_to(&self, to: &Self) -> (Bound<B256>, Bound<B256>) {
        (
            self.next.map_or(Bound::Unbounded, Bound::Included),
            to.next.map_or(Bound::Unbounded, Bound::Excluded),
        )
    }
}

// The coverage record as persisted, tied to the attempt that recorded it.
#[derive(Serialize, Deserialize)]
pub(crate) struct StoredCoverage {
    // Encoding version, checked before the rest is decoded.
    version: u32,
    // Attempt the coverage belongs to.
    attempt: SnapAttemptId,
    // How far that attempt has downloaded.
    coverage: AccountCoverage,
}

impl SnapRecord for StoredCoverage {
    const KEY: &'static str = "snap_account_coverage";
    const VERSION: u32 = 1;
}

impl StoredCoverage {
    // `coverage` for `attempt` at this build's version.
    const fn new(attempt: SnapAttemptId, coverage: AccountCoverage) -> Self {
        Self { version: Self::VERSION, attempt, coverage }
    }
}

impl<T: MetadataProvider> SnapAccountStore for T {
    // Resumes the recorded coverage, so resuming an attempt keeps its progress.
    fn start_account_coverage(&self, write: SnapWrite) -> Result<AccountCoverage, SnapSyncError>
    where
        Self: MetadataWriter,
    {
        if let Some(coverage) = self.account_coverage(write)? {
            return Ok(coverage)
        }
        let start = AccountCoverage::START;
        StoredCoverage::new(write.attempt(), start).write(self)?;
        Ok(start)
    }

    // A record left by another attempt reads as no coverage.
    fn account_coverage(&self, write: SnapWrite) -> Result<Option<AccountCoverage>, SnapSyncError> {
        self.authorize_snap_write(write)?;
        let Some(stored) = StoredCoverage::read(self)? else { return Ok(None) };
        Ok((stored.attempt == write.attempt()).then_some(stored.coverage))
    }

    // Every check runs before the first write, so a refused range changes nothing.
    fn commit_account_range(
        &self,
        write: SnapWrite,
        range: &VerifiedAccountRange,
        storages: B256Map<HashedStorage>,
        bytecodes: Vec<(B256, Bytecode)>,
    ) -> Result<AccountCoverage, SnapSyncError>
    where
        Self: MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
    {
        let attempt = self.authorize_snap_write(write)?;
        if range.state_root() != attempt.state_root() {
            return Err(SnapSyncError::RootMismatch {
                expected: attempt.state_root(),
                got: range.state_root(),
            })
        }
        let coverage = self.account_coverage(write)?.ok_or(SnapSyncError::NoCoverage)?;
        let advanced = coverage.advance(range)?;
        let progress = self.storage_progress(write, range.origin())?;
        // Slots persisted before the pivot moved hold its old values until the lists reach it.
        if progress != StorageProgress::START {
            let applied = self
                .catch_up_progress(write)?
                .ok_or(SnapSyncError::NoCatchUpProgress)?
                .applied()
                .number;
            if applied < attempt.pivot().number {
                return Err(SnapSyncError::CatchUpBehindPivot {
                    applied,
                    pivot: attempt.pivot().number,
                })
            }
        }
        let dependencies = RangeDependencies::new(range.accounts(), storages, bytecodes);
        let persisted = dependencies.verify(range.accounts(), &progress, self.tx_ref())?;

        let interval = coverage.interval_to(&advanced);
        self.remove::<tables::HashedAccounts>(interval)?;
        self.remove_storages_except(interval, &persisted)?;
        dependencies.write(self)?;
        StoredCoverage::new(write.attempt(), advanced).write(self)?;
        // A page ending before contracts with persisted storage leaves them to the next range.
        if let Some(next) = advanced.next() {
            progress.carry_to(self, write, next)?;
        }
        Ok(advanced)
    }

    fn remove_storages_except(
        &self,
        (mut start, end): (Bound<B256>, Bound<B256>),
        kept: &[B256],
    ) -> Result<(), SnapSyncError>
    where
        Self: DBProvider<Tx: DbTxMut>,
    {
        for account in kept {
            self.remove::<tables::HashedStorages>((start, Bound::Excluded(*account)))?;
            // Walks cannot start at an excluded key, so the next piece starts one key past it.
            let Some(after) = U256::from_be_bytes(account.0).checked_add(U256::ONE) else {
                return Ok(())
            };
            start = Bound::Included(after.into());
        }
        self.remove::<tables::HashedStorages>((start, end))?;
        Ok(())
    }

    // A record left by another attempt reads as nothing scheduled.
    fn snap_repairs(&self, write: SnapWrite) -> Result<StateRepairs, SnapSyncError> {
        self.authorize_snap_write(write)?;
        let Some(stored) = StoredRepairs::read(self)? else { return Ok(StateRepairs::default()) };
        Ok(if stored.attempt == write.attempt() { stored.repairs } else { StateRepairs::default() })
    }

    fn schedule_snap_repairs(
        &self,
        write: SnapWrite,
        repairs: StateRepairs,
    ) -> Result<(), SnapSyncError>
    where
        Self: MetadataWriter,
    {
        let mut scheduled = self.snap_repairs(write)?;
        scheduled.extend(repairs);
        StoredRepairs::store(self, write.attempt(), scheduled)
    }

    // Every check runs before the first write, so a refused repair changes nothing.
    #[allow(clippy::clone_on_copy)]
    fn commit_account_repair(
        &self,
        write: SnapWrite,
        range: &VerifiedAccountRange,
        slots: Vec<(B256, U256)>,
    ) -> Result<usize, SnapSyncError>
    where
        Self: BlockHashReader + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
    {
        // The repaired values belong to the pivot, so it must still be canonical.
        let attempt = self.authorize_canonical_snap_write(write)?;
        if range.state_root() != attempt.state_root() {
            return Err(SnapSyncError::RootMismatch {
                expected: attempt.state_root(),
                got: range.state_root(),
            })
        }
        let applied =
            self.catch_up_progress(write)?.ok_or(SnapSyncError::NoCatchUpProgress)?.applied();
        if applied != attempt.pivot() {
            return Err(SnapSyncError::CatchUpBehindPivot {
                applied: applied.number,
                pivot: attempt.pivot().number,
            })
        }
        let hashed_address = range.origin();
        // The proof starts at the origin, so an account keyed past it means none is there.
        let account = range
            .accounts()
            .first()
            .filter(|(hash, _)| *hash == hashed_address)
            .map(|(_, account)| account);
        if let Some(hash) = account.map(|account| account.code_hash) &&
            hash != KECCAK256_EMPTY &&
            self.tx_ref().get::<RawTable<tables::Bytecodes>>(RawKey::new(hash))?.is_none()
        {
            return Err(SnapSyncError::MissingCode { hash })
        }

        let has_storage = account.is_some_and(|account| account.storage_root != EMPTY_ROOT_HASH);
        let mut state = HashedPostState::default().with_accounts([(
            hashed_address,
            account.map(|account| Account::from(account.clone())),
        )]);
        if has_storage {
            // Zero values remove their slots.
            state = state
                .with_storages([(hashed_address, HashedStorage::from_iter(slots.iter().copied()))]);
        } else {
            self.remove::<tables::HashedStorages>(hashed_address..=hashed_address)?;
        }
        self.write_hashed_state(&state.into_sorted())?;

        let mut remaining = self.snap_repairs(write)?;
        remaining.resolve(hashed_address, has_storage.then_some(slots.as_slice()));
        let accounts = remaining.len();
        StoredRepairs::store(self, write.attempt(), remaining)?;
        Ok(accounts)
    }
}

// Accounts of a range with the storage and code supplied for them.
struct RangeDependencies {
    // Accounts and storage, sorted once for both the root checks and the write.
    state: HashedPostStateSorted,
    // Code keyed by the hash it was supplied under.
    bytecodes: Vec<(B256, Bytecode)>,
}

impl RangeDependencies {
    #[allow(clippy::clone_on_copy)]
    fn new(
        accounts: &[(B256, TrieAccount)],
        storages: B256Map<HashedStorage>,
        bytecodes: Vec<(B256, Bytecode)>,
    ) -> Self {
        let state = HashedPostState::default()
            .with_accounts(
                accounts
                    .iter()
                    .map(|(hash, account)| (*hash, Some(Account::from(account.clone())))),
            )
            .with_storages(storages)
            .into_sorted();
        Self { state, bytecodes }
    }

    // Checks storage against each account's root, and code against its hash or the table.
    // Returns the contracts whose storage was persisted ahead of the range, in key order.
    fn verify(
        &self,
        accounts: &[(B256, TrieAccount)],
        progress: &StorageProgress,
        tx: &impl DbTx,
    ) -> Result<Vec<B256>, SnapSyncError> {
        let mut available = self.supplied_code()?;
        let mut contracts = B256Set::default();
        let mut persisted = Vec::new();
        for (hash, account) in accounts {
            if account.storage_root != EMPTY_ROOT_HASH {
                contracts.insert(*hash);
                if self.verify_storage(*hash, account, progress, tx)? {
                    persisted.push(*hash);
                }
            }
            // Only presence matters, so stored code is not decoded.
            if account.code_hash != KECCAK256_EMPTY &&
                available.insert(account.code_hash) &&
                tx.get::<RawTable<tables::Bytecodes>>(RawKey::new(account.code_hash))?.is_none()
            {
                return Err(SnapSyncError::MissingCode { hash: account.code_hash })
            }
        }
        if let Some(account) =
            self.state.account_storages().keys().find(|hash| !contracts.contains(*hash))
        {
            return Err(SnapSyncError::UnexpectedStorage { account: *account })
        }
        Ok(persisted)
    }

    // Hashes of the supplied code, refusing code filed under a hash it does not hash to.
    fn supplied_code(&self) -> Result<B256Set, SnapSyncError> {
        let mut hashes = B256Set::default();
        for (hash, code) in &self.bytecodes {
            let got = code.hash_slow();
            if got != *hash {
                return Err(SnapSyncError::CodeMismatch { expected: *hash, got })
            }
            hashes.insert(*hash);
        }
        Ok(hashes)
    }

    // Checks the contract's storage, supplied or persisted, against its root. Returns whether it
    // was persisted.
    fn verify_storage(
        &self,
        hash: B256,
        account: &TrieAccount,
        progress: &StorageProgress,
        tx: &impl DbTx,
    ) -> Result<bool, SnapSyncError> {
        let (got, persisted) = match self.state.account_storages().get(&hash) {
            // Zero slots are deletions, which the trie does not hold.
            Some(storage) => (
                storage_root(
                    storage
                        .storage_slots_ref()
                        .iter()
                        .filter(|(_, value)| !value.is_zero())
                        .copied(),
                ),
                false,
            ),
            None if progress.is_complete(hash) => (persisted_storage_root(tx, hash)?, true),
            None => return Err(SnapSyncError::MissingStorage { account: hash }),
        };
        if got != account.storage_root {
            return Err(SnapSyncError::StorageRootMismatch {
                account: hash,
                expected: account.storage_root,
                got,
            })
        }
        Ok(persisted)
    }

    // Writes the accounts with their storage and code.
    fn write(self, writer: &impl StateWriter) -> Result<(), SnapSyncError> {
        writer.write_hashed_state(&self.state)?;
        writer.write_state_changes(StateChangeset {
            contracts: self.bytecodes,
            ..Default::default()
        })?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        test_utils::{
            account, generation, hashed_factory_with_roots, key, state_root, stored_slots,
            verified_range, verified_repair,
        },
        SnapGeneration, StorageChunk,
    };
    use alloy_primitives::{Bytes, U256};
    use reth_db_api::cursor::DbCursorRO;
    use reth_provider::{
        test_utils::MockNodeTypesWithDB, DatabaseProviderFactory, ProviderFactory,
    };
    use reth_trie_common::root::storage_root_unsorted;

    const FAR: B256 = B256::repeat_byte(0xaa);
    const SLOT: B256 = B256::repeat_byte(0x55);
    // A slot the stale state holds and the pivot does not.
    const OTHER: B256 = B256::repeat_byte(0x66);
    // Sorts between the contract and the far account, holding nothing at the pivot.
    const ABSENT: B256 = B256::repeat_byte(0x33);

    fn code() -> Bytecode {
        Bytecode::new_raw(Bytes::from_static(&[0x60, 0x00]))
    }

    // Storage holding one slot, with the root the contract fixture commits to.
    fn storage() -> (B256, HashedStorage) {
        let storage = HashedStorage::from_iter([(SLOT, U256::from(7))]);
        (
            storage_root_unsorted(storage.storage.iter().map(|(slot, value)| (*slot, *value))),
            storage,
        )
    }

    // A plain account, a contract with storage and code, and a far account.
    fn accounts() -> Vec<(B256, TrieAccount)> {
        let mut contract = account(2);
        contract.storage_root = storage().0;
        contract.code_hash = code().hash_slow();
        vec![(key(1), account(1)), (key(2), contract), (FAR, account(3))]
    }

    // Storage and code that satisfy the contract in `accounts`.
    fn dependencies() -> (B256Map<HashedStorage>, Vec<(B256, Bytecode)>) {
        (B256Map::from_iter([(key(2), storage().1)]), vec![(code().hash_slow(), code())])
    }

    // An attempt anchored to the trie's root, with its coverage recorded.
    fn started(
        accounts: &[(B256, TrieAccount)],
    ) -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite, SnapGeneration) {
        started_moving_to(accounts, accounts)
    }

    // `started`, with the later blocks committing to the state `moved`.
    fn started_moving_to(
        accounts: &[(B256, TrieAccount)],
        moved: &[(B256, TrieAccount)],
    ) -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite, SnapGeneration) {
        let factory = hashed_factory_with_roots(&[
            state_root(accounts),
            state_root(moved),
            state_root(moved),
        ]);
        let generation = generation(1, state_root(accounts));
        let provider = factory.database_provider_rw().unwrap();
        let write = provider.start_snap_attempt(generation).unwrap();
        provider.start_account_coverage(write).unwrap();
        provider.commit().unwrap();
        (factory, write, generation)
    }

    fn stored_accounts(provider: &impl DBProvider) -> Vec<B256> {
        let mut cursor = provider.tx_ref().cursor_read::<tables::HashedAccounts>().unwrap();
        cursor.walk(None).unwrap().map(|entry| entry.unwrap().0).collect()
    }

    fn stored(provider: &impl DBProvider) -> (Vec<B256>, bool, bool) {
        let tx = provider.tx_ref();
        (
            stored_accounts(provider),
            tx.get::<tables::HashedStorages>(key(2)).unwrap().is_some(),
            tx.get::<tables::Bytecodes>(code().hash_slow()).unwrap().is_some(),
        )
    }

    #[test]
    fn accounts_dependencies_and_coverage_commit_together() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (storages, bytecodes) = dependencies();

        let provider = factory.database_provider_rw().unwrap();
        let coverage = provider.commit_account_range(write, &range, storages, bytecodes).unwrap();
        provider.commit().unwrap();

        assert!(coverage.is_complete());
        let provider = factory.database_provider_rw().unwrap();
        assert_eq!(stored(&provider), (vec![key(1), key(2), FAR], true, true));
        assert_eq!(provider.account_coverage(write).unwrap(), Some(coverage));
    }

    #[test]
    fn a_partial_range_moves_the_coverage_to_its_next_key() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        // Only the first account, with a proof placing the next one at key 2.
        let range = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);

        let provider = factory.database_provider_rw().unwrap();
        let coverage =
            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();

        assert_eq!(coverage.next(), Some(key(2)));
        assert_eq!(stored_accounts(&provider), [key(1)]);
    }

    #[test]
    fn a_proven_empty_tail_completes_the_coverage() {
        let accounts = vec![(key(1), account(1)), (key(2), account(2))];
        let (factory, write, _) = started(&accounts);
        let tail = verified_range(&accounts, 0..0, key(3), &[key(3)]);
        assert!(tail.accounts().is_empty());
        let provider = factory.database_provider_rw().unwrap();
        let coverage = AccountCoverage { next: Some(key(3)) };
        StoredCoverage::new(write.attempt(), coverage).write(&provider).unwrap();

        let coverage =
            provider.commit_account_range(write, &tail, Default::default(), Vec::new()).unwrap();

        assert!(coverage.is_complete());
        assert!(stored_accounts(&provider).is_empty());
    }

    #[test]
    fn a_range_from_another_origin_changes_nothing() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let provider = factory.database_provider_rw().unwrap();
        let first = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
        provider.commit_account_range(write, &first, Default::default(), Vec::new()).unwrap();

        // The same range again, and one skipping ahead of the cursor.
        let duplicate =
            provider.commit_account_range(write, &first, Default::default(), Vec::new());
        let skipped = verified_range(&accounts, 2..3, FAR, &[FAR]);
        let ahead = provider.commit_account_range(write, &skipped, Default::default(), Vec::new());

        assert!(matches!(duplicate, Err(SnapSyncError::OutOfOrderRange { .. })));
        assert!(matches!(ahead, Err(SnapSyncError::OutOfOrderRange { .. })));
        assert_eq!(provider.account_coverage(write).unwrap().unwrap().next(), Some(key(2)));
        assert_eq!(stored_accounts(&provider), [key(1)]);
    }

    #[test]
    fn a_contract_without_its_storage_or_code_is_refused() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (storages, bytecodes) = dependencies();
        let provider = factory.database_provider_rw().unwrap();

        let no_storage =
            provider.commit_account_range(write, &range, Default::default(), bytecodes.clone());
        assert!(
            matches!(no_storage, Err(SnapSyncError::MissingStorage { account }) if account == key(2))
        );

        let no_code = provider.commit_account_range(write, &range, storages, Vec::new());
        assert!(matches!(no_code, Err(SnapSyncError::MissingCode { .. })));

        let stray = B256Map::from_iter([(key(1), storage().1), (key(2), storage().1)]);
        let unexpected = provider.commit_account_range(write, &range, stray, bytecodes);
        assert!(
            matches!(unexpected, Err(SnapSyncError::UnexpectedStorage { account }) if account == key(1))
        );

        assert_eq!(stored(&provider), (Vec::new(), false, false));
        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
    }

    #[test]
    fn missing_code_is_named_after_contracts_whose_code_is_stored() {
        let stored_code = code();
        let missing_code = Bytecode::new_raw(Bytes::from_static(&[0x60, 0x01]));
        let mut first = account(1);
        first.code_hash = stored_code.hash_slow();
        let mut second = account(2);
        second.code_hash = missing_code.hash_slow();
        let accounts = vec![(key(1), first), (key(2), second), (key(3), account(3))];
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let provider = factory.database_provider_rw().unwrap();
        provider
            .write_state_changes(StateChangeset {
                contracts: vec![(stored_code.hash_slow(), stored_code)],
                ..Default::default()
            })
            .unwrap();

        let refused = provider.commit_account_range(write, &range, Default::default(), Vec::new());

        assert!(matches!(
            refused,
            Err(SnapSyncError::MissingCode { hash }) if hash == missing_code.hash_slow()
        ));
        assert_eq!(stored_accounts(&provider), Vec::<B256>::new());
        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
    }

    #[test]
    fn storage_and_code_are_checked_against_what_the_account_commits_to() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (storages, bytecodes) = dependencies();
        let provider = factory.database_provider_rw().unwrap();

        // A slot short of the storage root, as a partial download would be.
        let partial = B256Map::from_iter([(key(2), HashedStorage::default())]);
        let short = provider.commit_account_range(write, &range, partial, bytecodes);
        assert!(matches!(
            short,
            Err(SnapSyncError::StorageRootMismatch { account, expected, .. })
                if account == key(2) && expected == storage().0
        ));

        // Code filed under a hash it does not hash to.
        let other = Bytecode::new_raw(Bytes::from_static(&[0x60, 0x01]));
        let relabelled = vec![(code().hash_slow(), other)];
        let wrong = provider.commit_account_range(write, &range, storages, relabelled);
        assert!(matches!(
            wrong,
            Err(SnapSyncError::CodeMismatch { expected, .. }) if expected == code().hash_slow()
        ));

        assert_eq!(stored(&provider), (Vec::new(), false, false));
        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
    }

    #[test]
    fn code_already_stored_need_not_be_supplied() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (storages, _) = dependencies();
        let provider = factory.database_provider_rw().unwrap();
        provider
            .write_state_changes(StateChangeset {
                contracts: vec![(code().hash_slow(), code())],
                ..Default::default()
            })
            .unwrap();

        let coverage = provider.commit_account_range(write, &range, storages, Vec::new()).unwrap();

        assert!(coverage.is_complete());
    }

    #[test]
    fn starting_coverage_again_resumes_where_the_attempt_left_off() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
        let provider = factory.database_provider_rw().unwrap();
        let committed =
            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();

        assert_eq!(provider.start_account_coverage(write).unwrap(), committed);
        assert_eq!(provider.account_coverage(write).unwrap(), Some(committed));
    }

    #[test]
    fn a_write_from_a_replaced_attempt_changes_nothing() {
        let accounts = accounts();
        let (factory, replaced, generation) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (storages, bytecodes) = dependencies();
        let provider = factory.database_provider_rw().unwrap();
        let current = provider.start_snap_attempt(generation).unwrap();

        let refused = provider.commit_account_range(replaced, &range, storages, bytecodes);

        assert!(matches!(refused, Err(SnapSyncError::StaleWrite { .. })));
        assert_eq!(stored(&provider), (Vec::new(), false, false));
        // The replaced attempt's coverage is not the new attempt's either.
        assert_eq!(provider.account_coverage(current).unwrap(), None);
        assert!(matches!(
            provider.account_coverage(replaced),
            Err(SnapSyncError::StaleWrite { .. })
        ));
    }

    #[test]
    fn a_new_attempt_replaces_what_an_earlier_one_left_in_the_interval() {
        let earlier = accounts();
        let current = vec![(key(1), account(9)), (FAR, account(3))];
        let (factory, write, _) = started_moving_to(&earlier, &current);
        let (storages, bytecodes) = dependencies();
        let provider = factory.database_provider_rw().unwrap();
        let range = verified_range(&earlier, 0..3, B256::ZERO, &[]);
        provider.commit_account_range(write, &range, storages, bytecodes).unwrap();
        provider.commit().unwrap();

        // At the new root the contract is gone and the first account changed.
        let provider = factory.database_provider_rw().unwrap();
        let write = provider.start_snap_attempt(generation(2, state_root(&current))).unwrap();
        provider.start_account_coverage(write).unwrap();
        let range = verified_range(&current, 0..2, B256::ZERO, &[]);
        let coverage =
            provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();

        assert!(coverage.is_complete());
        assert_eq!(stored(&provider), (vec![key(1), FAR], false, true));
        let first = provider.tx_ref().get::<tables::HashedAccounts>(key(1)).unwrap().unwrap();
        assert_eq!(first.nonce, 9);
    }

    #[test]
    fn a_range_proved_against_another_root_is_refused() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let provider = factory.database_provider_rw().unwrap();
        let other = vec![(key(7), account(7))];
        let range = verified_range(&other, 0..1, B256::ZERO, &[]);

        let refused = provider.commit_account_range(write, &range, Default::default(), Vec::new());

        assert!(matches!(refused, Err(SnapSyncError::RootMismatch { .. })));
        assert!(stored_accounts(&provider).is_empty());
    }

    #[test]
    fn a_range_without_recorded_coverage_is_refused() {
        let accounts = accounts();
        let factory = hashed_factory_with_roots(&[state_root(&accounts)]);
        let provider = factory.database_provider_rw().unwrap();
        let write = provider.start_snap_attempt(generation(1, state_root(&accounts))).unwrap();
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);

        assert!(matches!(
            provider.commit_account_range(write, &range, Default::default(), Vec::new()),
            Err(SnapSyncError::NoCoverage)
        ));
    }

    #[test]
    fn storage_persisted_ahead_of_the_range_commits_with_it() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (_, bytecodes) = dependencies();
        let provider = factory.database_provider_rw().unwrap();
        // Left behind for an account that has no storage at this root.
        let leftover = HashedPostState::default().with_storages([(key(1), storage().1)]);
        provider.write_hashed_state(&leftover.into_sorted()).unwrap();
        let slots = vec![(SLOT, U256::from(7))];
        let chunk = StorageChunk::new(key(2), storage().0, B256::ZERO, slots, None);
        provider.commit_storage_chunk(write, B256::ZERO, chunk).unwrap();

        let coverage =
            provider.commit_account_range(write, &range, Default::default(), bytecodes).unwrap();

        assert!(coverage.is_complete());
        assert_eq!(stored(&provider), (vec![key(1), key(2), FAR], true, true));
        assert!(provider.tx_ref().get::<tables::HashedStorages>(key(1)).unwrap().is_none());
    }

    #[test]
    fn persisted_storage_must_be_complete_and_match_the_account_root() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);
        let range = verified_range(&accounts, 0..3, B256::ZERO, &[]);
        let (_, bytecodes) = dependencies();

        // Part way through the contract.
        let provider = factory.database_provider_rw().unwrap();
        let slots = vec![(SLOT, U256::from(7))];
        let next = Some(B256::repeat_byte(0x66));
        let partial = StorageChunk::new(key(2), storage().0, B256::ZERO, slots, next);
        provider.commit_storage_chunk(write, B256::ZERO, partial).unwrap();
        let incomplete =
            provider.commit_account_range(write, &range, Default::default(), bytecodes.clone());
        assert!(
            matches!(incomplete, Err(SnapSyncError::MissingStorage { account }) if account == key(2))
        );
        drop(provider);

        // Complete, but not the storage the account commits to.
        let provider = factory.database_provider_rw().unwrap();
        let slots = vec![(SLOT, U256::from(8))];
        let wrong = StorageChunk::new(key(2), storage().0, B256::ZERO, slots, None);
        provider.commit_storage_chunk(write, B256::ZERO, wrong).unwrap();
        let mismatched =
            provider.commit_account_range(write, &range, Default::default(), bytecodes);
        assert!(matches!(
            mismatched,
            Err(SnapSyncError::StorageRootMismatch { account, .. }) if account == key(2)
        ));
        assert_eq!(provider.account_coverage(write).unwrap(), Some(AccountCoverage::START));
    }

    #[test]
    fn a_record_this_build_cannot_read_is_reported() {
        let accounts = accounts();
        let (factory, write, _) = started(&accounts);

        for record in [br#"{"version":999}"#.to_vec(), b"{}".to_vec()] {
            let provider = factory.database_provider_rw().unwrap();
            provider.write_metadata(StoredCoverage::KEY, record).unwrap();

            assert!(matches!(
                provider.account_coverage(write),
                Err(SnapSyncError::UnsupportedRecord { .. })
            ));
        }
    }

    // `started`, with headers so the pivot is canonical, and stale values for the contract and
    // for `ABSENT`, both scheduled for repair.
    fn repairing(
        accounts: &[(B256, TrieAccount)],
    ) -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite) {
        let (factory, write, _) = started(accounts);
        let provider = factory.database_provider_rw().unwrap();
        let stale = HashedPostState::default()
            .with_accounts([
                (key(2), Some(Account::from(account(9)))),
                (ABSENT, Some(Account::from(account(9)))),
            ])
            .with_storages([
                (key(2), HashedStorage::from_iter([(SLOT, U256::ONE), (OTHER, U256::from(2))])),
                (ABSENT, HashedStorage::from_iter([(SLOT, U256::ONE)])),
            ]);
        provider.write_hashed_state(&stale.into_sorted()).unwrap();
        let mut repairs = StateRepairs::default();
        repairs.insert_slot(key(2), SLOT);
        repairs.insert_slot(key(2), OTHER);
        repairs.insert_account(ABSENT);
        provider.schedule_snap_repairs(write, repairs).unwrap();
        provider.commit().unwrap();
        (factory, write)
    }

    fn repair(
        factory: &ProviderFactory<MockNodeTypesWithDB>,
        write: SnapWrite,
        range: &VerifiedAccountRange,
        slots: Vec<(B256, U256)>,
    ) -> Result<usize, SnapSyncError> {
        let provider = factory.database_provider_rw().unwrap();
        let remaining = provider.commit_account_repair(write, range, slots)?;
        provider.commit().unwrap();
        Ok(remaining)
    }

    #[test]
    #[allow(clippy::clone_on_copy)]
    fn a_repair_takes_the_pivot_account_and_scheduled_slots() {
        let accounts = accounts();
        let (factory, write) = repairing(&accounts);
        let range = verified_repair(&accounts, 1..2, key(2), &[key(2)]);
        // The pivot holds 7 at `SLOT` and nothing at `OTHER`.
        let slots = vec![(SLOT, U256::from(7)), (OTHER, U256::ZERO)];

        let refused = repair(&factory, write, &range, slots.clone());
        assert!(matches!(refused, Err(SnapSyncError::MissingCode { .. })));
        let provider = factory.database_provider_rw().unwrap();
        provider
            .write_state_changes(StateChangeset {
                contracts: vec![(code().hash_slow(), code())],
                ..Default::default()
            })
            .unwrap();
        provider.commit().unwrap();
        let remaining = repair(&factory, write, &range, slots).unwrap();

        assert_eq!(remaining, 1);
        let provider = factory.database_provider_ro().unwrap();
        let mut expected = StateRepairs::default();
        expected.insert_account(ABSENT);
        assert_eq!(provider.snap_repairs(write).unwrap(), expected);
        let stored = provider.tx_ref().get::<tables::HashedAccounts>(key(2)).unwrap();
        assert_eq!(stored, Some(Account::from(accounts[1].1.clone())));
        assert_eq!(stored_slots(&provider, key(2)), [(SLOT, U256::from(7))]);
    }

    #[test]
    fn an_account_the_pivot_lacks_is_removed_with_its_storage() {
        let accounts = accounts();
        let (factory, write) = repairing(&accounts);
        // The far account, the first past `ABSENT`, proves it absent.
        let range = verified_repair(&accounts, 2..3, ABSENT, &[ABSENT, FAR]);

        repair(&factory, write, &range, Vec::new()).unwrap();

        let provider = factory.database_provider_ro().unwrap();
        assert_eq!(provider.tx_ref().get::<tables::HashedAccounts>(ABSENT).unwrap(), None);
        assert!(stored_slots(&provider, ABSENT).is_empty());
        assert_eq!(provider.snap_repairs(write).unwrap().len(), 1);
    }

    #[test]
    fn a_repair_waits_for_catch_up_to_reach_the_pivot() {
        let accounts = accounts();
        let (factory, write) = repairing(&accounts);
        let provider = factory.database_provider_rw().unwrap();
        let write =
            provider.advance_snap_pivot(write, generation(2, state_root(&accounts))).unwrap();
        provider.commit().unwrap();
        let range = verified_repair(&accounts, 2..3, ABSENT, &[ABSENT, FAR]);

        let refused = repair(&factory, write, &range, Vec::new());

        assert!(matches!(refused, Err(SnapSyncError::CatchUpBehindPivot { applied: 1, pivot: 2 })));
    }

    #[test]
    fn a_new_attempt_starts_without_repairs() {
        let accounts = accounts();
        let (factory, _) = repairing(&accounts);
        let provider = factory.database_provider_rw().unwrap();

        let restarted = provider.start_snap_attempt(generation(1, state_root(&accounts))).unwrap();

        assert!(provider.snap_repairs(restarted).unwrap().is_empty());
        assert!(StoredRepairs::read(&provider).unwrap().is_none());
    }
}