reth_snap_sync/
attempt.rs

1//! Persistence for the attempt that owns downloaded snap state.
2//!
3//! Snap writes land in the canonical hashed state tables, so nothing in the data says which
4//! attempt produced it. Every write presents a [`SnapWrite`]; ones that no longer match are
5//! refused.
6
7use crate::{
8    account::StoredCoverage, common::SnapRecord, reorg::StoredAncestry, repair::StoredRepairs,
9    storage::StoredProgress, verify::StoredRebuild, CatchUpProgress, DownloadedAccount,
10    SnapAccountStore, SnapCatchUpStore, SnapGeneration, SnapReorg, SnapSyncError, StateRepairs,
11};
12use alloy_eips::{eip7928::bal::DecodedBal, BlockNumHash};
13use alloy_primitives::keccak256;
14use reth_primitives_traits::AlloyBlockHeader;
15use reth_storage_api::{
16    BlockHashReader, DBProvider, HeaderProvider, MetadataProvider, MetadataWriter, SnapAttempt,
17    SnapAttemptId, StorageSettings,
18};
19use serde::{Deserialize, Serialize};
20
21/// Persistence for the attempt that owns downloaded snap state.
22///
23/// Blanket-implemented over node metadata access, so these writes join the caller's transaction:
24/// state, bytecode and the attempt record commit together or not at all.
25pub trait SnapAttemptStore {
26    /// Starts an attempt anchored to `generation`, superseding any already recorded and the
27    /// progress it kept, with its catch-up progress at that pivot.
28    fn start_snap_attempt(&self, generation: SnapGeneration) -> Result<SnapWrite, SnapSyncError>
29    where
30        Self: MetadataWriter + HeaderProvider;
31
32    /// Returns the write an unfinished attempt accepts, if one owns the persisted state.
33    fn active_snap_write(&self) -> Result<Option<SnapWrite>, SnapSyncError>;
34
35    /// Returns the attempt when `write` still owns the persisted state, rejecting it otherwise.
36    fn authorize_snap_write(&self, write: SnapWrite) -> Result<SnapAttempt, SnapSyncError>;
37
38    /// Returns the attempt when `write` owns its state and its pivot is still canonical.
39    ///
40    /// Downloaded ranges are proved against the pivot, so they describe the fork it sits on, not
41    /// the canonical chain the applied blocks were checked against.
42    fn authorize_canonical_snap_write(
43        &self,
44        write: SnapWrite,
45    ) -> Result<SnapAttempt, SnapSyncError>
46    where
47        Self: BlockHashReader;
48
49    /// Re-anchors the attempt to `generation`, refusing writes proved against the previous root.
50    ///
51    /// Both pivots must be canonical and the new one past the old, so it descends from all
52    /// downloaded state. An orphaned pivot requires recovery or a fresh attempt.
53    fn advance_snap_pivot(
54        &self,
55        write: SnapWrite,
56        generation: SnapGeneration,
57    ) -> Result<SnapWrite, SnapSyncError>
58    where
59        Self: MetadataWriter + HeaderProvider + BlockHashReader;
60
61    /// Marks the attempt's downloaded state verified.
62    fn verify_snap_attempt(&self, write: SnapWrite) -> Result<(), SnapSyncError>
63    where
64        Self: MetadataWriter;
65
66    /// Gives up on an unfinished attempt, refusing its outstanding writes.
67    fn abandon_snap_attempt(&self) -> Result<(), SnapSyncError>
68    where
69        Self: MetadataWriter;
70
71    /// Returns where the canonical chain diverges from the branch the pivot of `write`'s attempt
72    /// is on.
73    ///
74    /// `Ok(None)` when the kept blocks do not reach back to where the branches part.
75    fn snap_reorg(&self, write: SnapWrite) -> Result<Option<SnapReorg>, SnapSyncError>
76    where
77        Self: BlockHashReader;
78
79    /// Schedules what `lists`, those of the orphaned blocks, changed in the downloaded state for
80    /// repair, and moves the attempt to `generation` with catch-up back at `ancestor` at most.
81    ///
82    /// Storage persisted ahead of its range may hold the orphaned values, so it is dropped.
83    fn commit_reorg_recovery(
84        &self,
85        write: SnapWrite,
86        ancestor: BlockNumHash,
87        lists: &[DecodedBal],
88        generation: SnapGeneration,
89    ) -> Result<SnapWrite, SnapSyncError>
90    where
91        Self: MetadataWriter + HeaderProvider + BlockHashReader + DBProvider;
92}
93
94/// What a write presents to prove it belongs to the attempt owning the persisted state.
95#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
96pub struct SnapWrite {
97    // Attempt this write belongs to.
98    attempt: SnapAttemptId,
99    // Pivot generation within that attempt.
100    state_version: u64,
101}
102
103impl SnapWrite {
104    // What `attempt` currently accepts.
105    const fn of(attempt: &SnapAttempt) -> Self {
106        Self { attempt: attempt.id(), state_version: attempt.state_version() }
107    }
108
109    /// Attempt this write belongs to.
110    pub const fn attempt(&self) -> SnapAttemptId {
111        self.attempt
112    }
113
114    /// Pivot generation this write was proved against.
115    pub const fn state_version(&self) -> u64 {
116        self.state_version
117    }
118}
119
120impl<T: MetadataProvider> SnapAttemptStore for T {
121    fn start_snap_attempt(&self, generation: SnapGeneration) -> Result<SnapWrite, SnapSyncError>
122    where
123        Self: MetadataWriter + HeaderProvider,
124    {
125        // Absent settings mean the legacy layout.
126        if !self.storage_settings()?.unwrap_or_else(StorageSettings::v1).use_hashed_state() {
127            return Err(SnapSyncError::UnsupportedStorage)
128        }
129        ensure_header_root(self, &generation)?;
130
131        let attempt =
132            SnapAttempt::start(self.snap_attempt()?, generation.target(), generation.state_root());
133        self.write_snap_attempt(&attempt)?;
134        // Ranges committed at this pivot need every list after it, however far the pivot moves.
135        CatchUpProgress::at_pivot(attempt.pivot()).write(self, attempt.id())?;
136        // Earlier attempts' progress never applies to this one, and a record another build wrote
137        // would fail every read of it.
138        StoredCoverage::clear(self)?;
139        StoredProgress::clear(self)?;
140        StoredRepairs::clear(self)?;
141        StoredRebuild::clear(self)?;
142        StoredAncestry::record(self, attempt.id(), attempt.pivot())?;
143        Ok(SnapWrite::of(&attempt))
144    }
145
146    fn active_snap_write(&self) -> Result<Option<SnapWrite>, SnapSyncError> {
147        Ok(self.snap_attempt()?.filter(SnapAttempt::is_unfinished).as_ref().map(SnapWrite::of))
148    }
149
150    fn authorize_snap_write(&self, write: SnapWrite) -> Result<SnapAttempt, SnapSyncError> {
151        let attempt = self.snap_attempt()?.ok_or(SnapSyncError::NoAttempt)?;
152        if !attempt.is_unfinished() || SnapWrite::of(&attempt) != write {
153            return Err(SnapSyncError::StaleWrite {
154                attempt: write.attempt,
155                state_version: write.state_version,
156            })
157        }
158        Ok(attempt)
159    }
160
161    fn authorize_canonical_snap_write(&self, write: SnapWrite) -> Result<SnapAttempt, SnapSyncError>
162    where
163        Self: BlockHashReader,
164    {
165        let attempt = self.authorize_snap_write(write)?;
166        let pivot = attempt.pivot();
167        if self.block_hash(pivot.number)? != Some(pivot.hash) {
168            return Err(SnapSyncError::NonCanonicalBlock { block: pivot.number, hash: pivot.hash })
169        }
170        Ok(attempt)
171    }
172
173    fn advance_snap_pivot(
174        &self,
175        write: SnapWrite,
176        generation: SnapGeneration,
177    ) -> Result<SnapWrite, SnapSyncError>
178    where
179        Self: MetadataWriter + HeaderProvider + BlockHashReader,
180    {
181        // Replacing an orphaned pivot would hide the fork that downloaded ranges belong to.
182        let mut attempt = self.authorize_canonical_snap_write(write)?;
183        let (pivot, target) = (attempt.pivot(), generation.target());
184        // Ranges already committed hold the old pivot's changes, which no list can undo.
185        if target.number <= pivot.number {
186            return Err(SnapSyncError::PivotNotAdvanced {
187                pivot: pivot.number,
188                target: target.number,
189            })
190        }
191        if self.block_hash(target.number)? != Some(target.hash) {
192            return Err(SnapSyncError::NonCanonicalBlock { block: target.number, hash: target.hash })
193        }
194        ensure_header_root(self, &generation)?;
195        attempt.re_anchor(target, generation.state_root());
196        self.write_snap_attempt(&attempt)?;
197        StoredAncestry::record(self, attempt.id(), target)?;
198        Ok(SnapWrite::of(&attempt))
199    }
200
201    fn verify_snap_attempt(&self, write: SnapWrite) -> Result<(), SnapSyncError>
202    where
203        Self: MetadataWriter,
204    {
205        let mut attempt = self.authorize_snap_write(write)?;
206        attempt.verify();
207        self.write_snap_attempt(&attempt)?;
208        Ok(())
209    }
210
211    fn abandon_snap_attempt(&self) -> Result<(), SnapSyncError>
212    where
213        Self: MetadataWriter,
214    {
215        if let Some(mut attempt) = self.snap_attempt()? &&
216            attempt.is_unfinished()
217        {
218            attempt.abandon();
219            self.write_snap_attempt(&attempt)?;
220        }
221        Ok(())
222    }
223
224    fn snap_reorg(&self, write: SnapWrite) -> Result<Option<SnapReorg>, SnapSyncError>
225    where
226        Self: BlockHashReader,
227    {
228        StoredAncestry::reorg(self, &self.authorize_snap_write(write)?)
229    }
230
231    fn commit_reorg_recovery(
232        &self,
233        write: SnapWrite,
234        ancestor: BlockNumHash,
235        lists: &[DecodedBal],
236        generation: SnapGeneration,
237    ) -> Result<SnapWrite, SnapSyncError>
238    where
239        Self: MetadataWriter + HeaderProvider + BlockHashReader + DBProvider,
240    {
241        let mut attempt = self.authorize_snap_write(write)?;
242        let target = generation.target();
243        for block in [ancestor, target] {
244            if self.block_hash(block.number)? != Some(block.hash) {
245                return Err(SnapSyncError::NonCanonicalBlock {
246                    block: block.number,
247                    hash: block.hash,
248                })
249            }
250        }
251        if target.number < ancestor.number {
252            return Err(SnapSyncError::PivotBelowAncestor {
253                ancestor: ancestor.number,
254                target: target.number,
255            })
256        }
257
258        ensure_header_root(self, &generation)?;
259
260        let coverage = self.account_coverage(write)?.ok_or(SnapSyncError::NoCoverage)?;
261        let mut repairs = StateRepairs::default();
262        for changes in lists.iter().flat_map(|list| list.as_bal().iter()) {
263            let hashed_address = keccak256(changes.address());
264            // Accounts not downloaded yet are downloaded whole at a canonical pivot later.
265            if !changes.has_changes() ||
266                self.downloaded_account(coverage, hashed_address)? == DownloadedAccount::Unknown
267            {
268                continue
269            }
270            repairs.insert_changes(hashed_address, changes);
271        }
272        self.schedule_snap_repairs(write, repairs)?;
273
274        let applied = self
275            .catch_up_progress(write)?
276            .ok_or(SnapSyncError::NoCatchUpProgress)?
277            .resume_after(ancestor);
278        attempt.re_anchor(target, generation.state_root());
279        self.write_snap_attempt(&attempt)?;
280        CatchUpProgress::at_pivot(applied).write(self, attempt.id())?;
281        StoredProgress::clear(self)?;
282        StoredAncestry::record(self, attempt.id(), target)?;
283        Ok(SnapWrite::of(&attempt))
284    }
285}
286
287// Rejects a generation whose root is not that of the canonical header at its target, which
288// ranges are proved against and which the attempt can only be verified with.
289fn ensure_header_root(
290    provider: &impl HeaderProvider,
291    generation: &SnapGeneration,
292) -> Result<(), SnapSyncError> {
293    let target = generation.target();
294    let header = provider
295        .sealed_header(target.number)?
296        .filter(|header| header.hash() == target.hash)
297        .ok_or(SnapSyncError::NonCanonicalBlock { block: target.number, hash: target.hash })?;
298    if header.state_root() != generation.state_root() {
299        return Err(SnapSyncError::HeaderRootMismatch {
300            block: target.number,
301            expected: header.state_root(),
302            got: generation.state_root(),
303        })
304    }
305    Ok(())
306}
307
308#[cfg(test)]
309mod tests {
310    use super::*;
311    use crate::test_utils::insert_generation_headers;
312    use alloy_eips::BlockNumHash;
313    use alloy_primitives::{Bytes, B256};
314    use reth_db_api::{tables, transaction::DbTx};
315    use reth_primitives_traits::Account;
316    use reth_provider::{
317        test_utils::{create_test_provider_factory, MockNodeTypesWithDB},
318        DBProvider, DatabaseProviderFactory, ProviderFactory,
319    };
320    use reth_storage_api::{metadata::keys, StateWriter};
321    use reth_trie_common::HashedPostState;
322    use revm::{bytecode::Bytecode, database::states::StateChangeset};
323
324    const HASHED_ADDRESS: B256 = B256::repeat_byte(0xbb);
325
326    fn code() -> Bytecode {
327        Bytecode::new_raw(Bytes::from_static(&[0x60, 0x00]))
328    }
329
330    fn generation(block: u64) -> SnapGeneration {
331        SnapGeneration::new(
332            BlockNumHash::new(block, B256::repeat_byte(block as u8)),
333            B256::repeat_byte(block as u8),
334        )
335    }
336
337    // A database using the hashed state layout snap writes into, holding the blocks the
338    // generations above anchor to.
339    fn factory() -> ProviderFactory<MockNodeTypesWithDB> {
340        let factory = create_test_provider_factory();
341        let provider = factory.database_provider_rw().unwrap();
342        provider.write_storage_settings(StorageSettings::v2()).unwrap();
343        provider.commit().unwrap();
344        insert_generation_headers(&factory, &[]);
345        factory
346    }
347
348    // Persists one account and its bytecode through the writers snap shares with execution.
349    fn download(provider: &impl StateWriter) {
350        let mut state = HashedPostState::default();
351        state.accounts.insert(HASHED_ADDRESS, Some(Account::default()));
352        provider.write_hashed_state(&state.into_sorted()).unwrap();
353        provider
354            .write_state_changes(StateChangeset {
355                contracts: vec![(code().hash_slow(), code())],
356                ..Default::default()
357            })
358            .unwrap();
359    }
360
361    fn downloaded(provider: &impl DBProvider) -> (bool, bool) {
362        let tx = provider.tx_ref();
363        (
364            tx.get::<tables::HashedAccounts>(HASHED_ADDRESS).unwrap().is_some(),
365            tx.get::<tables::Bytecodes>(code().hash_slow()).unwrap().is_some(),
366        )
367    }
368
369    #[test]
370    fn address_keyed_state_cannot_host_an_attempt() {
371        let factory = create_test_provider_factory();
372        let provider = factory.database_provider_rw().unwrap();
373
374        // Nothing recorded means the legacy layout.
375        assert!(matches!(
376            provider.start_snap_attempt(generation(1)),
377            Err(SnapSyncError::UnsupportedStorage)
378        ));
379        provider.write_storage_settings(StorageSettings::v1()).unwrap();
380        assert!(matches!(
381            provider.start_snap_attempt(generation(1)),
382            Err(SnapSyncError::UnsupportedStorage)
383        ));
384        assert_eq!(provider.snap_attempt().unwrap(), None);
385    }
386
387    #[test]
388    fn a_new_attempt_drops_progress_another_build_wrote() {
389        let factory = factory();
390        let provider = factory.database_provider_rw().unwrap();
391        let unreadable = br#"{"version":999}"#.to_vec();
392        for key in [StoredCoverage::KEY, StoredProgress::KEY, StoredRebuild::KEY] {
393            provider.write_metadata(key, unreadable.clone()).unwrap();
394        }
395
396        let write = provider.start_snap_attempt(generation(1)).unwrap();
397
398        assert_eq!(
399            crate::SnapAccountStore::start_account_coverage(&provider, write).unwrap(),
400            crate::AccountCoverage::START
401        );
402        for key in [StoredProgress::KEY, StoredRebuild::KEY] {
403            assert_eq!(provider.get_metadata(key).unwrap(), None);
404        }
405    }
406
407    #[test]
408    fn restarting_at_the_same_pivot_takes_a_new_identity() {
409        let factory = factory();
410        let provider = factory.database_provider_rw().unwrap();
411
412        let first = provider.start_snap_attempt(generation(1)).unwrap();
413        let second = provider.start_snap_attempt(generation(1)).unwrap();
414
415        assert_ne!(first.attempt(), second.attempt());
416        // The superseded attempt's outstanding downloads no longer own the state.
417        assert!(matches!(
418            provider.authorize_snap_write(first),
419            Err(SnapSyncError::StaleWrite { .. })
420        ));
421        provider.authorize_snap_write(second).unwrap();
422    }
423
424    #[test]
425    fn abandoning_an_attempt_keeps_its_identity_taken() {
426        let factory = factory();
427        let provider = factory.database_provider_rw().unwrap();
428
429        let abandoned = provider.start_snap_attempt(generation(1)).unwrap();
430        provider.abandon_snap_attempt().unwrap();
431        assert_eq!(provider.active_snap_write().unwrap(), None);
432        let started = provider.start_snap_attempt(generation(2)).unwrap();
433
434        assert_ne!(abandoned.attempt(), started.attempt());
435        // The abandoned attempt's outstanding downloads cannot pass as the new attempt's work.
436        assert!(matches!(
437            provider.authorize_snap_write(abandoned),
438            Err(SnapSyncError::StaleWrite { .. })
439        ));
440    }
441
442    #[test]
443    fn committing_keeps_downloaded_state_and_progress_together() {
444        let factory = factory();
445        let provider = factory.database_provider_rw().unwrap();
446        let write = provider.start_snap_attempt(generation(1)).unwrap();
447        download(&provider);
448        provider.commit().unwrap();
449
450        let provider = factory.database_provider_rw().unwrap();
451        assert_eq!(provider.active_snap_write().unwrap(), Some(write));
452        assert_eq!(downloaded(&provider), (true, true));
453    }
454
455    #[test]
456    fn rolling_back_drops_downloaded_state_and_progress_together() {
457        let factory = factory();
458        let provider = factory.database_provider_rw().unwrap();
459        provider.start_snap_attempt(generation(1)).unwrap();
460        download(&provider);
461        // Dropping without committing is the interrupted-commit case.
462        drop(provider);
463
464        let provider = factory.database_provider_rw().unwrap();
465        assert_eq!(provider.active_snap_write().unwrap(), None);
466        assert_eq!(downloaded(&provider), (false, false));
467    }
468
469    #[test]
470    fn the_pivot_advances_repeatedly_within_one_attempt() {
471        let factory = factory();
472        let provider = factory.database_provider_rw().unwrap();
473        let first = provider.start_snap_attempt(generation(1)).unwrap();
474
475        let second = provider.advance_snap_pivot(first, generation(2)).unwrap();
476        let third = provider.advance_snap_pivot(second, generation(3)).unwrap();
477
478        assert_eq!(third.attempt(), first.attempt());
479        assert_eq!(third.state_version(), first.state_version() + 2);
480        assert_eq!(provider.snap_attempt().unwrap().unwrap().pivot().number, 3);
481        for stale in [first, second] {
482            assert!(matches!(
483                provider.authorize_snap_write(stale),
484                Err(SnapSyncError::StaleWrite { .. })
485            ));
486        }
487    }
488
489    #[test]
490    fn the_pivot_only_moves_forward() {
491        let factory = factory();
492        let provider = factory.database_provider_rw().unwrap();
493        let write = provider.start_snap_attempt(generation(2)).unwrap();
494
495        // Committed ranges hold block 2's changes, which a lower pivot's lists cannot undo.
496        for target in [1, 2] {
497            assert!(matches!(
498                provider.advance_snap_pivot(write, generation(target)),
499                Err(SnapSyncError::PivotNotAdvanced { pivot: 2, target: refused })
500                    if refused == target
501            ));
502        }
503        provider.authorize_snap_write(write).unwrap();
504    }
505
506    #[test]
507    fn the_pivot_cannot_move_off_the_canonical_chain() {
508        let factory = factory();
509        let provider = factory.database_provider_rw().unwrap();
510        let write = provider.start_snap_attempt(generation(1)).unwrap();
511        let orphan = SnapGeneration::new(BlockNumHash::new(2, B256::repeat_byte(0xee)), B256::ZERO);
512
513        let refused = provider.advance_snap_pivot(write, orphan);
514
515        assert!(matches!(refused, Err(SnapSyncError::NonCanonicalBlock { block: 2, .. })));
516        provider.authorize_snap_write(write).unwrap();
517    }
518
519    #[test]
520    fn a_generation_must_carry_the_root_of_its_canonical_header() {
521        let factory = factory();
522        let provider = factory.database_provider_rw().unwrap();
523        let wrong =
524            |block: u64| SnapGeneration::new(generation(block).target(), B256::repeat_byte(0xcc));
525
526        assert!(matches!(
527            provider.start_snap_attempt(wrong(1)),
528            Err(SnapSyncError::HeaderRootMismatch { .. })
529        ));
530        let write = provider.start_snap_attempt(generation(1)).unwrap();
531        assert!(matches!(
532            provider.advance_snap_pivot(write, wrong(2)),
533            Err(SnapSyncError::HeaderRootMismatch { .. })
534        ));
535        provider.authorize_snap_write(write).unwrap();
536    }
537
538    #[test]
539    fn a_verified_attempt_accepts_no_further_writes() {
540        let factory = factory();
541        let provider = factory.database_provider_rw().unwrap();
542        let write = provider.start_snap_attempt(generation(1)).unwrap();
543
544        provider.verify_snap_attempt(write).unwrap();
545
546        assert!(!provider.snap_attempt().unwrap().unwrap().is_unfinished());
547        assert_eq!(provider.active_snap_write().unwrap(), None);
548        assert!(matches!(
549            provider.authorize_snap_write(write),
550            Err(SnapSyncError::StaleWrite { .. })
551        ));
552        // Verified state is complete, so abandoning cannot turn it into leftovers.
553        provider.abandon_snap_attempt().unwrap();
554        assert!(provider.snap_attempt().unwrap().unwrap().is_verified());
555    }
556
557    #[test]
558    fn a_record_this_build_cannot_read_is_reported_rather_than_ignored() {
559        let factory = factory();
560
561        for record in [br#"{"version":999}"#.to_vec(), b"{}".to_vec(), b"not json".to_vec()] {
562            let provider = factory.database_provider_rw().unwrap();
563            provider.write_metadata(keys::SNAP_ATTEMPT, record).unwrap();
564
565            assert!(provider.snap_attempt().is_err());
566            // Starting a new attempt cannot guarantee a distinct identity over an unreadable one.
567            assert!(provider.start_snap_attempt(generation(1)).is_err());
568        }
569    }
570}
reth_snap_sync/attempt.rs

reth_snap_sync/
attempt.rs
