feat(reth-bench): add generate-big-block command (#21082)

bin/reth-bench/Cargo.toml

@@ -17,21 +17,26 @@ workspace = true
 reth-cli-runner.workspace = true
 reth-cli-util.workspace = true
 reth-engine-primitives.workspace = true
+reth-ethereum-primitives.workspace = true
 reth-fs-util.workspace = true
 reth-node-api.workspace = true
 reth-node-core.workspace = true
 reth-primitives-traits.workspace = true
+reth-rpc-api.workspace = true
+
 reth-tracing.workspace = true
+reth-chainspec.workspace = true
 
 # alloy
 alloy-eips.workspace = true
 alloy-json-rpc.workspace = true
+alloy-consensus.workspace = true
 alloy-network.workspace = true
 alloy-primitives.workspace = true
 alloy-provider = { workspace = true, features = ["engine-api", "pubsub", "reqwest-rustls-tls"], default-features = false }
 alloy-pubsub.workspace = true
 alloy-rpc-client = { workspace = true, features = ["pubsub"] }
-alloy-rpc-types-engine.workspace = true
+alloy-rpc-types-engine = { workspace = true, features = ["kzg"] }
 alloy-transport-http.workspace = true
 alloy-transport-ipc.workspace = true
 alloy-transport-ws.workspace = true

bin/reth-bench/src/bench/gas_limit_ramp.rs

@@ -0,0 +1,160 @@
+//! Benchmarks empty block processing by ramping the block gas limit.
+
+use crate::{
+    authenticated_transport::AuthenticatedTransportConnect,
+    bench::{
+        helpers::{build_payload, prepare_payload_request, rpc_block_to_header},
+        output::GasRampPayloadFile,
+    },
+    valid_payload::{call_forkchoice_updated, call_new_payload, payload_to_new_payload},
+};
+use alloy_eips::BlockNumberOrTag;
+use alloy_provider::{network::AnyNetwork, Provider, RootProvider};
+use alloy_rpc_client::ClientBuilder;
+use alloy_rpc_types_engine::{ExecutionPayload, ForkchoiceState, JwtSecret};
+
+use clap::Parser;
+use reqwest::Url;
+use reth_chainspec::ChainSpec;
+use reth_cli_runner::CliContext;
+use reth_ethereum_primitives::TransactionSigned;
+use reth_primitives_traits::constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK};
+use std::{path::PathBuf, time::Instant};
+use tracing::info;
+
+/// `reth benchmark gas-limit-ramp` command.
+#[derive(Debug, Parser)]
+pub struct Command {
+    /// Number of blocks to generate.
+    #[arg(long, value_name = "BLOCKS")]
+    blocks: u64,
+
+    /// The Engine API RPC URL.
+    #[arg(long = "engine-rpc-url", value_name = "ENGINE_RPC_URL")]
+    engine_rpc_url: String,
+
+    /// Path to the JWT secret for Engine API authentication.
+    #[arg(long = "jwt-secret", value_name = "JWT_SECRET")]
+    jwt_secret: PathBuf,
+
+    /// Output directory for benchmark results and generated payloads.
+    #[arg(long, value_name = "OUTPUT")]
+    output: PathBuf,
+}
+
+impl Command {
+    /// Execute `benchmark gas-limit-ramp` command.
+    pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
+        if self.blocks == 0 {
+            return Err(eyre::eyre!("--blocks must be greater than 0"));
+        }
+
+        // Ensure output directory exists
+        if self.output.is_file() {
+            return Err(eyre::eyre!("Output path must be a directory"));
+        }
+        if !self.output.exists() {
+            std::fs::create_dir_all(&self.output)?;
+            info!("Created output directory: {:?}", self.output);
+        }
+
+        // Set up authenticated provider (used for both Engine API and eth_ methods)
+        let jwt = std::fs::read_to_string(&self.jwt_secret)?;
+        let jwt = JwtSecret::from_hex(jwt)?;
+        let auth_url = Url::parse(&self.engine_rpc_url)?;
+
+        info!("Connecting to Engine RPC at {}", auth_url);
+        let auth_transport = AuthenticatedTransportConnect::new(auth_url, jwt);
+        let client = ClientBuilder::default().connect_with(auth_transport).await?;
+        let provider = RootProvider::<AnyNetwork>::new(client);
+
+        // Get chain spec - required for fork detection
+        let chain_id = provider.get_chain_id().await?;
+        let chain_spec = ChainSpec::from_chain_id(chain_id)
+            .ok_or_else(|| eyre::eyre!("Unsupported chain id: {chain_id}"))?;
+
+        // Fetch the current head block as parent
+        let parent_block = provider
+            .get_block_by_number(BlockNumberOrTag::Latest)
+            .full()
+            .await?
+            .ok_or_else(|| eyre::eyre!("Failed to fetch latest block"))?;
+
+        let (mut parent_header, mut parent_hash) = rpc_block_to_header(parent_block);
+
+        let canonical_parent = parent_header.number;
+        let start_block = canonical_parent + 1;
+        let end_block = start_block + self.blocks - 1;
+
+        info!(canonical_parent, start_block, end_block, "Starting gas limit ramp benchmark");
+
+        let mut next_block_number = start_block;
+        let total_benchmark_duration = Instant::now();
+
+        while next_block_number <= end_block {
+            let timestamp = parent_header.timestamp.saturating_add(1);
+
+            let request = prepare_payload_request(&chain_spec, timestamp, parent_hash);
+            let new_payload_version = request.new_payload_version;
+
+            let (payload, sidecar) = build_payload(&provider, request).await?;
+
+            let mut block =
+                payload.clone().try_into_block_with_sidecar::<TransactionSigned>(&sidecar)?;
+
+            let max_increase = max_gas_limit_increase(parent_header.gas_limit);
+            let gas_limit =
+                parent_header.gas_limit.saturating_add(max_increase).min(MAXIMUM_GAS_LIMIT_BLOCK);
+
+            block.header.gas_limit = gas_limit;
+
+            let block_hash = block.header.hash_slow();
+            // Regenerate the payload from the modified block, but keep the original sidecar
+            // which contains the actual execution requests data (not just the hash)
+            let (payload, _) = ExecutionPayload::from_block_unchecked(block_hash, &block);
+            let (version, params) = payload_to_new_payload(
+                payload,
+                sidecar,
+                false,
+                block.header.withdrawals_root,
+                Some(new_payload_version),
+            )?;
+
+            // Save payload to file with version info for replay
+            let payload_path =
+                self.output.join(format!("payload_block_{}.json", block.header.number));
+            let file =
+                GasRampPayloadFile { version: version as u8, block_hash, params: params.clone() };
+            let payload_json = serde_json::to_string_pretty(&file)?;
+            std::fs::write(&payload_path, &payload_json)?;
+            info!(block_number = block.header.number, path = %payload_path.display(), "Saved payload");
+
+            call_new_payload(&provider, version, params).await?;
+
+            let forkchoice_state = ForkchoiceState {
+                head_block_hash: block_hash,
+                safe_block_hash: block_hash,
+                finalized_block_hash: block_hash,
+            };
+            call_forkchoice_updated(&provider, version, forkchoice_state, None).await?;
+
+            parent_header = block.header;
+            parent_hash = block_hash;
+            next_block_number += 1;
+        }
+
+        let final_gas_limit = parent_header.gas_limit;
+        info!(
+            total_duration=?total_benchmark_duration.elapsed(),
+            blocks_processed = self.blocks,
+            final_gas_limit,
+            "Benchmark complete"
+        );
+
+        Ok(())
+    }
+}
+
+const fn max_gas_limit_increase(parent_gas_limit: u64) -> u64 {
+    (parent_gas_limit / GAS_LIMIT_BOUND_DIVISOR).saturating_sub(1)
+}

bin/reth-bench/src/bench/generate_big_block.rs

