feat(stages): unwind on detached local head (#3066)

crates/stages/src/error.rs

@@ -20,6 +20,23 @@ pub enum StageError {
         #[source]
         error: consensus::ConsensusError,
     },
+    /// The stage encountered a downloader error where the responses cannot be attached to the
+    /// current head.
+    #[error(
+        "Stage encountered inconsistent chain. Downloaded header #{header_number} ({header_hash:?}) is detached from local head #{head_number} ({head_hash:?}). Details: {error}.",
+        header_number = header.number,
+        header_hash = header.hash,
+        head_number = local_head.number,
+        head_hash = local_head.hash,
+    )]
+    DetachedHead {
+        /// The local head we attempted to attach to.
+        local_head: SealedHeader,
+        /// The header we attempted to attach.
+        header: SealedHeader,
+        /// The error that occurred when attempting to attach the header.
+        error: Box<consensus::ConsensusError>,
+    },
     /// The stage encountered a database error.
     #[error("An internal database error occurred: {0}")]
     Database(#[from] DbError),

crates/stages/src/pipeline/mod.rs

@@ -2,7 +2,10 @@ use crate::{error::*, ExecInput, ExecOutput, Stage, StageError, UnwindInput};
 use futures_util::Future;
 use reth_db::database::Database;
 use reth_interfaces::executor::BlockExecutionError;
-use reth_primitives::{listener::EventListeners, stage::StageId, BlockNumber, H256};
+use reth_primitives::{
+    constants::BEACON_CONSENSUS_REORG_UNWIND_DEPTH, listener::EventListeners, stage::StageId,
+    BlockNumber, H256,
+};
 use reth_provider::{providers::get_stage_checkpoint, Transaction};
 use std::pin::Pin;
 use tokio::sync::watch;
@@ -377,7 +380,16 @@ where
                 Err(err) => {
                     self.listeners.notify(PipelineEvent::Error { stage_id });
 
-                    let out = if let StageError::Validation { block, error } = err {
+                    let out = if let StageError::DetachedHead { local_head, header, error } = err {
+                        warn!(target: "sync::pipeline", stage = %stage_id, ?local_head, ?header, ?error, "Stage encountered detached head");
+
+                        // We unwind because of a detached head.
+                        let unwind_to = local_head
+                            .number
+                            .saturating_sub(BEACON_CONSENSUS_REORG_UNWIND_DEPTH)
+                            .max(1);
+                        Ok(ControlFlow::Unwind { target: unwind_to, bad_block: local_head })
+                    } else if let StageError::Validation { block, error } = err {
                         warn!(
                             target: "sync::pipeline",
                             stage = %stage_id,

crates/stages/src/stages/headers.rs

@@ -7,7 +7,10 @@ use reth_db::{
     transaction::{DbTx, DbTxMut},
 };
 use reth_interfaces::{
-    p2p::headers::downloader::{HeaderDownloader, SyncTarget},
+    p2p::headers::{
+        downloader::{HeaderDownloader, SyncTarget},
+        error::HeadersDownloaderError,
+    },
     provider::ProviderError,
 };
 use reth_primitives::{
@@ -217,7 +220,14 @@ where
         // down to the local head (latest block in db).
         // Task downloader can return `None` only if the response relaying channel was closed. This
         // is a fatal error to prevent the pipeline from running forever.
-        let downloaded_headers = self.downloader.next().await.ok_or(StageError::ChannelClosed)?;
+        let downloaded_headers = match self.downloader.next().await {
+            Some(Ok(headers)) => headers,
+            Some(Err(HeadersDownloaderError::DetachedHead { local_head, header, error })) => {
+                error!(target: "sync::stages::headers", ?error, "Cannot attach header to head");
+                return Err(StageError::DetachedHead { local_head, header, error })
+            }
+            None => return Err(StageError::ChannelClosed),
+        };
 
         info!(target: "sync::stages::headers", len = downloaded_headers.len(), "Received headers");