chore: move stage command to reth-cli-commands (#9384)

Cargo.lock

@@ -6617,8 +6617,10 @@ dependencies = [
  "confy",
  "crossterm",
  "eyre",
+ "fdlimit",
  "human_bytes",
  "itertools 0.13.0",
+ "metrics-process",
  "proptest",
  "proptest-arbitrary-interop",
  "ratatui",
@@ -6627,11 +6629,13 @@ dependencies = [
  "reth-cli-runner",
  "reth-cli-util",
  "reth-config",
+ "reth-consensus",
  "reth-db",
  "reth-db-api",
  "reth-db-common",
  "reth-downloaders",
  "reth-evm",
+ "reth-exex",
  "reth-fs-util",
  "reth-network",
  "reth-network-p2p",

bin/reth/src/cli/mod.rs

@@ -8,13 +8,15 @@ use crate::{
     commands::{
         debug_cmd, import,
         node::{self, NoArgs},
-        stage,
     },
+    macros::block_executor,
     version::{LONG_VERSION, SHORT_VERSION},
 };
 use clap::{value_parser, Parser, Subcommand};
 use reth_chainspec::ChainSpec;
-use reth_cli_commands::{config_cmd, db, dump_genesis, init_cmd, init_state, p2p, prune, recover};
+use reth_cli_commands::{
+    config_cmd, db, dump_genesis, init_cmd, init_state, p2p, prune, recover, stage,
+};
 use reth_cli_runner::CliRunner;
 use reth_db::DatabaseEnv;
 use reth_node_builder::{NodeBuilder, WithLaunchContext};
@@ -159,7 +161,9 @@ impl<Ext: clap::Args + fmt::Debug> Cli<Ext> {
             }
             Commands::DumpGenesis(command) => runner.run_blocking_until_ctrl_c(command.execute()),
             Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute()),
-            Commands::Stage(command) => runner.run_command_until_exit(|ctx| command.execute(ctx)),
+            Commands::Stage(command) => runner.run_command_until_exit(|ctx| {
+                command.execute(ctx, |chain_spec| block_executor!(chain_spec))
+            }),
             Commands::P2P(command) => runner.run_until_ctrl_c(command.execute()),
             #[cfg(feature = "dev")]
             Commands::TestVectors(command) => runner.run_until_ctrl_c(command.execute()),

bin/reth/src/commands/mod.rs

@@ -3,4 +3,3 @@
 pub mod debug_cmd;
 pub mod import;
 pub mod node;
-pub mod stage;

crates/cli/commands/Cargo.toml

@@ -15,11 +15,13 @@ reth-chainspec.workspace = true
 reth-cli-runner.workspace = true
 reth-cli-util.workspace = true
 reth-config.workspace = true
+reth-consensus.workspace = true
 reth-db = { workspace = true, features = ["mdbx"] }
 reth-db-api.workspace = true
 reth-db-common.workspace = true
 reth-downloaders.workspace = true
 reth-evm.workspace = true
+reth-exex.workspace = true
 reth-fs-util.workspace = true
 reth-network = { workspace = true, features = ["serde"] }
 reth-network-p2p.workspace = true
@@ -46,6 +48,7 @@ tracing.workspace = true
 backon.workspace = true
 
 # io
+fdlimit.workspace = true
 confy.workspace = true
 toml = { workspace = true, features = ["display"] }
 
@@ -56,6 +59,9 @@ ratatui = { version = "0.27", default-features = false, features = [
     "crossterm",
 ] }
 
+# metrics
+metrics-process.workspace = true
+
 # reth test-vectors
 proptest = { workspace = true, optional = true }
 arbitrary = { workspace = true, optional = true }
@@ -69,4 +75,4 @@ dev = [
     "dep:proptest-arbitrary-interop",
     "reth-primitives/arbitrary",
     "reth-db-api/arbitrary"
-]
+]

crates/cli/commands/src/lib.rs

@@ -17,5 +17,6 @@ pub mod init_state;
 pub mod p2p;
 pub mod prune;
 pub mod recover;
+pub mod stage;
 #[cfg(feature = "dev")]
 pub mod test_vectors;

crates/cli/commands/src/stage/drop.rs

@@ -1,14 +1,14 @@
 //! Database debugging tool
-use crate::args::StageEnum;
+use crate::common::{AccessRights, Environment, EnvironmentArgs};
 use clap::Parser;
 use itertools::Itertools;
-use reth_cli_commands::common::{AccessRights, Environment, EnvironmentArgs};
 use reth_db::{static_file::iter_static_files, tables, DatabaseEnv};
 use reth_db_api::transaction::DbTxMut;
 use reth_db_common::{
     init::{insert_genesis_header, insert_genesis_history, insert_genesis_state},
     DbTool,
 };
+use reth_node_core::args::StageEnum;
 use reth_provider::{providers::StaticFileWriter, StaticFileProviderFactory};
 use reth_stages::StageId;
 use reth_static_file_types::{find_fixed_range, StaticFileSegment};

crates/cli/commands/src/stage/dump/execution.rs

@@ -1,22 +1,27 @@
 use super::setup;
-use crate::macros::block_executor;
 use reth_db::{tables, DatabaseEnv};
 use reth_db_api::{
     cursor::DbCursorRO, database::Database, table::TableImporter, transaction::DbTx,
 };
 use reth_db_common::DbTool;
+use reth_evm::{execute::BlockExecutorProvider, noop::NoopBlockExecutorProvider};
 use reth_node_core::dirs::{ChainPath, DataDirPath};
-use reth_provider::{providers::StaticFileProvider, ChainSpecProvider, ProviderFactory};
+use reth_provider::{providers::StaticFileProvider, ProviderFactory};
 use reth_stages::{stages::ExecutionStage, Stage, StageCheckpoint, UnwindInput};
 use tracing::info;
 
-pub(crate) async fn dump_execution_stage<DB: Database>(
+pub(crate) async fn dump_execution_stage<DB, E>(
     db_tool: &DbTool<DB>,
     from: u64,
     to: u64,
     output_datadir: ChainPath<DataDirPath>,
     should_run: bool,
-) -> eyre::Result<()> {
+    executor: E,
+) -> eyre::Result<()>
+where
+    DB: Database,
+    E: BlockExecutorProvider,
+{
     let (output_db, tip_block_number) = setup(from, to, &output_datadir.db(), db_tool)?;
 
     import_tables_with_range(&output_db, db_tool, from, to)?;
@@ -32,6 +37,7 @@ pub(crate) async fn dump_execution_stage<DB: Database>(
             ),
             to,
             from,
+            executor,
         )?;
     }
 
@@ -127,8 +133,7 @@ fn unwind_and_copy<DB: Database>(
 ) -> eyre::Result<()> {
     let provider = db_tool.provider_factory.provider_rw()?;
 
-    let executor = block_executor!(db_tool.chain());
+    let mut exec_stage = ExecutionStage::new_with_executor(NoopBlockExecutorProvider::default());
-    let mut exec_stage = ExecutionStage::new_with_executor(executor);
 
     exec_stage.unwind(
         &provider,
@@ -150,14 +155,18 @@ fn unwind_and_copy<DB: Database>(
 }
 
 /// Try to re-execute the stage without committing
-fn dry_run<DB: Database>(
+fn dry_run<DB, E>(
     output_provider_factory: ProviderFactory<DB>,
     to: u64,
     from: u64,
-) -> eyre::Result<()> {
+    executor: E,
+) -> eyre::Result<()>
+where
+    DB: Database,
+    E: BlockExecutorProvider,
+{
     info!(target: "reth::cli", "Executing stage. [dry-run]");
 
-    let executor = block_executor!(output_provider_factory.chain_spec());
     let mut exec_stage = ExecutionStage::new_with_executor(executor);
 
     let input =

crates/cli/commands/src/stage/dump/merkle.rs

@@ -1,10 +1,10 @@
 use super::setup;
-use crate::macros::block_executor;
 use eyre::Result;
 use reth_config::config::EtlConfig;
 use reth_db::{tables, DatabaseEnv};
 use reth_db_api::{database::Database, table::TableImporter};
 use reth_db_common::DbTool;
+use reth_evm::noop::NoopBlockExecutorProvider;
 use reth_exex::ExExManagerHandle;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_primitives::BlockNumber;
@@ -86,11 +86,9 @@ fn unwind_and_copy<DB: Database>(
 
     MerkleStage::default_unwind().unwind(&provider, unwind)?;
 
-    let executor = block_executor!(db_tool.chain());
-
     // Bring Plainstate to TO (hashing stage execution requires it)
     let mut exec_stage = ExecutionStage::new(
-        executor,
+        NoopBlockExecutorProvider::default(), // Not necessary for unwinding.
         ExecutionStageThresholds {
             max_blocks: Some(u64::MAX),
             max_changes: None,

crates/cli/commands/src/stage/dump/mod.rs

@@ -1,15 +1,19 @@
 //! Database debugging tool
-use crate::{args::DatadirArgs, dirs::DataDirPath};
+use crate::common::{AccessRights, Environment, EnvironmentArgs};
 use clap::Parser;
-use reth_cli_commands::common::{AccessRights, Environment, EnvironmentArgs};
+use reth_chainspec::ChainSpec;
 use reth_db::{init_db, mdbx::DatabaseArguments, tables, DatabaseEnv};
 use reth_db_api::{
     cursor::DbCursorRO, database::Database, models::ClientVersion, table::TableImporter,
     transaction::DbTx,
 };
 use reth_db_common::DbTool;
-use reth_node_core::dirs::PlatformPath;
+use reth_evm::execute::BlockExecutorProvider;
-use std::path::PathBuf;
+use reth_node_core::{
+    args::DatadirArgs,
+    dirs::{DataDirPath, PlatformPath},
+};
+use std::{path::PathBuf, sync::Arc};
 use tracing::info;
 
 mod hashing_storage;
@@ -72,16 +76,29 @@ macro_rules! handle_stage {
         let output_datadir = output_datadir.with_chain($tool.chain().chain, DatadirArgs::default());
         $stage_fn($tool, *from, *to, output_datadir, *dry_run).await?
     }};
+
+    ($stage_fn:ident, $tool:expr, $command:expr, $executor:expr) => {{
+        let StageCommand { output_datadir, from, to, dry_run, .. } = $command;
+        let output_datadir = output_datadir.with_chain($tool.chain().chain, DatadirArgs::default());
+        $stage_fn($tool, *from, *to, output_datadir, *dry_run, $executor).await?
+    }};
 }
 
 impl Command {
     /// Execute `dump-stage` command
-    pub async fn execute(self) -> eyre::Result<()> {
+    pub async fn execute<E, F>(self, executor: F) -> eyre::Result<()>
+    where
+        E: BlockExecutorProvider,
+        F: FnOnce(Arc<ChainSpec>) -> E,
+    {
         let Environment { provider_factory, .. } = self.env.init(AccessRights::RO)?;
         let tool = DbTool::new(provider_factory)?;
 
         match &self.command {
-            Stages::Execution(cmd) => handle_stage!(dump_execution_stage, &tool, cmd),
+            Stages::Execution(cmd) => {
+                let executor = executor(tool.chain());
+                handle_stage!(dump_execution_stage, &tool, cmd, executor)
+            }
             Stages::StorageHashing(cmd) => handle_stage!(dump_hashing_storage_stage, &tool, cmd),
             Stages::AccountHashing(cmd) => handle_stage!(dump_hashing_account_stage, &tool, cmd),
             Stages::Merkle(cmd) => handle_stage!(dump_merkle_stage, &tool, cmd),

crates/cli/commands/src/stage/mod.rs

@@ -1,7 +1,11 @@
 //! `reth stage` command
 
+use std::sync::Arc;
+
 use clap::{Parser, Subcommand};
+use reth_chainspec::ChainSpec;
 use reth_cli_runner::CliContext;
+use reth_evm::execute::BlockExecutorProvider;
 
 pub mod drop;
 pub mod dump;
@@ -35,11 +39,15 @@ pub enum Subcommands {
 
 impl Command {
     /// Execute `stage` command
-    pub async fn execute(self, ctx: CliContext) -> eyre::Result<()> {
+    pub async fn execute<E, F>(self, ctx: CliContext, executor: F) -> eyre::Result<()>
+    where
+        E: BlockExecutorProvider,
+        F: FnOnce(Arc<ChainSpec>) -> E,
+    {
         match self.command {
-            Subcommands::Run(command) => command.execute(ctx).await,
+            Subcommands::Run(command) => command.execute(ctx, executor).await,
             Subcommands::Drop(command) => command.execute().await,
-            Subcommands::Dump(command) => command.execute().await,
+            Subcommands::Dump(command) => command.execute(executor).await,
             Subcommands::Unwind(command) => command.execute().await,
         }
     }

crates/cli/commands/src/stage/run.rs

@@ -1,19 +1,20 @@
 //! Main `stage` command
 //!
 //! Stage debugging tool
-use crate::{
+use crate::common::{AccessRights, Environment, EnvironmentArgs};
-    args::{NetworkArgs, StageEnum},
-    macros::block_executor,
-    prometheus_exporter,
-};
 use clap::Parser;
 use reth_beacon_consensus::EthBeaconConsensus;
-use reth_cli_commands::common::{AccessRights, Environment, EnvironmentArgs};
+use reth_chainspec::ChainSpec;
 use reth_cli_runner::CliContext;
 use reth_cli_util::get_secret_key;
 use reth_config::config::{HashingConfig, SenderRecoveryConfig, TransactionLookupConfig};
 use reth_downloaders::bodies::bodies::BodiesDownloaderBuilder;
+use reth_evm::execute::BlockExecutorProvider;
 use reth_exex::ExExManagerHandle;
+use reth_node_core::{
+    args::{NetworkArgs, StageEnum},
+    prometheus_exporter,
+};
 use reth_provider::{
     ChainSpecProvider, StageCheckpointReader, StageCheckpointWriter, StaticFileProviderFactory,
     StaticFileWriter,
@@ -83,7 +84,11 @@ pub struct Command {
 
 impl Command {
     /// Execute `stage` command
-    pub async fn execute(self, ctx: CliContext) -> eyre::Result<()> {
+    pub async fn execute<E, F>(self, ctx: CliContext, executor: F) -> eyre::Result<()>
+    where
+        E: BlockExecutorProvider,
+        F: FnOnce(Arc<ChainSpec>) -> E,
+    {
         // Raise the fd limit of the process.
         // Does not do anything on windows.
         let _ = fdlimit::raise_fd_limit();
@@ -163,24 +168,21 @@ impl Command {
                     })),
                     None,
                 ),
-                StageEnum::Execution => {
+                StageEnum::Execution => (
-                    let executor = block_executor!(provider_factory.chain_spec());
+                    Box::new(ExecutionStage::new(
-                    (
+                        executor(provider_factory.chain_spec()),
-                        Box::new(ExecutionStage::new(
+                        ExecutionStageThresholds {
-                            executor,
+                            max_blocks: Some(batch_size),
-                            ExecutionStageThresholds {
+                            max_changes: None,
-                                max_blocks: Some(batch_size),
+                            max_cumulative_gas: None,
-                                max_changes: None,
+                            max_duration: None,
-                                max_cumulative_gas: None,
+                        },
-                                max_duration: None,
+                        config.stages.merkle.clean_threshold,
-                            },
+                        prune_modes,
-                            config.stages.merkle.clean_threshold,
+                        ExExManagerHandle::empty(),
-                            prune_modes,
+                    )),
-                            ExExManagerHandle::empty(),
+                    None,
-                        )),
+                ),
-                        None,
-                    )
-                }
                 StageEnum::TxLookup => (
                     Box::new(TransactionLookupStage::new(
                         TransactionLookupConfig { chunk_size: batch_size },

crates/cli/commands/src/stage/unwind.rs

@@ -1,13 +1,13 @@
 //! Unwinding a certain block range
 
-use crate::macros::block_executor;
+use crate::common::{AccessRights, Environment, EnvironmentArgs};
 use clap::{Parser, Subcommand};
 use reth_beacon_consensus::EthBeaconConsensus;
-use reth_cli_commands::common::{AccessRights, Environment, EnvironmentArgs};
 use reth_config::Config;
 use reth_consensus::Consensus;
 use reth_db_api::database::Database;
 use reth_downloaders::{bodies::noop::NoopBodiesDownloader, headers::noop::NoopHeaderDownloader};
+use reth_evm::noop::NoopBlockExecutorProvider;
 use reth_exex::ExExManagerHandle;
 use reth_node_core::args::NetworkArgs;
 use reth_primitives::{BlockHashOrNumber, BlockNumber, B256};
@@ -119,7 +119,9 @@ impl Command {
         let prune_modes = config.prune.clone().map(|prune| prune.segments).unwrap_or_default();
 
         let (tip_tx, tip_rx) = watch::channel(B256::ZERO);
-        let executor = block_executor!(provider_factory.chain_spec());
+
+        // Unwinding does not require a valid executor
+        let executor = NoopBlockExecutorProvider::default();
 
         let builder = if self.offline {
             Pipeline::builder().add_stages(