refactor: stage error fatal/recoverable variants

crates/stages/src/error.rs

@@ -36,15 +36,25 @@ pub enum StageError {
     /// rely on external downloaders
     #[error("Invalid download response: {0}")]
     Download(String),
-    /// The stage encountered an internal error.
+    /// The stage encountered a recoverable error.
+    ///
+    /// These types of errors are caught by the [Pipeline] and trigger a restart of the stage.
     #[error(transparent)]
-    Internal(Box<dyn std::error::Error + Send + Sync>),
+    Recoverable(Box<dyn std::error::Error + Send + Sync>),
+    /// The stage encountered a fatal error.
+    ///
+    /// These types of errors stop the pipeline.
+    #[error(transparent)]
+    Fatal(Box<dyn std::error::Error + Send + Sync>),
 }
 
 impl StageError {
     /// If the error is fatal the pipeline will stop.
     pub fn is_fatal(&self) -> bool {
-        matches!(self, StageError::Database(_) | StageError::DatabaseIntegrity(_))
+        matches!(
+            self,
+            StageError::Database(_) | StageError::DatabaseIntegrity(_) | StageError::Fatal(_)
+        )
     }
 }
 

crates/stages/src/pipeline.rs

@@ -255,7 +255,7 @@ impl<DB: Database> Pipeline<DB> {
                     }
                     Err(err) => {
                         self.events_sender.send(PipelineEvent::Error { stage_id }).await?;
-                        return Err(PipelineError::Stage(StageError::Internal(err)))
+                        return Err(PipelineError::Stage(StageError::Fatal(err)))
                     }
                 }
             }
@@ -739,7 +739,7 @@ mod tests {
         let result = Pipeline::<Env<WriteMap>>::new()
             .push(
                 TestStage::new(StageId("NonFatal"))
-                    .add_exec(Err(StageError::Internal(Box::new(std::fmt::Error))))
+                    .add_exec(Err(StageError::Recoverable(Box::new(std::fmt::Error))))
                     .add_exec(Ok(ExecOutput { stage_progress: 10, done: true, reached_tip: true })),
                 false,
             )

crates/stages/src/stages/bodies.rs

@@ -16,7 +16,7 @@ use reth_primitives::{
     BlockNumber, SealedHeader,
 };
 use std::{fmt::Debug, sync::Arc};
-use tracing::{error, warn};
+use tracing::warn;
 
 const BODIES: StageId = StageId("Bodies");
 
@@ -111,23 +111,14 @@ impl<DB: Database, D: BodyDownloader, C: Consensus> Stage<DB> for BodyStage<D, C
         let mut bodies_stream = self.downloader.bodies_stream(bodies_to_download.iter());
         let mut highest_block = previous_block;
         while let Some(result) = bodies_stream.next().await {
-            let block = match result {
-                Ok(block) => block,
-                Err(err) => {
-                    error!(
-                        "Encountered error downloading block {}. Details: {:?}",
-                        highest_block + 1,
-                        err
-                    );
-                    // Exit the stage early
-                    return Ok(ExecOutput {
-                        stage_progress: highest_block,
-                        done: false,
-                        reached_tip: false,
-                    })
-                }
+            let Ok(block) = result else {
+                warn!(
+                    "Encountered an error downloading block {}: {:?}",
+                    highest_block + 1,
+                    result.unwrap_err()
+                );
+                break
             };
-
             let block_number = block.number;
             // Write block
             let key = (block_number, block.hash()).into();

crates/stages/src/stages/senders.rs

@@ -37,7 +37,7 @@ enum SendersStageError {
 
 impl From<SendersStageError> for StageError {
     fn from(error: SendersStageError) -> Self {
-        StageError::Internal(Box::new(error))
+        StageError::Fatal(Box::new(error))
     }
 }