feat: run StaticFileProvider::check_consistency on start up

crates/config/src/config.rs

@@ -337,6 +337,13 @@ impl Default for PruneConfig {
     }
 }
 
+impl PruneConfig {
+    /// Returns whether there is any kind of receipt pruning configuration.
+    pub fn has_receipts_pruning(&self) -> bool {
+        self.segments.receipts.is_some() || !self.segments.receipts_log_filter.is_empty()
+    }
+}
+
 /// Helper type to support older versions of Duration deserialization.
 fn deserialize_duration<'de, D>(deserializer: D) -> Result<Option<Duration>, D::Error>
 where

crates/evm/src/execute.rs

@@ -176,6 +176,61 @@ pub trait BlockExecutorProvider: Send + Sync + Clone + Unpin + 'static {
         DB: Database<Error = ProviderError>;
 }
 
+/// A [BlockExecutorProvider] implementation that does nothing.
+#[derive(Debug, Default, Clone)]
+#[non_exhaustive]
+pub struct NoopBlockExecutorProvider;
+
+impl BlockExecutorProvider for NoopBlockExecutorProvider {
+    type Executor<DB: Database<Error = ProviderError>> = Self;
+
+    type BatchExecutor<DB: Database<Error = ProviderError>> = Self;
+
+    fn executor<DB>(&self, _: DB) -> Self::Executor<DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        Self
+    }
+
+    fn batch_executor<DB>(&self, _: DB, _: PruneModes) -> Self::BatchExecutor<DB>
+    where
+        DB: Database<Error = ProviderError>,
+    {
+        Self
+    }
+}
+
+impl<DB> Executor<DB> for NoopBlockExecutorProvider {
+    type Input<'a> = BlockExecutionInput<'a, BlockWithSenders>;
+    type Output = BlockExecutionOutput<Receipt>;
+    type Error = BlockExecutionError;
+
+    fn execute(self, _: Self::Input<'_>) -> Result<Self::Output, Self::Error> {
+        Err(BlockExecutionError::UnavailableForNoop)
+    }
+}
+
+impl<DB> BatchExecutor<DB> for NoopBlockExecutorProvider {
+    type Input<'a> = BlockExecutionInput<'a, BlockWithSenders>;
+    type Output = BatchBlockExecutionOutput;
+    type Error = BlockExecutionError;
+
+    fn execute_one(&mut self, _: Self::Input<'_>) -> Result<(), Self::Error> {
+        Err(BlockExecutionError::UnavailableForNoop)
+    }
+
+    fn finalize(self) -> Self::Output {
+        unreachable!()
+    }
+
+    fn set_tip(&mut self, _: BlockNumber) {}
+
+    fn size_hint(&self) -> Option<usize> {
+        None
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/interfaces/src/blockchain_tree/error.rs

@@ -307,7 +307,8 @@ impl InsertBlockErrorKind {
                     BlockExecutionError::CanonicalRevert { .. } |
                     BlockExecutionError::CanonicalCommit { .. } |
                     BlockExecutionError::AppendChainDoesntConnect { .. } |
-                    BlockExecutionError::UnavailableForTest => false,
+                    BlockExecutionError::UnavailableForTest |
+                    BlockExecutionError::UnavailableForNoop => false,
                     BlockExecutionError::Other(_) => false,
                 }
             }

crates/interfaces/src/executor.rs

@@ -115,6 +115,9 @@ pub enum BlockExecutionError {
     /// Note: this is not feature gated for convenience.
     #[error("execution unavailable for tests")]
     UnavailableForTest,
+    /// Only used for NoopBlockExecutorProvider
+    #[error("execution unavailable for noop")]
+    UnavailableForNoop,
     /// Error when fetching latest block state.
     #[error(transparent)]
     LatestBlock(#[from] ProviderError),

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

@@ -4,22 +4,30 @@ use std::{cmp::max, sync::Arc, thread::available_parallelism};
 
 use eyre::Context;
 use rayon::ThreadPoolBuilder;
-use tokio::sync::mpsc::Receiver;
+use tokio::sync::{mpsc::Receiver, oneshot};
 
 use reth_auto_seal_consensus::MiningMode;
+use reth_beacon_consensus::EthBeaconConsensus;
 use reth_config::{config::EtlConfig, PruneConfig};
 use reth_db::{database::Database, database_metrics::DatabaseMetrics};
+use reth_downloaders::{bodies::noop::NoopBodiesDownloader, headers::noop::NoopHeaderDownloader};
+use reth_evm::execute::NoopBlockExecutorProvider;
 use reth_interfaces::p2p::headers::client::HeadersClient;
 use reth_node_core::{
     cli::config::RethRpcConfig,
     dirs::{ChainPath, DataDirPath},
     init::{init_genesis, InitDatabaseError},
     node_config::NodeConfig,
 };
-use reth_primitives::{BlockNumber, Chain, ChainSpec, Head, PruneModes, B256};
-use reth_provider::{providers::StaticFileProvider, ProviderFactory, StaticFileProviderFactory};
+use reth_primitives::{
+    stage::PipelineTarget, BlockNumber, Chain, ChainSpec, Head, PruneModes, B256,
+};
+use reth_provider::{
+    providers::StaticFileProvider, HeaderSyncMode, ProviderFactory, StaticFileProviderFactory,
+};
 use reth_prune::PrunerBuilder;
 use reth_rpc_layer::JwtSecret;
+use reth_stages::{sets::DefaultStages, Pipeline};
 use reth_static_file::StaticFileProducer;
 use reth_tasks::TaskExecutor;
 use reth_tracing::tracing::{error, info, warn};
@@ -312,25 +320,80 @@ impl<R> LaunchContextWith<Attached<WithConfigs, R>> {
 
 impl<DB> LaunchContextWith<Attached<WithConfigs, DB>>
 where
-    DB: Clone,
+    DB: Database + Clone + 'static,
 {
     /// Returns the [ProviderFactory] for the attached database.
-    pub fn create_provider_factory(&self) -> eyre::Result<ProviderFactory<DB>> {
+    pub async fn create_provider_factory(&self) -> eyre::Result<ProviderFactory<DB>> {
         let factory = ProviderFactory::new(
             self.right().clone(),
             self.chain_spec(),
             self.data_dir().static_files(),
         )?
         .with_static_files_metrics();
 
+        let has_receipt_pruning =
+            self.toml_config().prune.as_ref().map_or(false, |a| a.has_receipts_pruning());
+
+        info!(target: "reth::cli", "Verifying storage consistency.");
+
+        // Check for consistency between database and static files. If it fails, it unwinds to
+        // the first block that's consistent between database and static files.
+        if let Some(unwind_target) = factory.static_file_provider().check_consistency(
+            &factory.provider()?,
+            has_receipt_pruning,
+            false,
+        )? {
+            // Highly unlikely to happen, and given its destructive nature, it's better to panic
+            // instead.
+            if PipelineTarget::Unwind(0) == unwind_target {
+                panic!("A static file <> database inconsistency was found that would trigger an unwind to block 0.")
+            }
+
+            info!(target: "reth::cli", unwind_target = %unwind_target, "Executing an unwind after a failed storage consistency check.");
+
+            // Builds an unwind-only pipeline
+            let pipeline = Pipeline::builder()
+                .add_stages(DefaultStages::new(
+                    factory.clone(),
+                    HeaderSyncMode::Continuous,
+                    Arc::new(EthBeaconConsensus::new(self.chain_spec())),
+                    NoopHeaderDownloader::default(),
+                    NoopBodiesDownloader::default(),
+                    NoopBlockExecutorProvider::default(),
+                    self.toml_config().stages.clone(),
+                    self.prune_modes().unwrap_or_default(),
+                ))
+                .build(
+                    factory.clone(),
+                    StaticFileProducer::new(
+                        factory.clone(),
+                        factory.static_file_provider(),
+                        self.prune_modes().unwrap_or_default(),
+                    ),
+                );
+
+            // Unwinds to block
+            let (tx, rx) = oneshot::channel();
+
+            // Pipeline should be run as blocking and panic if it fails.
+            self.task_executor().spawn_critical_blocking(
+                "pipeline task",
+                Box::pin(async move {
+                    let (_, result) = pipeline.run_as_fut(Some(unwind_target)).await;
+                    let _ = tx.send(result);
+                }),
+            );
+            rx.await??;
+        }
+
         Ok(factory)
     }
 
     /// Creates a new [ProviderFactory] and attaches it to the launch context.
-    pub fn with_provider_factory(
+    pub async fn with_provider_factory(
         self,
     ) -> eyre::Result<LaunchContextWith<Attached<WithConfigs, ProviderFactory<DB>>>> {
-        let factory = self.create_provider_factory()?;
+        let factory = self.create_provider_factory().await?;
         let ctx = LaunchContextWith {
             inner: self.inner,
             attachment: self.attachment.map_right(|_| factory),

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

@@ -98,7 +98,7 @@ where
             // ensure certain settings take effect
             .with_adjusted_configs()
             // Create the provider factory
-            .with_provider_factory()?
+            .with_provider_factory().await?
             .inspect(|_| {
                 info!(target: "reth::cli", "Database opened");
             })

crates/primitives/src/stage/mod.rs

@@ -12,7 +12,7 @@ pub use checkpoints::{
 };
 
 /// Direction and target block for pipeline operations.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum PipelineTarget {
     /// Target for forward synchronization, indicating a block hash to sync to.
     Sync(BlockHash),
@@ -53,3 +53,14 @@ impl From<BlockHash> for PipelineTarget {
         Self::Sync(hash)
     }
 }
+
+impl std::fmt::Display for PipelineTarget {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            PipelineTarget::Sync(block) => {
+                write!(f, "Sync({block})")
+            }
+            PipelineTarget::Unwind(block) => write!(f, "Unwind({block})"),
+        }
+    }
+}

crates/primitives/src/static_file/segment.rs

@@ -138,6 +138,11 @@ impl StaticFileSegment {
     pub fn is_headers(&self) -> bool {
         matches!(self, StaticFileSegment::Headers)
     }
+
+    /// Returns `true` if the segment is `StaticFileSegment::Receipts`.
+    pub fn is_receipts(&self) -> bool {
+        matches!(self, StaticFileSegment::Receipts)
+    }
 }
 
 /// A segment header that contains information common to all segments. Used for storage.