feat(cli): make stopping on invalid block the default for reth import (#21403)

Co-authored-by: Amp <amp@ampcode.com>

crates/cli/commands/src/import.rs

@@ -26,6 +26,14 @@ pub struct ImportCommand<C: ChainSpecParser> {
     #[arg(long, value_name = "CHUNK_LEN", verbatim_doc_comment)]
     chunk_len: Option<u64>,
 
+    /// Fail immediately when an invalid block is encountered.
+    ///
+    /// By default, the import will stop at the last valid block if an invalid block is
+    /// encountered during execution or validation, leaving the database at the last valid
+    /// block state. When this flag is set, the import will instead fail with an error.
+    #[arg(long, verbatim_doc_comment)]
+    fail_on_invalid_block: bool,
+
     /// The path(s) to block file(s) for import.
     ///
     /// The online stages (headers and bodies) are replaced by a file import, after which the
@@ -52,7 +60,11 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportComm
 
         info!(target: "reth::cli", "Starting import of {} file(s)", self.paths.len());
 
-        let import_config = ImportConfig { no_state: self.no_state, chunk_len: self.chunk_len };
+        let import_config = ImportConfig {
+            no_state: self.no_state,
+            chunk_len: self.chunk_len,
+            fail_on_invalid_block: self.fail_on_invalid_block,
+        };
 
         let executor = components.evm_config().clone();
         let consensus = Arc::new(components.consensus().clone());
@@ -81,7 +93,20 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportComm
             total_decoded_blocks += result.total_decoded_blocks;
             total_decoded_txns += result.total_decoded_txns;
 
-            if !result.is_complete() {
+            // Check if we stopped due to an invalid block
+            if result.stopped_on_invalid_block {
+                info!(target: "reth::cli",
+                      "Stopped at last valid block {} due to invalid block {} in file: {}. Imported {} blocks, {} transactions",
+                      result.last_valid_block.unwrap_or(0),
+                      result.bad_block.unwrap_or(0),
+                      path.display(),
+                      result.total_imported_blocks,
+                      result.total_imported_txns);
+                // Stop importing further files and exit successfully
+                break;
+            }
+
+            if !result.is_successful() {
                 return Err(eyre::eyre!(
                     "Chain was partially imported from file: {}. Imported {}/{} blocks, {}/{} transactions",
                     path.display(),
@@ -98,7 +123,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> ImportComm
         }
 
         info!(target: "reth::cli",
-              "All files imported successfully. Total: {}/{} blocks, {}/{} transactions",
+              "Import complete. Total: {}/{} blocks, {}/{} transactions",
               total_imported_blocks, total_decoded_blocks, total_imported_txns, total_decoded_txns);
 
         Ok(())
@@ -139,4 +164,20 @@ mod tests {
         assert_eq!(args.paths[1], PathBuf::from("file2.rlp"));
         assert_eq!(args.paths[2], PathBuf::from("file3.rlp"));
     }
+
+    #[test]
+    fn parse_import_command_with_fail_on_invalid_block() {
+        let args: ImportCommand<EthereumChainSpecParser> =
+            ImportCommand::parse_from(["reth", "--fail-on-invalid-block", "chain.rlp"]);
+        assert!(args.fail_on_invalid_block);
+        assert_eq!(args.paths.len(), 1);
+        assert_eq!(args.paths[0], PathBuf::from("chain.rlp"));
+    }
+
+    #[test]
+    fn parse_import_command_default_stops_on_invalid_block() {
+        let args: ImportCommand<EthereumChainSpecParser> =
+            ImportCommand::parse_from(["reth", "chain.rlp"]);
+        assert!(!args.fail_on_invalid_block);
+    }
 }

crates/cli/commands/src/import_core.rs

@@ -22,11 +22,11 @@ use reth_provider::{
     StageCheckpointReader,
 };
 use reth_prune::PruneModes;
-use reth_stages::{prelude::*, Pipeline, StageId, StageSet};
+use reth_stages::{prelude::*, ControlFlow, Pipeline, StageId, StageSet};
 use reth_static_file::StaticFileProducer;
 use std::{path::Path, sync::Arc};
 use tokio::sync::watch;
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
 
 /// Configuration for importing blocks from RLP files.
 #[derive(Debug, Clone, Default)]
@@ -35,6 +35,9 @@ pub struct ImportConfig {
     pub no_state: bool,
     /// Chunk byte length to read from file.
     pub chunk_len: Option<u64>,
+    /// If true, fail immediately when an invalid block is encountered.
+    /// By default (false), the import stops at the last valid block and exits successfully.
+    pub fail_on_invalid_block: bool,
 }
 
 /// Result of an import operation.
@@ -48,6 +51,12 @@ pub struct ImportResult {
     pub total_imported_blocks: usize,
     /// Total number of transactions imported into the database.
     pub total_imported_txns: usize,
+    /// Whether the import was stopped due to an invalid block.
+    pub stopped_on_invalid_block: bool,
+    /// The block number that was invalid, if any.
+    pub bad_block: Option<u64>,
+    /// The last valid block number when stopped due to invalid block.
+    pub last_valid_block: Option<u64>,
 }
 
 impl ImportResult {
@@ -56,6 +65,14 @@ impl ImportResult {
         self.total_decoded_blocks == self.total_imported_blocks &&
             self.total_decoded_txns == self.total_imported_txns
     }
+
+    /// Returns true if the import was successful, considering stop-on-invalid-block mode.
+    ///
+    /// In stop-on-invalid-block mode, a partial import is considered successful if we
+    /// stopped due to an invalid block (leaving the DB at the last valid block).
+    pub fn is_successful(&self) -> bool {
+        self.is_complete() || self.stopped_on_invalid_block
+    }
 }
 
 /// Imports blocks from an RLP-encoded file into the database.
@@ -103,6 +120,11 @@ where
     let static_file_producer =
         StaticFileProducer::new(provider_factory.clone(), PruneModes::default());
 
+    // Track if we stopped due to an invalid block
+    let mut stopped_on_invalid_block = false;
+    let mut bad_block_number: Option<u64> = None;
+    let mut last_valid_block_number: Option<u64> = None;
+
     while let Some(file_client) =
         reader.next_chunk::<BlockTy<N>>(consensus.clone(), Some(sealed_header)).await?
     {
@@ -137,12 +159,51 @@ where
 
         // Run pipeline
         info!(target: "reth::import", "Starting sync pipeline");
-        tokio::select! {
-            res = pipeline.run() => res?,
-            _ = tokio::signal::ctrl_c() => {
-                info!(target: "reth::import", "Import interrupted by user");
-                break;
-            },
+        if import_config.fail_on_invalid_block {
+            // Original behavior: fail on unwind
+            tokio::select! {
+                res = pipeline.run() => res?,
+                _ = tokio::signal::ctrl_c() => {
+                    info!(target: "reth::import", "Import interrupted by user");
+                    break;
+                },
+            }
+        } else {
+            // Default behavior: Use run_loop() to handle unwinds gracefully
+            let result = tokio::select! {
+                res = pipeline.run_loop() => res,
+                _ = tokio::signal::ctrl_c() => {
+                    info!(target: "reth::import", "Import interrupted by user");
+                    break;
+                },
+            };
+
+            match result {
+                Ok(ControlFlow::Unwind { target, bad_block }) => {
+                    // An invalid block was encountered; stop at last valid block
+                    let bad = bad_block.block.number;
+                    warn!(
+                        target: "reth::import",
+                        bad_block = bad,
+                        last_valid_block = target,
+                        "Invalid block encountered during import; stopping at last valid block"
+                    );
+                    stopped_on_invalid_block = true;
+                    bad_block_number = Some(bad);
+                    last_valid_block_number = Some(target);
+                    break;
+                }
+                Ok(ControlFlow::Continue { block_number }) => {
+                    debug!(target: "reth::import", block_number, "Pipeline chunk completed");
+                }
+                Ok(ControlFlow::NoProgress { block_number }) => {
+                    debug!(target: "reth::import", ?block_number, "Pipeline made no progress");
+                }
+                Err(e) => {
+                    // Propagate other pipeline errors
+                    return Err(e.into());
+                }
+            }
         }
 
         sealed_header = provider_factory
@@ -160,9 +221,20 @@ where
         total_decoded_txns,
         total_imported_blocks,
         total_imported_txns,
+        stopped_on_invalid_block,
+        bad_block: bad_block_number,
+        last_valid_block: last_valid_block_number,
     };
 
-    if !result.is_complete() {
+    if result.stopped_on_invalid_block {
+        info!(target: "reth::import",
+            total_imported_blocks,
+            total_imported_txns,
+            bad_block = ?result.bad_block,
+            last_valid_block = ?result.last_valid_block,
+            "Import stopped at last valid block due to invalid block"
+        );
+    } else if !result.is_complete() {
         error!(target: "reth::import",
             total_decoded_blocks,
             total_imported_blocks,

docs/vocs/docs/pages/cli/reth/import.mdx

@@ -187,6 +187,13 @@ RocksDB:
       --chunk-len <CHUNK_LEN>
           Chunk byte length to read from file.
 
+      --fail-on-invalid-block
+          Fail immediately when an invalid block is encountered.
+
+          By default, the import will stop at the last valid block if an invalid block is
+          encountered during execution or validation, leaving the database at the last valid
+          block state. When this flag is set, the import will instead fail with an error.
+
   <IMPORT_PATH>...
           The path(s) to block file(s) for import.