@@ -0,0 +1,617 @@
+//! Command for generating large blocks by packing transactions from real blocks.
+//!
+//! This command fetches transactions from existing blocks and packs them into a single
+//! large block using the `testing_buildBlockV1` RPC endpoint.
+
+use crate::authenticated_transport::AuthenticatedTransportConnect;
+use alloy_eips::{BlockNumberOrTag, Typed2718};
+use alloy_primitives::{Bytes, B256};
+use alloy_provider::{ext::EngineApi, network::AnyNetwork, Provider, RootProvider};
+use alloy_rpc_client::ClientBuilder;
+use alloy_rpc_types_engine::{
+    ExecutionPayloadEnvelopeV4, ExecutionPayloadEnvelopeV5, ForkchoiceState, JwtSecret,
+    PayloadAttributes,
+};
+use alloy_transport::layers::RetryBackoffLayer;
+use clap::Parser;
+use eyre::Context;
+use reqwest::Url;
+use reth_cli_runner::CliContext;
+use reth_rpc_api::TestingBuildBlockRequestV1;
+use std::future::Future;
+use tokio::sync::mpsc;
+use tracing::{info, warn};
+
+/// A single transaction with its gas used and raw encoded bytes.
+#[derive(Debug, Clone)]
+pub struct RawTransaction {
+    /// The actual gas used by the transaction (from receipt).
+    pub gas_used: u64,
+    /// The transaction type (e.g., 3 for EIP-4844 blob txs).
+    pub tx_type: u8,
+    /// The raw RLP-encoded transaction bytes.
+    pub raw: Bytes,
+}
+
+/// Abstraction over sources of transactions for big block generation.
+///
+/// Implementors provide transactions from different sources (RPC, database, files, etc.)
+pub trait TransactionSource {
+    /// Fetch transactions from a specific block number.
+    ///
+    /// Returns `Ok(None)` if the block doesn't exist.
+    /// Returns `Ok(Some((transactions, gas_used)))` with the block's transactions and total gas.
+    fn fetch_block_transactions(
+        &self,
+        block_number: u64,
+    ) -> impl Future<Output = eyre::Result<Option<(Vec<RawTransaction>, u64)>>> + Send;
+}
+
+/// RPC-based transaction source that fetches from a remote node.
+#[derive(Debug)]
+pub struct RpcTransactionSource {
+    provider: RootProvider<AnyNetwork>,
+}
+
+impl RpcTransactionSource {
+    /// Create a new RPC transaction source.
+    pub const fn new(provider: RootProvider<AnyNetwork>) -> Self {
+        Self { provider }
+    }
+
+    /// Create from an RPC URL with retry backoff.
+    pub fn from_url(rpc_url: &str) -> eyre::Result<Self> {
+        let client = ClientBuilder::default()
+            .layer(RetryBackoffLayer::new(10, 800, u64::MAX))
+            .http(rpc_url.parse()?);
+        let provider = RootProvider::<AnyNetwork>::new(client);
+        Ok(Self { provider })
+    }
+}
+
+impl TransactionSource for RpcTransactionSource {
+    async fn fetch_block_transactions(
+        &self,
+        block_number: u64,
+    ) -> eyre::Result<Option<(Vec<RawTransaction>, u64)>> {
+        // Fetch block and receipts in parallel
+        let (block, receipts) = tokio::try_join!(
+            self.provider.get_block_by_number(block_number.into()).full(),
+            self.provider.get_block_receipts(block_number.into())
+        )?;
+
+        let Some(block) = block else {
+            return Ok(None);
+        };
+
+        let Some(receipts) = receipts else {
+            return Err(eyre::eyre!("Receipts not found for block {}", block_number));
+        };
+
+        let block_gas_used = block.header.gas_used;
+
+        // Convert cumulative gas from receipts to per-tx gas_used
+        let mut prev_cumulative = 0u64;
+        let transactions: Vec<RawTransaction> = block
+            .transactions
+            .txns()
+            .zip(receipts.iter())
+            .map(|(tx, receipt)| {
+                let cumulative = receipt.inner.inner.inner.receipt.cumulative_gas_used;
+                let gas_used = cumulative - prev_cumulative;
+                prev_cumulative = cumulative;
+
+                let with_encoded = tx.inner.inner.clone().into_encoded();
+                RawTransaction {
+                    gas_used,
+                    tx_type: tx.inner.ty(),
+                    raw: with_encoded.encoded_bytes().clone(),
+                }
+            })
+            .collect();
+
+        Ok(Some((transactions, block_gas_used)))
+    }
+}
+
+/// Collects transactions from a source up to a target gas usage.
+#[derive(Debug)]
+pub struct TransactionCollector<S> {
+    source: S,
+    target_gas: u64,
+}
+
+impl<S: TransactionSource> TransactionCollector<S> {
+    /// Create a new transaction collector.
+    pub const fn new(source: S, target_gas: u64) -> Self {
+        Self { source, target_gas }
+    }
+
+    /// Collect transactions starting from the given block number.
+    ///
+    /// Skips blob transactions (type 3) and collects until target gas is reached.
+    /// Returns the collected raw transaction bytes, total gas used, and the next block number.
+    pub async fn collect(&self, start_block: u64) -> eyre::Result<(Vec<Bytes>, u64, u64)> {
+        let mut transactions: Vec<Bytes> = Vec::new();
+        let mut total_gas: u64 = 0;
+        let mut current_block = start_block;
+
+        while total_gas < self.target_gas {
+            let Some((block_txs, _)) = self.source.fetch_block_transactions(current_block).await?
+            else {
+                warn!(block = current_block, "Block not found, stopping");
+                break;
+            };
+
+            for tx in block_txs {
+                // Skip blob transactions (EIP-4844, type 3)
+                if tx.tx_type == 3 {
+                    continue;
+                }
+
+                if total_gas + tx.gas_used <= self.target_gas {
+                    transactions.push(tx.raw);
+                    total_gas += tx.gas_used;
+                }
+
+                if total_gas >= self.target_gas {
+                    break;
+                }
+            }
+
+            current_block += 1;
+
+            // Stop early if remaining gas is under 1M (close enough to target)
+            let remaining_gas = self.target_gas.saturating_sub(total_gas);
+            if remaining_gas < 1_000_000 {
+                break;
+            }
+        }
+
+        info!(
+            total_txs = transactions.len(),
+            total_gas,
+            next_block = current_block,
+            "Finished collecting transactions"
+        );
+
+        Ok((transactions, total_gas, current_block))
+    }
+}
+
+/// `reth bench generate-big-block` command
+///
+/// Generates a large block by fetching transactions from existing blocks and packing them
+/// into a single block using the `testing_buildBlockV1` RPC endpoint.
+#[derive(Debug, Parser)]
+pub struct Command {
+    /// The RPC URL to use for fetching blocks (can be an external archive node).
+    #[arg(long, value_name = "RPC_URL")]
+    rpc_url: String,
+
+    /// The engine RPC URL (with JWT authentication).
+    #[arg(long, value_name = "ENGINE_RPC_URL", default_value = "http://localhost:8551")]
+    engine_rpc_url: String,
+
+    /// The RPC URL for `testing_buildBlockV1` calls (same node as engine, regular RPC port).
+    #[arg(long, value_name = "TESTING_RPC_URL", default_value = "http://localhost:8545")]
+    testing_rpc_url: String,
+
+    /// Path to the JWT secret file for engine API authentication.
+    #[arg(long, value_name = "JWT_SECRET")]
+    jwt_secret: std::path::PathBuf,
+
+    /// Target gas to pack into the block.
+    #[arg(long, value_name = "TARGET_GAS", default_value = "30000000")]
+    target_gas: u64,
+
+    /// Starting block number to fetch transactions from.
+    /// If not specified, starts from the engine's latest block.
+    #[arg(long, value_name = "FROM_BLOCK")]
+    from_block: Option<u64>,
+
+    /// Execute the payload (call newPayload + forkchoiceUpdated).
+    /// If false, only builds the payload and prints it.
+    #[arg(long, default_value = "false")]
+    execute: bool,
+
+    /// Number of payloads to generate. Each payload uses the previous as parent.
+    /// When count == 1, the payload is only generated and saved, not executed.
+    /// When count > 1, each payload is executed before building the next.
+    #[arg(long, default_value = "1")]
+    count: u64,
+
+    /// Number of transaction batches to prefetch in background when count > 1.
+    /// Higher values reduce latency but use more memory.
+    #[arg(long, default_value = "4")]
+    prefetch_buffer: usize,
+
+    /// Output directory for generated payloads. Each payload is saved as `payload_block_N.json`.
+    #[arg(long, value_name = "OUTPUT_DIR")]
+    output_dir: std::path::PathBuf,
+}
+
+/// A built payload ready for execution.
+struct BuiltPayload {
+    block_number: u64,
+    envelope: ExecutionPayloadEnvelopeV4,
+    block_hash: B256,
+    timestamp: u64,
+}
+
+impl Command {
+    /// Execute the `generate-big-block` command
+    pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
+        info!(target_gas = self.target_gas, count = self.count, "Generating big block(s)");
+
+        // Set up authenticated engine provider
+        let jwt =
+            std::fs::read_to_string(&self.jwt_secret).wrap_err("Failed to read JWT secret file")?;
+        let jwt = JwtSecret::from_hex(jwt.trim())?;
+        let auth_url = Url::parse(&self.engine_rpc_url)?;
+
+        info!("Connecting to Engine RPC at {}", auth_url);
+        let auth_transport = AuthenticatedTransportConnect::new(auth_url.clone(), jwt);
+        let auth_client = ClientBuilder::default().connect_with(auth_transport).await?;
+        let auth_provider = RootProvider::<AnyNetwork>::new(auth_client);
+
+        // Set up testing RPC provider (for testing_buildBlockV1)
+        info!("Connecting to Testing RPC at {}", self.testing_rpc_url);
+        let testing_client = ClientBuilder::default()
+            .layer(RetryBackoffLayer::new(10, 800, u64::MAX))
+            .http(self.testing_rpc_url.parse()?);
+        let testing_provider = RootProvider::<AnyNetwork>::new(testing_client);
+
+        // Get the parent block (latest canonical block)
+        info!(endpoint = "engine", method = "eth_getBlockByNumber", block = "latest", "RPC call");
+        let parent_block = auth_provider
+            .get_block_by_number(BlockNumberOrTag::Latest)
+            .await?
+            .ok_or_else(|| eyre::eyre!("Failed to fetch latest block"))?;
+
+        let parent_hash = parent_block.header.hash;
+        let parent_number = parent_block.header.number;
+        let parent_timestamp = parent_block.header.timestamp;
+
+        info!(
+            parent_hash = %parent_hash,
+            parent_number = parent_number,
+            "Using initial parent block"
+        );
+
+        // Create output directory
+        std::fs::create_dir_all(&self.output_dir).wrap_err_with(|| {
+            format!("Failed to create output directory: {:?}", self.output_dir)
+        })?;
+
+        let start_block = self.from_block.unwrap_or(parent_number);
+
+        // Use pipelined execution when generating multiple payloads
+        if self.count > 1 {
+            self.execute_pipelined(
+                &auth_provider,
+                &testing_provider,
+                start_block,
+                parent_hash,
+                parent_timestamp,
+            )
+            .await?;
+        } else {
+            // Single payload - collect transactions and build
+            let tx_source = RpcTransactionSource::from_url(&self.rpc_url)?;
+            let collector = TransactionCollector::new(tx_source, self.target_gas);
+            let (transactions, _total_gas, _next_block) = collector.collect(start_block).await?;
+
+            if transactions.is_empty() {
+                return Err(eyre::eyre!("No transactions collected"));
+            }
+
+            self.execute_sequential(
+                &auth_provider,
+                &testing_provider,
+                transactions,
+                parent_hash,
+                parent_timestamp,
+            )
+            .await?;
+        }
+
+        info!(count = self.count, output_dir = %self.output_dir.display(), "All payloads generated");
+        Ok(())
+    }
+
+    /// Sequential execution path for single payload or no-execute mode.
+    async fn execute_sequential(
+        &self,
+        auth_provider: &RootProvider<AnyNetwork>,
+        testing_provider: &RootProvider<AnyNetwork>,
+        transactions: Vec<Bytes>,
+        mut parent_hash: B256,
+        mut parent_timestamp: u64,
+    ) -> eyre::Result<()> {
+        for i in 0..self.count {
+            info!(
+                payload = i + 1,
+                total = self.count,
+                parent_hash = %parent_hash,
+                parent_timestamp = parent_timestamp,
+                "Building payload via testing_buildBlockV1"
+            );
+
+            let built = self
+                .build_payload(testing_provider, &transactions, i, parent_hash, parent_timestamp)
+                .await?;
+
+            self.save_payload(&built)?;
+
+            if self.execute || self.count > 1 {
+                info!(payload = i + 1, block_hash = %built.block_hash, "Executing payload (newPayload + FCU)");
+                self.execute_payload_v4(auth_provider, built.envelope, parent_hash).await?;
+                info!(payload = i + 1, "Payload executed successfully");
+            }
+
+            parent_hash = built.block_hash;
+            parent_timestamp = built.timestamp;
+        }
+        Ok(())
+    }
+
+    /// Pipelined execution - fetches transactions and builds payloads in background.
+    async fn execute_pipelined(
+        &self,
+        auth_provider: &RootProvider<AnyNetwork>,
+        testing_provider: &RootProvider<AnyNetwork>,
+        start_block: u64,
+        initial_parent_hash: B256,
+        initial_parent_timestamp: u64,
+    ) -> eyre::Result<()> {
+        // Create channel for transaction batches (one batch per payload)
+        let (tx_sender, mut tx_receiver) = mpsc::channel::<Vec<Bytes>>(self.prefetch_buffer);
+
+        // Spawn background task to continuously fetch transaction batches
+        let rpc_url = self.rpc_url.clone();
+        let target_gas = self.target_gas;
+        let count = self.count;
+
+        let fetcher_handle = tokio::spawn(async move {
+            let tx_source = match RpcTransactionSource::from_url(&rpc_url) {
+                Ok(source) => source,
+                Err(e) => {
+                    warn!(error = %e, "Failed to create transaction source");
+                    return;
+                }
+            };
+
+            let collector = TransactionCollector::new(tx_source, target_gas);
+            let mut current_block = start_block;
+
+            for payload_idx in 0..count {
+                match collector.collect(current_block).await {
+                    Ok((transactions, total_gas, next_block)) => {
+                        info!(
+                            payload = payload_idx + 1,
+                            tx_count = transactions.len(),
+                            total_gas,
+                            blocks = format!("{}..{}", current_block, next_block),
+                            "Fetched transactions"
+                        );
+                        current_block = next_block;
+
+                        if tx_sender.send(transactions).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(e) => {
+                        warn!(payload = payload_idx + 1, error = %e, "Failed to fetch transactions");
+                        break;
+                    }
+                }
+            }
+        });
+
+        let mut parent_hash = initial_parent_hash;
+        let mut parent_timestamp = initial_parent_timestamp;
+        let mut pending_build: Option<tokio::task::JoinHandle<eyre::Result<BuiltPayload>>> = None;
+
+        for i in 0..self.count {
+            let is_last = i == self.count - 1;
+
+            // Get current payload (either from pending build or build now)
+            let current_payload = if let Some(handle) = pending_build.take() {
+                handle.await??
+            } else {
+                // First payload - wait for transactions and build synchronously
+                let transactions = tx_receiver
+                    .recv()
+                    .await
+                    .ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
+
+                if transactions.is_empty() {
+                    return Err(eyre::eyre!("No transactions collected for payload {}", i + 1));
+                }
+
+                info!(
+                    payload = i + 1,
+                    total = self.count,
+                    parent_hash = %parent_hash,
+                    parent_timestamp = parent_timestamp,
+                    tx_count = transactions.len(),
+                    "Building payload via testing_buildBlockV1"
+                );
+                self.build_payload(
+                    testing_provider,
+                    &transactions,
+                    i,
+                    parent_hash,
+                    parent_timestamp,
+                )
+                .await?
+            };
+
+            self.save_payload(&current_payload)?;
+
+            let current_block_hash = current_payload.block_hash;
+            let current_timestamp = current_payload.timestamp;
+
+            // Execute current payload first
+            info!(payload = i + 1, block_hash = %current_block_hash, "Executing payload (newPayload + FCU)");
+            self.execute_payload_v4(auth_provider, current_payload.envelope, parent_hash).await?;
+            info!(payload = i + 1, "Payload executed successfully");
+
+            // Start building next payload in background (if not last) - AFTER execution
+            if !is_last {
+                // Get transactions for next payload (should already be fetched or fetching)
+                let next_transactions = tx_receiver
+                    .recv()
+                    .await
+                    .ok_or_else(|| eyre::eyre!("Transaction fetcher stopped unexpectedly"))?;
+
+                if next_transactions.is_empty() {
+                    return Err(eyre::eyre!("No transactions collected for payload {}", i + 2));
+                }
+
+                let testing_provider = testing_provider.clone();
+                let next_index = i + 1;
+                let total = self.count;
+
+                pending_build = Some(tokio::spawn(async move {
+                    info!(
+                        payload = next_index + 1,
+                        total = total,
+                        parent_hash = %current_block_hash,
+                        parent_timestamp = current_timestamp,
+                        tx_count = next_transactions.len(),
+                        "Building payload via testing_buildBlockV1"
+                    );
+
+                    Self::build_payload_static(
+                        &testing_provider,
+                        &next_transactions,
+                        next_index,
+                        current_block_hash,
+                        current_timestamp,
+                    )
+                    .await
+                }));
+            }
+
+            parent_hash = current_block_hash;
+            parent_timestamp = current_timestamp;
+        }
+
+        // Clean up the fetcher task
+        drop(tx_receiver);
+        let _ = fetcher_handle.await;
+
+        Ok(())
+    }
+
+    /// Build a single payload via `testing_buildBlockV1`.
+    async fn build_payload(
+        &self,
+        testing_provider: &RootProvider<AnyNetwork>,
+        transactions: &[Bytes],
+        index: u64,
+        parent_hash: B256,
+        parent_timestamp: u64,
+    ) -> eyre::Result<BuiltPayload> {
+        Self::build_payload_static(
+            testing_provider,
+            transactions,
+            index,
+            parent_hash,
+            parent_timestamp,
+        )
+        .await
+    }
+
+    /// Static version for use in spawned tasks.
+    async fn build_payload_static(
+        testing_provider: &RootProvider<AnyNetwork>,
+        transactions: &[Bytes],
+        index: u64,
+        parent_hash: B256,
+        parent_timestamp: u64,
+    ) -> eyre::Result<BuiltPayload> {
+        let request = TestingBuildBlockRequestV1 {
+            parent_block_hash: parent_hash,
+            payload_attributes: PayloadAttributes {
+                timestamp: parent_timestamp + 12,
+                prev_randao: B256::ZERO,
+                suggested_fee_recipient: alloy_primitives::Address::ZERO,
+                withdrawals: Some(vec![]),
+                parent_beacon_block_root: Some(B256::ZERO),
+            },
+            transactions: transactions.to_vec(),
+            extra_data: None,
+        };
+
+        let total_tx_bytes: usize = transactions.iter().map(|tx| tx.len()).sum();
+        info!(
+            payload = index + 1,
+            tx_count = transactions.len(),
+            total_tx_bytes = total_tx_bytes,
+            parent_hash = %parent_hash,
+            "Sending to testing_buildBlockV1"
+        );
+        let envelope: ExecutionPayloadEnvelopeV5 =
+            testing_provider.client().request("testing_buildBlockV1", [request]).await?;
+
+        let v4_envelope = envelope.try_into_v4()?;
+
+        let inner = &v4_envelope.envelope_inner.execution_payload.payload_inner.payload_inner;
+        let block_hash = inner.block_hash;
+        let block_number = inner.block_number;
+        let timestamp = inner.timestamp;
+
+        Ok(BuiltPayload { block_number, envelope: v4_envelope, block_hash, timestamp })
+    }
+
+    /// Save a payload to disk.
+    fn save_payload(&self, payload: &BuiltPayload) -> eyre::Result<()> {
+        let filename = format!("payload_block_{}.json", payload.block_number);
+        let filepath = self.output_dir.join(&filename);
+        let json = serde_json::to_string_pretty(&payload.envelope)?;
+        std::fs::write(&filepath, &json)
+            .wrap_err_with(|| format!("Failed to write payload to {:?}", filepath))?;
+        info!(block_number = payload.block_number, block_hash = %payload.block_hash, path = %filepath.display(), "Payload saved");
+        Ok(())
+    }
+
+    async fn execute_payload_v4(
+        &self,
+        provider: &RootProvider<AnyNetwork>,
+        envelope: ExecutionPayloadEnvelopeV4,
+        parent_hash: B256,
+    ) -> eyre::Result<()> {
+        let block_hash =
+            envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
+
+        let status = provider
+            .new_payload_v4(
+                envelope.envelope_inner.execution_payload,
+                vec![],
+                B256::ZERO,
+                envelope.execution_requests.to_vec(),
+            )
+            .await?;
+
+        if !status.is_valid() {
+            return Err(eyre::eyre!("Payload rejected: {:?}", status));
+        }
+
+        let fcu_state = ForkchoiceState {
+            head_block_hash: block_hash,
+            safe_block_hash: parent_hash,
+            finalized_block_hash: parent_hash,
+        };
+
+        let fcu_result = provider.fork_choice_updated_v3(fcu_state, None).await?;
+
+        if !fcu_result.is_valid() {
+            return Err(eyre::eyre!("FCU rejected: {:?}", fcu_result));
+        }
+
+        Ok(())
+    }
+}

