feat: use buffered ancestor to determine sync target (#2802)

Co-authored-by: Roman Krasiuk <rokrassyuk@gmail.com>

crates/stages/src/error.rs

@@ -3,7 +3,7 @@ use reth_interfaces::{
     consensus, db::DatabaseError as DbError, executor, p2p::error::DownloadError,
     provider::ProviderError,
 };
-use reth_primitives::BlockNumber;
+use reth_primitives::SealedHeader;
 use reth_provider::TransactionError;
 use thiserror::Error;
 use tokio::sync::mpsc::error::SendError;
@@ -12,10 +12,10 @@ use tokio::sync::mpsc::error::SendError;
 #[derive(Error, Debug)]
 pub enum StageError {
     /// The stage encountered a state validation error.
-    #[error("Stage encountered a validation error in block {block}: {error}.")]
+    #[error("Stage encountered a validation error in block {number}: {error}.", number = block.number)]
     Validation {
         /// The block that failed validation.
-        block: BlockNumber,
+        block: SealedHeader,
         /// The underlying consensus error.
         #[source]
         error: consensus::ConsensusError,
@@ -23,12 +23,12 @@ pub enum StageError {
     /// The stage encountered a database error.
     #[error("An internal database error occurred: {0}")]
     Database(#[from] DbError),
-    #[error("Stage encountered a execution error in block {block}: {error}.")]
+    #[error("Stage encountered a execution error in block {number}: {error}.", number = block.number)]
     /// The stage encountered a execution error
     // TODO: Probably redundant, should be rolled into `Validation`
     ExecutionError {
         /// The block that failed execution.
-        block: BlockNumber,
+        block: SealedHeader,
         /// The underlying execution error.
         #[source]
         error: executor::BlockExecutionError,
@@ -71,7 +71,6 @@ impl StageError {
                 StageError::Download(_) |
                 StageError::DatabaseIntegrity(_) |
                 StageError::StageCheckpoint(_) |
-                StageError::ExecutionError { .. } |
                 StageError::ChannelClosed |
                 StageError::Fatal(_) |
                 StageError::Transaction(_)

crates/stages/src/pipeline/ctrl.rs

@@ -1,4 +1,4 @@
-use reth_primitives::BlockNumber;
+use reth_primitives::{BlockNumber, SealedHeader};
 
 /// Determines the control flow during pipeline execution.
 #[derive(Debug, Eq, PartialEq)]
@@ -8,7 +8,7 @@ pub enum ControlFlow {
         /// The block to unwind to.
         target: BlockNumber,
         /// The block that caused the unwind.
-        bad_block: Option<BlockNumber>,
+        bad_block: SealedHeader,
     },
     /// The pipeline is allowed to continue executing stages.
     Continue {

crates/stages/src/pipeline/mod.rs

@@ -221,7 +221,7 @@ where
                 }
                 ControlFlow::Continue { progress } => self.progress.update(progress),
                 ControlFlow::Unwind { target, bad_block } => {
-                    self.unwind(target, bad_block).await?;
+                    self.unwind(target, Some(bad_block.number)).await?;
                     return Ok(ControlFlow::Unwind { target, bad_block })
                 }
             }
@@ -371,7 +371,7 @@ where
                         warn!(
                             target: "sync::pipeline",
                             stage = %stage_id,
-                            bad_block = %block,
+                            bad_block = %block.number,
                             "Stage encountered a validation error: {error}"
                         );
 
@@ -380,7 +380,7 @@ where
                         // beginning.
                         Ok(ControlFlow::Unwind {
                             target: prev_checkpoint.unwrap_or_default().block_number,
-                            bad_block: Some(block),
+                            bad_block: block,
                         })
                     } else if err.is_fatal() {
                         error!(
@@ -422,7 +422,9 @@ mod tests {
     use crate::{test_utils::TestStage, UnwindOutput};
     use assert_matches::assert_matches;
     use reth_db::mdbx::{self, test_utils, EnvKind};
-    use reth_interfaces::{consensus, provider::ProviderError};
+    use reth_interfaces::{
+        consensus, provider::ProviderError, test_utils::generators::random_header,
+    };
     use tokio_stream::StreamExt;
 
     #[test]
@@ -676,7 +678,7 @@ mod tests {
             .add_stage(
                 TestStage::new(StageId::Other("B"))
                     .add_exec(Err(StageError::Validation {
-                        block: 5,
+                        block: random_header(5, Default::default()),
                         error: consensus::ConsensusError::BaseFeeMissing,
                     }))
                     .add_unwind(Ok(UnwindOutput { checkpoint: StageCheckpoint::new(0) }))

crates/stages/src/stages/execution.rs

@@ -156,7 +156,10 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
             let (block, senders) = block.into_components();
             let block_state = executor
                 .execute_and_verify_receipt(&block, td, Some(senders))
-                .map_err(|error| StageError::ExecutionError { block: block_number, error })?;
+                .map_err(|error| StageError::ExecutionError {
+                    block: block.header.clone().seal_slow(),
+                    error,
+                })?;
 
             // Gas metrics
             self.metrics

crates/stages/src/stages/merkle.rs

@@ -10,7 +10,7 @@ use reth_primitives::{
     hex,
     stage::{MerkleCheckpoint, StageCheckpoint, StageId},
     trie::StoredSubNode,
-    BlockNumber, H256,
+    BlockNumber, SealedHeader, H256,
 };
 use reth_provider::Transaction;
 use reth_trie::{IntermediateStateRootState, StateRoot, StateRootProgress};
@@ -66,20 +66,23 @@ impl MerkleStage {
         Self::Unwind
     }
 
-    /// Check that the computed state root matches the expected.
+    /// Check that the computed state root matches the root in the expected header.
     fn validate_state_root(
         &self,
         got: H256,
-        expected: H256,
+        expected: SealedHeader,
         target_block: BlockNumber,
     ) -> Result<(), StageError> {
-        if got == expected {
+        if got == expected.state_root {
             Ok(())
         } else {
             warn!(target: "sync::stages::merkle", ?target_block, ?got, ?expected, "Block's root state failed verification");
             Err(StageError::Validation {
-                block: target_block,
-                error: consensus::ConsensusError::BodyStateRootDiff { got, expected },
+                block: expected.clone(),
+                error: consensus::ConsensusError::BodyStateRootDiff {
+                    got,
+                    expected: expected.state_root,
+                },
             })
         }
     }
@@ -154,7 +157,8 @@ impl<DB: Database> Stage<DB> for MerkleStage {
         let (from_block, to_block) = range.clone().into_inner();
         let current_block = input.previous_stage_checkpoint().block_number;
 
-        let block_root = tx.get_header(current_block)?.state_root;
+        let block = tx.get_header(current_block)?;
+        let block_root = block.state_root;
 
         let mut checkpoint = self.get_execution_checkpoint(tx)?;
 
@@ -219,7 +223,7 @@ impl<DB: Database> Stage<DB> for MerkleStage {
         // Reset the checkpoint
         self.save_execution_checkpoint(tx, None)?;
 
-        self.validate_state_root(trie_root, block_root, to_block)?;
+        self.validate_state_root(trie_root, block.seal_slow(), to_block)?;
 
         info!(target: "sync::stages::merkle::exec", stage_progress = to_block, is_final_range = true, "Stage iteration finished");
         Ok(ExecOutput { checkpoint: StageCheckpoint::new(to_block), done: true })
@@ -251,8 +255,8 @@ impl<DB: Database> Stage<DB> for MerkleStage {
                     .map_err(|e| StageError::Fatal(Box::new(e)))?;
 
             // Validate the calulated state root
-            let target_root = tx.get_header(input.unwind_to)?.state_root;
-            self.validate_state_root(block_root, target_root, input.unwind_to)?;
+            let target = tx.get_header(input.unwind_to)?;
+            self.validate_state_root(block_root, target.seal_slow(), input.unwind_to)?;
 
             // Validation passed, apply unwind changes to the database.
             updates.flush(tx.deref_mut())?;

crates/stages/src/stages/total_difficulty.rs

@@ -78,7 +78,7 @@ impl<DB: Database> Stage<DB> for TotalDifficultyStage {
 
             self.consensus
                 .validate_header_with_total_difficulty(&header, td)
-                .map_err(|error| StageError::Validation { block: header.number, error })?;
+                .map_err(|error| StageError::Validation { block: header.seal_slow(), error })?;
             cursor_td.append(block_number, td.into())?;
         }
         info!(target: "sync::stages::total_difficulty", stage_progress = end_block, is_final_range, "Stage iteration finished");