fix(stages): Fix two bugs related to stage checkpoints and pipeline syncs (#20521)

crates/node/builder/src/launch/common.rs

@@ -938,9 +938,13 @@ where
     ///
     /// A target block hash if the pipeline is inconsistent, otherwise `None`.
     pub fn check_pipeline_consistency(&self) -> ProviderResult<Option<B256>> {
+        // We skip the era stage if it's not enabled
+        let era_enabled = self.era_import_source().is_some();
+        let mut all_stages =
+            StageId::ALL.into_iter().filter(|id| era_enabled || id != &StageId::Era);
+
         // Get the expected first stage based on config.
-        let first_stage =
-            if self.era_import_source().is_some() { StageId::Era } else { StageId::Headers };
+        let first_stage = all_stages.next().expect("there must be at least one stage");
 
         // If no target was provided, check if the stages are congruent - check if the
         // checkpoint of the last stage matches the checkpoint of the first.
@@ -950,20 +954,28 @@ where
             .unwrap_or_default()
             .block_number;
 
-        // Skip the first stage as we've already retrieved it and comparing all other checkpoints
-        // against it.
-        for stage_id in StageId::ALL.iter().skip(1) {
+        // Compare all other stages against the first
+        for stage_id in all_stages {
             let stage_checkpoint = self
                 .blockchain_db()
-                .get_stage_checkpoint(*stage_id)?
+                .get_stage_checkpoint(stage_id)?
                 .unwrap_or_default()
                 .block_number;
 
             // If the checkpoint of any stage is less than the checkpoint of the first stage,
             // retrieve and return the block hash of the latest header and use it as the target.
+            debug!(
+                target: "consensus::engine",
+                first_stage_id = %first_stage,
+                first_stage_checkpoint,
+                stage_id = %stage_id,
+                stage_checkpoint = stage_checkpoint,
+                "Checking stage against first stage",
+            );
             if stage_checkpoint < first_stage_checkpoint {
                 debug!(
                     target: "consensus::engine",
+                    first_stage_id = %first_stage,
                     first_stage_checkpoint,
                     inconsistent_stage_id = %stage_id,
                     inconsistent_stage_checkpoint = stage_checkpoint,

crates/stages/stages/src/stages/merkle_changesets.rs

@@ -5,7 +5,7 @@ use reth_consensus::ConsensusError;
 use reth_primitives_traits::{GotExpected, SealedHeader};
 use reth_provider::{
     ChainStateBlockReader, DBProvider, HeaderProvider, ProviderError, PruneCheckpointReader,
-    PruneCheckpointWriter, StageCheckpointReader, TrieWriter,
+    PruneCheckpointWriter, StageCheckpointReader, StageCheckpointWriter, TrieWriter,
 };
 use reth_prune_types::{
     PruneCheckpoint, PruneMode, PruneSegment, MERKLE_CHANGESETS_RETENTION_BLOCKS,
@@ -300,6 +300,7 @@ where
         + DBProvider
         + HeaderProvider
         + ChainStateBlockReader
+        + StageCheckpointWriter
         + PruneCheckpointReader
         + PruneCheckpointWriter,
 {
@@ -404,6 +405,28 @@ where
             computed_range.start = computed_range.end;
         }
 
+        // If we've unwound so far that there are no longer enough trie changesets available then
+        // simply clear them and the checkpoints, so that on next pipeline startup they will be
+        // regenerated.
+        debug!(
+            target: "sync::stages::merkle_changesets",
+            ?computed_range,
+            retention_blocks=?self.retention_blocks,
+            "Checking if computed range is over retention threshold",
+        );
+        if computed_range.end - computed_range.start < self.retention_blocks {
+            debug!(
+                target: "sync::stages::merkle_changesets",
+                ?computed_range,
+                retention_blocks=?self.retention_blocks,
+                "Clearing checkpoints completely",
+            );
+            provider.clear_trie_changesets()?;
+            provider
+                .save_stage_checkpoint(StageId::MerkleChangeSets, StageCheckpoint::default())?;
+            return Ok(UnwindOutput { checkpoint: StageCheckpoint::default() })
+        }
+
         // `computed_range.end` is exclusive
         let checkpoint = StageCheckpoint::new(computed_range.end.saturating_sub(1));