feat(stages): duration threshold for Execution stage (#6073)

bin/reth/src/builder/mod.rs

@@ -912,6 +912,7 @@ impl NodeConfig {
                             max_blocks: stage_config.execution.max_blocks,
                             max_changes: stage_config.execution.max_changes,
                             max_cumulative_gas: stage_config.execution.max_cumulative_gas,
+                            max_duration: stage_config.execution.max_duration,
                         },
                         stage_config
                             .merkle

bin/reth/src/commands/debug_cmd/execution.rs

@@ -136,6 +136,7 @@ impl Command {
                         max_blocks: None,
                         max_changes: None,
                         max_cumulative_gas: None,
+                        max_duration: None,
                     },
                     stage_conf
                         .merkle

bin/reth/src/commands/debug_cmd/merkle.rs

@@ -205,6 +205,7 @@ impl Command {
                 max_blocks: Some(1),
                 max_changes: None,
                 max_cumulative_gas: None,
+                max_duration: None,
             },
             MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD,
             PruneModes::all(),

bin/reth/src/commands/import.rs

@@ -183,6 +183,7 @@ impl ImportCommand {
                         max_blocks: config.stages.execution.max_blocks,
                         max_changes: config.stages.execution.max_changes,
                         max_cumulative_gas: config.stages.execution.max_cumulative_gas,
+                        max_duration: config.stages.execution.max_duration,
                     },
                     config
                         .stages

bin/reth/src/commands/node/events.rs

@@ -122,7 +122,7 @@ impl<DB> NodeState<DB> {
                             %target,
                             %stage_progress,
                             %stage_eta,
-                            message,
+                            "{message}",
                         )
                     } else {
                         info!(
@@ -131,7 +131,7 @@ impl<DB> NodeState<DB> {
                             checkpoint = %checkpoint.block_number,
                             %target,
                             %stage_progress,
-                            message,
+                            "{message}",
                         )
                     }
                 }

bin/reth/src/commands/stage/dump/merkle.rs

@@ -73,6 +73,7 @@ async fn unwind_and_copy<DB: Database>(
             max_blocks: Some(u64::MAX),
             max_changes: None,
             max_cumulative_gas: None,
+            max_duration: None,
         },
         MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD,
         PruneModes::all(),

bin/reth/src/commands/stage/run.rs

@@ -204,6 +204,7 @@ impl Command {
                                 max_blocks: Some(batch_size),
                                 max_changes: None,
                                 max_cumulative_gas: None,
+                                max_duration: None,
                             },
                             config.stages.merkle.clean_threshold,
                             config.prune.map(|prune| prune.segments).unwrap_or_default(),

book/run/config.md

@@ -131,22 +131,21 @@ The execution stage executes historical transactions. This stage is generally ve
 
 Each executed transaction also generates a number of changesets, and mutates the current state of accounts and storage.
 
-For this reason, there are two ways to control how much work to perform before the results are written to disk.
+For this reason, there are several ways to control how much work to perform before the results are written to disk.
 
 ```toml
 [stages.execution]
-# The maximum amount of blocks to execute before writing the results to disk.
+# The maximum number of blocks to process before the execution stage commits.
 max_blocks = 500000
-# The maximum amount of account and storage changes to collect before writing
-# the results to disk.
+# The maximum number of state changes to keep in memory before the execution stage commits.
 max_changes = 5000000
+# The maximum cumulative amount of gas to process before the execution stage commits.
+max_cumulative_gas = 1500000000000 # 30_000_000 * 50_000_000
+# The maximum time spent on blocks processing before the execution stage commits.
+max_duration = '10m'
 ```
 
-Either one of `max_blocks` or `max_changes` must be specified, and both can also be specified at the same time:
-
-- If only `max_blocks` is specified, reth will execute (up to) that amount of blocks before writing to disk.
-- If only `max_changes` is specified, reth will execute as many blocks as possible until the target amount of state transitions have occurred before writing to disk.
-- If both are specified, then the first threshold to be hit will determine when the results are written to disk.
+For all thresholds specified, the first to be hit will determine when the results are written to disk.
 
 Lower values correspond to more frequent disk writes, but also lower memory consumption. A lower value also negatively impacts sync speed, since reth keeps a cache around for the entire duration of blocks executed in the same range.
 

crates/config/src/config.rs

@@ -5,7 +5,7 @@ use reth_network::{NetworkConfigBuilder, PeersConfig, SessionsConfig};
 use reth_primitives::PruneModes;
 use secp256k1::SecretKey;
 use serde::{Deserialize, Serialize};
-use std::path::PathBuf;
+use std::{path::PathBuf, time::Duration};
 
 /// Configuration for the reth node.
 #[derive(Debug, Clone, Default, Deserialize, PartialEq, Serialize)]