bin/reth-bench/src/bench/helpers.rs

@@ -0,0 +1,196 @@
+//! Common helpers for reth-bench commands.
+
+use crate::valid_payload::call_forkchoice_updated;
+use alloy_consensus::Header;
+use alloy_eips::eip4844::kzg_to_versioned_hash;
+use alloy_primitives::{Address, B256};
+use alloy_provider::{ext::EngineApi, network::AnyNetwork, RootProvider};
+use alloy_rpc_types_engine::{
+    CancunPayloadFields, ExecutionPayload, ExecutionPayloadSidecar, ForkchoiceState,
+    PayloadAttributes, PayloadId, PraguePayloadFields,
+};
+use eyre::OptionExt;
+use reth_chainspec::{ChainSpec, EthereumHardforks};
+use reth_node_api::EngineApiMessageVersion;
+use tracing::debug;
+
+/// Prepared payload request data for triggering block building.
+pub(crate) struct PayloadRequest {
+    /// The payload attributes for the new block.
+    pub(crate) attributes: PayloadAttributes,
+    /// The forkchoice state pointing to the parent block.
+    pub(crate) forkchoice_state: ForkchoiceState,
+    /// The engine API version for FCU calls.
+    pub(crate) fcu_version: EngineApiMessageVersion,
+    /// The getPayload version to use (1-5).
+    pub(crate) get_payload_version: u8,
+    /// The newPayload version to use.
+    pub(crate) new_payload_version: EngineApiMessageVersion,
+}
+
+/// Prepare payload attributes and forkchoice state for a new block.
+pub(crate) fn prepare_payload_request(
+    chain_spec: &ChainSpec,
+    timestamp: u64,
+    parent_hash: B256,
+) -> PayloadRequest {
+    let shanghai_active = chain_spec.is_shanghai_active_at_timestamp(timestamp);
+    let cancun_active = chain_spec.is_cancun_active_at_timestamp(timestamp);
+    let prague_active = chain_spec.is_prague_active_at_timestamp(timestamp);
+    let osaka_active = chain_spec.is_osaka_active_at_timestamp(timestamp);
+
+    // FCU version: V3 for Cancun+Prague+Osaka, V2 for Shanghai, V1 otherwise
+    let fcu_version = if cancun_active {
+        EngineApiMessageVersion::V3
+    } else if shanghai_active {
+        EngineApiMessageVersion::V2
+    } else {
+        EngineApiMessageVersion::V1
+    };
+
+    // getPayload version: 5 for Osaka, 4 for Prague, 3 for Cancun, 2 for Shanghai, 1 otherwise
+    // newPayload version: 4 for Prague+Osaka (no V5), 3 for Cancun, 2 for Shanghai, 1 otherwise
+    let (get_payload_version, new_payload_version) = if osaka_active {
+        (5, EngineApiMessageVersion::V4) // Osaka uses getPayloadV5 but newPayloadV4
+    } else if prague_active {
+        (4, EngineApiMessageVersion::V4)
+    } else if cancun_active {
+        (3, EngineApiMessageVersion::V3)
+    } else if shanghai_active {
+        (2, EngineApiMessageVersion::V2)
+    } else {
+        (1, EngineApiMessageVersion::V1)
+    };
+
+    PayloadRequest {
+        attributes: PayloadAttributes {
+            timestamp,
+            prev_randao: B256::ZERO,
+            suggested_fee_recipient: Address::ZERO,
+            withdrawals: shanghai_active.then(Vec::new),
+            parent_beacon_block_root: cancun_active.then_some(B256::ZERO),
+        },
+        forkchoice_state: ForkchoiceState {
+            head_block_hash: parent_hash,
+            safe_block_hash: parent_hash,
+            finalized_block_hash: parent_hash,
+        },
+        fcu_version,
+        get_payload_version,
+        new_payload_version,
+    }
+}
+
+/// Trigger payload building via FCU and retrieve the built payload.
+///
+/// This sends a forkchoiceUpdated with payload attributes to start building,
+/// then calls getPayload to retrieve the result.
+pub(crate) async fn build_payload(
+    provider: &RootProvider<AnyNetwork>,
+    request: PayloadRequest,
+) -> eyre::Result<(ExecutionPayload, ExecutionPayloadSidecar)> {
+    let fcu_result = call_forkchoice_updated(
+        provider,
+        request.fcu_version,
+        request.forkchoice_state,
+        Some(request.attributes.clone()),
+    )
+    .await?;
+
+    let payload_id =
+        fcu_result.payload_id.ok_or_eyre("Payload builder did not return a payload id")?;
+
+    get_payload_with_sidecar(
+        provider,
+        request.get_payload_version,
+        payload_id,
+        request.attributes.parent_beacon_block_root,
+    )
+    .await
+}
+
+/// Convert an RPC block to a consensus header and block hash.
+pub(crate) fn rpc_block_to_header(block: alloy_provider::network::AnyRpcBlock) -> (Header, B256) {
+    let block_hash = block.header.hash;
+    let header = block.header.inner.clone().into_header_with_defaults();
+    (header, block_hash)
+}
+
+/// Compute versioned hashes from KZG commitments.
+fn versioned_hashes_from_commitments(
+    commitments: &[alloy_primitives::FixedBytes<48>],
+) -> Vec<B256> {
+    commitments.iter().map(|c| kzg_to_versioned_hash(c.as_ref())).collect()
+}
+
+/// Fetch an execution payload using the appropriate engine API version.
+pub(crate) async fn get_payload_with_sidecar(
+    provider: &RootProvider<AnyNetwork>,
+    version: u8,
+    payload_id: PayloadId,
+    parent_beacon_block_root: Option<B256>,
+) -> eyre::Result<(ExecutionPayload, ExecutionPayloadSidecar)> {
+    debug!(get_payload_version = ?version, ?payload_id, "Sending getPayload");
+
+    match version {
+        1 => {
+            let payload = provider.get_payload_v1(payload_id).await?;
+            Ok((ExecutionPayload::V1(payload), ExecutionPayloadSidecar::none()))
+        }
+        2 => {
+            let envelope = provider.get_payload_v2(payload_id).await?;
+            let payload = match envelope.execution_payload {
+                alloy_rpc_types_engine::ExecutionPayloadFieldV2::V1(p) => ExecutionPayload::V1(p),
+                alloy_rpc_types_engine::ExecutionPayloadFieldV2::V2(p) => ExecutionPayload::V2(p),
+            };
+            Ok((payload, ExecutionPayloadSidecar::none()))
+        }
+        3 => {
+            let envelope = provider.get_payload_v3(payload_id).await?;
+            let versioned_hashes =
+                versioned_hashes_from_commitments(&envelope.blobs_bundle.commitments);
+            let cancun_fields = CancunPayloadFields {
+                parent_beacon_block_root: parent_beacon_block_root
+                    .ok_or_eyre("parent_beacon_block_root required for V3")?,
+                versioned_hashes,
+            };
+            Ok((
+                ExecutionPayload::V3(envelope.execution_payload),
+                ExecutionPayloadSidecar::v3(cancun_fields),
+            ))
+        }
+        4 => {
+            let envelope = provider.get_payload_v4(payload_id).await?;
+            let versioned_hashes = versioned_hashes_from_commitments(
+                &envelope.envelope_inner.blobs_bundle.commitments,
+            );
+            let cancun_fields = CancunPayloadFields {
+                parent_beacon_block_root: parent_beacon_block_root
+                    .ok_or_eyre("parent_beacon_block_root required for V4")?,
+                versioned_hashes,
+            };
+            let prague_fields = PraguePayloadFields::new(envelope.execution_requests);
+            Ok((
+                ExecutionPayload::V3(envelope.envelope_inner.execution_payload),
+                ExecutionPayloadSidecar::v4(cancun_fields, prague_fields),
+            ))
+        }
+        5 => {
+            // V5 (Osaka) - use raw request since alloy doesn't have get_payload_v5 yet
+            let envelope = provider.get_payload_v5(payload_id).await?;
+            let versioned_hashes =
+                versioned_hashes_from_commitments(&envelope.blobs_bundle.commitments);
+            let cancun_fields = CancunPayloadFields {
+                parent_beacon_block_root: parent_beacon_block_root
+                    .ok_or_eyre("parent_beacon_block_root required for V5")?,
+                versioned_hashes,
+            };
+            let prague_fields = PraguePayloadFields::new(envelope.execution_requests);
+            Ok((
+                ExecutionPayload::V3(envelope.execution_payload),
+                ExecutionPayloadSidecar::v4(cancun_fields, prague_fields),
+            ))
+        }
+        _ => panic!("This tool does not support getPayload versions past v5"),
+    }
+}

