reth_node_ethereum/snap/
run.rs

1//! Alternates header catch-up and snap bootstrap until the state is downloaded.
2
3use super::{
4    context::NodeSnapContext,
5    handoff::{HandoffOutcome, RebuildOutcome, SnapHandoff},
6};
7use alloy_consensus::{BlockHeader, Sealable};
8use alloy_eips::BlockNumHash;
9use alloy_primitives::B256;
10use reth_errors::RethError;
11use reth_network_p2p::{headers::client::HeadersClient, snap::client::SnapClient};
12use reth_provider::{
13    providers::ProviderNodeTypes, BlockHashReader, BlockNumReader, ProviderError, ProviderFactory,
14};
15use reth_snap_sync::{SnapBootstrap, SnapBootstrapOutcome, SnapWrite};
16use reth_stages::{
17    ControlFlow, Pipeline, PipelineError, PipelineTarget, PipelineWithResult, StageId,
18};
19use reth_tasks::Runtime;
20use reth_tracing::tracing::{debug, info, warn};
21use std::{pin::pin, time::Duration};
22use tokio::sync::watch;
23use tokio_util::sync::CancellationToken;
24
25// Minimum time between header refreshes while forkchoice moves.
26//
27// Ten slots keeps the head well inside the ~128 recent blocks peers serve state for, without
28// stopping the download every slot.
29const HEADER_REFRESH: Duration = Duration::from_secs(120);
30
31// Retry unresolved targets without flooding peers while the head remains unchanged.
32const HEADER_LOOKUP_RETRY: Duration = Duration::from_secs(1);
33
34// Returning without progress hands control back to the engine without a fatal error.
35const STOPPED: ControlFlow = ControlFlow::NoProgress { block_number: None };
36
37// Root mismatches retried before the run fails. Every range, list and bytecode is authenticated
38// against the pivot, so a mismatch is a local fault a fresh attempt rarely avoids.
39const ROOT_MISMATCH_RETRIES: u32 = 1;
40
41// Everything one spawned run needs, moved into its task.
42pub(super) struct SnapRun<N: ProviderNodeTypes, C> {
43    client: C,
44    factory: ProviderFactory<N>,
45    runtime: Runtime,
46    header_refresh: Duration,
47    // Root mismatches retried before the run fails.
48    root_mismatch_retries: u32,
49    // Cancelled when the backfill is dropped.
50    stop: CancellationToken,
51    // Latest finalized block the engine reported.
52    finalized: watch::Receiver<B256>,
53}
54
55impl<N: ProviderNodeTypes, C> SnapRun<N, C> {
56    pub(super) const fn new(
57        client: C,
58        factory: ProviderFactory<N>,
59        runtime: Runtime,
60        stop: CancellationToken,
61        finalized: watch::Receiver<B256>,
62    ) -> Self {
63        Self {
64            client,
65            factory,
66            runtime,
67            header_refresh: HEADER_REFRESH,
68            root_mismatch_retries: ROOT_MISMATCH_RETRIES,
69            stop,
70            finalized,
71        }
72    }
73
74    #[cfg(test)]
75    const fn with_header_refresh(mut self, header_refresh: Duration) -> Self {
76        self.header_refresh = header_refresh;
77        self
78    }
79
80    #[cfg(test)]
81    const fn with_root_mismatch_retries(mut self, root_mismatch_retries: u32) -> Self {
82        self.root_mismatch_retries = root_mismatch_retries;
83        self
84    }
85}
86
87impl<N, C> SnapRun<N, C>
88where
89    N: ProviderNodeTypes,
90    C: SnapClient + HeadersClient + Clone + Unpin + 'static,
91{
92    // Hands the pipeline back with the result, so the backfill can run it again.
93    pub(super) async fn run(
94        self,
95        mut pipeline: Pipeline<N>,
96        targets: watch::Receiver<B256>,
97    ) -> PipelineWithResult<N> {
98        let result = self.bootstrap(&mut pipeline, targets).await;
99        (pipeline, result)
100    }
101
102    // Alternates header catch-up and snap bootstrap until the state is downloaded or the run
103    // stops. Every bootstrap step commits, so each loop resumes where the last one stopped.
104    async fn bootstrap(
105        &self,
106        pipeline: &mut Pipeline<N>,
107        mut targets: watch::Receiver<B256>,
108    ) -> Result<ControlFlow, PipelineError> {
109        let mut mismatches = 0;
110        let verified = loop {
111            match self.catch_up_headers(pipeline, &mut targets).await? {
112                Pass::Stopped => return Ok(STOPPED),
113                Pass::Again | Pass::RootMismatch => continue,
114                Pass::Done => {}
115            }
116            match self.download(&mut targets).await? {
117                // A closed targets channel means the backfill is gone, so no head will follow.
118                SnapBootstrapOutcome::Stopped
119                    if self.stop.is_cancelled() || targets.has_changed().is_err() =>
120                {
121                    return Ok(STOPPED)
122                }
123                SnapBootstrapOutcome::Stopped => {
124                    debug!(target: "sync::snap", "Refreshing headers before resuming snap sync");
125                }
126                SnapBootstrapOutcome::TrieRebuild { write, pivot } => {
127                    match self.rebuild_and_hand_off(pipeline, &mut targets, write, pivot).await? {
128                        Pass::Stopped => return Ok(STOPPED),
129                        Pass::Again => {}
130                        Pass::RootMismatch => {
131                            mismatches += 1;
132                            if mismatches > self.root_mismatch_retries {
133                                return Err(PipelineError::Internal(RethError::msg(format!(
134                                    "snap state root mismatched {mismatches} times, last at block {}",
135                                    pivot.number
136                                ))))
137                            }
138                        }
139                        Pass::Done => break Some(pivot),
140                    }
141                }
142                SnapBootstrapOutcome::Verified { pivot } => break Some(pivot),
143                SnapBootstrapOutcome::BeforeBlockAccessLists => break None,
144            }
145        };
146        match verified {
147            Some(pivot) => {
148                info!(target: "sync::snap", ?pivot, "Snap state verified, resuming the pipeline");
149            }
150            None => {
151                info!(target: "sync::snap", "Chain predates block access lists, syncing with the staged pipeline");
152            }
153        }
154        self.finish(pipeline, &mut targets).await
155    }
156
157    // Runs one bootstrap until the state is downloaded, the backfill stops or headers are due.
158    async fn download(
159        &self,
160        targets: &mut watch::Receiver<B256>,
161    ) -> Result<SnapBootstrapOutcome, PipelineError> {
162        let run_stop = self.stop.child_token();
163        let context = NodeSnapContext::new(
164            self.factory.clone(),
165            self.client.clone(),
166            targets.clone(),
167            self.finalized.clone(),
168        );
169        let mut session = SnapBootstrap::new(
170            self.client.clone(),
171            self.factory.clone(),
172            self.runtime.clone(),
173            context,
174        )
175        .with_cancellation(run_stop.clone())
176        .with_shutdown(self.stop.clone());
177        let mut run = pin!(session.run());
178        let outcome = tokio::select! {
179            biased;
180            outcome = &mut run => outcome,
181            () = self.refresh_due(targets) => {
182                // Stops at the next step boundary, keeping committed progress.
183                run_stop.cancel();
184                run.await
185            }
186        };
187        match outcome {
188            Ok(outcome) => Ok(outcome),
189            Err(error) if error.is_fatal() => Err(PipelineError::Internal(RethError::other(error))),
190            // A failed sync waits for the next head, and committed progress lets the next bootstrap
191            // resume.
192            Err(error) => {
193                warn!(target: "sync::snap", %error, "Snap sync failed, retrying once forkchoice moves");
194                if let Some(Err(_)) = self.stop.run_until_cancelled(targets.changed()).await {
195                    debug!(target: "sync::snap", "Targets channel closed, stopping snap sync");
196                }
197                Ok(SnapBootstrapOutcome::Stopped)
198            }
199        }
200    }
201
202    // Rebuilds the trie at the pivot of `write`'s attempt, then hands the state off. The rebuild
203    // can take hours, so it stops with the backfill.
204    async fn rebuild_and_hand_off(
205        &self,
206        pipeline: &mut Pipeline<N>,
207        targets: &mut watch::Receiver<B256>,
208        write: SnapWrite,
209        pivot: BlockNumHash,
210    ) -> Result<Pass, PipelineError> {
211        match self.with_handoff(move |handoff, stop| handoff.rebuild(write, stop)).await? {
212            RebuildOutcome::Rebuilt => self.hand_off(pipeline, targets, write, pivot).await,
213            RebuildOutcome::RootMismatch => Ok(Pass::RootMismatch),
214            RebuildOutcome::Stopped => Ok(Pass::Stopped),
215        }
216    }
217
218    // Catches headers up, since forkchoice may have moved during the rebuild, then publishes the
219    // state if its pivot is still canonical.
220    async fn hand_off(
221        &self,
222        pipeline: &mut Pipeline<N>,
223        targets: &mut watch::Receiver<B256>,
224        write: SnapWrite,
225        pivot: BlockNumHash,
226    ) -> Result<Pass, PipelineError> {
227        let headers = self.catch_up_headers(pipeline, targets).await?;
228        if headers != Pass::Done {
229            return Ok(headers)
230        }
231        Ok(match self.with_handoff(move |handoff, stop| handoff.hand_off(write, stop)).await? {
232            HandoffOutcome::Completed => Pass::Done,
233            HandoffOutcome::PivotReorged => {
234                info!(target: "sync::snap", ?pivot, "Snap pivot was reorged before the handoff, restarting");
235                Pass::Again
236            }
237            HandoffOutcome::RootMismatch => Pass::RootMismatch,
238            HandoffOutcome::Stopped => Pass::Stopped,
239        })
240    }
241
242    // Runs `f` on the blocking pool, since rebuilding and publishing read every account.
243    async fn with_handoff<T: Send + 'static>(
244        &self,
245        f: impl FnOnce(SnapHandoff<N>, &CancellationToken) -> Result<T, PipelineError> + Send + 'static,
246    ) -> Result<T, PipelineError> {
247        let handoff = SnapHandoff::new(self.factory.clone());
248        let stop = self.stop.clone();
249        self.runtime
250            .spawn_blocking(move || f(handoff, &stop))
251            .await
252            .map_err(|error| PipelineError::Internal(RethError::other(error)))?
253    }
254
255    // Runs every stage to the latest target, above the pivot once snap state is published.
256    async fn finish(
257        &self,
258        pipeline: &mut Pipeline<N>,
259        targets: &mut watch::Receiver<B256>,
260    ) -> Result<ControlFlow, PipelineError> {
261        let Some((target, ahead)) = self.resolve_target(targets).await? else { return Ok(STOPPED) };
262        // A stale target never completes the header stage, so the stages run to the local
263        // head and the engine follows forkchoice from there.
264        let target = if ahead { target } else { self.local_head()? };
265        let stages = pipeline.run_until(StageId::Finish, Some(PipelineTarget::Sync(target)));
266        self.stop.run_until_cancelled(stages).await.unwrap_or(Ok(STOPPED))
267    }
268
269    // Syncs headers to the latest target. A detached head unwinds headers short of the target,
270    // and a target that moved during the pass is not synced yet, so both need another pass.
271    async fn catch_up_headers(
272        &self,
273        pipeline: &mut Pipeline<N>,
274        targets: &mut watch::Receiver<B256>,
275    ) -> Result<Pass, PipelineError> {
276        let Some((target, ahead)) = self.resolve_target(targets).await? else {
277            return Ok(Pass::Stopped)
278        };
279        if !ahead {
280            debug!(target: "sync::snap", %target, "Forkchoice target is not above the local headers, skipping the header pass");
281            return Ok(Pass::Done)
282        }
283        // Snap needs canonical headers and their BAL hashes, but nothing below the pivot may
284        // execute, so only the header stage runs. Finish the pass before retargeting: dropping it
285        // retains partially downloaded headers in the stage's ETL collectors.
286        let headers = pipeline.run_until(StageId::Headers, Some(PipelineTarget::Sync(target)));
287        let Some(headers) = self.stop.run_until_cancelled(headers).await else {
288            return Ok(Pass::Stopped)
289        };
290        if headers?.is_unwind() || targets.has_changed().unwrap_or(false) {
291            return Ok(Pass::Again)
292        }
293        Ok(Pass::Done)
294    }
295
296    // Whether `target` is above the local headers, or `None` when no peer serves its header. The
297    // header stage only finishes once it downloads the block after the local head, so a pass to a
298    // sibling or an older block never completes.
299    async fn is_ahead(&self, target: B256) -> Result<Option<bool>, PipelineError> {
300        let head = {
301            let provider = self.factory.provider()?;
302            if provider.block_number(target)?.is_some() {
303                return Ok(Some(false))
304            }
305            provider.last_block_number()?
306        };
307        let header = self.client.get_header(target.into()).await;
308        // A peer may answer with another block's header.
309        Ok(header
310            .ok()
311            .and_then(|header| header.into_data())
312            .filter(|header| header.hash_slow() == target)
313            .map(|header| header.number() > head))
314    }
315
316    // Returns the hash of the highest local header.
317    fn local_head(&self) -> Result<B256, PipelineError> {
318        let provider = self.factory.provider()?;
319        let number = provider.last_block_number()?;
320        Ok(provider.block_hash(number)?.ok_or(ProviderError::HeaderNotFound(number.into()))?)
321    }
322
323    // Resolves once the refresh interval has passed and forkchoice has moved, or once the
324    // backfill is gone.
325    async fn refresh_due(&self, targets: &mut watch::Receiver<B256>) {
326        tokio::time::sleep(self.header_refresh).await;
327        let _ = targets.changed().await;
328    }
329
330    // Resolve the target before starting a pipeline pass, which cannot safely be retargeted with
331    // headers buffered. Only the peer lookup and retry delay are interrupted by forkchoice.
332    async fn resolve_target(
333        &self,
334        targets: &mut watch::Receiver<B256>,
335    ) -> Result<Option<(B256, bool)>, PipelineError> {
336        loop {
337            let target = *targets.borrow_and_update();
338            let ahead = tokio::select! {
339                biased;
340                () = self.stop.cancelled() => return Ok(None),
341                Ok(()) = targets.changed() => continue,
342                ahead = self.is_ahead(target) => ahead?,
343            };
344            if let Some(ahead) = ahead {
345                return Ok(Some((target, ahead)))
346            }
347            tokio::select! {
348                biased;
349                () = self.stop.cancelled() => return Ok(None),
350                Ok(()) = targets.changed() => {},
351                () = tokio::time::sleep(HEADER_LOOKUP_RETRY) => {},
352            }
353        }
354    }
355}
356
357// What a header or handoff pass leaves the run to do.
358#[derive(Debug, PartialEq, Eq)]
359enum Pass {
360    // The backfill stopped, so the run returns without progress.
361    Stopped,
362    // Headers moved or the pivot was reorged out, so the run starts over from headers.
363    Again,
364    // The rebuilt root differs from the pivot's header, so the attempt was abandoned and the run
365    // starts over from headers.
366    RootMismatch,
367    // The pass finished, so the run moves on to its next step.
368    Done,
369}
370
371#[cfg(test)]
372pub(crate) mod tests {
373    use super::*;
374    use crate::snap::{
375        handoff::tests::{downloaded_attempt, PIVOT},
376        tests::{
377            hashed_factory, headers_done, headers_reach, next_target, pipeline, pipeline_on,
378            pipeline_with, target,
379        },
380    };
381    use alloy_consensus::Header;
382    use alloy_eips::eip1898::BlockWithParent;
383    use futures::future::{pending, ready, Either, Pending, Ready};
384    use reth_consensus::ConsensusError;
385    use reth_db::{tables, transaction::DbTxMut};
386    use reth_eth_wire_types::snap::{
387        GetAccountRangeMessage, GetBlockAccessListsMessage, GetByteCodesMessage,
388        GetStorageRangesMessage,
389    };
390    use reth_network_p2p::{
391        download::DownloadClient, error::PeerRequestResult, headers::client::HeadersRequest,
392        priority::Priority, NoopFullBlockClient,
393    };
394    use reth_network_peers::{PeerId, WithPeerId};
395    use reth_primitives_traits::{Account, SealedHeader};
396    use reth_provider::{
397        test_utils::{insert_headers, MockNodeTypesWithDB},
398        DBProvider, DatabaseProviderFactory, HeaderProvider, MetadataProvider,
399    };
400    use reth_snap_sync::{SnapAttemptStore, SnapStateVerifier, DEFAULT_SCAN_CHUNK};
401    use reth_stages::{
402        ExecInput, ExecOutput, Stage, StageError, StageSetBuilder, UnwindInput, UnwindOutput,
403    };
404    use reth_stages_api::test_utils::TestStage;
405    use std::{
406        sync::{
407            atomic::{AtomicBool, Ordering},
408            Arc, Mutex,
409        },
410        task::{Context, Poll},
411    };
412
413    // Answers a header request with the requested one of `headers`, or with all of them when none
414    // matches, as a faulty peer would. Snap requests fail like the noop client's.
415    #[derive(Clone, Debug)]
416    pub(crate) struct ServesHeaders {
417        headers: Vec<Header>,
418        noop: NoopFullBlockClient,
419        empty_response: Arc<AtomicBool>,
420        pending_target: Option<B256>,
421    }
422
423    impl Default for ServesHeaders {
424        fn default() -> Self {
425            Self {
426                headers: (100..=101)
427                    .map(|number| Header { number, ..Default::default() })
428                    .collect(),
429                noop: NoopFullBlockClient::default(),
430                empty_response: Arc::default(),
431                pending_target: None,
432            }
433        }
434    }
435
436    impl DownloadClient for ServesHeaders {
437        fn report_bad_message(&self, _peer_id: PeerId) {}
438
439        fn num_connected_peers(&self) -> usize {
440            1
441        }
442    }
443
444    impl HeadersClient for ServesHeaders {
445        type Header = Header;
446        type Output =
447            Either<Ready<PeerRequestResult<Vec<Header>>>, Pending<PeerRequestResult<Vec<Header>>>>;
448
449        fn get_headers_with_priority(
450            &self,
451            request: HeadersRequest,
452            _priority: Priority,
453        ) -> Self::Output {
454            if self.pending_target.is_some_and(|hash| request.start == hash.into()) {
455                return Either::Right(pending())
456            }
457            if self.empty_response.swap(false, Ordering::Relaxed) {
458                return Either::Left(ready(Ok(WithPeerId::new(PeerId::random(), vec![]))))
459            }
460            let requested = self
461                .headers
462                .iter()
463                .find(|header| request.start == header.hash_slow().into())
464                .map_or_else(|| self.headers.clone(), |header| vec![header.clone()]);
465            Either::Left(ready(Ok(WithPeerId::new(PeerId::random(), requested))))
466        }
467    }
468
469    impl SnapClient for ServesHeaders {
470        type Output = <NoopFullBlockClient as SnapClient>::Output;
471
472        fn get_account_range_with_priority(
473            &self,
474            request: GetAccountRangeMessage,
475            priority: Priority,
476        ) -> Self::Output {
477            self.noop.get_account_range_with_priority(request, priority)
478        }
479
480        fn get_storage_ranges(&self, request: GetStorageRangesMessage) -> Self::Output {
481            self.noop.get_storage_ranges(request)
482        }
483
484        fn get_storage_ranges_with_priority(
485            &self,
486            request: GetStorageRangesMessage,
487            priority: Priority,
488        ) -> Self::Output {
489            self.noop.get_storage_ranges_with_priority(request, priority)
490        }
491
492        fn get_byte_codes(&self, request: GetByteCodesMessage) -> Self::Output {
493            self.noop.get_byte_codes(request)
494        }
495
496        fn get_byte_codes_with_priority(
497            &self,
498            request: GetByteCodesMessage,
499            priority: Priority,
500        ) -> Self::Output {
501            self.noop.get_byte_codes_with_priority(request, priority)
502        }
503
504        fn get_block_access_lists_with_priority(
505            &self,
506            request: GetBlockAccessListsMessage,
507            priority: Priority,
508        ) -> Self::Output {
509            self.noop.get_block_access_lists_with_priority(request, priority)
510        }
511    }
512
513    // A header stage that waits for `gate`, then records whether a snap attempt exists at each
514    // pass.
515    #[derive(Debug)]
516    struct GatedHeaders {
517        gate: watch::Receiver<bool>,
518        attempts: Arc<Mutex<Vec<bool>>>,
519        stage: TestStage,
520    }
521
522    impl<Provider: MetadataProvider> Stage<Provider> for GatedHeaders {
523        fn id(&self) -> StageId {
524            StageId::Headers
525        }
526
527        fn poll_execute_ready(
528            &mut self,
529            cx: &mut Context<'_>,
530            _input: ExecInput,
531        ) -> Poll<Result<(), StageError>> {
532            if *self.gate.borrow() {
533                return Poll::Ready(Ok(()))
534            }
535            cx.waker().wake_by_ref();
536            Poll::Pending
537        }
538
539        fn execute(
540            &mut self,
541            provider: &Provider,
542            input: ExecInput,
543        ) -> Result<ExecOutput, StageError> {
544            self.attempts.lock().unwrap().push(provider.snap_attempt()?.is_some());
545            self.stage.execute(provider, input)
546        }
547
548        fn unwind(
549            &mut self,
550            _provider: &Provider,
551            _input: UnwindInput,
552        ) -> Result<UnwindOutput, StageError> {
553            unreachable!("nothing unwinds in this test")
554        }
555    }
556
557    // A run over `factory` that refreshes headers as soon as forkchoice moves.
558    fn snap_run(
559        factory: &ProviderFactory<MockNodeTypesWithDB>,
560    ) -> (SnapRun<MockNodeTypesWithDB, ServesHeaders>, CancellationToken) {
561        snap_run_with(factory, ServesHeaders::default())
562    }
563
564    // A run over `factory` requesting from `client` that refreshes headers as soon as forkchoice
565    // moves.
566    fn snap_run_with<C>(
567        factory: &ProviderFactory<MockNodeTypesWithDB>,
568        client: C,
569    ) -> (SnapRun<MockNodeTypesWithDB, C>, CancellationToken) {
570        let stop = CancellationToken::new();
571        let run = SnapRun::new(
572            client,
573            factory.clone(),
574            Runtime::test(),
575            stop.clone(),
576            watch::channel(B256::ZERO).1,
577        )
578        .with_header_refresh(Duration::ZERO);
579        (run, stop)
580    }
581
582    // A downloaded attempt with its trie rebuild started, and a pipeline over the same database
583    // running `headers`.
584    fn handoff_ready(
585        headers: TestStage,
586    ) -> (
587        Pipeline<MockNodeTypesWithDB>,
588        ProviderFactory<MockNodeTypesWithDB>,
589        SnapWrite,
590        BlockNumHash,
591    ) {
592        let (factory, write) = downloaded_attempt();
593        let pivot = factory.sealed_header(PIVOT).unwrap().unwrap().num_hash();
594        let provider = factory.database_provider_rw().unwrap();
595        provider.start_trie_rebuild(write, DEFAULT_SCAN_CHUNK, &CancellationToken::new()).unwrap();
596        provider.commit().unwrap();
597        let pipeline = pipeline_on(
598            &factory,
599            StageSetBuilder::default().add_stage(headers),
600            watch::channel(B256::ZERO).0,
601        );
602        (pipeline, factory, write, pivot)
603    }
604
605    // A header stage error that unwinds the pipeline; only the unwind matters here.
606    fn detached_head() -> StageError {
607        StageError::DetachedHead {
608            local_head: Box::new(BlockWithParent {
609                parent: B256::ZERO,
610                block: BlockNumHash::default(),
611            }),
612            header: Box::new(BlockWithParent {
613                parent: B256::ZERO,
614                block: BlockNumHash::default(),
615            }),
616            error: Box::new(ConsensusError::BaseFeeMissing),
617        }
618    }
619
620    #[tokio::test]
621    async fn a_new_target_refreshes_headers_then_resumes_until_stopped() {
622        let (pipeline, factory) = pipeline(
623            TestStage::new(StageId::Headers).add_exec(headers_done(0)).add_exec(headers_done(1)),
624        );
625        let (targets, receiver) = watch::channel(target());
626        let (run, stop) = snap_run(&factory);
627        let mut events = pipeline.events();
628        let run = tokio::spawn(run.run(pipeline, receiver));
629        headers_reach(&mut events, 0).await;
630
631        targets.send(next_target()).unwrap();
632
633        // The bootstrap stopped, headers caught up to the new target and a new bootstrap resumed.
634        headers_reach(&mut events, 1).await;
635        assert!(!run.is_finished());
636
637        stop.cancel();
638        let (_pipeline, result) = run.await.unwrap();
639        assert!(matches!(result, Ok(ControlFlow::NoProgress { block_number: None })));
640    }
641
642    #[tokio::test]
643    async fn a_target_moved_during_headers_is_synced_before_the_bootstrap() {
644        let (open, gate) = watch::channel(false);
645        let attempts = Arc::new(Mutex::new(Vec::new()));
646        let headers = GatedHeaders {
647            gate,
648            attempts: attempts.clone(),
649            stage: TestStage::new(StageId::Headers)
650                .add_exec(headers_done(1))
651                .add_exec(headers_done(2)),
652        };
653        let (tip, _) = watch::channel(B256::ZERO);
654        let mut pipeline_tip = tip.subscribe();
655        let (pipeline, factory) = pipeline_with(headers, tip);
656        // Headers with block access list hashes, so a bootstrap can anchor a pivot.
657        let mut parent = B256::ZERO;
658        let stored: Vec<_> = (0..=64)
659            .map(|number| {
660                let header = SealedHeader::seal_slow(Header {
661                    number,
662                    parent_hash: parent,
663                    block_access_list_hash: Some(B256::ZERO),
664                    ..Default::default()
665                });
666                parent = header.hash();
667                header
668            })
669            .collect();
670        insert_headers(&factory, &stored);
671        // Targets above the stored headers that peers serve, so each pass runs to completion.
672        let target = Header { number: 65, parent_hash: parent, ..Default::default() };
673        let next = Header { number: 66, parent_hash: target.hash_slow(), ..Default::default() };
674        let (target_hash, next_hash) = (target.hash_slow(), next.hash_slow());
675        let client = ServesHeaders { headers: vec![target, next], ..Default::default() };
676        let (targets, receiver) = watch::channel(target_hash);
677        let (run, stop) = snap_run_with(&factory, client);
678        let mut events = pipeline.events();
679        let run = tokio::spawn(run.run(pipeline, receiver));
680
681        // Forkchoice moves while the first header pass is still downloading.
682        pipeline_tip.wait_for(|tip| *tip == target_hash).await.unwrap();
683        targets.send(next_hash).unwrap();
684        open.send(true).unwrap();
685
686        tokio::time::timeout(Duration::from_secs(5), headers_reach(&mut events, 2))
687            .await
688            .expect("headers are synced to the moved target");
689        stop.cancel();
690        let (_pipeline, result) = run.await.unwrap();
691        assert!(matches!(result, Ok(ControlFlow::NoProgress { block_number: None })));
692        // No bootstrap recorded an attempt on the outdated headers between the two passes.
693        assert_eq!(*attempts.lock().unwrap(), [false, false]);
694    }
695
696    #[tokio::test]
697    async fn a_detached_head_syncs_headers_again_before_the_bootstrap() {
698        let (pipeline, factory) = pipeline(
699            TestStage::new(StageId::Headers)
700                .add_exec(Err(detached_head()))
701                .add_exec(headers_done(2)),
702        );
703        let (_targets, receiver) = watch::channel(target());
704        let (run, stop) = snap_run(&factory);
705        let mut events = pipeline.events();
706        let run = tokio::spawn(run.run(pipeline, receiver));
707
708        // Forkchoice never moves, so only the unwind itself can trigger the second header run.
709        tokio::time::timeout(Duration::from_secs(5), headers_reach(&mut events, 2))
710            .await
711            .expect("headers are synced again after the unwind");
712        stop.cancel();
713        let (_pipeline, result) = run.await.unwrap();
714        assert!(matches!(result, Ok(ControlFlow::NoProgress { block_number: None })));
715    }
716
717    #[tokio::test]
718    async fn the_pipeline_finishes_to_the_latest_target() {
719        let (tip, _) = watch::channel(B256::ZERO);
720        let pipeline_tip = tip.subscribe();
721        let (mut pipeline, factory) =
722            pipeline_with(TestStage::new(StageId::Finish).add_exec(headers_done(1)), tip);
723        let (targets, mut receiver) = watch::channel(target());
724        let (run, _stop) = snap_run(&factory);
725
726        // Forkchoice moved while the state was downloading.
727        targets.send(next_target()).unwrap();
728        run.finish(&mut pipeline, &mut receiver).await.unwrap();
729
730        assert_eq!(*pipeline_tip.borrow(), next_target());
731    }
732
733    #[tokio::test]
734    async fn a_moved_target_syncs_headers_before_the_handoff() {
735        let (mut pipeline, factory, write, pivot) =
736            handoff_ready(TestStage::new(StageId::Headers).add_exec(headers_done(PIVOT)));
737        let (targets, mut receiver) = watch::channel(target());
738        let (run, _stop) = snap_run(&factory);
739
740        // Forkchoice moved while the trie was rebuilding.
741        targets.send(next_target()).unwrap();
742        let pass =
743            run.rebuild_and_hand_off(&mut pipeline, &mut receiver, write, pivot).await.unwrap();
744
745        assert_eq!(pass, Pass::Done);
746        assert!(factory.provider().unwrap().snap_attempt().unwrap().unwrap().is_verified());
747    }
748
749    #[tokio::test]
750    async fn a_reorg_during_the_rebuild_syncs_headers_again_without_a_handoff() {
751        let (mut pipeline, factory, write, pivot) =
752            handoff_ready(TestStage::new(StageId::Headers).add_exec(Err(detached_head())));
753        let (_targets, mut receiver) = watch::channel(target());
754        let (run, _stop) = snap_run(&factory);
755
756        let pass =
757            run.rebuild_and_hand_off(&mut pipeline, &mut receiver, write, pivot).await.unwrap();
758
759        assert_eq!(pass, Pass::Again);
760        assert!(!factory.provider().unwrap().snap_attempt().unwrap().unwrap().is_verified());
761    }
762
763    #[tokio::test]
764    async fn a_root_mismatch_starts_a_new_attempt_instead_of_failing() {
765        let (mut pipeline, factory, write, pivot) = handoff_ready(TestStage::new(StageId::Headers));
766        // State whose root differs from the one the pivot header commits to.
767        let provider = factory.database_provider_rw().unwrap();
768        provider
769            .tx_ref()
770            .put::<tables::HashedAccounts>(
771                B256::repeat_byte(3),
772                Account { nonce: 1, ..Default::default() },
773            )
774            .unwrap();
775        provider.commit().unwrap();
776        let (_targets, mut receiver) = watch::channel(target());
777        let (run, _stop) = snap_run(&factory);
778
779        let pass =
780            run.rebuild_and_hand_off(&mut pipeline, &mut receiver, write, pivot).await.unwrap();
781
782        assert_eq!(pass, Pass::RootMismatch);
783        assert!(factory.provider().unwrap().active_snap_write().unwrap().is_none());
784    }
785
786    #[tokio::test]
787    async fn repeated_root_mismatches_fail_the_run() {
788        let (mut pipeline, factory, _, _) =
789            handoff_ready(TestStage::new(StageId::Headers).add_exec(headers_done(PIVOT)));
790        // State whose root differs from the one the pivot header commits to.
791        let provider = factory.database_provider_rw().unwrap();
792        provider
793            .tx_ref()
794            .put::<tables::HashedAccounts>(
795                B256::repeat_byte(3),
796                Account { nonce: 1, ..Default::default() },
797            )
798            .unwrap();
799        provider.commit().unwrap();
800        let (_targets, receiver) = watch::channel(target());
801        let (run, _stop) = snap_run(&factory);
802
803        let result = run.with_root_mismatch_retries(0).bootstrap(&mut pipeline, receiver).await;
804
805        assert!(matches!(result, Err(PipelineError::Internal(_))), "{result:?}");
806        assert!(factory.provider().unwrap().active_snap_write().unwrap().is_none());
807    }
808
809    #[tokio::test]
810    async fn a_chain_before_block_access_lists_syncs_with_the_staged_pipeline() {
811        let factory = hashed_factory();
812        let stages = StageSetBuilder::default()
813            .add_stage(
814                TestStage::new(StageId::Headers)
815                    .add_exec(headers_done(0))
816                    .add_exec(headers_done(0)),
817            )
818            .add_stage(TestStage::new(StageId::Finish).add_exec(headers_done(0)));
819        let mut pipeline = pipeline_on(&factory, stages, watch::channel(B256::ZERO).0);
820        insert_headers(&factory, &[SealedHeader::seal_slow(Header::default())]);
821        let (_targets, receiver) = watch::channel(target());
822        let (run, _stop) = snap_run(&factory);
823
824        // The second header run is the staged pipeline finishing to the target.
825        let result = run.bootstrap(&mut pipeline, receiver).await.unwrap();
826
827        assert_ne!(result, STOPPED);
828        assert!(factory.provider().unwrap().snap_attempt().unwrap().is_none());
829    }
830
831    #[tokio::test]
832    async fn a_sibling_of_the_local_head_skips_the_header_pass() {
833        // A header stage with no outputs, so running it fails the test.
834        let (mut pipeline, factory) =
835            pipeline_with(TestStage::new(StageId::Headers), watch::channel(B256::ZERO).0);
836        let (run, _, mut receiver) = sibling_of_head(&factory);
837
838        let pass = run.catch_up_headers(&mut pipeline, &mut receiver).await.unwrap();
839
840        assert_eq!(pass, Pass::Done);
841    }
842
843    #[tokio::test]
844    async fn a_sibling_of_the_local_head_finishes_to_the_local_head() {
845        let factory = hashed_factory();
846        let stages = StageSetBuilder::default()
847            .add_stage(TestStage::new(StageId::Headers).add_exec(headers_done(1)))
848            .add_stage(TestStage::new(StageId::Finish).add_exec(headers_done(1)));
849        let (tip, pipeline_tip) = watch::channel(B256::ZERO);
850        let mut pipeline = pipeline_on(&factory, stages, tip);
851        let (run, head, mut receiver) = sibling_of_head(&factory);
852
853        let result = run.finish(&mut pipeline, &mut receiver).await.unwrap();
854
855        assert_ne!(result, STOPPED);
856        assert_eq!(*pipeline_tip.borrow(), head);
857    }
858
859    #[tokio::test]
860    async fn a_header_other_than_the_target_leaves_it_unresolved() {
861        let factory = hashed_factory();
862        let genesis = Header::default();
863        insert_headers(&factory, &[SealedHeader::seal_slow(genesis.clone())]);
864        // The peer answers with the genesis header, which isn't the requested target.
865        let client = ServesHeaders { headers: vec![genesis], ..Default::default() };
866        let (run, _stop) = snap_run_with(&factory, client);
867
868        assert_eq!(run.is_ahead(target()).await.unwrap(), None);
869    }
870
871    #[tokio::test]
872    async fn a_new_target_replaces_an_unresolved_one_before_headers_start() {
873        let (tip, pipeline_tip) = watch::channel(B256::ZERO);
874        let (mut pipeline, factory) =
875            pipeline_with(TestStage::new(StageId::Headers).add_exec(headers_done(1)), tip);
876        let (targets, mut receiver) = watch::channel(B256::repeat_byte(0xff));
877        let (run, _stop) = snap_run(&factory);
878        let mut pass = pin!(run.catch_up_headers(&mut pipeline, &mut receiver));
879
880        assert!(futures::poll!(&mut pass).is_pending());
881        assert_eq!(*pipeline_tip.borrow(), B256::ZERO);
882        targets.send(next_target()).unwrap();
883
884        let pass = tokio::time::timeout(Duration::from_secs(5), pass)
885            .await
886            .expect("the new target resolves without starting the old header pass");
887        assert_eq!(pass.unwrap(), Pass::Done);
888        assert_eq!(*pipeline_tip.borrow(), next_target());
889    }
890
891    #[tokio::test]
892    async fn a_new_target_replaces_an_unresolved_one_before_finishing() {
893        let factory = hashed_factory();
894        let stages = StageSetBuilder::default()
895            .add_stage(TestStage::new(StageId::Headers).add_exec(headers_done(1)))
896            .add_stage(TestStage::new(StageId::Finish).add_exec(headers_done(1)));
897        let (tip, pipeline_tip) = watch::channel(B256::ZERO);
898        let mut pipeline = pipeline_on(&factory, stages, tip);
899        let (targets, mut receiver) = watch::channel(B256::repeat_byte(0xff));
900        let (run, _stop) = snap_run(&factory);
901        let mut finished = pin!(run.finish(&mut pipeline, &mut receiver));
902
903        assert!(futures::poll!(&mut finished).is_pending());
904        assert_eq!(*pipeline_tip.borrow(), B256::ZERO);
905        targets.send(next_target()).unwrap();
906
907        let result = tokio::time::timeout(Duration::from_secs(5), finished)
908            .await
909            .expect("the new target resolves before the pipeline starts");
910        assert_ne!(result.unwrap(), STOPPED);
911        assert_eq!(*pipeline_tip.borrow(), next_target());
912    }
913
914    #[tokio::test]
915    async fn a_new_target_interrupts_a_pending_header_lookup() {
916        let (tip, pipeline_tip) = watch::channel(B256::ZERO);
917        let (mut pipeline, factory) =
918            pipeline_with(TestStage::new(StageId::Headers).add_exec(headers_done(1)), tip);
919        let (targets, mut receiver) = watch::channel(target());
920        let client = ServesHeaders { pending_target: Some(target()), ..Default::default() };
921        let (run, _stop) = snap_run_with(&factory, client);
922        let mut pass = pin!(run.catch_up_headers(&mut pipeline, &mut receiver));
923
924        assert!(futures::poll!(&mut pass).is_pending());
925        assert_eq!(*pipeline_tip.borrow(), B256::ZERO);
926        targets.send(next_target()).unwrap();
927
928        let pass = tokio::time::timeout(Duration::from_secs(5), pass)
929            .await
930            .expect("forkchoice interrupts the pending peer request");
931        assert_eq!(pass.unwrap(), Pass::Done);
932        assert_eq!(*pipeline_tip.borrow(), next_target());
933    }
934
935    #[tokio::test]
936    async fn an_empty_response_retries_a_stale_target_without_starting_headers() {
937        let (tip, pipeline_tip) = watch::channel(B256::ZERO);
938        let (mut pipeline, factory) = pipeline_with(TestStage::new(StageId::Headers), tip);
939        let (run, _, mut receiver) = sibling_of_head(&factory);
940        run.client.empty_response.store(true, Ordering::Relaxed);
941        let mut pass = pin!(run.catch_up_headers(&mut pipeline, &mut receiver));
942
943        assert!(futures::poll!(&mut pass).is_pending());
944        assert_eq!(*pipeline_tip.borrow(), B256::ZERO);
945        let pass = tokio::time::timeout(Duration::from_secs(5), pass)
946            .await
947            .expect("the same target is retried without another forkchoice update");
948
949        assert_eq!(pass.unwrap(), Pass::Done);
950        assert_eq!(*pipeline_tip.borrow(), B256::ZERO);
951    }
952
953    #[tokio::test]
954    async fn stopping_an_unresolved_target_leaves_headers_untouched() {
955        let (tip, pipeline_tip) = watch::channel(B256::ZERO);
956        let (mut pipeline, factory) = pipeline_with(TestStage::new(StageId::Headers), tip);
957        let (_targets, mut receiver) = watch::channel(target());
958        let client = ServesHeaders { pending_target: Some(target()), ..Default::default() };
959        let (run, stop) = snap_run_with(&factory, client);
960        let mut pass = pin!(run.catch_up_headers(&mut pipeline, &mut receiver));
961
962        assert!(futures::poll!(&mut pass).is_pending());
963        stop.cancel();
964
965        assert_eq!(pass.await.unwrap(), Pass::Stopped);
966        assert_eq!(*pipeline_tip.borrow(), B256::ZERO);
967    }
968
969    // Headers through block 1 in `factory`, and a run whose forkchoice target is a sibling of
970    // block 1 that peers serve. Returns the run, the local head's hash and the target receiver.
971    fn sibling_of_head(
972        factory: &ProviderFactory<MockNodeTypesWithDB>,
973    ) -> (SnapRun<MockNodeTypesWithDB, ServesHeaders>, B256, watch::Receiver<B256>) {
974        let genesis = SealedHeader::seal_slow(Header::default());
975        let head = SealedHeader::seal_slow(Header {
976            number: 1,
977            parent_hash: genesis.hash(),
978            ..Default::default()
979        });
980        insert_headers(factory, &[genesis.clone(), head.clone()]);
981        let sibling =
982            Header { number: 1, parent_hash: genesis.hash(), gas_limit: 1, ..Default::default() };
983        let receiver = watch::channel(sibling.hash_slow()).1;
984        let client = ServesHeaders { headers: vec![sibling], ..Default::default() };
985        (snap_run_with(factory, client).0, head.hash(), receiver)
986    }
987}
reth_node_ethereum/snap/run.rs

reth_node_ethereum/snap/
run.rs
