chore(sync): db integrity error (#148)

* db integrity error

* rm unused import

crates/stages/src/error.rs

@@ -1,6 +1,6 @@
-use crate::{pipeline::PipelineEvent, stages::headers::HeaderStageError};
+use crate::pipeline::PipelineEvent;
 use reth_interfaces::db::Error as DbError;
-use reth_primitives::BlockNumber;
+use reth_primitives::{BlockNumber, H256};
 use thiserror::Error;
 use tokio::sync::mpsc::error::SendError;
 
@@ -16,16 +16,42 @@ pub enum StageError {
         block: BlockNumber,
     },
     /// The stage encountered a database error.
-    #[error("A database error occurred.")]
+    #[error("An internal database error occurred.")]
     Database(#[from] DbError),
-    /// The headers stage encountered an error.
-    #[error("Headers stage error.")]
-    HeadersStage(#[from] HeaderStageError),
+    /// The stage encountered a database integrity error.
+    #[error("A database integrity error occurred.")]
+    DatabaseIntegrity(#[from] DatabaseIntegrityError),
     /// The stage encountered an internal error.
     #[error(transparent)]
     Internal(Box<dyn std::error::Error + Send + Sync>),
 }
 
+/// A database integrity error.
+/// The sender stage error
+#[derive(Error, Debug)]
+pub enum DatabaseIntegrityError {
+    /// Cannonical hash is missing from db
+    #[error("no cannonical hash for block #{number}")]
+    NoCannonicalHash {
+        /// The block number key
+        number: BlockNumber,
+    },
+    /// Cannonical header is missing from db
+    #[error("no cannonical hash for block #{number}")]
+    NoCannonicalHeader {
+        /// The block number key
+        number: BlockNumber,
+    },
+    /// Header is missing from db
+    #[error("no header for block #{number} ({hash})")]
+    NoHeader {
+        /// The block number key
+        number: BlockNumber,
+        /// The block hash key
+        hash: H256,
+    },
+}
+
 /// A pipeline execution error.
 #[derive(Error, Debug)]
 pub enum PipelineError {

crates/stages/src/stages/headers.rs

@@ -1,4 +1,7 @@
-use crate::{ExecInput, ExecOutput, Stage, StageError, StageId, UnwindInput, UnwindOutput};
+use crate::{
+    DatabaseIntegrityError, ExecInput, ExecOutput, Stage, StageError, StageId, UnwindInput,
+    UnwindOutput,
+};
 use reth_interfaces::{
     consensus::{Consensus, ForkchoiceState},
     db::{
@@ -12,7 +15,6 @@ use reth_interfaces::{
 };
 use reth_primitives::{rpc::BigEndianHash, BlockNumber, HeaderLocked, H256, U256};
 use std::fmt::Debug;
-use thiserror::Error;
 use tracing::*;
 
 const HEADERS: StageId = StageId("HEADERS");
@@ -28,31 +30,6 @@ pub struct HeaderStage<D: Downloader, C: Consensus, H: HeadersClient> {
     pub client: H,
 }
 
-/// The header stage error
-#[derive(Error, Debug)]
-pub enum HeaderStageError {
-    /// Cannonical hash is missing from db
-    #[error("no cannonical hash for block #{number}")]
-    NoCannonicalHash {
-        /// The block number key
-        number: BlockNumber,
-    },
-    /// Cannonical header is missing from db
-    #[error("no cannonical hash for block #{number}")]
-    NoCannonicalHeader {
-        /// The block number key
-        number: BlockNumber,
-    },
-    /// Header is missing from db
-    #[error("no header for block #{number} ({hash})")]
-    NoHeader {
-        /// The block number key
-        number: BlockNumber,
-        /// The block hash key
-        hash: H256,
-    },
-}
-
 #[async_trait::async_trait]
 impl<DB: Database, D: Downloader, C: Consensus, H: HeadersClient> Stage<DB>
     for HeaderStage<D, C, H>
@@ -78,12 +55,12 @@ impl<DB: Database, D: Downloader, C: Consensus, H: HeadersClient> Stage<DB>
         // TODO: handle input.max_block
         let last_hash =
             tx.get::<tables::CanonicalHeaders>(last_block_num)?.ok_or_else(|| -> StageError {
-                HeaderStageError::NoCannonicalHash { number: last_block_num }.into()
+                DatabaseIntegrityError::NoCannonicalHash { number: last_block_num }.into()
             })?;
         let last_header = tx
             .get::<tables::Headers>((last_block_num, last_hash).into())?
             .ok_or_else(|| -> StageError {
-                HeaderStageError::NoHeader { number: last_block_num, hash: last_hash }.into()
+                DatabaseIntegrityError::NoHeader { number: last_block_num, hash: last_hash }.into()
             })?;
         let head = HeaderLocked::new(last_header, last_hash);
 
@@ -144,7 +121,7 @@ impl<D: Downloader, C: Consensus, H: HeadersClient> HeaderStage<D, C, H> {
         height: BlockNumber,
     ) -> Result<(), StageError> {
         let hash = tx.get::<tables::CanonicalHeaders>(height)?.ok_or_else(|| -> StageError {
-            HeaderStageError::NoCannonicalHeader { number: height }.into()
+            DatabaseIntegrityError::NoCannonicalHeader { number: height }.into()
         })?;
         let td: Vec<u8> = tx.get::<tables::HeaderTD>((height, hash).into())?.unwrap(); // TODO:
         self.client.update_status(height, hash, H256::from_slice(&td)).await;
@@ -252,7 +229,7 @@ pub(crate) mod tests {
         let rx = execute_stage(db.inner(), input, Ok(vec![]));
         assert_matches!(
             rx.await.unwrap(),
-            Err(StageError::HeadersStage(HeaderStageError::NoCannonicalHeader { .. }))
+            Err(StageError::DatabaseIntegrity(DatabaseIntegrityError::NoCannonicalHeader { .. }))
         );
     }