bin/reth-bench/src/bench/mod.rs

@@ -6,9 +6,16 @@ use reth_node_core::args::LogArgs;
 use reth_tracing::FileWorkerGuard;
 
 mod context;
+mod gas_limit_ramp;
+mod generate_big_block;
+pub(crate) mod helpers;
+pub use generate_big_block::{
+    RawTransaction, RpcTransactionSource, TransactionCollector, TransactionSource,
+};
 mod new_payload_fcu;
 mod new_payload_only;
 mod output;
+mod replay_payloads;
 mod send_payload;
 
 /// `reth bench` command
@@ -27,6 +34,9 @@ pub enum Subcommands {
     /// Benchmark which calls `newPayload`, then `forkchoiceUpdated`.
     NewPayloadFcu(new_payload_fcu::Command),
 
+    /// Benchmark which builds empty blocks with a ramped gas limit.
+    GasLimitRamp(gas_limit_ramp::Command),
+
     /// Benchmark which only calls subsequent `newPayload` calls.
     NewPayloadOnly(new_payload_only::Command),
 
@@ -41,6 +51,29 @@ pub enum Subcommands {
     /// `cast block latest --full --json | reth-bench send-payload --rpc-url localhost:5000
     /// --jwt-secret $(cat ~/.local/share/reth/mainnet/jwt.hex)`
     SendPayload(send_payload::Command),
+
+    /// Generate a large block by packing transactions from existing blocks.
+    ///
+    /// This command fetches transactions from real blocks and packs them into a single
+    /// block using the `testing_buildBlockV1` RPC endpoint.
+    ///
+    /// Example:
+    ///
+    /// `reth-bench generate-big-block --rpc-url http://localhost:8545 --engine-rpc-url
+    /// http://localhost:8551 --jwt-secret ~/.local/share/reth/mainnet/jwt.hex --target-gas
+    /// 30000000`
+    GenerateBigBlock(generate_big_block::Command),
+
+    /// Replay pre-generated payloads from a directory.
+    ///
+    /// This command reads payload files from a previous `generate-big-block` run and replays
+    /// them in sequence using `newPayload` followed by `forkchoiceUpdated`.
+    ///
+    /// Example:
+    ///
+    /// `reth-bench replay-payloads --payload-dir ./payloads --engine-rpc-url
+    /// http://localhost:8551 --jwt-secret ~/.local/share/reth/mainnet/jwt.hex`
+    ReplayPayloads(replay_payloads::Command),
 }
 
 impl BenchmarkCommand {
@@ -51,8 +84,11 @@ impl BenchmarkCommand {
 
         match self.command {
             Subcommands::NewPayloadFcu(command) => command.execute(ctx).await,
+            Subcommands::GasLimitRamp(command) => command.execute(ctx).await,
             Subcommands::NewPayloadOnly(command) => command.execute(ctx).await,
             Subcommands::SendPayload(command) => command.execute(ctx).await,
+            Subcommands::GenerateBigBlock(command) => command.execute(ctx).await,
+            Subcommands::ReplayPayloads(command) => command.execute(ctx).await,
         }
     }
 

bin/reth-bench/src/bench/new_payload_fcu.rs

@@ -13,8 +13,7 @@ use crate::{
     bench::{
         context::BenchContext,
         output::{
-            CombinedResult, NewPayloadResult, TotalGasOutput, TotalGasRow, COMBINED_OUTPUT_SUFFIX,
-            GAS_OUTPUT_SUFFIX,
+            write_benchmark_results, CombinedResult, NewPayloadResult, TotalGasOutput, TotalGasRow,
         },
     },
     valid_payload::{block_to_new_payload, call_forkchoice_updated, call_new_payload},
@@ -27,7 +26,6 @@ use alloy_rpc_client::RpcClient;
 use alloy_rpc_types_engine::ForkchoiceState;
 use alloy_transport_ws::WsConnect;
 use clap::Parser;
-use csv::Writer;
 use eyre::{Context, OptionExt};
 use futures::StreamExt;
 use humantime::parse_duration;
@@ -123,6 +121,7 @@ impl Command {
             auth_provider,
             mut next_block,
             is_optimism,
+            ..
         } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
 
         let buffer_size = self.rpc_block_buffer_size;
@@ -188,6 +187,7 @@ impl Command {
             result
         } {
             let gas_used = block.header.gas_used;
+            let gas_limit = block.header.gas_limit;
             let block_number = block.header.number;
             let transaction_count = block.transactions.len() as u64;
 
@@ -211,6 +211,7 @@ impl Command {
             let fcu_latency = total_latency - new_payload_result.latency;
             let combined_result = CombinedResult {
                 block_number,
+                gas_limit,
                 transaction_count,
                 new_payload_result,
                 fcu_latency,
@@ -240,28 +241,11 @@ impl Command {
         // since the benchmark goal is measuring Ggas/s of newPayload/FCU, not persistence.
         drop(waiter);
 
-        let (gas_output_results, combined_results): (_, Vec<CombinedResult>) =
+        let (gas_output_results, combined_results): (Vec<TotalGasRow>, Vec<CombinedResult>) =
             results.into_iter().unzip();
 
-        // Write CSV output files
         if let Some(ref path) = self.benchmark.output {
-            let output_path = path.join(COMBINED_OUTPUT_SUFFIX);
-            info!("Writing engine api call latency output to file: {:?}", output_path);
-            let mut writer = Writer::from_path(&output_path)?;
-            for result in combined_results {
-                writer.serialize(result)?;
-            }
-            writer.flush()?;
-
-            let output_path = path.join(GAS_OUTPUT_SUFFIX);
-            info!("Writing total gas output to file: {:?}", output_path);
-            let mut writer = Writer::from_path(&output_path)?;
-            for row in &gas_output_results {
-                writer.serialize(row)?;
-            }
-            writer.flush()?;
-
-            info!("Finished writing benchmark output files to {:?}.", path);
+            write_benchmark_results(path, &gas_output_results, combined_results)?;
         }
 
         let gas_output = TotalGasOutput::new(gas_output_results)?;

bin/reth-bench/src/bench/new_payload_only.rs

@@ -49,6 +49,7 @@ impl Command {
             auth_provider,
             mut next_block,
             is_optimism,
+            ..
         } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
 
         let buffer_size = self.rpc_block_buffer_size;
@@ -96,11 +97,7 @@ impl Command {
             let transaction_count = block.transactions.len() as u64;
             let gas_used = block.header.gas_used;
 
-            debug!(
-                target: "reth-bench",
-                number=?block.header.number,
-                "Sending payload to engine",
-            );
+            debug!(number=?block.header.number, "Sending payload to engine");
 
             let (version, params) = block_to_new_payload(block, is_optimism)?;
 

bin/reth-bench/src/bench/output.rs

@@ -1,10 +1,13 @@
 //! Contains various benchmark output formats, either for logging or for
 //! serialization to / from files.
 
+use alloy_primitives::B256;
+use csv::Writer;
 use eyre::OptionExt;
 use reth_primitives_traits::constants::GIGAGAS;
-use serde::{ser::SerializeStruct, Serialize};
-use std::time::Duration;
+use serde::{ser::SerializeStruct, Deserialize, Serialize};
+use std::{path::Path, time::Duration};
+use tracing::info;
 
 /// This is the suffix for gas output csv files.
 pub(crate) const GAS_OUTPUT_SUFFIX: &str = "total_gas.csv";
@@ -15,6 +18,17 @@ pub(crate) const COMBINED_OUTPUT_SUFFIX: &str = "combined_latency.csv";
 /// This is the suffix for new payload output csv files.
 pub(crate) const NEW_PAYLOAD_OUTPUT_SUFFIX: &str = "new_payload_latency.csv";
 
+/// Serialized format for gas ramp payloads on disk.
+#[derive(Debug, Serialize, Deserialize)]
+pub(crate) struct GasRampPayloadFile {
+    /// Engine API version (1-5).
+    pub(crate) version: u8,
+    /// The block hash for FCU.
+    pub(crate) block_hash: B256,
+    /// The params to pass to newPayload.
+    pub(crate) params: serde_json::Value,
+}
+
 /// This represents the results of a single `newPayload` call in the benchmark, containing the gas
 /// used and the `newPayload` latency.
 #[derive(Debug)]
@@ -67,6 +81,8 @@ impl Serialize for NewPayloadResult {
 pub(crate) struct CombinedResult {
     /// The block number of the block being processed.
     pub(crate) block_number: u64,
+    /// The gas limit of the block.
+    pub(crate) gas_limit: u64,
     /// The number of transactions in the block.
     pub(crate) transaction_count: u64,
     /// The `newPayload` result.
@@ -88,7 +104,7 @@ impl std::fmt::Display for CombinedResult {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "Payload {} processed at {:.4} Ggas/s, used {} total gas. Combined gas per second: {:.4} Ggas/s. fcu latency: {:?}, newPayload latency: {:?}",
+            "Block {} processed at {:.4} Ggas/s, used {} total gas. Combined: {:.4} Ggas/s. fcu: {:?}, newPayload: {:?}",
             self.block_number,
             self.new_payload_result.gas_per_second() / GIGAGAS as f64,
             self.new_payload_result.gas_used,
@@ -110,10 +126,11 @@ impl Serialize for CombinedResult {
         let fcu_latency = self.fcu_latency.as_micros();
         let new_payload_latency = self.new_payload_result.latency.as_micros();
         let total_latency = self.total_latency.as_micros();
-        let mut state = serializer.serialize_struct("CombinedResult", 6)?;
+        let mut state = serializer.serialize_struct("CombinedResult", 7)?;
 
         // flatten the new payload result because this is meant for CSV writing
         state.serialize_field("block_number", &self.block_number)?;
+        state.serialize_field("gas_limit", &self.gas_limit)?;
         state.serialize_field("transaction_count", &self.transaction_count)?;
         state.serialize_field("gas_used", &self.new_payload_result.gas_used)?;
         state.serialize_field("new_payload_latency", &new_payload_latency)?;
@@ -167,6 +184,36 @@ impl TotalGasOutput {
     }
 }
 
+/// Write benchmark results to CSV files.
+///
+/// Writes two files to the output directory:
+/// - `combined_latency.csv`: Per-block latency results
+/// - `total_gas.csv`: Per-block gas usage over time
+pub(crate) fn write_benchmark_results(
+    output_dir: &Path,
+    gas_results: &[TotalGasRow],
+    combined_results: Vec<CombinedResult>,
+) -> eyre::Result<()> {
+    let output_path = output_dir.join(COMBINED_OUTPUT_SUFFIX);
+    info!("Writing engine api call latency output to file: {:?}", output_path);
+    let mut writer = Writer::from_path(&output_path)?;
+    for result in combined_results {
+        writer.serialize(result)?;
+    }
+    writer.flush()?;
+
+    let output_path = output_dir.join(GAS_OUTPUT_SUFFIX);
+    info!("Writing total gas output to file: {:?}", output_path);
+    let mut writer = Writer::from_path(&output_path)?;
+    for row in gas_results {
+        writer.serialize(row)?;
+    }
+    writer.flush()?;
+
+    info!("Finished writing benchmark output files to {:?}.", output_dir);
+    Ok(())
+}
+
 /// This serializes the `time` field of the [`TotalGasRow`] to microseconds.
 ///
 /// This is essentially just for the csv writer, which would have headers

bin/reth-bench/src/bench/replay_payloads.rs

@@ -0,0 +1,332 @@
+//! Command for replaying pre-generated payloads from disk.
+//!
+//! This command reads `ExecutionPayloadEnvelopeV4` files from a directory and replays them
+//! in sequence using `newPayload` followed by `forkchoiceUpdated`.
+
+use crate::{
+    authenticated_transport::AuthenticatedTransportConnect,
+    bench::output::GasRampPayloadFile,
+    valid_payload::{call_forkchoice_updated, call_new_payload},
+};
+use alloy_primitives::B256;
+use alloy_provider::{ext::EngineApi, network::AnyNetwork, Provider, RootProvider};
+use alloy_rpc_client::ClientBuilder;
+use alloy_rpc_types_engine::{ExecutionPayloadEnvelopeV4, ForkchoiceState, JwtSecret};
+use clap::Parser;
+use eyre::Context;
+use reqwest::Url;
+use reth_cli_runner::CliContext;
+use reth_node_api::EngineApiMessageVersion;
+use std::path::PathBuf;
+use tracing::{debug, info};
+
+/// `reth bench replay-payloads` command
+///
+/// Replays pre-generated payloads from a directory by calling `newPayload` followed by
+/// `forkchoiceUpdated` for each payload in sequence.
+#[derive(Debug, Parser)]
+pub struct Command {
+    /// The engine RPC URL (with JWT authentication).
+    #[arg(long, value_name = "ENGINE_RPC_URL", default_value = "http://localhost:8551")]
+    engine_rpc_url: String,
+
+    /// Path to the JWT secret file for engine API authentication.
+    #[arg(long, value_name = "JWT_SECRET")]
+    jwt_secret: PathBuf,
+
+    /// Directory containing payload files (`payload_block_N.json`).
+    #[arg(long, value_name = "PAYLOAD_DIR")]
+    payload_dir: PathBuf,
+
+    /// Optional limit on the number of payloads to replay.
+    /// If not specified, replays all payloads in the directory.
+    #[arg(long, value_name = "COUNT")]
+    count: Option<usize>,
+
+    /// Skip the first N payloads.
+    #[arg(long, value_name = "SKIP", default_value = "0")]
+    skip: usize,
+
+    /// Optional directory containing gas ramp payloads to replay first.
+    /// These are replayed before the main payloads to warm up the gas limit.
+    #[arg(long, value_name = "GAS_RAMP_DIR")]
+    gas_ramp_dir: Option<PathBuf>,
+}
+
+/// A loaded payload ready for execution.
+struct LoadedPayload {
+    /// The index (from filename).
+    index: u64,
+    /// The payload envelope.
+    envelope: ExecutionPayloadEnvelopeV4,
+    /// The block hash.
+    block_hash: B256,
+}
+
+/// A gas ramp payload loaded from disk.
+struct GasRampPayload {
+    /// Block number from filename.
+    block_number: u64,
+    /// Engine API version for newPayload.
+    version: EngineApiMessageVersion,
+    /// The file contents.
+    file: GasRampPayloadFile,
+}
+
+impl Command {
+    /// Execute the `replay-payloads` command.
+    pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
+        info!(payload_dir = %self.payload_dir.display(), "Replaying payloads");
+
+        // Set up authenticated engine provider
+        let jwt =
+            std::fs::read_to_string(&self.jwt_secret).wrap_err("Failed to read JWT secret file")?;
+        let jwt = JwtSecret::from_hex(jwt.trim())?;
+        let auth_url = Url::parse(&self.engine_rpc_url)?;
+
+        info!("Connecting to Engine RPC at {}", auth_url);
+        let auth_transport = AuthenticatedTransportConnect::new(auth_url.clone(), jwt);
+        let auth_client = ClientBuilder::default().connect_with(auth_transport).await?;
+        let auth_provider = RootProvider::<AnyNetwork>::new(auth_client);
+
+        // Get parent block (latest canonical block) - we need this for the first FCU
+        let parent_block = auth_provider
+            .get_block_by_number(alloy_eips::BlockNumberOrTag::Latest)
+            .await?
+            .ok_or_else(|| eyre::eyre!("Failed to fetch latest block"))?;
+
+        let initial_parent_hash = parent_block.header.hash;
+        let initial_parent_number = parent_block.header.number;
+
+        info!(
+            parent_hash = %initial_parent_hash,
+            parent_number = initial_parent_number,
+            "Using initial parent block"
+        );
+
+        // Load all payloads upfront to avoid I/O delays between phases
+        let gas_ramp_payloads = if let Some(ref gas_ramp_dir) = self.gas_ramp_dir {
+            let payloads = self.load_gas_ramp_payloads(gas_ramp_dir)?;
+            if payloads.is_empty() {
+                return Err(eyre::eyre!("No gas ramp payload files found in {:?}", gas_ramp_dir));
+            }
+            info!(count = payloads.len(), "Loaded gas ramp payloads from disk");
+            payloads
+        } else {
+            Vec::new()
+        };
+
+        let payloads = self.load_payloads()?;
+        if payloads.is_empty() {
+            return Err(eyre::eyre!("No payload files found in {:?}", self.payload_dir));
+        }
+        info!(count = payloads.len(), "Loaded main payloads from disk");
+
+        let mut parent_hash = initial_parent_hash;
+
+        // Replay gas ramp payloads first
+        for (i, payload) in gas_ramp_payloads.iter().enumerate() {
+            info!(
+                gas_ramp_payload = i + 1,
+                total = gas_ramp_payloads.len(),
+                block_number = payload.block_number,
+                block_hash = %payload.file.block_hash,
+                "Executing gas ramp payload (newPayload + FCU)"
+            );
+
+            call_new_payload(&auth_provider, payload.version, payload.file.params.clone()).await?;
+
+            let fcu_state = ForkchoiceState {
+                head_block_hash: payload.file.block_hash,
+                safe_block_hash: parent_hash,
+                finalized_block_hash: parent_hash,
+            };
+            call_forkchoice_updated(&auth_provider, payload.version, fcu_state, None).await?;
+
+            info!(gas_ramp_payload = i + 1, "Gas ramp payload executed successfully");
+            parent_hash = payload.file.block_hash;
+        }
+
+        if !gas_ramp_payloads.is_empty() {
+            info!(count = gas_ramp_payloads.len(), "All gas ramp payloads replayed");
+        }
+
+        for (i, payload) in payloads.iter().enumerate() {
+            info!(
+                payload = i + 1,
+                total = payloads.len(),
+                index = payload.index,
+                block_hash = %payload.block_hash,
+                "Executing payload (newPayload + FCU)"
+            );
+
+            self.execute_payload_v4(&auth_provider, &payload.envelope, parent_hash).await?;
+
+            info!(payload = i + 1, "Payload executed successfully");
+            parent_hash = payload.block_hash;
+        }
+
+        info!(count = payloads.len(), "All payloads replayed successfully");
+        Ok(())
+    }
+
+    /// Load and parse all payload files from the directory.
+    fn load_payloads(&self) -> eyre::Result<Vec<LoadedPayload>> {
+        let mut payloads = Vec::new();
+
+        // Read directory entries
+        let entries: Vec<_> = std::fs::read_dir(&self.payload_dir)
+            .wrap_err_with(|| format!("Failed to read directory {:?}", self.payload_dir))?
+            .filter_map(|e| e.ok())
+            .filter(|e| {
+                e.path().extension().and_then(|s| s.to_str()) == Some("json") &&
+                    e.file_name().to_string_lossy().starts_with("payload_")
+            })
+            .collect();
+
+        // Parse filenames to get indices and sort
+        let mut indexed_paths: Vec<(u64, PathBuf)> = entries
+            .into_iter()
+            .filter_map(|e| {
+                let name = e.file_name();
+                let name_str = name.to_string_lossy();
+                // Extract index from "payload_NNN.json"
+                let index_str = name_str.strip_prefix("payload_")?.strip_suffix(".json")?;
+                let index: u64 = index_str.parse().ok()?;
+                Some((index, e.path()))
+            })
+            .collect();
+
+        indexed_paths.sort_by_key(|(idx, _)| *idx);
+
+        // Apply skip and count
+        let indexed_paths: Vec<_> = indexed_paths.into_iter().skip(self.skip).collect();
+        let indexed_paths: Vec<_> = match self.count {
+            Some(count) => indexed_paths.into_iter().take(count).collect(),
+            None => indexed_paths,
+        };
+
+        // Load each payload
+        for (index, path) in indexed_paths {
+            let content = std::fs::read_to_string(&path)
+                .wrap_err_with(|| format!("Failed to read {:?}", path))?;
+            let envelope: ExecutionPayloadEnvelopeV4 = serde_json::from_str(&content)
+                .wrap_err_with(|| format!("Failed to parse {:?}", path))?;
+
+            let block_hash =
+                envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
+
+            info!(
+                index = index,
+                block_hash = %block_hash,
+                path = %path.display(),
+                "Loaded payload"
+            );
+
+            payloads.push(LoadedPayload { index, envelope, block_hash });
+        }
+
+        Ok(payloads)
+    }
+
+    /// Load and parse gas ramp payload files from a directory.
+    fn load_gas_ramp_payloads(&self, dir: &PathBuf) -> eyre::Result<Vec<GasRampPayload>> {
+        let mut payloads = Vec::new();
+
+        let entries: Vec<_> = std::fs::read_dir(dir)
+            .wrap_err_with(|| format!("Failed to read directory {:?}", dir))?
+            .filter_map(|e| e.ok())
+            .filter(|e| {
+                e.path().extension().and_then(|s| s.to_str()) == Some("json") &&
+                    e.file_name().to_string_lossy().starts_with("payload_block_")
+            })
+            .collect();
+
+        // Parse filenames to get block numbers and sort
+        let mut indexed_paths: Vec<(u64, PathBuf)> = entries
+            .into_iter()
+            .filter_map(|e| {
+                let name = e.file_name();
+                let name_str = name.to_string_lossy();
+                // Extract block number from "payload_block_NNN.json"
+                let block_str = name_str.strip_prefix("payload_block_")?.strip_suffix(".json")?;
+                let block_number: u64 = block_str.parse().ok()?;
+                Some((block_number, e.path()))
+            })
+            .collect();
+
+        indexed_paths.sort_by_key(|(num, _)| *num);
+
+        for (block_number, path) in indexed_paths {
+            let content = std::fs::read_to_string(&path)
+                .wrap_err_with(|| format!("Failed to read {:?}", path))?;
+            let file: GasRampPayloadFile = serde_json::from_str(&content)
+                .wrap_err_with(|| format!("Failed to parse {:?}", path))?;
+
+            let version = match file.version {
+                1 => EngineApiMessageVersion::V1,
+                2 => EngineApiMessageVersion::V2,
+                3 => EngineApiMessageVersion::V3,
+                4 => EngineApiMessageVersion::V4,
+                5 => EngineApiMessageVersion::V5,
+                v => return Err(eyre::eyre!("Invalid version {} in {:?}", v, path)),
+            };
+
+            info!(
+                block_number,
+                block_hash = %file.block_hash,
+                path = %path.display(),
+                "Loaded gas ramp payload"
+            );
+
+            payloads.push(GasRampPayload { block_number, version, file });
+        }
+
+        Ok(payloads)
+    }
+
+    async fn execute_payload_v4(
+        &self,
+        provider: &RootProvider<AnyNetwork>,
+        envelope: &ExecutionPayloadEnvelopeV4,
+        parent_hash: B256,
+    ) -> eyre::Result<()> {
+        let block_hash =
+            envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
+
+        debug!(
+            method = "engine_newPayloadV4",
+            block_hash = %block_hash,
+            "Sending newPayload"
+        );
+
+        let status = provider
+            .new_payload_v4(
+                envelope.envelope_inner.execution_payload.clone(),
+                vec![],
+                B256::ZERO,
+                envelope.execution_requests.to_vec(),
+            )
+            .await?;
+
+        info!(?status, "newPayloadV4 response");
+
+        if !status.is_valid() {
+            return Err(eyre::eyre!("Payload rejected: {:?}", status));
+        }
+
+        let fcu_state = ForkchoiceState {
+            head_block_hash: block_hash,
+            safe_block_hash: parent_hash,
+            finalized_block_hash: parent_hash,
+        };
+
+        debug!(method = "engine_forkchoiceUpdatedV3", ?fcu_state, "Sending forkchoiceUpdated");
+
+        let fcu_result = provider.fork_choice_updated_v3(fcu_state, None).await?;
+
+        info!(?fcu_result, "forkchoiceUpdatedV3 response");
+
+        Ok(())
+    }
+}

bin/reth-bench/src/valid_payload.rs

@@ -3,15 +3,16 @@
 //! before sending additional calls.
 
 use alloy_eips::eip7685::Requests;
+use alloy_primitives::B256;
 use alloy_provider::{ext::EngineApi, network::AnyRpcBlock, Network, Provider};
 use alloy_rpc_types_engine::{
-    ExecutionPayload, ExecutionPayloadInputV2, ForkchoiceState, ForkchoiceUpdated,
-    PayloadAttributes, PayloadStatus,
+    ExecutionPayload, ExecutionPayloadInputV2, ExecutionPayloadSidecar, ForkchoiceState,
+    ForkchoiceUpdated, PayloadAttributes, PayloadStatus,
 };
 use alloy_transport::TransportResult;
 use op_alloy_rpc_types_engine::OpExecutionPayloadV4;
 use reth_node_api::EngineApiMessageVersion;
-use tracing::error;
+use tracing::{debug, error};
 
 /// An extension trait for providers that implement the engine API, to wait for a VALID response.
 #[async_trait::async_trait]
@@ -52,6 +53,13 @@ where
         fork_choice_state: ForkchoiceState,
         payload_attributes: Option<PayloadAttributes>,
     ) -> TransportResult<ForkchoiceUpdated> {
+        debug!(
+            method = "engine_forkchoiceUpdatedV1",
+            ?fork_choice_state,
+            ?payload_attributes,
+            "Sending forkchoiceUpdated"
+        );
+
         let mut status =
             self.fork_choice_updated_v1(fork_choice_state, payload_attributes.clone()).await?;
 
@@ -82,6 +90,13 @@ where
         fork_choice_state: ForkchoiceState,
         payload_attributes: Option<PayloadAttributes>,
     ) -> TransportResult<ForkchoiceUpdated> {
+        debug!(
+            method = "engine_forkchoiceUpdatedV2",
+            ?fork_choice_state,
+            ?payload_attributes,
+            "Sending forkchoiceUpdated"
+        );
+
         let mut status =
             self.fork_choice_updated_v2(fork_choice_state, payload_attributes.clone()).await?;
 
@@ -112,6 +127,13 @@ where
         fork_choice_state: ForkchoiceState,
         payload_attributes: Option<PayloadAttributes>,
     ) -> TransportResult<ForkchoiceUpdated> {
+        debug!(
+            method = "engine_forkchoiceUpdatedV3",
+            ?fork_choice_state,
+            ?payload_attributes,
+            "Sending forkchoiceUpdated"
+        );
+
         let mut status =
             self.fork_choice_updated_v3(fork_choice_state, payload_attributes.clone()).await?;
 
@@ -148,33 +170,51 @@ pub(crate) fn block_to_new_payload(
 
     // Convert to execution payload
     let (payload, sidecar) = ExecutionPayload::from_block_slow(&block);
+    payload_to_new_payload(payload, sidecar, is_optimism, block.withdrawals_root, None)
+}
 
+pub(crate) fn payload_to_new_payload(
+    payload: ExecutionPayload,
+    sidecar: ExecutionPayloadSidecar,
+    is_optimism: bool,
+    withdrawals_root: Option<B256>,
+    target_version: Option<EngineApiMessageVersion>,
+) -> eyre::Result<(EngineApiMessageVersion, serde_json::Value)> {
     let (version, params) = match payload {
         ExecutionPayload::V3(payload) => {
             let cancun = sidecar.cancun().unwrap();
 
             if let Some(prague) = sidecar.prague() {
+                // Use target version if provided (for Osaka), otherwise default to V4
+                let version = target_version.unwrap_or(EngineApiMessageVersion::V4);
+
                 if is_optimism {
+                    let withdrawals_root = withdrawals_root.ok_or_else(|| {
+                        eyre::eyre!("Missing withdrawals root for Optimism payload")
+                    })?;
                     (
-                        EngineApiMessageVersion::V4,
+                        version,
                         serde_json::to_value((
-                            OpExecutionPayloadV4 {
-                                payload_inner: payload,
-                                withdrawals_root: block.withdrawals_root.unwrap(),
-                            },
+                            OpExecutionPayloadV4 { payload_inner: payload, withdrawals_root },
                             cancun.versioned_hashes.clone(),
                             cancun.parent_beacon_block_root,
                             Requests::default(),
                         ))?,
                     )
                 } else {
+                    // Extract actual Requests from RequestsOrHash
+                    let requests = prague
+                        .requests
+                        .requests()
+                        .cloned()
+                        .ok_or_else(|| eyre::eyre!("Prague sidecar has hash, not requests"))?;
                     (
-                        EngineApiMessageVersion::V4,
+                        version,
                         serde_json::to_value((
                             payload,
                             cancun.versioned_hashes.clone(),
                             cancun.parent_beacon_block_root,
-                            prague.requests.requests_hash(),
+                            requests,
                         ))?,
                     )
                 }
@@ -217,6 +257,8 @@ pub(crate) async fn call_new_payload<N: Network, P: Provider<N>>(
 ) -> TransportResult<()> {
     let method = version.method_name();
 
+    debug!(method, "Sending newPayload");
+
     let mut status: PayloadStatus = provider.client().request(method, &params).await?;
 
     while !status.is_valid() {
@@ -237,12 +279,15 @@ pub(crate) async fn call_new_payload<N: Network, P: Provider<N>>(
 /// Calls the correct `engine_forkchoiceUpdated` method depending on the given
 /// `EngineApiMessageVersion`, using the provided forkchoice state and payload attributes for the
 /// actual engine api message call.
+///
+/// Note: For Prague (V4), we still use forkchoiceUpdatedV3 as there is no V4.
 pub(crate) async fn call_forkchoice_updated<N, P: EngineApiValidWaitExt<N>>(
     provider: P,
     message_version: EngineApiMessageVersion,
     forkchoice_state: ForkchoiceState,
     payload_attributes: Option<PayloadAttributes>,
 ) -> TransportResult<ForkchoiceUpdated> {
+    // FCU V3 is used for both Cancun and Prague (there is no FCU V4)
     match message_version {
         EngineApiMessageVersion::V3 | EngineApiMessageVersion::V4 | EngineApiMessageVersion::V5 => {
             provider.fork_choice_updated_v3_wait(forkchoice_state, payload_attributes).await

crates/chainspec/src/spec.rs

@@ -460,6 +460,18 @@ impl ChainSpec {
     pub fn builder() -> ChainSpecBuilder {
         ChainSpecBuilder::default()
     }
+
+    /// Map a chain ID to a known chain spec, if available.
+    pub fn from_chain_id(chain_id: u64) -> Option<Arc<Self>> {
+        match NamedChain::try_from(chain_id).ok()? {
+            NamedChain::Mainnet => Some(MAINNET.clone()),
+            NamedChain::Sepolia => Some(SEPOLIA.clone()),
+            NamedChain::Holesky => Some(HOLESKY.clone()),
+            NamedChain::Hoodi => Some(HOODI.clone()),
+            NamedChain::Dev => Some(DEV.clone()),
+            _ => None,
+        }
+    }
 }
 
 impl<H: BlockHeader> ChainSpec<H> {

crates/ethereum/payload/Cargo.toml

@@ -31,8 +31,8 @@ reth-payload-validator.workspace = true
 
 # ethereum
 alloy-rlp.workspace = true
-revm.workspace = true
 alloy-rpc-types-engine.workspace = true
+revm.workspace = true
 
 # alloy
 alloy-eips.workspace = true

crates/rpc/rpc-api/Cargo.toml

@@ -36,7 +36,6 @@ alloy-serde.workspace = true
 alloy-rpc-types-beacon.workspace = true
 alloy-rpc-types-engine.workspace = true
 alloy-genesis.workspace = true
-serde = { workspace = true, features = ["derive"] }
 
 # misc
 jsonrpsee = { workspace = true, features = ["server", "macros"] }

crates/rpc/rpc-api/src/testing.rs

@@ -5,32 +5,24 @@
 //! disabled by default and never be exposed on public-facing RPC without an
 //! explicit operator flag.
 
-use alloy_primitives::{Bytes, B256};
-use alloy_rpc_types_engine::{
-    ExecutionPayloadEnvelopeV5, PayloadAttributes as EthPayloadAttributes,
-};
+use alloy_rpc_types_engine::ExecutionPayloadEnvelopeV5;
 use jsonrpsee::proc_macros::rpc;
-use serde::{Deserialize, Serialize};
 
-/// Capability string for `testing_buildBlockV1`.
-pub const TESTING_BUILD_BLOCK_V1: &str = "testing_buildBlockV1";
-
-/// Request payload for `testing_buildBlockV1`.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub struct TestingBuildBlockRequestV1 {
-    /// Parent block hash of the block to build.
-    pub parent_block_hash: B256,
-    /// Payload attributes (Cancun version).
-    pub payload_attributes: EthPayloadAttributes,
-    /// Raw signed transactions to force-include in order.
-    pub transactions: Vec<Bytes>,
-    /// Optional extra data for the block header.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub extra_data: Option<Bytes>,
-}
+pub use alloy_rpc_types_engine::{TestingBuildBlockRequestV1, TESTING_BUILD_BLOCK_V1};
 
 /// Testing RPC interface for building a block in a single call.
+///
+/// # Enabling
+///
+/// This namespace is disabled by default for security reasons. To enable it,
+/// add `testing` to the `--http.api` flag:
+///
+/// ```sh
+/// reth node --http --http.api eth,testing
+/// ```
+///
+/// **Warning:** Never expose this on public-facing RPC endpoints without proper
+/// authentication.
 #[cfg_attr(not(feature = "client"), rpc(server, namespace = "testing"))]
 #[cfg_attr(feature = "client", rpc(server, client, namespace = "testing"))]
 pub trait TestingApi {

crates/rpc/rpc/src/testing.rs

@@ -1,6 +1,18 @@
 //! Implementation of the `testing` namespace.
 //!
 //! This exposes `testing_buildBlockV1`, intended for non-production/debug use.
+//!
+//! # Enabling the testing namespace
+//!
+//! The `testing_` namespace is disabled by default for security reasons.
+//! To enable it, add `testing` to the `--http.api` flag when starting the node:
+//!
+//! ```sh
+//! reth node --http --http.api eth,testing
+//! ```
+//!
+//! **Warning:** This namespace allows building arbitrary blocks. Never expose it
+//! on public-facing RPC endpoints without proper authentication.
 
 use alloy_consensus::{Header, Transaction};
 use alloy_evm::Evm;
@@ -94,8 +106,8 @@ where
 
                 let mut invalid_senders: HashSet<Address, DefaultHashBuilder> = HashSet::default();
 
-                for tx in request.transactions {
-                    let tx: Recovered<TxTy<Evm::Primitives>> = recover_raw_transaction(&tx)?;
+                for (idx, tx) in request.transactions.iter().enumerate() {
+                    let tx: Recovered<TxTy<Evm::Primitives>> = recover_raw_transaction(tx)?;
                     let sender = tx.signer();
 
                     if skip_invalid_transactions && invalid_senders.contains(&sender) {
@@ -109,6 +121,7 @@ where
                             if skip_invalid_transactions {
                                 debug!(
                                     target: "rpc::testing",
+                                    tx_idx = idx,
                                     ?sender,
                                     error = ?err,
                                     "Skipping invalid transaction"
@@ -116,6 +129,13 @@ where
                                 invalid_senders.insert(sender);
                                 continue;
                             }
+                            debug!(
+                                target: "rpc::testing",
+                                tx_idx = idx,
+                                ?sender,
+                                error = ?err,
+                                "Transaction execution failed"
+                            );
                             return Err(Eth::Error::from_eth_err(err));
                         }
                     };