reth_node_ethereum/snap/
handoff.rs

1//! Hands snap-downloaded state over to the staged pipeline.
2//!
3//! The merkle stage rebuilds the trie from the downloaded state, and only a matching root on a
4//! still canonical pivot is published and accepted.
5
6use alloy_eips::BlockNumHash;
7use reth_errors::{ConsensusError, RethError};
8use reth_provider::{
9    providers::ProviderNodeTypes, DBProvider, DatabaseProviderFactory, HeaderProvider,
10    ProviderFactory, ProviderResult, StageCheckpointReader, StageCheckpointWriter,
11};
12use reth_snap_sync::{SnapAttemptStore, SnapStateVerifier, SnapWrite};
13use reth_stages::{
14    stages::MerkleStage, BlockErrorKind, ExecInput, PipelineError, Stage, StageError, StageId,
15};
16use reth_tracing::tracing::{error, info};
17use tokio_util::sync::CancellationToken;
18
19/// Hands one attempt's downloaded state over to the staged pipeline.
20///
21/// Every step commits on its own and repeats safely, so an interrupted handoff runs again.
22#[derive(Debug)]
23pub struct SnapHandoff<N: ProviderNodeTypes> {
24    // Database the state was downloaded into.
25    factory: ProviderFactory<N>,
26}
27
28impl<N: ProviderNodeTypes> SnapHandoff<N> {
29    /// Creates a handoff over the node's database.
30    pub const fn new(factory: ProviderFactory<N>) -> Self {
31        Self { factory }
32    }
33
34    /// Rebuilds the trie at the pivot of `write`'s attempt and checks its root, abandoning the
35    /// attempt on a mismatch so the next run starts a new one. Progress commits as it goes, and
36    /// `stop` ends it early.
37    ///
38    /// The attempt's state must have been handed to the merkle stage with
39    /// [`SnapStateVerifier::start_trie_rebuild`].
40    pub fn rebuild(
41        &self,
42        write: SnapWrite,
43        stop: &CancellationToken,
44    ) -> Result<RebuildOutcome, PipelineError> {
45        let pivot = self.pivot(write)?;
46        match self.rebuild_trie(pivot, stop) {
47            Err(PipelineError::Stage(StageError::Block {
48                error: BlockErrorKind::Validation(ConsensusError::BodyStateRootDiff(diff)),
49                ..
50            })) => {
51                // Downloaded state is authenticated against the pivot, so this is a local fault.
52                error!(target: "sync::snap", ?pivot, %diff, "Snap state root mismatch, abandoning the attempt");
53                self.abandon()?;
54                Ok(RebuildOutcome::RootMismatch)
55            }
56            rebuilt => rebuilt,
57        }
58    }
59
60    /// Publishes the state downloaded under `write` at its attempt's pivot and accepts it, so the
61    /// pipeline continues above the pivot. The pivot must still be canonical, and its trie is
62    /// rebuilt first if [`Self::rebuild`] hasn't finished, until `stop` fires.
63    ///
64    /// The attempt's state must have been handed to the merkle stage with
65    /// [`SnapStateVerifier::start_trie_rebuild`].
66    pub fn hand_off(
67        &self,
68        write: SnapWrite,
69        stop: &CancellationToken,
70    ) -> Result<HandoffOutcome, PipelineError> {
71        let pivot = self.pivot(write)?;
72        if !self.is_canonical(pivot)? {
73            self.abandon()?;
74            return Ok(HandoffOutcome::PivotReorged)
75        }
76
77        // A trie already rebuilt to the pivot returns at once.
78        match self.rebuild(write, stop)? {
79            RebuildOutcome::Rebuilt => {}
80            RebuildOutcome::RootMismatch => return Ok(HandoffOutcome::RootMismatch),
81            RebuildOutcome::Stopped => return Ok(HandoffOutcome::Stopped),
82        }
83
84        self.publish(write)?;
85        info!(target: "sync::snap", pivot = pivot.number, "Snap state published; history below it is unavailable");
86        Ok(HandoffOutcome::Completed)
87    }
88
89    /// Publishes an attempt whose trie is rebuilt at a still canonical pivot, finishing a publish
90    /// that stopped before its database commit. An attempt rebuilt but not yet handed off is
91    /// published as well. Runs before the consistency check, which would otherwise unwind the
92    /// static files below their anchor.
93    pub fn resume_interrupted_publish(&self) -> Result<(), PipelineError> {
94        let Some((write, pivot)) = self.interrupted_publish()? else { return Ok(()) };
95        info!(target: "sync::snap", pivot, "Resuming an interrupted snap state publish");
96        self.publish(write)
97    }
98
99    // An unfinished attempt whose trie is rebuilt at a still canonical pivot. Read from the
100    // database alone, since a publish deletes the static files before that commit.
101    fn interrupted_publish(&self) -> Result<Option<(SnapWrite, u64)>, PipelineError> {
102        let provider = self.factory.provider()?;
103        let Some(write) = provider.active_snap_write().map_err(RethError::other)? else {
104            return Ok(None)
105        };
106        let pivot = provider.authorize_snap_write(write).map_err(RethError::other)?.pivot();
107        let rebuilt = provider.is_trie_rebuild_started(write).map_err(RethError::other)? &&
108            provider
109                .get_stage_checkpoint(StageId::MerkleExecute)?
110                .is_some_and(|checkpoint| checkpoint.block_number == pivot.number);
111        Ok((rebuilt && self.is_canonical(pivot)?).then_some((write, pivot.number)))
112    }
113
114    // Pivot of the attempt `write` belongs to, refusing a write the attempt no longer accepts.
115    fn pivot(&self, write: SnapWrite) -> Result<BlockNumHash, PipelineError> {
116        let provider = self.factory.provider()?;
117        Ok(provider.authorize_snap_write(write).map_err(RethError::other)?.pivot())
118    }
119
120    // Forkchoice can reorg the pivot out while its state downloads, and publishing anchors the
121    // node to it, so a pivot that is no longer canonical is dropped instead of published.
122    fn is_canonical(&self, pivot: BlockNumHash) -> ProviderResult<bool> {
123        Ok(self
124            .factory
125            .provider()?
126            .sealed_header(pivot.number)?
127            .is_some_and(|header| header.hash() == pivot.hash))
128    }
129
130    // Drops the attempt, so the next run starts a new one.
131    fn abandon(&self) -> Result<(), PipelineError> {
132        let provider = self.factory.database_provider_rw()?;
133        provider.abandon_snap_attempt().map_err(RethError::other)?;
134        provider.commit()?;
135        Ok(())
136    }
137
138    // Accepts the state, anchors the static files at the verified pivot and moves the checkpoints
139    // there in one database commit, so a stop before it leaves the attempt to resume.
140    fn publish(&self, write: SnapWrite) -> Result<(), PipelineError> {
141        let provider = self.factory.database_provider_rw()?;
142        // Deleted static files don't roll back with the database, so a rejected state must fail
143        // before them.
144        let pivot = provider.verify_state_root(write).map_err(RethError::other)?.target().number;
145        // History below the pivot was never downloaded, so it counts as pruned and the static
146        // files start there.
147        provider.anchor_pruned_static_files(pivot)?;
148        provider.publish_snap_state(pivot).map_err(RethError::other)?;
149        provider.commit()?;
150        Ok(())
151    }
152
153    // Rebuilds the trie from the downloaded state up to `pivot`, committing the stage's progress
154    // in chunks so a restart resumes it. The stage checks the root against the pivot's header.
155    fn rebuild_trie(
156        &self,
157        pivot: BlockNumHash,
158        stop: &CancellationToken,
159    ) -> Result<RebuildOutcome, PipelineError> {
160        let mut stage = MerkleStage::default_execution();
161        loop {
162            if stop.is_cancelled() {
163                return Ok(RebuildOutcome::Stopped)
164            }
165            let provider = self.factory.database_provider_rw()?;
166            let checkpoint = provider.get_stage_checkpoint(StageId::MerkleExecute)?;
167            let output =
168                stage.execute(&provider, ExecInput { target: Some(pivot.number), checkpoint })?;
169            provider.save_stage_checkpoint(StageId::MerkleExecute, output.checkpoint)?;
170            provider.commit()?;
171            if output.done {
172                return Ok(RebuildOutcome::Rebuilt)
173            }
174        }
175    }
176}
177
178/// What a handoff did with the downloaded state.
179#[must_use]
180#[derive(Debug, Eq, PartialEq)]
181pub enum HandoffOutcome {
182    /// The state is published, its trie rebuilt and its root verified.
183    Completed,
184    /// The pivot left the canonical chain, so the attempt was abandoned for a new one.
185    PivotReorged,
186    /// The rebuilt root differs from the pivot's header, so the attempt was abandoned for a new
187    /// one.
188    RootMismatch,
189    /// The trie rebuild was stopped before reaching the pivot, so nothing was published.
190    Stopped,
191}
192
193/// How a trie rebuild ended.
194#[must_use]
195#[derive(Debug, Eq, PartialEq)]
196pub enum RebuildOutcome {
197    /// The trie is rebuilt at the pivot and its root matches the pivot's header.
198    Rebuilt,
199    /// The rebuilt root differs from the pivot's header, so the attempt was abandoned for a new
200    /// one.
201    RootMismatch,
202    /// The rebuild was stopped before reaching the pivot. Its progress so far is committed.
203    Stopped,
204}
205
206#[cfg(test)]
207pub(crate) mod tests {
208    use super::*;
209    use crate::snap::tests::hashed_factory;
210    use alloy_consensus::Header;
211    use alloy_primitives::B256;
212    use futures::future::{ready, Ready};
213    use reth_db::{tables, transaction::DbTxMut};
214    use reth_downloaders::snap::{AccountRangeDownloader, AccountRangeOutcome};
215    use reth_eth_wire_types::snap::{
216        AccountData, AccountRangeMessage, GetAccountRangeMessage, GetBlockAccessListsMessage,
217        GetByteCodesMessage, GetStorageRangesMessage,
218    };
219    use reth_network_p2p::{
220        download::DownloadClient,
221        error::PeerRequestResult,
222        priority::Priority,
223        snap::client::{SnapClient, SnapResponse},
224    };
225    use reth_network_peers::{PeerId, WithPeerId};
226    use reth_primitives_traits::Account;
227    use reth_provider::{
228        test_utils::MockNodeTypesWithDB, BlockWriter, MetadataProvider, StaticFileProviderFactory,
229        StaticFileSegment, StaticFileWriter,
230    };
231    use reth_snap_sync::{
232        SnapAccountStore, SnapGeneration, DEFAULT_RESPONSE_BYTES, DEFAULT_SCAN_CHUNK, MAX_HASH,
233    };
234    use reth_stages::StageCheckpoint;
235    use reth_tasks::Runtime;
236    use reth_trie_common::{root::state_root_unsorted, TrieAccount};
237
238    pub(crate) const PIVOT: u64 = 1;
239
240    // Serves every account range from one trie, as a peer holding all of it would. A complete
241    // trie needs no proof.
242    #[derive(Debug)]
243    struct WholeTrieClient(Vec<(B256, TrieAccount)>);
244
245    impl DownloadClient for WholeTrieClient {
246        fn report_bad_message(&self, _peer_id: PeerId) {}
247
248        fn num_connected_peers(&self) -> usize {
249            1
250        }
251    }
252
253    impl SnapClient for WholeTrieClient {
254        type Output = Ready<PeerRequestResult<SnapResponse>>;
255
256        fn get_account_range_with_priority(
257            &self,
258            request: GetAccountRangeMessage,
259            _priority: Priority,
260        ) -> Self::Output {
261            let accounts = self
262                .0
263                .iter()
264                .map(|(key, account)| AccountData::from_trie_account(*key, account))
265                .collect();
266            let message =
267                AccountRangeMessage { request_id: request.request_id, accounts, proof: Vec::new() };
268            ready(Ok(WithPeerId::new(PeerId::random(), SnapResponse::AccountRange(message))))
269        }
270
271        fn get_storage_ranges(&self, _request: GetStorageRangesMessage) -> Self::Output {
272            unreachable!("the accounts have no storage")
273        }
274
275        fn get_storage_ranges_with_priority(
276            &self,
277            _request: GetStorageRangesMessage,
278            _priority: Priority,
279        ) -> Self::Output {
280            unreachable!("the accounts have no storage")
281        }
282
283        fn get_byte_codes(&self, _request: GetByteCodesMessage) -> Self::Output {
284            unreachable!("the accounts have no code")
285        }
286
287        fn get_byte_codes_with_priority(
288            &self,
289            _request: GetByteCodesMessage,
290            _priority: Priority,
291        ) -> Self::Output {
292            unreachable!("the accounts have no code")
293        }
294
295        fn get_block_access_lists_with_priority(
296            &self,
297            _request: GetBlockAccessListsMessage,
298            _priority: Priority,
299        ) -> Self::Output {
300            unreachable!("the attempt stays at its pivot")
301        }
302    }
303
304    // Headers through the pivot on storage v2.
305    fn with_headers() -> ProviderFactory<MockNodeTypesWithDB> {
306        with_headers_committing_to(Header::default().state_root)
307    }
308
309    // Headers through the pivot on storage v2, each committing to `state_root`.
310    fn with_headers_committing_to(state_root: B256) -> ProviderFactory<MockNodeTypesWithDB> {
311        let factory = hashed_factory();
312        let static_files = factory.static_file_provider();
313        let mut writer = static_files.latest_writer(StaticFileSegment::Headers).unwrap();
314        let mut parent = B256::ZERO;
315        for number in 0..=PIVOT {
316            let header = Header { number, parent_hash: parent, state_root, ..Default::default() };
317            let hash = header.hash_slow();
318            writer.append_header(&header, &hash).unwrap();
319            parent = hash;
320        }
321        writer.commit().unwrap();
322        drop(writer);
323        factory
324    }
325
326    // Headers through the pivot, with an unverified attempt downloading state at it.
327    fn downloading() -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite) {
328        let factory = with_headers();
329        let provider = factory.database_provider_rw().unwrap();
330        let header = provider.sealed_header(PIVOT).unwrap().unwrap();
331        let write = provider
332            .start_snap_attempt(SnapGeneration::new(header.num_hash(), header.state_root))
333            .unwrap();
334        provider.commit().unwrap();
335        (factory, write)
336    }
337
338    // Headers committing to two plain accounts, with an attempt at the pivot that downloaded both.
339    pub(crate) fn downloaded_attempt() -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite) {
340        let accounts: Vec<_> = (1..=2)
341            .map(|nonce| {
342                (
343                    B256::with_last_byte(nonce),
344                    TrieAccount { nonce: nonce.into(), ..Default::default() },
345                )
346            })
347            .collect();
348        let root = state_root_unsorted(accounts.clone());
349        let factory = with_headers_committing_to(root);
350        let provider = factory.database_provider_rw().unwrap();
351        let pivot = provider.sealed_header(PIVOT).unwrap().unwrap().num_hash();
352        let write = provider.start_snap_attempt(SnapGeneration::new(pivot, root)).unwrap();
353        provider.start_account_coverage(write).unwrap();
354        let request = GetAccountRangeMessage {
355            request_id: 1,
356            root_hash: root,
357            starting_hash: B256::ZERO,
358            limit_hash: MAX_HASH,
359            response_bytes: DEFAULT_RESPONSE_BYTES,
360        };
361        let downloader =
362            AccountRangeDownloader::new(WholeTrieClient(accounts), request, Runtime::test())
363                .unwrap();
364        let AccountRangeOutcome::Verified(range) = futures::executor::block_on(downloader).unwrap()
365        else {
366            panic!("the client serves the requested root")
367        };
368        provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
369        provider.commit().unwrap();
370        (factory, write)
371    }
372
373    // An attempt that downloaded all of its state, handed to the merkle stage.
374    fn handed_to_merkle() -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite) {
375        let (factory, write) = downloaded_attempt();
376        let provider = factory.database_provider_rw().unwrap();
377        provider.start_trie_rebuild(write, DEFAULT_SCAN_CHUNK, &CancellationToken::new()).unwrap();
378        provider.commit().unwrap();
379        (factory, write)
380    }
381
382    // An attempt that downloaded all of its state, with its trie rebuilt at the pivot.
383    fn rebuilt() -> (ProviderFactory<MockNodeTypesWithDB>, SnapWrite) {
384        let (factory, write) = handed_to_merkle();
385        let rebuild = SnapHandoff::new(factory.clone()).rebuild(write, &CancellationToken::new());
386        assert_eq!(rebuild.unwrap(), RebuildOutcome::Rebuilt);
387        (factory, write)
388    }
389
390    // A publish whose static files committed before the node stopped, ahead of its database
391    // commit.
392    fn interrupt_publish(factory: &ProviderFactory<MockNodeTypesWithDB>) {
393        let provider = factory.database_provider_rw().unwrap();
394        provider.anchor_pruned_static_files(PIVOT).unwrap();
395        provider.publish_snap_state(PIVOT).unwrap();
396        factory.static_file_provider().finalize().unwrap();
397    }
398
399    // Asserts the state is published at the pivot and accepted.
400    fn assert_published(factory: &ProviderFactory<MockNodeTypesWithDB>) {
401        assert_eq!(factory.check_consistency().unwrap(), (None, None));
402        let static_files = factory.static_file_provider();
403        for segment in StaticFileSegment::iter().filter(|segment| !segment.is_headers()) {
404            assert_eq!(
405                static_files.get_highest_static_file_block(segment),
406                Some(PIVOT),
407                "{segment}"
408            );
409        }
410        let provider = factory.database_provider_ro().unwrap();
411        assert_eq!(
412            provider.get_stage_checkpoint(StageId::Execution).unwrap(),
413            Some(StageCheckpoint::new(PIVOT))
414        );
415        assert!(provider.snap_attempt().unwrap().unwrap().is_verified());
416    }
417
418    // Headers through the pivot, with the state published at it.
419    fn published() -> ProviderFactory<MockNodeTypesWithDB> {
420        let factory = with_headers();
421        let provider = factory.database_provider_rw().unwrap();
422        provider.anchor_pruned_static_files(PIVOT).unwrap();
423        provider.publish_snap_state(PIVOT).unwrap();
424        provider.commit().unwrap();
425        factory
426    }
427
428    #[test]
429    fn resuming_without_an_interrupted_publish_changes_nothing() {
430        let (factory, ..) = downloading();
431
432        SnapHandoff::new(factory.clone()).resume_interrupted_publish().unwrap();
433
434        let provider = factory.database_provider_ro().unwrap();
435        assert_eq!(provider.get_stage_checkpoint(StageId::Execution).unwrap(), None);
436        assert_eq!(factory.check_consistency().unwrap(), (None, None));
437    }
438
439    #[test]
440    fn the_block_after_the_pivot_appends_to_the_anchored_files() {
441        let factory = published();
442        let provider = factory.database_provider_rw().unwrap();
443
444        // Appending fails unless the anchored segments start right after the pivot.
445        provider.append_block_bodies(vec![(PIVOT + 1, Some(&Default::default()))]).unwrap();
446        provider.commit().unwrap();
447    }
448
449    #[test]
450    fn a_pivot_reorged_out_is_abandoned_instead_of_published() {
451        let factory = with_headers();
452        let provider = factory.database_provider_rw().unwrap();
453        let pivot = provider.sealed_header(PIVOT).unwrap().unwrap();
454        let write = provider
455            .start_snap_attempt(SnapGeneration::new(pivot.num_hash(), pivot.state_root))
456            .unwrap();
457        provider.commit().unwrap();
458
459        // The canonical header at the pivot's number is a different block now.
460        let replacement = Header { extra_data: vec![1].into(), ..pivot.header().clone() };
461        let static_files = factory.static_file_provider();
462        let mut writer = static_files.latest_writer(StaticFileSegment::Headers).unwrap();
463        writer.prune_headers(1).unwrap();
464        writer.commit().unwrap();
465        writer.append_header(&replacement, &replacement.hash_slow()).unwrap();
466        writer.commit().unwrap();
467        drop(writer);
468        let handoff =
469            SnapHandoff::new(factory.clone()).hand_off(write, &CancellationToken::new()).unwrap();
470
471        assert_eq!(handoff, HandoffOutcome::PivotReorged);
472        let provider = factory.database_provider_ro().unwrap();
473        assert!(provider.active_snap_write().unwrap().is_none());
474    }
475
476    #[test]
477    fn an_abandoned_attempt_is_not_an_interrupted_publish() {
478        let (factory, ..) = downloading();
479        let provider = factory.database_provider_rw().unwrap();
480        provider.abandon_snap_attempt().unwrap();
481        provider.anchor_pruned_static_files(PIVOT).unwrap();
482        provider.commit().unwrap();
483
484        SnapHandoff::new(factory.clone()).resume_interrupted_publish().unwrap();
485
486        let provider = factory.database_provider_ro().unwrap();
487        assert_eq!(provider.get_stage_checkpoint(StageId::Execution).unwrap(), None);
488    }
489
490    #[test]
491    fn a_root_mismatch_abandons_the_attempt_before_publishing() {
492        let (factory, write) = downloading();
493        let provider = factory.database_provider_rw().unwrap();
494        // State whose root differs from the empty root the pivot header commits to.
495        provider
496            .tx_ref()
497            .put::<tables::HashedAccounts>(
498                B256::repeat_byte(3),
499                Account { nonce: 1, ..Default::default() },
500            )
501            .unwrap();
502        provider.commit().unwrap();
503
504        let handoff = SnapHandoff::new(factory.clone()).hand_off(write, &CancellationToken::new());
505
506        assert_eq!(handoff.unwrap(), HandoffOutcome::RootMismatch);
507
508        let provider = factory.database_provider_ro().unwrap();
509        assert!(provider.active_snap_write().unwrap().is_none());
510        assert_eq!(provider.get_stage_checkpoint(StageId::Execution).unwrap(), None);
511        assert_eq!(
512            factory.static_file_provider().get_lowest_range_start(StaticFileSegment::Transactions),
513            None
514        );
515    }
516
517    #[test]
518    fn a_handoff_accepts_the_state_with_its_publish() {
519        let (factory, write) = rebuilt();
520
521        let handoff =
522            SnapHandoff::new(factory.clone()).hand_off(write, &CancellationToken::new()).unwrap();
523
524        assert_eq!(handoff, HandoffOutcome::Completed);
525        assert_published(&factory);
526    }
527
528    #[test]
529    fn a_publish_interrupted_before_its_checkpoints_resumes_at_startup() {
530        let (factory, ..) = rebuilt();
531        interrupt_publish(&factory);
532
533        // Startup finishes the publish before checking consistency, which would otherwise unwind
534        // the anchored files to the old checkpoints.
535        SnapHandoff::new(factory.clone()).resume_interrupted_publish().unwrap();
536
537        assert_published(&factory);
538    }
539
540    #[test]
541    fn a_rejected_publish_keeps_the_static_files() {
542        let (factory, write) = rebuilt();
543        let provider = factory.database_provider_rw().unwrap();
544        provider.abandon_snap_attempt().unwrap();
545        provider.commit().unwrap();
546
547        assert!(SnapHandoff::new(factory.clone()).publish(write).is_err());
548
549        assert_eq!(factory.static_file_provider().earliest_history_height(), 0);
550    }
551
552    #[test]
553    fn a_resume_cut_short_resumes_again() {
554        let (factory, ..) = rebuilt();
555        interrupt_publish(&factory);
556
557        // The resume deletes the anchored files at once, then stops before anything commits.
558        factory.database_provider_rw().unwrap().anchor_pruned_static_files(PIVOT).unwrap();
559
560        SnapHandoff::new(factory.clone()).resume_interrupted_publish().unwrap();
561
562        assert_published(&factory);
563    }
564
565    #[test]
566    fn a_stopped_handoff_publishes_nothing() {
567        let (factory, write) = handed_to_merkle();
568        let stop = CancellationToken::new();
569        stop.cancel();
570
571        let handoff = SnapHandoff::new(factory.clone()).hand_off(write, &stop).unwrap();
572
573        assert_eq!(handoff, HandoffOutcome::Stopped);
574        let provider = factory.database_provider_ro().unwrap();
575        assert_eq!(provider.active_snap_write().unwrap(), Some(write));
576        assert_eq!(provider.get_stage_checkpoint(StageId::Execution).unwrap(), None);
577        assert_eq!(factory.static_file_provider().earliest_history_height(), 0);
578    }
579}
reth_node_ethereum/snap/handoff.rs

reth_node_ethereum/snap/
handoff.rs
