feat(cli): add stage unwind command (#2913)

bin/reth/src/stage/mod.rs

@@ -4,6 +4,7 @@ use clap::{Parser, Subcommand};
 pub mod drop;
 pub mod dump;
 pub mod run;
+pub mod unwind;
 
 /// `reth stage` command
 #[derive(Debug, Parser)]
@@ -26,6 +27,8 @@ pub enum Subcommands {
     Drop(drop::Command),
     /// Dumps a stage from a range into a new database.
     Dump(dump::Command),
+    /// Unwinds a certain block range, deleting it from the database.
+    Unwind(unwind::Command),
 }
 
 impl Command {
@@ -35,6 +38,7 @@ impl Command {
             Subcommands::Run(command) => command.execute().await,
             Subcommands::Drop(command) => command.execute().await,
             Subcommands::Dump(command) => command.execute().await,
+            Subcommands::Unwind(command) => command.execute().await,
         }
     }
 }

bin/reth/src/stage/unwind.rs

@@ -0,0 +1,130 @@
+//! Unwinding a certain block range
+
+use crate::dirs::{DataDirPath, MaybePlatformPath};
+use clap::{Parser, Subcommand};
+use reth_db::{
+    database::Database,
+    mdbx::{Env, WriteMap},
+    tables,
+    transaction::DbTx,
+};
+use reth_primitives::{BlockHashOrNumber, ChainSpec};
+use reth_provider::Transaction;
+use reth_staged_sync::utils::chainspec::genesis_value_parser;
+use std::{ops::RangeInclusive, sync::Arc};
+
+use reth_db::cursor::DbCursorRO;
+
+/// `reth stage unwind` command
+#[derive(Debug, Parser)]
+pub struct Command {
+    /// The path to the data dir for all reth files and subdirectories.
+    ///
+    /// Defaults to the OS-specific data directory:
+    ///
+    /// - Linux: `$XDG_DATA_HOME/reth/` or `$HOME/.local/share/reth/`
+    /// - Windows: `{FOLDERID_RoamingAppData}/reth/`
+    /// - macOS: `$HOME/Library/Application Support/reth/`
+    #[arg(long, value_name = "DATA_DIR", verbatim_doc_comment, default_value_t, global = true)]
+    datadir: MaybePlatformPath<DataDirPath>,
+
+    /// The chain this node is running.
+    ///
+    /// Possible values are either a built-in chain or the path to a chain specification file.
+    ///
+    /// Built-in chains:
+    /// - mainnet
+    /// - goerli
+    /// - sepolia
+    #[arg(
+        long,
+        value_name = "CHAIN_OR_PATH",
+        verbatim_doc_comment,
+        default_value = "mainnet",
+        value_parser = genesis_value_parser,
+        global = true
+    )]
+    chain: Arc<ChainSpec>,
+
+    #[clap(subcommand)]
+    command: Subcommands,
+}
+
+impl Command {
+    /// Execute `db stage unwind` command
+    pub async fn execute(self) -> eyre::Result<()> {
+        // add network name to data dir
+        let data_dir = self.datadir.unwrap_or_chain_default(self.chain.chain);
+        let db_path = data_dir.db_path();
+        if !db_path.exists() {
+            eyre::bail!("Database {db_path:?} does not exist.")
+        }
+
+        let db = Env::<WriteMap>::open(db_path.as_ref(), reth_db::mdbx::EnvKind::RW)?;
+
+        let range = self.command.unwind_range(&db)?;
+
+        if *range.start() == 0 {
+            eyre::bail!("Cannot unwind genesis block")
+        }
+
+        let mut tx = Transaction::new(&db)?;
+
+        let blocks_and_execution = tx
+            .take_block_and_execution_range(&self.chain, range)
+            .map_err(|err| eyre::eyre!("Transaction error on unwind: {err:?}"))?;
+
+        tx.commit()?;
+
+        println!("Unwound {} blocks", blocks_and_execution.len());
+
+        Ok(())
+    }
+}
+
+/// `reth stage unwind` subcommand
+#[derive(Subcommand, Debug, Eq, PartialEq)]
+enum Subcommands {
+    /// Unwinds the database until the given block number (range is inclusive).
+    #[clap(name = "to-block")]
+    ToBlock { target: BlockHashOrNumber },
+    /// Unwinds the given number of blocks from the database.
+    #[clap(name = "num-blocks")]
+    NumBlocks { amount: u64 },
+}
+
+impl Subcommands {
+    /// Returns the block range to unwind.
+    ///
+    /// This returns an inclusive range: [target..=latest]
+    fn unwind_range<DB: Database>(&self, db: DB) -> eyre::Result<RangeInclusive<u64>> {
+        let tx = db.tx()?;
+        let mut cursor = tx.cursor_read::<tables::CanonicalHeaders>()?;
+        let last = cursor.last()?.ok_or_else(|| eyre::eyre!("No blocks in database"))?;
+
+        let target = match self {
+            Subcommands::ToBlock { target } => match target {
+                BlockHashOrNumber::Hash(hash) => tx
+                    .get::<tables::HeaderNumbers>(*hash)?
+                    .ok_or_else(|| eyre::eyre!("Block hash not found in database: {hash:?}"))?,
+                BlockHashOrNumber::Number(num) => *num,
+            },
+            Subcommands::NumBlocks { amount } => last.0.saturating_sub(*amount),
+        };
+        Ok(target..=last.0)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_unwind() {
+        let cmd = Command::parse_from(["reth", "--datadir", "dir", "to-block", "100"]);
+        assert_eq!(cmd.command, Subcommands::ToBlock { target: BlockHashOrNumber::Number(100) });
+
+        let cmd = Command::parse_from(["reth", "--datadir", "dir", "num-blocks", "100"]);
+        assert_eq!(cmd.command, Subcommands::NumBlocks { amount: 100 });
+    }
+}

crates/primitives/src/block.rs

@@ -2,6 +2,7 @@ use crate::{
     Address, BlockHash, BlockNumber, Header, SealedHeader, TransactionSigned, Withdrawal, H256,
 };
 use ethers_core::types::{BlockNumber as EthersBlockNumber, U64};
+use fixed_hash::rustc_hex::FromHexError;
 use reth_codecs::derive_arbitrary;
 use reth_rlp::{Decodable, DecodeError, Encodable, RlpDecodable, RlpEncodable};
 use serde::{
@@ -296,6 +297,32 @@ impl Decodable for BlockHashOrNumber {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+#[error("Failed to parse `{input}` as integer: {pares_int_error} or as hex: {hex_error}")]
+pub struct ParseBlockHashOrNumberError {
+    input: String,
+    pares_int_error: ParseIntError,
+    hex_error: FromHexError,
+}
+
+impl FromStr for BlockHashOrNumber {
+    type Err = ParseBlockHashOrNumberError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match u64::from_str(s) {
+            Ok(val) => Ok(val.into()),
+            Err(pares_int_error) => match H256::from_str(s) {
+                Ok(val) => Ok(val.into()),
+                Err(hex_error) => Err(ParseBlockHashOrNumberError {
+                    input: s.to_string(),
+                    pares_int_error,
+                    hex_error,
+                }),
+            },
+        }
+    }
+}
+
 /// A Block Identifier
 /// <https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1898.md>
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]

crates/storage/db/src/tables/mod.rs

@@ -141,7 +141,7 @@ table!(
 );
 
 table!(
-    /// Stores the block number corresponding to an header.
+    /// Stores the block number corresponding to a header.
     ( HeaderNumbers ) BlockHash | BlockNumber
 );