@@ -181,10 +181,12 @@ impl Default for SenderRecoveryConfig {
 pub struct ExecutionConfig {
     /// The maximum number of blocks to process before the execution stage commits.
     pub max_blocks: Option<u64>,
-    /// The maximum amount of state changes to keep in memory before the execution stage commits.
+    /// The maximum number of state changes to keep in memory before the execution stage commits.
     pub max_changes: Option<u64>,
-    /// The maximum gas to process before the execution stage commits.
+    /// The maximum cumulative amount of gas to process before the execution stage commits.
     pub max_cumulative_gas: Option<u64>,
+    /// The maximum time spent on blocks processing before the execution stage commits.
+    pub max_duration: Option<Duration>,
 }
 
 impl Default for ExecutionConfig {
@@ -194,6 +196,8 @@ impl Default for ExecutionConfig {
             max_changes: Some(5_000_000),
             // 50k full blocks of 30M gas
             max_cumulative_gas: Some(30_000_000 * 50_000),
+            // 10 minutes
+            max_duration: Some(Duration::from_secs(10 * 60)),
         }
     }
 }

crates/stages/src/stages/execution.rs

@@ -135,12 +135,15 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
         let mut fetch_block_duration = Duration::default();
         let mut execution_duration = Duration::default();
         debug!(target: "sync::stages::execution", start = start_block, end = max_block, "Executing range");
-        // Execute block range
 
+        // Execute block range
         let mut cumulative_gas = 0;
+        let batch_start = Instant::now();
 
         for block_number in start_block..=max_block {
-            let time = Instant::now();
+            // Fetch the block
+            let fetch_block_start = Instant::now();
+
             let td = provider
                 .header_td_by_number(block_number)?
                 .ok_or_else(|| ProviderError::HeaderNotFound(block_number.into()))?;
@@ -150,21 +153,20 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
                 .block_with_senders(block_number.into(), TransactionVariant::NoHash)?
                 .ok_or_else(|| ProviderError::BlockNotFound(block_number.into()))?;
 
-            fetch_block_duration += time.elapsed();
+            fetch_block_duration += fetch_block_start.elapsed();
 
             cumulative_gas += block.gas_used;
 
             // Configure the executor to use the current state.
             trace!(target: "sync::stages::execution", number = block_number, txs = block.body.len(), "Executing block");
 
-            let time = Instant::now();
             // Execute the block
+            let execute_start = Instant::now();
             executor.execute_and_verify_receipt(&block, td).map_err(|error| StageError::Block {
                 block: Box::new(block.header.clone().seal_slow()),
                 error: BlockErrorKind::Execution(error),
             })?;
-
-            execution_duration += time.elapsed();
+            execution_duration += execute_start.elapsed();
 
             // Gas metrics
             if let Some(metrics_tx) = &mut self.metrics_tx {
@@ -182,6 +184,7 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
                 block_number - start_block,
                 bundle_size_hint,
                 cumulative_gas,
+                batch_start.elapsed(),
             ) {
                 break
             }
@@ -451,12 +454,14 @@ impl<EF: ExecutorFactory, DB: Database> Stage<DB> for ExecutionStage<EF> {
 /// current database transaction, which frees up memory.
 #[derive(Debug, Clone)]
 pub struct ExecutionStageThresholds {
-    /// The maximum number of blocks to process before the execution stage commits.
+    /// The maximum number of blocks to execute before the execution stage commits.
     pub max_blocks: Option<u64>,
-    /// The maximum amount of state changes to keep in memory before the execution stage commits.
+    /// The maximum number of state changes to keep in memory before the execution stage commits.
     pub max_changes: Option<u64>,
-    /// The maximum amount of cumultive gas used in the batch.
+    /// The maximum cumulative amount of gas to process before the execution stage commits.
     pub max_cumulative_gas: Option<u64>,
+    /// The maximum spent on blocks processing before the execution stage commits.
+    pub max_duration: Option<Duration>,
 }
 
 impl Default for ExecutionStageThresholds {
@@ -464,8 +469,10 @@ impl Default for ExecutionStageThresholds {
         Self {
             max_blocks: Some(500_000),
             max_changes: Some(5_000_000),
-            // 30M gas per block on 50k blocks
+            // 50k full blocks of 30M gas
             max_cumulative_gas: Some(30_000_000 * 50_000),
+            // 10 minutes
+            max_duration: Some(Duration::from_secs(10 * 60)),
         }
     }
 }
@@ -478,10 +485,12 @@ impl ExecutionStageThresholds {
         blocks_processed: u64,
         changes_processed: u64,
         cumulative_gas_used: u64,
+        elapsed: Duration,
     ) -> bool {
         blocks_processed >= self.max_blocks.unwrap_or(u64::MAX) ||
             changes_processed >= self.max_changes.unwrap_or(u64::MAX) ||
-            cumulative_gas_used >= self.max_cumulative_gas.unwrap_or(u64::MAX)
+            cumulative_gas_used >= self.max_cumulative_gas.unwrap_or(u64::MAX) ||
+            elapsed >= self.max_duration.unwrap_or(Duration::MAX)
     }
 }
 
@@ -511,6 +520,7 @@ mod tests {
                 max_blocks: Some(100),
                 max_changes: None,
                 max_cumulative_gas: None,
+                max_duration: None,
             },
             MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD,
             PruneModes::none(),

crates/stages/src/stages/mod.rs

@@ -135,6 +135,7 @@ mod tests {
                     max_blocks: Some(100),
                     max_changes: None,
                     max_cumulative_gas: None,
+                    max_duration: None,
                 },
                 MERKLE_STAGE_DEFAULT_CLEAN_THRESHOLD,
                 prune_modes.clone(),