reth_stages_api/pipeline/
mod.rs

1mod ctrl;
2mod event;
3pub use crate::pipeline::ctrl::ControlFlow;
4use crate::{PipelineTarget, StageCheckpoint, StageId};
5use alloy_primitives::{BlockNumber, B256};
6pub use event::*;
7use futures_util::Future;
8use reth_primitives_traits::constants::BEACON_CONSENSUS_REORG_UNWIND_DEPTH;
9use reth_provider::{
10    providers::ProviderNodeTypes, BlockHashReader, BlockNumReader, ChainStateBlockReader,
11    ChainStateBlockWriter, DBProvider, DatabaseProviderFactory, MetadataProvider, MetadataWriter,
12    PipelineUnwind, ProviderError, ProviderFactory, PruneCheckpointReader, StageCheckpointReader,
13    StageCheckpointWriter, StaticFileProviderFactory, StorageSettingsCache,
14};
15use reth_prune::PrunerBuilder;
16use reth_static_file::StaticFileProducer;
17use reth_tokio_util::{EventSender, EventStream};
18use std::{
19    pin::Pin,
20    time::{Duration, Instant},
21};
22use tokio::sync::watch;
23use tracing::*;
24
25mod builder;
26mod progress;
27mod set;
28
29use crate::{
30    BlockErrorKind, ExecInput, ExecOutput, MetricEvent, MetricEventsSender, PipelineError, Stage,
31    StageError, StageExt, UnwindInput,
32};
33pub use builder::*;
34use progress::*;
35use reth_errors::RethResult;
36pub use set::*;
37
38/// A container for a queued stage.
39pub(crate) type BoxedStage<DB> = Box<dyn Stage<DB>>;
40
41/// The future that returns the owned pipeline and the result of the pipeline run. See
42/// [`Pipeline::run_as_fut`].
43pub type PipelineFut<N> = Pin<Box<dyn Future<Output = PipelineWithResult<N>> + Send>>;
44
45/// The pipeline type itself with the result of [`Pipeline::run_as_fut`]
46pub type PipelineWithResult<N> = (Pipeline<N>, Result<ControlFlow, PipelineError>);
47
48#[cfg_attr(doc, aquamarine::aquamarine)]
49/// A staged sync pipeline.
50///
51/// The pipeline executes queued [stages][Stage] serially. An external component determines the tip
52/// of the chain and the pipeline then executes each stage in order from the current local chain tip
53/// and the external chain tip. When a stage is executed, it will run until it reaches the chain
54/// tip.
55///
56/// After the entire pipeline has been run, it will run again unless asked to stop (see
57/// [`Pipeline::set_max_block`]).
58///
59/// `include_mmd!("docs/mermaid/pipeline.mmd`")
60///
61/// # Unwinding
62///
63/// In case of a validation error (as determined by the consensus engine) in one of the stages, the
64/// pipeline will unwind the stages in reverse order of execution. It is also possible to
65/// request an unwind manually (see [`Pipeline::unwind`]).
66///
67/// # Defaults
68///
69/// The [`DefaultStages`](crate::sets::DefaultStages) are used to fully sync reth.
70pub struct Pipeline<N: ProviderNodeTypes> {
71    /// Provider factory.
72    provider_factory: ProviderFactory<N>,
73    /// All configured stages in the order they will be executed.
74    stages: Vec<BoxedStage<<ProviderFactory<N> as DatabaseProviderFactory>::ProviderRW>>,
75    /// The maximum block number to sync to.
76    max_block: Option<BlockNumber>,
77    static_file_producer: StaticFileProducer<ProviderFactory<N>>,
78    /// Sender for events the pipeline emits.
79    event_sender: EventSender<PipelineEvent>,
80    /// Keeps track of the progress of the pipeline.
81    progress: PipelineProgress,
82    /// A Sender for the current chain tip to sync to.
83    ///
84    /// This is used to notify the headers stage about a new sync target.
85    tip_tx: Option<watch::Sender<B256>>,
86    metrics_tx: Option<MetricEventsSender>,
87    /// Whether an unwind should fail the syncing process. Should only be set when downloading
88    /// blocks from trusted sources and expecting them to be valid.
89    fail_on_unwind: bool,
90    /// Block that was chosen as a target of the last unwind triggered by
91    /// [`StageError::DetachedHead`] error.
92    last_detached_head_unwind_target: Option<B256>,
93    /// Number of consecutive unwind attempts due to [`StageError::DetachedHead`] for the current
94    /// fork.
95    detached_head_attempts: u64,
96}
97
98impl<N: ProviderNodeTypes> Pipeline<N> {
99    /// Construct a pipeline using a [`PipelineBuilder`].
100    pub fn builder() -> PipelineBuilder<<ProviderFactory<N> as DatabaseProviderFactory>::ProviderRW>
101    {
102        PipelineBuilder::default()
103    }
104
105    /// Return the minimum block number achieved by
106    /// any stage during the execution of the pipeline.
107    pub const fn minimum_block_number(&self) -> Option<u64> {
108        self.progress.minimum_block_number
109    }
110
111    /// Set tip for reverse sync.
112    #[track_caller]
113    pub fn set_tip(&self, tip: B256) {
114        let _ = self.tip_tx.as_ref().expect("tip sender is set").send(tip).map_err(|_| {
115            warn!(target: "sync::pipeline", "Chain tip channel closed");
116        });
117    }
118
119    /// Listen for events on the pipeline.
120    pub fn events(&self) -> EventStream<PipelineEvent> {
121        self.event_sender.new_listener()
122    }
123
124    /// Get a mutable reference to a stage by index.
125    pub fn stage(
126        &mut self,
127        idx: usize,
128    ) -> &mut dyn Stage<<ProviderFactory<N> as DatabaseProviderFactory>::ProviderRW> {
129        &mut self.stages[idx]
130    }
131}
132
133impl<N: ProviderNodeTypes> Pipeline<N> {
134    /// Registers progress metrics for each registered stage
135    pub fn register_metrics(&mut self) -> Result<(), PipelineError> {
136        let Some(metrics_tx) = &mut self.metrics_tx else { return Ok(()) };
137        let provider = self.provider_factory.provider()?;
138
139        for stage in &self.stages {
140            let stage_id = stage.id();
141            let _ = metrics_tx.send(MetricEvent::StageCheckpoint {
142                stage_id,
143                checkpoint: provider.get_stage_checkpoint(stage_id)?.unwrap_or_default(),
144                max_block_number: None,
145                elapsed: Duration::default(),
146            });
147        }
148        Ok(())
149    }
150
151    /// Consume the pipeline and run it until it reaches the provided tip, if set. Return the
152    /// pipeline and its result as a future.
153    #[track_caller]
154    pub fn run_as_fut(mut self, target: Option<PipelineTarget>) -> PipelineFut<N> {
155        let _ = self.register_metrics();
156        Box::pin(async move {
157            // NOTE: the tip should only be None if we are in continuous sync mode.
158            if let Some(target) = target {
159                match target {
160                    PipelineTarget::Sync(tip) => self.set_tip(tip),
161                    PipelineTarget::Unwind(target) => {
162                        if let Err(err) = self.move_to_static_files() {
163                            return (self, Err(err.into()))
164                        }
165                        if let Err(err) = self.unwind(target, None) {
166                            return (self, Err(err))
167                        }
168                        self.progress.update(target);
169
170                        return (self, Ok(ControlFlow::Continue { block_number: target }))
171                    }
172                }
173            }
174
175            let result = self.run_loop().await;
176            trace!(target: "sync::pipeline", ?target, ?result, "Pipeline finished");
177            (self, result)
178        })
179    }
180
181    /// Run the pipeline in an infinite loop. Will terminate early if the user has specified
182    /// a `max_block` in the pipeline.
183    pub async fn run(&mut self) -> Result<(), PipelineError> {
184        let _ = self.register_metrics(); // ignore error
185
186        loop {
187            let next_action = self.run_loop().await?;
188
189            if next_action.is_unwind() && self.fail_on_unwind {
190                return Err(PipelineError::UnexpectedUnwind)
191            }
192
193            // Terminate the loop early if it's reached the maximum user
194            // configured block.
195            if next_action.should_continue() &&
196                self.progress
197                    .minimum_block_number
198                    .zip(self.max_block)
199                    .is_some_and(|(progress, target)| progress >= target)
200            {
201                trace!(
202                    target: "sync::pipeline",
203                    ?next_action,
204                    minimum_block_number = ?self.progress.minimum_block_number,
205                    max_block = ?self.max_block,
206                    "Terminating pipeline."
207                );
208                return Ok(())
209            }
210        }
211    }
212
213    /// Performs one pass of the pipeline across all stages. After successful
214    /// execution of each stage, it proceeds to commit it to the database.
215    ///
216    /// If any stage is unsuccessful at execution, we proceed to
217    /// unwind. This will undo the progress across the entire pipeline
218    /// up to the block that caused the error.
219    ///
220    /// Returns the control flow after it ran the pipeline.
221    /// This will be [`ControlFlow::Continue`] or [`ControlFlow::NoProgress`] of the _last_ stage in
222    /// the pipeline (for example the `Finish` stage). Or [`ControlFlow::Unwind`] of the stage
223    /// that caused the unwind.
224    pub async fn run_loop(&mut self) -> Result<ControlFlow, PipelineError> {
225        self.run_loop_until(None).await
226    }
227
228    /// Runs one pipeline pass through `last_stage`, including that stage.
229    ///
230    /// Each stage in the selected prefix runs to completion; later stages are not executed.
231    /// A [`PipelineTarget::Sync`] updates the tip before running the prefix, while a
232    /// [`PipelineTarget::Unwind`] unwinds the entire pipeline and returns without executing stages.
233    /// Stage errors that request an unwind also unwind the entire pipeline. Normal static-file
234    /// migration and pruning still run.
235    ///
236    /// Returns the control flow of the partial pass, or [`PipelineError::MissingStage`] if
237    /// `last_stage` is absent from the pipeline.
238    pub async fn run_until(
239        &mut self,
240        last_stage: StageId,
241        target: Option<PipelineTarget>,
242    ) -> Result<ControlFlow, PipelineError> {
243        let _ = self.register_metrics();
244        if !self.stages.iter().any(|stage| stage.id() == last_stage) {
245            return Err(PipelineError::MissingStage(last_stage))
246        }
247
248        match target {
249            Some(PipelineTarget::Sync(tip)) => self.set_tip(tip),
250            // An unwind rewinds every stage, so stopping part way through has no meaning.
251            Some(PipelineTarget::Unwind(target)) => {
252                self.move_to_static_files()?;
253                self.unwind(target, None)?;
254                self.progress.update(target);
255                return Ok(ControlFlow::Continue { block_number: target })
256            }
257            None => {}
258        }
259
260        self.run_loop_until(Some(last_stage)).await
261    }
262
263    // Shared stage loop. `last_stage` stops the pass once that stage has been executed.
264    async fn run_loop_until(
265        &mut self,
266        last_stage: Option<StageId>,
267    ) -> Result<ControlFlow, PipelineError> {
268        self.resume_unwind()?;
269        self.move_to_static_files()?;
270
271        let mut previous_stage = None;
272        for stage_index in 0..self.stages.len() {
273            let stage = &self.stages[stage_index];
274            let stage_id = stage.id();
275
276            trace!(target: "sync::pipeline", stage = %stage_id, "Executing stage");
277            let next = self.execute_stage_to_completion(previous_stage, stage_index).await?;
278
279            trace!(target: "sync::pipeline", stage = %stage_id, ?next, "Completed stage");
280
281            match next {
282                ControlFlow::NoProgress { block_number } => {
283                    if let Some(block_number) = block_number {
284                        self.progress.update(block_number);
285                    }
286                }
287                ControlFlow::Continue { block_number } => self.progress.update(block_number),
288                ControlFlow::Unwind { target, bad_block } => {
289                    self.unwind(target, Some(bad_block.block.number))?;
290                    return Ok(ControlFlow::Unwind { target, bad_block })
291                }
292            }
293
294            previous_stage = Some(
295                self.provider_factory
296                    .provider()?
297                    .get_stage_checkpoint(stage_id)?
298                    .unwrap_or_default()
299                    .block_number,
300            );
301
302            if last_stage == Some(stage_id) {
303                break
304            }
305        }
306
307        Ok(self.progress.next_ctrl())
308    }
309
310    /// Run [static file producer](StaticFileProducer) and [pruner](reth_prune::Pruner) to **move**
311    /// all data from the database to static files for corresponding
312    /// [segments](reth_static_file_types::StaticFileSegment), according to their [stage
313    /// checkpoints](StageCheckpoint):
314    /// - [`StaticFileSegment::Headers`](reth_static_file_types::StaticFileSegment::Headers) ->
315    ///   [`StageId::Headers`]
316    /// - [`StaticFileSegment::Receipts`](reth_static_file_types::StaticFileSegment::Receipts) ->
317    ///   [`StageId::Execution`]
318    /// - [`StaticFileSegment::Transactions`](reth_static_file_types::StaticFileSegment::Transactions)
319    ///   -> [`StageId::Bodies`]
320    ///
321    /// This is a legacy storage.v1 backfill step. Storage.v2 writes directly to static files and
322    /// `RocksDB`, so there is no MDBX -> static-file migration to perform.
323    ///
324    /// CAUTION: This method locks the static file producer Mutex, hence can block the thread if the
325    /// lock is occupied.
326    pub fn move_to_static_files(&self) -> RethResult<()> {
327        if self.provider_factory.cached_storage_settings().is_v2() {
328            return Ok(())
329        }
330
331        // Copies data from database to static files
332        let lowest_static_file_height =
333            self.static_file_producer.lock().copy_to_static_files()?.min_block_num();
334
335        // Deletes data which has been copied to static files.
336        if let Some(prune_tip) = lowest_static_file_height {
337            // Run the pruner so we don't potentially end up with higher height in the database vs
338            // static files during a pipeline unwind
339            let mut pruner = PrunerBuilder::new(Default::default())
340                .delete_limit(usize::MAX)
341                .build_with_provider_factory(self.provider_factory.clone());
342
343            pruner.run(prune_tip)?;
344        }
345
346        Ok(())
347    }
348
349    /// Unwind the stages to the target block (exclusive).
350    ///
351    /// If the unwind is due to a bad block the number of that block should be specified.
352    pub fn unwind(
353        &mut self,
354        to: BlockNumber,
355        bad_block: Option<BlockNumber>,
356    ) -> Result<(), PipelineError> {
357        self.resume_unwind()?;
358        self.unwind_stages(PipelineUnwind {
359            target: to,
360            stages: self.stages.iter().map(|stage| stage.id().to_string()).collect(),
361            bad_block,
362        })
363    }
364
365    /// Completes a recorded unwind without unwinding stages outside its original scope.
366    pub fn resume_unwind(&mut self) -> Result<(), PipelineError> {
367        let unwind = self.provider_factory.provider()?.pipeline_unwind()?;
368        if let Some(unwind) = unwind {
369            self.unwind_stages(unwind)?;
370        }
371        Ok(())
372    }
373
374    fn unwind_stages(&mut self, unwind: PipelineUnwind) -> Result<(), PipelineError> {
375        // Validate the entire scope before committing anything. A different pipeline must not
376        // silently clear work belonging to a disabled or custom stage.
377        for id in &unwind.stages {
378            if !self.stages.iter().any(|stage| stage.id().as_str() == id) {
379                return Err(ProviderError::other(std::io::Error::other(format!(
380                    "cannot resume interrupted unwind to block {}: pipeline is missing stage {id}; \
381                     restart the node with the original stages enabled or rerun `reth stage unwind` \
382                     without `--offline`",
383                    unwind.target,
384                )))
385                .into())
386            }
387        }
388        let to = unwind.target;
389        let bad_block = unwind.bad_block;
390        let (latest_block, prune_modes, checkpoints) = {
391            let provider = self.provider_factory.provider()?;
392            (
393                provider.last_block_number()?,
394                provider.prune_modes_ref().clone(),
395                provider.get_prune_checkpoints()?,
396            )
397        };
398        prune_modes.ensure_unwind_target_unpruned(latest_block, to, &checkpoints)?;
399
400        // Legacy Engine: Prevent the static file producer from racing the unwind.
401        let _locked_sf_producer = self.static_file_producer.lock();
402        let mut provider_rw =
403            self.provider_factory.unwind_provider_rw()?.disable_long_read_transaction_safety();
404
405        // Reject unreachable targets before recording intent or committing any stage. Only
406        // validate the recorded scope: recovery must not adopt unrelated stages.
407        for stage in &mut self.stages {
408            let stage_id = stage.id();
409            if unwind.stages.iter().any(|id| id == stage_id.as_str()) {
410                let checkpoint = provider_rw.get_stage_checkpoint(stage_id)?.unwrap_or_default();
411                if checkpoint.block_number > to &&
412                    let Err(err) = stage.validate_unwind(
413                        &provider_rw,
414                        UnwindInput { checkpoint, unwind_to: to, bad_block },
415                    )
416                {
417                    self.event_sender.notify(PipelineEvent::Error { stage_id });
418                    return Err(PipelineError::Stage(StageError::Fatal(Box::new(err))))
419                }
420            }
421        }
422
423        // The intent must be durable before the first stage commits independently.
424        provider_rw.write_pipeline_unwind(Some(&unwind))?;
425        provider_rw.commit()?;
426        provider_rw = self.provider_factory.unwind_provider_rw()?;
427
428        for id in unwind.stages.iter().rev() {
429            let stage = self
430                .stages
431                .iter_mut()
432                .find(|stage| stage.id().as_str() == id)
433                .expect("unwind stages validated above");
434            let stage_id = stage.id();
435            let span = info_span!("Unwinding", stage = %stage_id);
436            let _enter = span.enter();
437
438            let mut checkpoint = provider_rw.get_stage_checkpoint(stage_id)?.unwrap_or_default();
439            if checkpoint.block_number < to {
440                debug!(
441                    target: "sync::pipeline",
442                    from = %checkpoint.block_number,
443                    %to,
444                    "Unwind point too far for stage"
445                );
446                self.event_sender.notify(PipelineEvent::Skipped { stage_id });
447
448                continue
449            }
450
451            info!(
452                target: "sync::pipeline",
453                from = %checkpoint.block_number,
454                %to,
455                ?bad_block,
456                "Starting unwind"
457            );
458            while checkpoint.block_number > to {
459                let unwind_started_at = Instant::now();
460                let input = UnwindInput { checkpoint, unwind_to: to, bad_block };
461                self.event_sender.notify(PipelineEvent::Unwind { stage_id, input });
462
463                let output = stage.unwind(&provider_rw, input);
464                match output {
465                    Ok(unwind_output) => {
466                        checkpoint = unwind_output.checkpoint;
467                        info!(
468                            target: "sync::pipeline",
469                            stage = %stage_id,
470                            unwind_to = to,
471                            progress = checkpoint.block_number,
472                            done = checkpoint.block_number == to,
473                            "Stage unwound"
474                        );
475
476                        provider_rw.save_stage_checkpoint(stage_id, checkpoint)?;
477
478                        // Notify event listeners and update metrics.
479                        self.event_sender
480                            .notify(PipelineEvent::Unwound { stage_id, result: unwind_output });
481
482                        if let Some(metrics_tx) = &mut self.metrics_tx {
483                            let _ = metrics_tx.send(MetricEvent::StageCheckpoint {
484                                stage_id,
485                                checkpoint,
486                                // We assume it was set in the previous execute iteration, so it
487                                // doesn't change when we unwind.
488                                max_block_number: None,
489                                elapsed: unwind_started_at.elapsed(),
490                            });
491                        }
492
493                        // update finalized and safe block if needed
494                        let last_saved_finalized_block_number =
495                            provider_rw.last_finalized_block_number()?;
496
497                        // If None, that means the finalized block is not written so we should
498                        // always save in that case
499                        if last_saved_finalized_block_number.is_none() ||
500                            Some(checkpoint.block_number) < last_saved_finalized_block_number
501                        {
502                            provider_rw.save_finalized_block_number(BlockNumber::from(
503                                checkpoint.block_number,
504                            ))?;
505                        }
506
507                        let last_saved_safe_block_number = provider_rw.last_safe_block_number()?;
508
509                        if last_saved_safe_block_number.is_none() ||
510                            Some(checkpoint.block_number) < last_saved_safe_block_number
511                        {
512                            provider_rw.save_safe_block_number(BlockNumber::from(
513                                checkpoint.block_number,
514                            ))?;
515                        }
516
517                        provider_rw.commit()?;
518
519                        stage.post_unwind_commit()?;
520
521                        provider_rw = self.provider_factory.unwind_provider_rw()?;
522                    }
523                    Err(err) => {
524                        self.event_sender.notify(PipelineEvent::Error { stage_id });
525
526                        return Err(PipelineError::Stage(StageError::Fatal(Box::new(err))))
527                    }
528                }
529            }
530        }
531
532        provider_rw.write_pipeline_unwind(None)?;
533        provider_rw.commit()?;
534        Ok(())
535    }
536
537    async fn execute_stage_to_completion(
538        &mut self,
539        previous_stage: Option<BlockNumber>,
540        stage_index: usize,
541    ) -> Result<ControlFlow, PipelineError> {
542        let total_stages = self.stages.len();
543
544        let stage_id = self.stage(stage_index).id();
545        let mut made_progress = false;
546        let target = self.max_block.or(previous_stage);
547
548        loop {
549            let prev_checkpoint = self.provider_factory.get_stage_checkpoint(stage_id)?;
550
551            let stage_reached_max_block = prev_checkpoint
552                .zip(self.max_block)
553                .is_some_and(|(prev_progress, target)| prev_progress.block_number >= target);
554            if stage_reached_max_block {
555                warn!(
556                    target: "sync::pipeline",
557                    stage = %stage_id,
558                    max_block = self.max_block,
559                    prev_block = prev_checkpoint.map(|progress| progress.block_number),
560                    "Stage reached target block, skipping."
561                );
562                self.event_sender.notify(PipelineEvent::Skipped { stage_id });
563
564                // We reached the maximum block, so we skip the stage
565                return Ok(ControlFlow::NoProgress {
566                    block_number: prev_checkpoint.map(|progress| progress.block_number),
567                })
568            }
569
570            let exec_input = ExecInput { target, checkpoint: prev_checkpoint };
571
572            self.event_sender.notify(PipelineEvent::Prepare {
573                pipeline_stages_progress: PipelineStagesProgress {
574                    current: stage_index + 1,
575                    total: total_stages,
576                },
577                stage_id,
578                checkpoint: prev_checkpoint,
579                target,
580            });
581
582            if let Err(err) = self.stage(stage_index).execute_ready(exec_input).await {
583                self.event_sender.notify(PipelineEvent::Error { stage_id });
584                match self.on_stage_error(stage_id, prev_checkpoint, err)? {
585                    Some(ctrl) => return Ok(ctrl),
586                    None => continue,
587                };
588            }
589
590            let stage_started_at = Instant::now();
591            let provider_rw = self.provider_factory.database_provider_rw()?;
592
593            self.event_sender.notify(PipelineEvent::Run {
594                pipeline_stages_progress: PipelineStagesProgress {
595                    current: stage_index + 1,
596                    total: total_stages,
597                },
598                stage_id,
599                checkpoint: prev_checkpoint,
600                target,
601            });
602
603            match self.stage(stage_index).execute(&provider_rw, exec_input) {
604                Ok(out @ ExecOutput { checkpoint, done }) => {
605                    // Update stage checkpoint.
606                    provider_rw.save_stage_checkpoint(stage_id, checkpoint)?;
607
608                    // Commit processed data to the database.
609                    provider_rw.commit()?;
610
611                    // Invoke stage post commit hook.
612                    self.stage(stage_index).post_execute_commit()?;
613
614                    // Notify event listeners and update metrics.
615                    self.event_sender.notify(PipelineEvent::Ran {
616                        pipeline_stages_progress: PipelineStagesProgress {
617                            current: stage_index + 1,
618                            total: total_stages,
619                        },
620                        stage_id,
621                        result: out,
622                    });
623                    if let Some(metrics_tx) = &mut self.metrics_tx {
624                        let _ = metrics_tx.send(MetricEvent::StageCheckpoint {
625                            stage_id,
626                            checkpoint,
627                            max_block_number: target,
628                            elapsed: stage_started_at.elapsed(),
629                        });
630                    }
631
632                    let block_number = checkpoint.block_number;
633                    let prev_block_number = prev_checkpoint.unwrap_or_default().block_number;
634                    made_progress |= block_number != prev_block_number;
635                    if done {
636                        return Ok(if made_progress {
637                            ControlFlow::Continue { block_number }
638                        } else {
639                            ControlFlow::NoProgress { block_number: Some(block_number) }
640                        })
641                    }
642                }
643                Err(err) => {
644                    drop(provider_rw);
645                    self.event_sender.notify(PipelineEvent::Error { stage_id });
646
647                    if let Some(ctrl) = self.on_stage_error(stage_id, prev_checkpoint, err)? {
648                        return Ok(ctrl)
649                    }
650                }
651            }
652        }
653    }
654
655    fn on_stage_error(
656        &mut self,
657        stage_id: StageId,
658        prev_checkpoint: Option<StageCheckpoint>,
659        err: StageError,
660    ) -> Result<Option<ControlFlow>, PipelineError> {
661        if let StageError::DetachedHead { local_head, header, error } = err {
662            warn!(target: "sync::pipeline", stage = %stage_id, ?local_head, ?header, %error, "Stage encountered detached head");
663
664            if let Some(last_detached_head_unwind_target) = self.last_detached_head_unwind_target {
665                if local_head.block.hash == last_detached_head_unwind_target &&
666                    header.block.number == local_head.block.number + 1
667                {
668                    self.detached_head_attempts += 1;
669                } else {
670                    self.detached_head_attempts = 1;
671                }
672            } else {
673                self.detached_head_attempts = 1;
674            }
675
676            // We unwind because of a detached head.
677            let unwind_to = local_head
678                .block
679                .number
680                .saturating_sub(
681                    BEACON_CONSENSUS_REORG_UNWIND_DEPTH.saturating_mul(self.detached_head_attempts),
682                )
683                .max(self.provider_factory.static_file_provider().genesis_block_number());
684
685            self.last_detached_head_unwind_target = self.provider_factory.block_hash(unwind_to)?;
686            Ok(Some(ControlFlow::Unwind { target: unwind_to, bad_block: local_head }))
687        } else if let StageError::Block { block, error } = err {
688            match error {
689                BlockErrorKind::Validation(validation_error) => {
690                    error!(
691                        target: "sync::pipeline",
692                        stage = %stage_id,
693                        bad_block = %block.block.number,
694                        bad_block_hash = %block.block.hash,
695                        "Stage encountered a validation error: {validation_error}"
696                    );
697
698                    // FIXME: When handling errors, we do not commit the database transaction. This
699                    // leads to the Merkle stage not clearing its checkpoint, and restarting from an
700                    // invalid place.
701                    // Only reset MerkleExecute checkpoint if MerkleExecute itself failed
702                    if stage_id == StageId::MerkleExecute {
703                        let provider_rw = self.provider_factory.database_provider_rw()?;
704                        provider_rw
705                            .save_stage_checkpoint_progress(StageId::MerkleExecute, vec![])?;
706                        provider_rw.save_stage_checkpoint(
707                            StageId::MerkleExecute,
708                            prev_checkpoint.unwrap_or_default(),
709                        )?;
710
711                        provider_rw.commit()?;
712                    }
713                }
714                BlockErrorKind::Execution(execution_error) => {
715                    error!(
716                        target: "sync::pipeline",
717                        stage = %stage_id,
718                        bad_block = %block.block.number,
719                        bad_block_hash = %block.block.hash,
720                        "Stage encountered an execution error: {execution_error}"
721                    );
722                }
723            }
724
725            // We unwind because of a block error. If the unwind itself fails, we bail entirely;
726            // otherwise, we restart the execution loop from the beginning.
727            Ok(Some(ControlFlow::Unwind {
728                target: prev_checkpoint.unwrap_or_default().block_number,
729                bad_block: block,
730            }))
731        } else if let StageError::MissingStaticFileData { block, segment } = err {
732            error!(
733                target: "sync::pipeline",
734                stage = %stage_id,
735                bad_block = %block.block.number,
736                segment = %segment,
737                "Stage is missing static file data."
738            );
739
740            Ok(Some(ControlFlow::Unwind {
741                target: block.block.number.saturating_sub(1),
742                bad_block: block,
743            }))
744        } else if err.is_fatal() {
745            error!(target: "sync::pipeline", stage = %stage_id, "Stage encountered a fatal error: {err}");
746            Err(err.into())
747        } else {
748            // On other errors we assume they are recoverable if we discard the
749            // transaction and run the stage again.
750            warn!(
751                target: "sync::pipeline",
752                stage = %stage_id,
753                "Stage encountered a non-fatal error: {err}. Retrying..."
754            );
755            Ok(None)
756        }
757    }
758}
759
760impl<N: ProviderNodeTypes> std::fmt::Debug for Pipeline<N> {
761    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
762        f.debug_struct("Pipeline")
763            .field("stages", &self.stages.iter().map(|stage| stage.id()).collect::<Vec<StageId>>())
764            .field("max_block", &self.max_block)
765            .field("event_sender", &self.event_sender)
766            .field("fail_on_unwind", &self.fail_on_unwind)
767            .finish()
768    }
769}
770
771#[cfg(test)]
772mod tests {
773    use std::sync::atomic::Ordering;
774
775    use super::*;
776    use crate::{test_utils::TestStage, UnwindOutput};
777    use alloy_eips::{eip1898::BlockWithParent, BlockNumHash};
778    use assert_matches::assert_matches;
779    use reth_consensus::ConsensusError;
780    use reth_errors::ProviderError;
781    use reth_provider::test_utils::{
782        create_test_provider_factory, create_test_provider_factory_with_genesis_block_number,
783        MockNodeTypesWithDB,
784    };
785    use reth_prune::PruneModes;
786    use reth_testing_utils::generators::{self, random_block_with_parent};
787    use tokio_stream::StreamExt;
788
789    #[test]
790    fn record_progress_calculates_outliers() {
791        let mut progress = PipelineProgress::default();
792
793        progress.update(10);
794        assert_eq!(progress.minimum_block_number, Some(10));
795        assert_eq!(progress.maximum_block_number, Some(10));
796
797        progress.update(20);
798        assert_eq!(progress.minimum_block_number, Some(10));
799        assert_eq!(progress.maximum_block_number, Some(20));
800
801        progress.update(1);
802        assert_eq!(progress.minimum_block_number, Some(1));
803        assert_eq!(progress.maximum_block_number, Some(20));
804    }
805
806    #[test]
807    fn progress_ctrl_flow() {
808        let mut progress = PipelineProgress::default();
809
810        assert_eq!(progress.next_ctrl(), ControlFlow::NoProgress { block_number: None });
811
812        progress.update(1);
813        assert_eq!(progress.next_ctrl(), ControlFlow::Continue { block_number: 1 });
814    }
815
816    /// Runs only the requested prefix of the pipeline.
817    #[tokio::test]
818    async fn run_until_stops_after_the_named_stage() {
819        let provider_factory = create_test_provider_factory();
820
821        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
822            .add_stage(
823                TestStage::new(StageId::Other("A"))
824                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10)))),
825            )
826            // No execution is queued for B, so running it would panic on the empty script.
827            .add_stage(TestStage::new(StageId::Other("B")))
828            .with_max_block(10)
829            .build(
830                provider_factory.clone(),
831                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
832            );
833
834        pipeline.run_until(StageId::Other("A"), None).await.unwrap();
835
836        let provider = provider_factory.provider().unwrap();
837        assert_eq!(
838            provider.get_stage_checkpoint(StageId::Other("A")).unwrap(),
839            Some(StageCheckpoint::new(10))
840        );
841        assert_eq!(provider.get_stage_checkpoint(StageId::Other("B")).unwrap(), None);
842    }
843
844    /// A prefix run naming a stage the pipeline does not have is an error, not a full run.
845    #[tokio::test]
846    async fn run_until_rejects_an_unknown_stage() {
847        let provider_factory = create_test_provider_factory();
848
849        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
850            .add_stage(TestStage::new(StageId::Other("A")))
851            .build(
852                provider_factory.clone(),
853                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
854            );
855
856        let result = pipeline.run_until(StageId::Other("Z"), None).await;
857
858        assert_matches!(result, Err(PipelineError::MissingStage(StageId::Other("Z"))));
859    }
860
861    /// Runs a simple pipeline.
862    #[tokio::test]
863    async fn run_pipeline() {
864        let provider_factory = create_test_provider_factory();
865
866        let stage_a = TestStage::new(StageId::Other("A"))
867            .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(20))));
868        let (stage_a, post_execute_commit_counter_a) = stage_a.with_post_execute_commit_counter();
869        let (stage_a, post_unwind_commit_counter_a) = stage_a.with_post_unwind_commit_counter();
870
871        let stage_b = TestStage::new(StageId::Other("B"))
872            .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10))));
873        let (stage_b, post_execute_commit_counter_b) = stage_b.with_post_execute_commit_counter();
874        let (stage_b, post_unwind_commit_counter_b) = stage_b.with_post_unwind_commit_counter();
875
876        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
877            .add_stage(stage_a)
878            .add_stage(stage_b)
879            .with_max_block(10)
880            .build(
881                provider_factory.clone(),
882                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
883            );
884        let events = pipeline.events();
885
886        // Run pipeline
887        tokio::spawn(async move {
888            pipeline.run().await.unwrap();
889        });
890
891        // Check that the stages were run in order
892        assert_eq!(
893            events.collect::<Vec<PipelineEvent>>().await,
894            vec![
895                PipelineEvent::Prepare {
896                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
897                    stage_id: StageId::Other("A"),
898                    checkpoint: None,
899                    target: Some(10),
900                },
901                PipelineEvent::Run {
902                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
903                    stage_id: StageId::Other("A"),
904                    checkpoint: None,
905                    target: Some(10),
906                },
907                PipelineEvent::Ran {
908                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
909                    stage_id: StageId::Other("A"),
910                    result: ExecOutput::done(StageCheckpoint::new(20)),
911                },
912                PipelineEvent::Prepare {
913                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
914                    stage_id: StageId::Other("B"),
915                    checkpoint: None,
916                    target: Some(10),
917                },
918                PipelineEvent::Run {
919                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
920                    stage_id: StageId::Other("B"),
921                    checkpoint: None,
922                    target: Some(10),
923                },
924                PipelineEvent::Ran {
925                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
926                    stage_id: StageId::Other("B"),
927                    result: ExecOutput::done(StageCheckpoint::new(10)),
928                },
929            ]
930        );
931
932        assert_eq!(post_execute_commit_counter_a.load(Ordering::Relaxed), 1);
933        assert_eq!(post_unwind_commit_counter_a.load(Ordering::Relaxed), 0);
934
935        assert_eq!(post_execute_commit_counter_b.load(Ordering::Relaxed), 1);
936        assert_eq!(post_unwind_commit_counter_b.load(Ordering::Relaxed), 0);
937    }
938
939    /// Unwinds a simple pipeline.
940    #[tokio::test]
941    async fn unwind_pipeline() {
942        let provider_factory = create_test_provider_factory();
943
944        let stage_a = TestStage::new(StageId::Other("A"))
945            .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(100))))
946            .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(1) }));
947        let (stage_a, post_execute_commit_counter_a) = stage_a.with_post_execute_commit_counter();
948        let (stage_a, post_unwind_commit_counter_a) = stage_a.with_post_unwind_commit_counter();
949
950        let stage_b = TestStage::new(StageId::Other("B"))
951            .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10))))
952            .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(1) }));
953        let (stage_b, post_execute_commit_counter_b) = stage_b.with_post_execute_commit_counter();
954        let (stage_b, post_unwind_commit_counter_b) = stage_b.with_post_unwind_commit_counter();
955
956        let stage_c = TestStage::new(StageId::Other("C"))
957            .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(20))))
958            .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(1) }));
959        let (stage_c, post_execute_commit_counter_c) = stage_c.with_post_execute_commit_counter();
960        let (stage_c, post_unwind_commit_counter_c) = stage_c.with_post_unwind_commit_counter();
961
962        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
963            .add_stage(stage_a)
964            .add_stage(stage_b)
965            .add_stage(stage_c)
966            .with_max_block(10)
967            .build(
968                provider_factory.clone(),
969                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
970            );
971        let events = pipeline.events();
972
973        // Run pipeline
974        tokio::spawn(async move {
975            // Sync first
976            pipeline.run().await.expect("Could not run pipeline");
977
978            // Unwind
979            pipeline.unwind(1, None).expect("Could not unwind pipeline");
980        });
981
982        // Check that the stages were unwound in reverse order
983        assert_eq!(
984            events.collect::<Vec<PipelineEvent>>().await,
985            vec![
986                // Executing
987                PipelineEvent::Prepare {
988                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 3 },
989                    stage_id: StageId::Other("A"),
990                    checkpoint: None,
991                    target: Some(10),
992                },
993                PipelineEvent::Run {
994                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 3 },
995                    stage_id: StageId::Other("A"),
996                    checkpoint: None,
997                    target: Some(10),
998                },
999                PipelineEvent::Ran {
1000                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 3 },
1001                    stage_id: StageId::Other("A"),
1002                    result: ExecOutput::done(StageCheckpoint::new(100)),
1003                },
1004                PipelineEvent::Prepare {
1005                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 3 },
1006                    stage_id: StageId::Other("B"),
1007                    checkpoint: None,
1008                    target: Some(10),
1009                },
1010                PipelineEvent::Run {
1011                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 3 },
1012                    stage_id: StageId::Other("B"),
1013                    checkpoint: None,
1014                    target: Some(10),
1015                },
1016                PipelineEvent::Ran {
1017                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 3 },
1018                    stage_id: StageId::Other("B"),
1019                    result: ExecOutput::done(StageCheckpoint::new(10)),
1020                },
1021                PipelineEvent::Prepare {
1022                    pipeline_stages_progress: PipelineStagesProgress { current: 3, total: 3 },
1023                    stage_id: StageId::Other("C"),
1024                    checkpoint: None,
1025                    target: Some(10),
1026                },
1027                PipelineEvent::Run {
1028                    pipeline_stages_progress: PipelineStagesProgress { current: 3, total: 3 },
1029                    stage_id: StageId::Other("C"),
1030                    checkpoint: None,
1031                    target: Some(10),
1032                },
1033                PipelineEvent::Ran {
1034                    pipeline_stages_progress: PipelineStagesProgress { current: 3, total: 3 },
1035                    stage_id: StageId::Other("C"),
1036                    result: ExecOutput::done(StageCheckpoint::new(20)),
1037                },
1038                // Unwinding
1039                PipelineEvent::Unwind {
1040                    stage_id: StageId::Other("C"),
1041                    input: UnwindInput {
1042                        checkpoint: StageCheckpoint::new(20),
1043                        unwind_to: 1,
1044                        bad_block: None
1045                    }
1046                },
1047                PipelineEvent::Unwound {
1048                    stage_id: StageId::Other("C"),
1049                    result: UnwindOutput { checkpoint: StageCheckpoint::new(1) },
1050                },
1051                PipelineEvent::Unwind {
1052                    stage_id: StageId::Other("B"),
1053                    input: UnwindInput {
1054                        checkpoint: StageCheckpoint::new(10),
1055                        unwind_to: 1,
1056                        bad_block: None
1057                    }
1058                },
1059                PipelineEvent::Unwound {
1060                    stage_id: StageId::Other("B"),
1061                    result: UnwindOutput { checkpoint: StageCheckpoint::new(1) },
1062                },
1063                PipelineEvent::Unwind {
1064                    stage_id: StageId::Other("A"),
1065                    input: UnwindInput {
1066                        checkpoint: StageCheckpoint::new(100),
1067                        unwind_to: 1,
1068                        bad_block: None
1069                    }
1070                },
1071                PipelineEvent::Unwound {
1072                    stage_id: StageId::Other("A"),
1073                    result: UnwindOutput { checkpoint: StageCheckpoint::new(1) },
1074                },
1075            ]
1076        );
1077
1078        assert_eq!(post_execute_commit_counter_a.load(Ordering::Relaxed), 1);
1079        assert_eq!(post_unwind_commit_counter_a.load(Ordering::Relaxed), 1);
1080
1081        assert_eq!(post_execute_commit_counter_b.load(Ordering::Relaxed), 1);
1082        assert_eq!(post_unwind_commit_counter_b.load(Ordering::Relaxed), 1);
1083
1084        assert_eq!(post_execute_commit_counter_c.load(Ordering::Relaxed), 1);
1085        assert_eq!(post_unwind_commit_counter_c.load(Ordering::Relaxed), 1);
1086    }
1087
1088    /// Unwinds a pipeline with intermediate progress.
1089    #[tokio::test]
1090    async fn unwind_pipeline_with_intermediate_progress() {
1091        let provider_factory = create_test_provider_factory();
1092
1093        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
1094            .add_stage(
1095                TestStage::new(StageId::Other("A"))
1096                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(100))))
1097                    .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(50) })),
1098            )
1099            .add_stage(
1100                TestStage::new(StageId::Other("B"))
1101                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10)))),
1102            )
1103            .with_max_block(10)
1104            .build(
1105                provider_factory.clone(),
1106                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
1107            );
1108        let events = pipeline.events();
1109
1110        // Run pipeline
1111        tokio::spawn(async move {
1112            // Sync first
1113            pipeline.run().await.expect("Could not run pipeline");
1114
1115            // Unwind
1116            pipeline.unwind(50, None).expect("Could not unwind pipeline");
1117        });
1118
1119        // Check that the stages were unwound in reverse order
1120        assert_eq!(
1121            events.collect::<Vec<PipelineEvent>>().await,
1122            vec![
1123                // Executing
1124                PipelineEvent::Prepare {
1125                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1126                    stage_id: StageId::Other("A"),
1127                    checkpoint: None,
1128                    target: Some(10),
1129                },
1130                PipelineEvent::Run {
1131                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1132                    stage_id: StageId::Other("A"),
1133                    checkpoint: None,
1134                    target: Some(10),
1135                },
1136                PipelineEvent::Ran {
1137                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1138                    stage_id: StageId::Other("A"),
1139                    result: ExecOutput::done(StageCheckpoint::new(100)),
1140                },
1141                PipelineEvent::Prepare {
1142                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1143                    stage_id: StageId::Other("B"),
1144                    checkpoint: None,
1145                    target: Some(10),
1146                },
1147                PipelineEvent::Run {
1148                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1149                    stage_id: StageId::Other("B"),
1150                    checkpoint: None,
1151                    target: Some(10),
1152                },
1153                PipelineEvent::Ran {
1154                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1155                    stage_id: StageId::Other("B"),
1156                    result: ExecOutput::done(StageCheckpoint::new(10)),
1157                },
1158                // Unwinding
1159                // Nothing to unwind in stage "B"
1160                PipelineEvent::Skipped { stage_id: StageId::Other("B") },
1161                PipelineEvent::Unwind {
1162                    stage_id: StageId::Other("A"),
1163                    input: UnwindInput {
1164                        checkpoint: StageCheckpoint::new(100),
1165                        unwind_to: 50,
1166                        bad_block: None
1167                    }
1168                },
1169                PipelineEvent::Unwound {
1170                    stage_id: StageId::Other("A"),
1171                    result: UnwindOutput { checkpoint: StageCheckpoint::new(50) },
1172                },
1173            ]
1174        );
1175    }
1176
1177    /// Runs a pipeline that unwinds during sync.
1178    ///
1179    /// The flow is:
1180    ///
1181    /// - Stage A syncs to block 10
1182    /// - Stage B triggers an unwind, marking block 5 as bad
1183    /// - Stage B unwinds to its previous progress, block 0 but since it is still at block 0, it is
1184    ///   skipped entirely (there is nothing to unwind)
1185    /// - Stage A unwinds to its previous progress, block 0
1186    /// - Stage A syncs back up to block 10
1187    /// - Stage B syncs to block 10
1188    /// - The pipeline finishes
1189    #[tokio::test]
1190    async fn run_pipeline_with_unwind() {
1191        let provider_factory = create_test_provider_factory();
1192
1193        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
1194            .add_stage(
1195                TestStage::new(StageId::Other("A"))
1196                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10))))
1197                    .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(0) }))
1198                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10)))),
1199            )
1200            .add_stage(
1201                TestStage::new(StageId::Other("B"))
1202                    .add_exec(Err(StageError::Block {
1203                        block: Box::new(random_block_with_parent(
1204                            &mut generators::rng(),
1205                            5,
1206                            Default::default(),
1207                        )),
1208                        error: BlockErrorKind::Validation(ConsensusError::BaseFeeMissing),
1209                    }))
1210                    .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(0) }))
1211                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10)))),
1212            )
1213            .with_max_block(10)
1214            .build(
1215                provider_factory.clone(),
1216                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
1217            );
1218        let events = pipeline.events();
1219
1220        // Run pipeline
1221        tokio::spawn(async move {
1222            pipeline.run().await.expect("Could not run pipeline");
1223        });
1224
1225        // Check that the stages were unwound in reverse order
1226        assert_eq!(
1227            events.collect::<Vec<PipelineEvent>>().await,
1228            vec![
1229                PipelineEvent::Prepare {
1230                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1231                    stage_id: StageId::Other("A"),
1232                    checkpoint: None,
1233                    target: Some(10),
1234                },
1235                PipelineEvent::Run {
1236                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1237                    stage_id: StageId::Other("A"),
1238                    checkpoint: None,
1239                    target: Some(10),
1240                },
1241                PipelineEvent::Ran {
1242                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1243                    stage_id: StageId::Other("A"),
1244                    result: ExecOutput::done(StageCheckpoint::new(10)),
1245                },
1246                PipelineEvent::Prepare {
1247                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1248                    stage_id: StageId::Other("B"),
1249                    checkpoint: None,
1250                    target: Some(10),
1251                },
1252                PipelineEvent::Run {
1253                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1254                    stage_id: StageId::Other("B"),
1255                    checkpoint: None,
1256                    target: Some(10),
1257                },
1258                PipelineEvent::Error { stage_id: StageId::Other("B") },
1259                PipelineEvent::Unwind {
1260                    stage_id: StageId::Other("A"),
1261                    input: UnwindInput {
1262                        checkpoint: StageCheckpoint::new(10),
1263                        unwind_to: 0,
1264                        bad_block: Some(5)
1265                    }
1266                },
1267                PipelineEvent::Unwound {
1268                    stage_id: StageId::Other("A"),
1269                    result: UnwindOutput { checkpoint: StageCheckpoint::new(0) },
1270                },
1271                PipelineEvent::Prepare {
1272                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1273                    stage_id: StageId::Other("A"),
1274                    checkpoint: Some(StageCheckpoint::new(0)),
1275                    target: Some(10),
1276                },
1277                PipelineEvent::Run {
1278                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1279                    stage_id: StageId::Other("A"),
1280                    checkpoint: Some(StageCheckpoint::new(0)),
1281                    target: Some(10),
1282                },
1283                PipelineEvent::Ran {
1284                    pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 2 },
1285                    stage_id: StageId::Other("A"),
1286                    result: ExecOutput::done(StageCheckpoint::new(10)),
1287                },
1288                PipelineEvent::Prepare {
1289                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1290                    stage_id: StageId::Other("B"),
1291                    checkpoint: None,
1292                    target: Some(10),
1293                },
1294                PipelineEvent::Run {
1295                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1296                    stage_id: StageId::Other("B"),
1297                    checkpoint: None,
1298                    target: Some(10),
1299                },
1300                PipelineEvent::Ran {
1301                    pipeline_stages_progress: PipelineStagesProgress { current: 2, total: 2 },
1302                    stage_id: StageId::Other("B"),
1303                    result: ExecOutput::done(StageCheckpoint::new(10)),
1304                },
1305            ]
1306        );
1307    }
1308
1309    /// Checks that the pipeline re-runs stages on non-fatal errors and stops on fatal ones.
1310    #[tokio::test]
1311    async fn pipeline_error_handling() {
1312        // Non-fatal
1313        let provider_factory = create_test_provider_factory();
1314        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
1315            .add_stage(
1316                TestStage::new(StageId::Other("NonFatal"))
1317                    .add_exec(Err(StageError::Recoverable(Box::new(std::fmt::Error))))
1318                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10)))),
1319            )
1320            .with_max_block(10)
1321            .build(
1322                provider_factory.clone(),
1323                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
1324            );
1325        let result = pipeline.run().await;
1326        assert_matches!(result, Ok(()));
1327
1328        // Fatal
1329        let provider_factory = create_test_provider_factory();
1330        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
1331            .add_stage(TestStage::new(StageId::Other("Fatal")).add_exec(Err(
1332                StageError::DatabaseIntegrity(ProviderError::BlockBodyIndicesNotFound(5)),
1333            )))
1334            .build(
1335                provider_factory.clone(),
1336                StaticFileProducer::new(provider_factory.clone(), PruneModes::default()),
1337            );
1338        let result = pipeline.run().await;
1339        assert_matches!(
1340            result,
1341            Err(PipelineError::Stage(StageError::DatabaseIntegrity(
1342                ProviderError::BlockBodyIndicesNotFound(5)
1343            )))
1344        );
1345    }
1346
1347    #[test]
1348    fn detached_head_unwinds_no_lower_than_genesis() {
1349        for genesis in [0, 5] {
1350            let factory = create_test_provider_factory_with_genesis_block_number(genesis);
1351            let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
1352                .build(factory.clone(), StaticFileProducer::new(factory, PruneModes::default()));
1353            let block = |number| {
1354                Box::new(BlockWithParent {
1355                    parent: B256::ZERO,
1356                    block: BlockNumHash::new(number, B256::repeat_byte(1)),
1357                })
1358            };
1359            let error = StageError::DetachedHead {
1360                local_head: block(genesis + 1),
1361                header: block(genesis + 2),
1362                error: Box::new(ConsensusError::BaseFeeMissing),
1363            };
1364
1365            let flow = pipeline.on_stage_error(StageId::Headers, None, error).unwrap();
1366
1367            assert_eq!(
1368                flow,
1369                Some(ControlFlow::Unwind { target: genesis, bad_block: block(genesis + 1) })
1370            );
1371        }
1372    }
1373
1374    #[test]
1375    fn interrupted_unwind_preserves_scope_and_target() {
1376        for target in [0, 2] {
1377            let factory = create_test_provider_factory();
1378            let provider = factory.database_provider_rw().unwrap();
1379            for id in
1380                [StageId::Headers, StageId::Execution, StageId::StorageHashing, StageId::Finish]
1381            {
1382                provider.save_stage_checkpoint(id, StageCheckpoint::new(10)).unwrap();
1383            }
1384            provider.commit().unwrap();
1385            let build = |builder: PipelineBuilder<_>| {
1386                builder.build(
1387                    factory.clone(),
1388                    StaticFileProducer::new(factory.clone(), PruneModes::default()),
1389                )
1390            };
1391            // Like an offline unwind, this pipeline does not own Headers or Finish.
1392            let mut pipeline =
1393                build(
1394                    Pipeline::builder()
1395                        .add_stage(
1396                            TestStage::new(StageId::Execution)
1397                                .add_unwind(Err(StageError::MissingDownloadBuffer)),
1398                        )
1399                        .add_stage(TestStage::new(StageId::StorageHashing).add_unwind(Ok(
1400                            UnwindOutput { checkpoint: StageCheckpoint::new(target) },
1401                        ))),
1402                );
1403            assert!(pipeline.unwind(target, None).is_err());
1404            drop(pipeline);
1405            let pending = factory.provider().unwrap().pipeline_unwind().unwrap().unwrap();
1406            assert_eq!(pending.target, target);
1407            assert_eq!(pending.stages, ["Execution", "StorageHashing"]);
1408
1409            // A pipeline with missing stages cannot adopt and clear the unfinished unwind.
1410            assert_eq!(
1411                build(Pipeline::builder()).resume_unwind().unwrap_err().to_string(),
1412                format!(
1413                    "cannot resume interrupted unwind to block {target}: pipeline is missing stage Execution; \
1414                     restart the node with the original stages enabled or rerun `reth stage unwind` without `--offline`"
1415                )
1416            );
1417            assert_eq!(
1418                factory.provider().unwrap().pipeline_unwind().unwrap(),
1419                Some(pending.clone())
1420            );
1421
1422            let mut pipeline = build(
1423                Pipeline::builder()
1424                    .add_stage(
1425                        TestStage::new(StageId::Headers)
1426                            .add_validate_unwind(Err(StageError::MissingDownloadBuffer)),
1427                    )
1428                    .add_stage(
1429                        TestStage::new(StageId::Execution)
1430                            .add_unwind(Err(StageError::MissingDownloadBuffer))
1431                            .add_unwind(Ok(UnwindOutput {
1432                                checkpoint: StageCheckpoint::new(target),
1433                            })),
1434                    )
1435                    .add_stage(TestStage::new(StageId::StorageHashing))
1436                    .add_stage(TestStage::new(StageId::Finish)),
1437            );
1438            assert!(pipeline.resume_unwind().is_err());
1439            assert_eq!(factory.provider().unwrap().pipeline_unwind().unwrap(), Some(pending));
1440            pipeline.resume_unwind().unwrap();
1441            let provider = factory.provider().unwrap();
1442            assert_eq!(provider.pipeline_unwind().unwrap(), None);
1443            for id in [StageId::Execution, StageId::StorageHashing] {
1444                assert_eq!(
1445                    provider.get_stage_checkpoint(id).unwrap().unwrap().block_number,
1446                    target
1447                );
1448            }
1449            for id in [StageId::Headers, StageId::Finish] {
1450                assert_eq!(provider.get_stage_checkpoint(id).unwrap().unwrap().block_number, 10);
1451            }
1452        }
1453    }
1454
1455    #[tokio::test]
1456    async fn forward_sync_resumes_unwind_first() {
1457        let factory = create_test_provider_factory();
1458        factory.set_storage_settings_cache(reth_provider::StorageSettings::v2());
1459        let provider = factory.database_provider_rw().unwrap();
1460        provider.save_stage_checkpoint(StageId::Execution, StageCheckpoint::new(10)).unwrap();
1461        provider
1462            .write_pipeline_unwind(Some(&PipelineUnwind {
1463                target: 2,
1464                stages: vec!["Execution".into()],
1465                bad_block: None,
1466            }))
1467            .unwrap();
1468        provider.commit().unwrap();
1469        let mut pipeline = Pipeline::builder()
1470            .with_max_block(10)
1471            .add_stage(
1472                TestStage::new(StageId::Execution)
1473                    .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(2) }))
1474                    .add_exec(Ok(ExecOutput::done(StageCheckpoint::new(10)))),
1475            )
1476            .build(
1477                factory.clone(),
1478                StaticFileProducer::new(factory.clone(), PruneModes::default()),
1479            );
1480        pipeline.run_loop().await.unwrap();
1481        assert_eq!(factory.provider().unwrap().pipeline_unwind().unwrap(), None);
1482        assert_eq!(
1483            factory.get_stage_checkpoint(StageId::Execution).unwrap().unwrap().block_number,
1484            10
1485        );
1486    }
1487
1488    #[test]
1489    fn malformed_unwind_is_not_ignored() {
1490        let factory = create_test_provider_factory();
1491        let provider = factory.database_provider_rw().unwrap();
1492        provider.write_metadata("pipeline_unwind", vec![0xff]).unwrap();
1493        provider.commit().unwrap();
1494        assert!(factory.provider().unwrap().pipeline_unwind().is_err());
1495    }
1496    /// A target rejected by an earlier stage must not leave the later stages unwound.
1497    #[test]
1498    fn rejected_unwind_target_leaves_all_stages_intact() {
1499        let factory = create_test_provider_factory();
1500        let provider = factory.database_provider_rw().unwrap();
1501        for id in [StageId::Other("A"), StageId::Other("B")] {
1502            provider.save_stage_checkpoint(id, StageCheckpoint::new(10)).unwrap();
1503        }
1504        provider.commit().unwrap();
1505
1506        let mut pipeline = Pipeline::<MockNodeTypesWithDB>::builder()
1507            .add_stage(
1508                TestStage::new(StageId::Other("A"))
1509                    .add_validate_unwind(Err(StageError::MissingDownloadBuffer)),
1510            )
1511            .add_stage(
1512                TestStage::new(StageId::Other("B"))
1513                    .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(1) })),
1514            )
1515            .build(
1516                factory.clone(),
1517                StaticFileProducer::new(factory.clone(), PruneModes::default()),
1518            );
1519
1520        assert_matches!(pipeline.unwind(1, None), Err(PipelineError::Stage(StageError::Fatal(_))));
1521        let provider = factory.provider().unwrap();
1522        assert_eq!(provider.pipeline_unwind().unwrap(), None);
1523        for id in [StageId::Other("A"), StageId::Other("B")] {
1524            assert_eq!(provider.get_stage_checkpoint(id).unwrap(), Some(StageCheckpoint::new(10)));
1525        }
1526    }
1527}
reth_stages_api/pipeline/mod.rs

reth_stages_api/pipeline/
mod.rs
