wip: benching dashmap with big blocks

Adds a new reth-bench subcommand that generates large blocks by:
1. Fetching transactions from existing blocks via RPC
2. Packing them until target gas is reached
3. Calling engine_getPayloadV3Hacked or V4 to build the payload
4. Optionally executing via newPayload + forkchoiceUpdated

Usage:
  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 \
    --execute

Cargo.toml

@@ -497,33 +497,33 @@ alloy-trie = { version = "0.9.1", default-features = false }
 
 alloy-hardforks = "0.4.5"
 
-alloy-consensus = { version = "1.4.1", default-features = false }
-alloy-contract = { version = "1.4.1", default-features = false }
-alloy-eips = { version = "1.4.1", default-features = false }
-alloy-genesis = { version = "1.4.1", default-features = false }
-alloy-json-rpc = { version = "1.4.1", default-features = false }
-alloy-network = { version = "1.4.1", default-features = false }
-alloy-network-primitives = { version = "1.4.1", default-features = false }
-alloy-provider = { version = "1.4.1", features = ["reqwest", "debug-api"], default-features = false }
-alloy-pubsub = { version = "1.4.1", default-features = false }
-alloy-rpc-client = { version = "1.4.1", default-features = false }
-alloy-rpc-types = { version = "1.4.1", features = ["eth"], default-features = false }
-alloy-rpc-types-admin = { version = "1.4.1", default-features = false }
-alloy-rpc-types-anvil = { version = "1.4.1", default-features = false }
-alloy-rpc-types-beacon = { version = "1.4.1", default-features = false }
-alloy-rpc-types-debug = { version = "1.4.1", default-features = false }
-alloy-rpc-types-engine = { version = "1.4.1", default-features = false }
-alloy-rpc-types-eth = { version = "1.4.1", default-features = false }
-alloy-rpc-types-mev = { version = "1.4.1", default-features = false }
-alloy-rpc-types-trace = { version = "1.4.1", default-features = false }
-alloy-rpc-types-txpool = { version = "1.4.1", default-features = false }
-alloy-serde = { version = "1.4.1", default-features = false }
-alloy-signer = { version = "1.4.1", default-features = false }
-alloy-signer-local = { version = "1.4.1", default-features = false }
-alloy-transport = { version = "1.4.1" }
-alloy-transport-http = { version = "1.4.1", features = ["reqwest-rustls-tls"], default-features = false }
-alloy-transport-ipc = { version = "1.4.1", default-features = false }
-alloy-transport-ws = { version = "1.4.1", default-features = false }
+alloy-consensus = { version = "1.4.0", default-features = false }
+alloy-contract = { version = "1.4.0", default-features = false }
+alloy-eips = { version = "1.4.0", default-features = false }
+alloy-genesis = { version = "1.4.0", default-features = false }
+alloy-json-rpc = { version = "1.4.0", default-features = false }
+alloy-network = { version = "1.4.0", default-features = false }
+alloy-network-primitives = { version = "1.4.0", default-features = false }
+alloy-provider = { version = "1.4.0", features = ["reqwest", "debug-api"], default-features = false }
+alloy-pubsub = { version = "1.4.0", default-features = false }
+alloy-rpc-client = { version = "1.4.0", default-features = false }
+alloy-rpc-types = { version = "1.4.0", features = ["eth"], default-features = false }
+alloy-rpc-types-admin = { version = "1.4.0", default-features = false }
+alloy-rpc-types-anvil = { version = "1.4.0", default-features = false }
+alloy-rpc-types-beacon = { version = "1.4.0", default-features = false }
+alloy-rpc-types-debug = { version = "1.4.0", default-features = false }
+alloy-rpc-types-engine = { version = "1.4.0", default-features = false }
+alloy-rpc-types-eth = { version = "1.4.0", default-features = false }
+alloy-rpc-types-mev = { version = "1.4.0", default-features = false }
+alloy-rpc-types-trace = { version = "1.4.0", default-features = false }
+alloy-rpc-types-txpool = { version = "1.4.0", default-features = false }
+alloy-serde = { version = "1.4.0", default-features = false }
+alloy-signer = { version = "1.4.0", default-features = false }
+alloy-signer-local = { version = "1.4.0", default-features = false }
+alloy-transport = { version = "1.4.0" }
+alloy-transport-http = { version = "1.4.0", features = ["reqwest-rustls-tls"], default-features = false }
+alloy-transport-ipc = { version = "1.4.0", default-features = false }
+alloy-transport-ws = { version = "1.4.0", default-features = false }
 
 # op
 alloy-op-evm = { version = "0.25.0", default-features = false }
@@ -746,34 +746,37 @@ vergen-git2 = "1.0.5"
 # networking
 ipnet = "2.11"
 
-# [patch.crates-io]
-# alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-contract = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-eips = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-genesis = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-json-rpc = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-network = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-network-primitives = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-provider = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-pubsub = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-client = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-admin = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-anvil = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-beacon = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-debug = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-engine = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-eth = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-mev = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-trace = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-txpool = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-serde = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-signer = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-signer-local = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport-http = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport-ipc = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport-ws = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
+[patch.crates-io]
+alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-consensus-any = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-contract = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-eips = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-genesis = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-json-rpc = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-network = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-network-primitives = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-provider = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-pubsub = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-client = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-admin = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-anvil = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-any = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-beacon = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-debug = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-engine = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-eth = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-mev = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-trace = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-rpc-types-txpool = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-serde = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-signer = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-signer-local = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-transport = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-transport-http = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-transport-ipc = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-transport-ws = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
+alloy-tx-macros = { git = "https://github.com/alloy-rs/alloy", rev = "82055ca44d660508a3614a1fb9acfd9ece2c24d7" }
 
 # op-alloy-consensus = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }
 # op-alloy-network = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }

analyze_pinned_thread.py

@@ -0,0 +1,27 @@
+from playwright.sync_api import sync_playwright
+import time
+
+# The pinned thread URL - switch to call tree view
+BASE_URL = "https://profiler.firefox.com/public/q0h10v5sjtqstm2sj90azpzxpmhjd8r3agnx058"
+CALL_TREE_URL = BASE_URL + "/calltree/?globalTrackOrder=201&hiddenGlobalTracks=01&hiddenLocalTracksByPid=1153071.2-0wh~1153071.1-xgwxiz7z9wA5A7wAp&localTrackOrderByPid=1153071.1-xhxiz7z9wA5A7wApz8xjwz1Di0wxfz2wz6A6AqAswDfArDgDhxg&range=19841m4475&symbolServer=http%3A%2F%2F127.0.0.1%3A3001%2F24w2lg79pyph7f2pbcpqnhsfk1rrhp6yl4pr074&thread=E5&v=12"
+
+with sync_playwright() as p:
+    browser = p.chromium.launch(headless=True)
+    page = browser.new_page()
+    
+    print(f"Loading: {CALL_TREE_URL}")
+    page.goto(CALL_TREE_URL, timeout=60000)
+    page.wait_for_load_state('networkidle')
+    time.sleep(8)  # Extra time for profile to load
+    
+    page.screenshot(path='/tmp/pinned_thread_calltree.png', full_page=True)
+    print("Screenshot saved to /tmp/pinned_thread_calltree.png")
+    
+    # Get the full page text
+    body_text = page.locator('body').inner_text()
+    print("\n" + "="*80)
+    print("FULL PAGE TEXT:")
+    print("="*80)
+    print(body_text[:8000])
+    
+    browser.close()

analyze_profiler.py

@@ -0,0 +1,59 @@
+from playwright.sync_api import sync_playwright
+import time
+
+URLS = {
+    "slow_block_main": "https://share.firefox.dev/45a9S1G",
+    "normal_blocks": "https://share.firefox.dev/4j7tadO",
+    "pinned_thread": "https://share.firefox.dev/3LylcxY",
+    "proof_fetch_tasks": "https://share.firefox.dev/4pAZyac",
+}
+
+with sync_playwright() as p:
+    browser = p.chromium.launch(headless=True)
+    
+    for name, url in URLS.items():
+        print(f"\n{'='*60}")
+        print(f"PROFILE: {name}")
+        print(f"URL: {url}")
+        print('='*60)
+        
+        page = browser.new_page()
+        page.goto(url, timeout=60000)
+        
+        # Wait for the profiler to fully load
+        page.wait_for_load_state('networkidle')
+        time.sleep(5)  # Extra time for JS rendering
+        
+        # Get the final URL (profiler redirects)
+        final_url = page.url
+        print(f"Final URL: {final_url}")
+        
+        # Take a screenshot
+        page.screenshot(path=f'/tmp/{name}.png', full_page=True)
+        print(f"Screenshot saved to /tmp/{name}.png")
+        
+        # Try to extract call tree content
+        try:
+            # Look for the call tree table
+            call_tree = page.locator('.treeView, .call-tree, [class*="CallTree"], [class*="callTree"]').first
+            if call_tree:
+                text = call_tree.inner_text()
+                print(f"\nCall Tree Content (first 3000 chars):\n{text[:3000]}")
+        except Exception as e:
+            print(f"Could not extract call tree: {e}")
+        
+        # Get visible text content
+        try:
+            body_text = page.locator('body').inner_text()
+            # Filter for relevant content
+            lines = body_text.split('\n')
+            relevant = [l for l in lines if any(kw in l.lower() for kw in 
+                ['%', 'reth', 'proof', 'trie', 'hash', 'fetch', 'state', 'root', 
+                 'rayon', 'tokio', 'engine', 'execute', 'sload', 'journal'])]
+            print(f"\nRelevant lines:\n" + '\n'.join(relevant[:100]))
+        except Exception as e:
+            print(f"Could not extract body: {e}")
+        
+        page.close()
+    
+    browser.close()

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,220 @@
+//! 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::{
+            write_benchmark_results, CombinedResult, NewPayloadResult, TotalGasOutput, TotalGasRow,
+        },
+    },
+    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, NamedChain, DEV, HOLESKY, HOODI, MAINNET, SEPOLIA};
+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, sync::Arc, time::Instant};
+use tracing::{debug, 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,
+
+    /// Optional output directory for benchmark results.
+    #[arg(long, value_name = "OUTPUT")]
+    output: Option<PathBuf>,
+}
+
+/// Map a chain ID to a known chain spec.
+fn chain_spec_from_id(chain_id: u64) -> Option<Arc<ChainSpec>> {
+    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 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 let Some(ref output) = self.output {
+            if output.is_file() {
+                return Err(eyre::eyre!("Output path must be a directory"));
+            }
+            if !output.exists() {
+                std::fs::create_dir_all(output)?;
+                info!("Created output directory: {:?}", 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 = chain_spec_from_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"))?;
+
+        debug!(?parent_block, "Raw RPC parent block");
+
+        let (mut parent_header, mut parent_hash) = rpc_block_to_header(parent_block);
+
+        let start_block = parent_header.number + 1;
+        let end_block = start_block + self.blocks - 1;
+
+        debug!(?parent_header, "Converted parent header");
+
+        info!(start_block, end_block, "Starting gas limit ramp benchmark");
+
+        let mut next_block_number = start_block;
+        let mut results = Vec::new();
+        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 = next_gas_limit(parent_header.gas_limit, max_increase);
+
+            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),
+            )?;
+
+            debug!(
+                target: "reth-bench",
+                block_number = block.header.number,
+                gas_limit,
+                max_increase,
+                "Sending empty payload"
+            );
+
+            let start = Instant::now();
+            call_new_payload(&provider, version, params).await?;
+
+            let new_payload_result =
+                NewPayloadResult { gas_used: block.header.gas_used, latency: start.elapsed() };
+
+            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?;
+
+            let total_latency = start.elapsed();
+            let fcu_latency = total_latency - new_payload_result.latency;
+            let transaction_count = block.body.transactions.len() as u64;
+            let combined_result = CombinedResult {
+                block_number: block.header.number,
+                gas_limit,
+                transaction_count,
+                new_payload_result,
+                fcu_latency,
+                total_latency,
+            };
+
+            info!(%combined_result);
+
+            let gas_row = TotalGasRow {
+                block_number: block.header.number,
+                transaction_count,
+                gas_used: block.header.gas_used,
+                time: total_benchmark_duration.elapsed(),
+            };
+            results.push((gas_row, combined_result));
+
+            parent_header = block.header;
+            parent_hash = block_hash;
+
+            debug!(?parent_header, "RPC parent header");
+            next_block_number += 1;
+        }
+
+        let (gas_output_results, combined_results): (Vec<TotalGasRow>, Vec<CombinedResult>) =
+            results.into_iter().unzip();
+
+        if let Some(ref path) = self.output {
+            write_benchmark_results(path, &gas_output_results, combined_results)?;
+        }
+
+        let final_gas_limit = parent_header.gas_limit;
+        let gas_output = TotalGasOutput::new(gas_output_results)?;
+        info!(
+            total_duration=?gas_output.total_duration,
+            total_gas_used=?gas_output.total_gas_used,
+            blocks_processed=?gas_output.blocks_processed,
+            final_gas_limit,
+            "Total Ggas/s: {:.4}",
+            gas_output.total_gigagas_per_second()
+        );
+
+        Ok(())
+    }
+}
+
+const fn max_gas_limit_increase(parent_gas_limit: u64) -> u64 {
+    (parent_gas_limit / GAS_LIMIT_BOUND_DIVISOR).saturating_sub(1)
+}
+
+fn next_gas_limit(parent_gas_limit: u64, max_increase: u64) -> u64 {
+    parent_gas_limit.saturating_add(max_increase).min(MAXIMUM_GAS_LIMIT_BLOCK)
+}

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

@@ -0,0 +1,658 @@
+//! 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_packBlock` RPC endpoint.
+
+use crate::authenticated_transport::AuthenticatedTransportConnect;
+use alloy_consensus::Transaction;
+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::{debug, info, warn};
+
+/// A single transaction with its gas limit and raw encoded bytes.
+#[derive(Debug, Clone)]
+pub struct RawTransaction {
+    /// The gas limit of the transaction.
+    pub gas_limit: 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 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)>> {
+        let block = self.provider.get_block_by_number(block_number.into()).full().await?;
+
+        let Some(block) = block else {
+            return Ok(None);
+        };
+
+        let gas_used = block.header.gas_used;
+        let transactions = block
+            .transactions
+            .txns()
+            .map(|tx| {
+                let with_encoded = tx.inner.inner.clone().into_encoded();
+                RawTransaction {
+                    gas_limit: tx.gas_limit(),
+                    tx_type: tx.inner.ty(),
+                    raw: with_encoded.encoded_bytes().clone(),
+                }
+            })
+            .collect();
+
+        Ok(Some((transactions, gas_used)))
+    }
+}
+
+/// Collects transactions from a source up to a target gas limit.
+#[derive(Debug)]
+pub struct TransactionCollector<S> {
+    source: S,
+    target_gas: u64,
+}
+
+impl<S: TransactionSource> TransactionCollector<S> {
+    /// Create a new transaction collector.
+    pub 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 collected, 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;
+
+        info!(start_block = start_block, "Fetching transactions from blocks");
+
+        while total_gas < self.target_gas {
+            info!(
+                block = current_block,
+                total_gas = total_gas,
+                target_gas = self.target_gas,
+                progress_pct = (total_gas as f64 / self.target_gas as f64 * 100.0) as u32,
+                "Fetching block"
+            );
+
+            let Some((block_txs, block_gas_used)) =
+                self.source.fetch_block_transactions(current_block).await?
+            else {
+                warn!(block = current_block, "Block not found, stopping");
+                break;
+            };
+
+            let block_tx_count = block_txs.len();
+            let mut block_gas_added: u64 = 0;
+            let mut block_txs_added: usize = 0;
+
+            for tx in block_txs {
+                // Skip blob transactions (EIP-4844, type 3)
+                if tx.tx_type == 3 {
+                    debug!("Skipping blob transaction");
+                    continue;
+                }
+
+                if total_gas + tx.gas_limit <= self.target_gas {
+                    transactions.push(tx.raw);
+                    total_gas += tx.gas_limit;
+                    block_gas_added += tx.gas_limit;
+                    block_txs_added += 1;
+                }
+
+                if total_gas >= self.target_gas {
+                    break;
+                }
+            }
+
+            info!(
+                block = current_block,
+                block_txs = block_tx_count,
+                block_gas_used = block_gas_used,
+                added_txs = block_txs_added,
+                added_gas = block_gas_added,
+                total_txs = transactions.len(),
+                total_gas = total_gas,
+                "Processed block"
+            );
+
+            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 {
+                info!(remaining_gas = remaining_gas, "Stopping: within 1M of target gas");
+                break;
+            }
+        }
+
+        info!(
+            total_txs = transactions.len(),
+            total_gas = 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_packBlock` 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_packBlock 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.
+    #[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_NNN.json.
+    #[arg(long, value_name = "OUTPUT_DIR")]
+    output_dir: std::path::PathBuf,
+}
+
+/// A built payload ready for execution.
+struct BuiltPayload {
+    index: 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_packBlock)
+        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_packBlock"
+            );
+
+            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");
+
+                // Small delay to allow in-memory state to propagate before building next payload
+                tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            }
+
+            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 {
+                info!(
+                    payload = payload_idx + 1,
+                    start_block = current_block,
+                    "Fetching transactions for payload"
+                );
+
+                match collector.collect(current_block).await {
+                    Ok((transactions, total_gas, next_block)) => {
+                        info!(
+                            payload = payload_idx + 1,
+                            tx_count = transactions.len(),
+                            total_gas = total_gas,
+                            next_block = next_block,
+                            "Fetched transactions for payload"
+                        );
+                        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_packBlock"
+                );
+                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");
+
+            // Small delay to allow in-memory state to propagate before building next payload
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+
+            // 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_packBlock"
+                    );
+
+                    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_packBlock.
+    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,
+        };
+
+        info!(
+            endpoint = "testing",
+            method = "testing_packBlock",
+            payload = index + 1,
+            parent_hash = %parent_hash,
+            "RPC call"
+        );
+        let envelope: ExecutionPayloadEnvelopeV5 =
+            testing_provider.client().request("testing_packBlock", [request]).await?;
+        info!(payload = index + 1, "testing_packBlock response received, converting to V4 envelope");
+
+        let v4_envelope = envelope.try_into_v4()?;
+
+        let block_hash =
+            v4_envelope.envelope_inner.execution_payload.payload_inner.payload_inner.block_hash;
+        let timestamp =
+            v4_envelope.envelope_inner.execution_payload.payload_inner.payload_inner.timestamp;
+
+        Ok(BuiltPayload { index, envelope: v4_envelope, block_hash, timestamp })
+    }
+
+    /// Save a payload to disk.
+    fn save_payload(&self, payload: &BuiltPayload) -> eyre::Result<()> {
+        let filename = format!("payload_{:03}.json", payload.index + 1);
+        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!(payload = payload.index + 1, 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;
+
+        info!(
+            endpoint = "engine",
+            method = "engine_newPayloadV4",
+            block_hash = %block_hash,
+            "RPC call"
+        );
+
+        let status = provider
+            .new_payload_v4(
+                envelope.envelope_inner.execution_payload,
+                vec![],
+                B256::ZERO,
+                envelope.execution_requests.to_vec(),
+            )
+            .await?;
+
+        info!(endpoint = "engine", method = "engine_newPayloadV4", ?status, "RPC 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,
+        };
+
+        info!(
+            endpoint = "engine",
+            method = "engine_forkchoiceUpdatedV3",
+            head = %block_hash,
+            safe = %parent_hash,
+            finalized = %parent_hash,
+            "RPC call"
+        );
+
+        let fcu_result = provider.fork_choice_updated_v3(fcu_state, None).await?;
+
+        info!(endpoint = "engine", method = "engine_forkchoiceUpdatedV3", ?fcu_result, "RPC response");
+
+        Ok(())
+    }
+}

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

@@ -0,0 +1,214 @@
+//! 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)> {
+    let method = match version {
+        1 => "engine_getPayloadV1",
+        2 => "engine_getPayloadV2",
+        3 => "engine_getPayloadV3",
+        4 => "engine_getPayloadV4",
+        5 => "engine_getPayloadV5",
+        _ => return Err(eyre::eyre!("Unsupported getPayload version: {}", version)),
+    };
+
+    debug!(
+        target: "reth-bench",
+        method,
+        ?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
+            use alloy_provider::Provider;
+            use alloy_rpc_types_engine::ExecutionPayloadEnvelopeV5;
+
+            let envelope: ExecutionPayloadEnvelopeV5 =
+                provider.client().request(method, (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),
+            ))
+        }
+        _ => unreachable!(),
+    }
+}

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_packBlock` 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;

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

@@ -1,10 +1,12 @@
 //! Contains various benchmark output formats, either for logging or for
 //! serialization to / from files.
 
+use csv::Writer;
 use eyre::OptionExt;
 use reth_primitives_traits::constants::GIGAGAS;
 use serde::{ser::SerializeStruct, Serialize};
-use std::time::Duration;
+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";
@@ -67,6 +69,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,8 +92,9 @@ 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 {} (gas_limit: {}) at {:.4} Ggas/s, used {} gas. Combined: {:.4} Ggas/s. fcu: {:?}, newPayload: {:?}",
             self.block_number,
+            self.gas_limit,
             self.new_payload_result.gas_per_second() / GIGAGAS as f64,
             self.new_payload_result.gas_used,
             self.combined_gas_per_second() / GIGAGAS as f64,
@@ -110,10 +115,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 +173,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,221 @@
+//! 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;
+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 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_001.json, payload_002.json, etc.).
+    #[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,
+}
+
+/// 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,
+}
+
+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"
+        );
+
+        // Discover and load payloads
+        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 payloads from disk");
+
+        // Execute payloads in sequence
+        let mut parent_hash = initial_parent_hash;
+
+        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)
+    }
+
+    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,14 @@ where
         fork_choice_state: ForkchoiceState,
         payload_attributes: Option<PayloadAttributes>,
     ) -> TransportResult<ForkchoiceUpdated> {
+        debug!(
+            target: "reth-bench",
+            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 +91,14 @@ where
         fork_choice_state: ForkchoiceState,
         payload_attributes: Option<PayloadAttributes>,
     ) -> TransportResult<ForkchoiceUpdated> {
+        debug!(
+            target: "reth-bench",
+            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 +129,14 @@ where
         fork_choice_state: ForkchoiceState,
         payload_attributes: Option<PayloadAttributes>,
     ) -> TransportResult<ForkchoiceUpdated> {
+        debug!(
+            target: "reth-bench",
+            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 +173,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 +260,13 @@ pub(crate) async fn call_new_payload<N: Network, P: Provider<N>>(
 ) -> TransportResult<()> {
     let method = version.method_name();
 
+    debug!(
+        target: "reth-bench",
+        method,
+        params = %serde_json::to_string_pretty(&params).unwrap_or_default(),
+        "Sending newPayload"
+    );
+
     let mut status: PayloadStatus = provider.client().request(method, &params).await?;
 
     while !status.is_valid() {
@@ -237,12 +287,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/engine/tree/Cargo.toml

@@ -44,6 +44,8 @@ alloy-primitives.workspace = true
 alloy-rlp.workspace = true
 alloy-rpc-types-engine.workspace = true
 
+parking_lot.workspace = true
+
 revm.workspace = true
 revm-primitives.workspace = true
 
@@ -65,7 +67,6 @@ schnellru.workspace = true
 rayon.workspace = true
 tracing.workspace = true
 derive_more.workspace = true
-parking_lot.workspace = true
 crossbeam-channel.workspace = true
 
 # optional deps for test-utils

crates/engine/tree/src/tree/payload_processor/multiproof.rs

@@ -225,7 +225,7 @@ struct MultiproofInput {
     proof_targets: MultiProofTargets,
     proof_sequence_number: u64,
     state_root_message_sender: CrossbeamSender<MultiProofMessage>,
-    multi_added_removed_keys: Option<Arc<MultiAddedRemovedKeys>>,
+    multi_added_removed_keys: Option<MultiAddedRemovedKeys>,
 }
 
 impl MultiproofInput {
@@ -608,9 +608,6 @@ impl MultiProofTask {
         // we still want to optimistically fetch extension children for the leaf addition case.
         self.multi_added_removed_keys.touch_accounts(targets.keys().copied());
 
-        // Clone+Arc MultiAddedRemovedKeys for sharing with the dispatched multiproof tasks
-        let multi_added_removed_keys = Arc::new(self.multi_added_removed_keys.clone());
-
         self.metrics.prefetch_proof_targets_accounts_histogram.record(targets.len() as f64);
         self.metrics
             .prefetch_proof_targets_storages_histogram
@@ -636,7 +633,7 @@ impl MultiProofTask {
                     proof_targets,
                     proof_sequence_number: self.proof_sequencer.next_sequence(),
                     state_root_message_sender: self.tx.clone(),
-                    multi_added_removed_keys: Some(multi_added_removed_keys.clone()),
+                    multi_added_removed_keys: Some(self.multi_added_removed_keys.clone()),
                 });
             },
         );
@@ -704,9 +701,6 @@ impl MultiProofTask {
             state_updates += 1;
         }
 
-        // Clone+Arc MultiAddedRemovedKeys for sharing with the dispatched multiproof tasks
-        let multi_added_removed_keys = Arc::new(self.multi_added_removed_keys.clone());
-
         let chunking_len = not_fetched_state_update.chunking_length();
         let mut spawned_proof_targets = MultiProofTargets::default();
         let available_account_workers =
@@ -725,7 +719,7 @@ impl MultiProofTask {
                 let proof_targets = get_proof_targets(
                     &hashed_state_update,
                     &self.fetched_proof_targets,
-                    &multi_added_removed_keys,
+                    &self.multi_added_removed_keys,
                 );
                 spawned_proof_targets.extend_ref(&proof_targets);
 
@@ -735,7 +729,7 @@ impl MultiProofTask {
                     proof_targets,
                     proof_sequence_number: self.proof_sequencer.next_sequence(),
                     state_root_message_sender: self.tx.clone(),
-                    multi_added_removed_keys: Some(multi_added_removed_keys.clone()),
+                    multi_added_removed_keys: Some(self.multi_added_removed_keys.clone()),
                 });
             },
         );
@@ -1217,7 +1211,7 @@ fn get_proof_targets(
             .keys()
             .filter(|slot| {
                 !fetched.is_some_and(|f| f.contains(*slot)) ||
-                    storage_added_removed_keys.is_some_and(|k| k.is_removed(slot))
+                    storage_added_removed_keys.as_ref().is_some_and(|k| k.is_removed(slot))
             })
             .peekable();
 

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/ethereum/payload/src/validator.rs

@@ -6,6 +6,7 @@ use reth_chainspec::EthereumHardforks;
 use reth_payload_validator::{cancun, prague, shanghai};
 use reth_primitives_traits::{Block as _, SealedBlock, SignedTransaction};
 use std::sync::Arc;
+use tracing::info;
 
 /// Execution payload validator.
 #[derive(Clone, Debug)]
@@ -78,6 +79,15 @@ where
     // First parse the block
     let sealed_block = payload.try_into_block_with_sidecar(&sidecar)?.seal_slow();
 
+    info!(
+        target: "engine::payload_validator",
+        header = ?sealed_block.header(),
+        computed_hash = ?sealed_block.hash(),
+        expected_hash = ?expected_hash,
+        ?sidecar,
+        "Validating payload block hash"
+    );
+
     // Ensure the hash included in the payload matches the block hash
     if expected_hash != sealed_block.hash() {
         return Err(PayloadError::BlockHash {

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

@@ -42,4 +42,13 @@ pub trait TestingApi {
         &self,
         request: TestingBuildBlockRequestV1,
     ) -> jsonrpsee::core::RpcResult<ExecutionPayloadEnvelopeV5>;
+
+    /// Builds a block using the provided parent, payload attributes, and transactions.
+    ///
+    /// Like testing_buildBlockV1 but skips invalid transactions.
+    #[method(name = "packBlock")]
+    async fn pack_block(
+        &self,
+        request: TestingBuildBlockRequestV1,
+    ) -> jsonrpsee::core::RpcResult<ExecutionPayloadEnvelopeV5>;
 }

crates/rpc/rpc/src/testing.rs

@@ -4,7 +4,7 @@
 
 use alloy_consensus::{Header, Transaction};
 use alloy_evm::Evm;
-use alloy_primitives::U256;
+use alloy_primitives::{map::HashSet, U256};
 use alloy_rpc_types_engine::ExecutionPayloadEnvelopeV5;
 use async_trait::async_trait;
 use jsonrpsee::core::RpcResult;
@@ -17,9 +17,11 @@ use reth_revm::{database::StateProviderDatabase, db::State};
 use reth_rpc_api::{TestingApiServer, TestingBuildBlockRequestV1};
 use reth_rpc_eth_api::{helpers::Call, FromEthApiError};
 use reth_rpc_eth_types::{utils::recover_raw_transaction, EthApiError};
-use reth_storage_api::{BlockReader, HeaderProvider};
+use reth_storage_api::{BlockReader, HeaderProvider, StateProvider};
 use revm::context::Block;
+use revm_primitives::map::DefaultHashBuilder;
 use std::sync::Arc;
+use tracing::{debug, info};
 
 /// Testing API handler.
 #[derive(Debug, Clone)]
@@ -79,11 +81,40 @@ where
                 let mut total_fees = U256::ZERO;
                 let base_fee = builder.evm_mut().block().basefee();
 
-                for tx in request.transactions {
-                    let tx: Recovered<TxTy<Evm::Primitives>> = recover_raw_transaction(&tx)?;
+                debug!(
+                    target: "rpc::testing",
+                    parent_hash = ?request.parent_block_hash,
+                    parent_number = parent.number(),
+                    num_transactions = request.transactions.len(),
+                    ?base_fee,
+                    "Starting block build"
+                );
+
+                for (idx, tx) in request.transactions.iter().enumerate() {
+                    let tx: Recovered<TxTy<Evm::Primitives>> = recover_raw_transaction(tx)?;
+                    let sender = tx.signer();
+                    let tx_nonce = tx.nonce();
+                    let tx_hash = *tx.tx_hash();
+
+                    let account_nonce = state
+                        .account_nonce(&sender)
+                        .map_err(RethError::other)
+                        .map_err(Eth::Error::from_eth_err)?;
+
                     let tip = tx.effective_tip_per_gas(base_fee).unwrap_or_default();
-                    let gas_used =
-                        builder.execute_transaction(tx).map_err(Eth::Error::from_eth_err)?;
+                    let gas_used = builder.execute_transaction(tx).map_err(|e| {
+                        debug!(
+                            target: "rpc::testing",
+                            tx_idx = idx,
+                            ?tx_hash,
+                            ?sender,
+                            tx_nonce,
+                            account_nonce = ?account_nonce,
+                            error = ?e,
+                            "Transaction execution failed"
+                        );
+                        Eth::Error::from_eth_err(e)
+                    })?;
 
                     total_fees += U256::from(tip) * U256::from(gas_used);
                 }
@@ -107,6 +138,132 @@ where
             })
             .await
     }
+
+    async fn pack_block(
+        &self,
+        request: TestingBuildBlockRequestV1,
+    ) -> Result<ExecutionPayloadEnvelopeV5, Eth::Error> {
+        let evm_config = self.evm_config.clone();
+        self.eth_api
+            .spawn_with_state_at_block(request.parent_block_hash, move |eth_api, state| {
+                let state = state.database.0;
+                let mut db = State::builder()
+                    .with_bundle_update()
+                    .with_database(StateProviderDatabase::new(&state))
+                    .build();
+                let parent = eth_api
+                    .provider()
+                    .sealed_header_by_hash(request.parent_block_hash)?
+                    .ok_or_else(|| {
+                    EthApiError::HeaderNotFound(request.parent_block_hash.into())
+                })?;
+
+                let env_attrs = NextBlockEnvAttributes {
+                    timestamp: request.payload_attributes.timestamp,
+                    suggested_fee_recipient: request.payload_attributes.suggested_fee_recipient,
+                    prev_randao: request.payload_attributes.prev_randao,
+                    gas_limit: parent.gas_limit(),
+                    parent_beacon_block_root: request.payload_attributes.parent_beacon_block_root,
+                    withdrawals: request.payload_attributes.withdrawals.map(Into::into),
+                    extra_data: request.extra_data.unwrap_or_default(),
+                };
+
+                let mut builder = evm_config
+                    .builder_for_next_block(&mut db, &parent, env_attrs)
+                    .map_err(RethError::other)
+                    .map_err(Eth::Error::from_eth_err)?;
+                builder.apply_pre_execution_changes().map_err(Eth::Error::from_eth_err)?;
+
+                let mut total_fees = U256::ZERO;
+                let base_fee = builder.evm_mut().block().basefee();
+
+                debug!(
+                    target: "rpc::testing",
+                    parent_hash = ?request.parent_block_hash,
+                    parent_number = parent.number(),
+                    num_transactions = request.transactions.len(),
+                    ?base_fee,
+                    "Starting block build"
+                );
+
+                // Keeps track of invalid senders - if one tx is invalid then all future txs from
+                // that sender are invalid as well
+                let mut invalid_senders_map = HashSet::<_, DefaultHashBuilder>::default();
+
+                for (idx, tx) in request.transactions.iter().enumerate() {
+                    let tx: Recovered<TxTy<Evm::Primitives>> = recover_raw_transaction(tx)?;
+                    let sender = tx.signer();
+                    let tx_nonce = tx.nonce();
+                    let tx_hash = *tx.tx_hash();
+
+                    if invalid_senders_map.contains(&sender) {
+                        // skip any senders which we have ignored previously
+                        continue;
+                    }
+
+                    let account_nonce = state
+                        .account_nonce(&sender)
+                        .map_err(RethError::other)
+                        .map_err(Eth::Error::from_eth_err)?;
+
+                    let tip = tx.effective_tip_per_gas(base_fee).unwrap_or_default();
+                    let gas_used = match builder.execute_transaction(tx) {
+                        Err(err) => {
+                            invalid_senders_map.insert(sender);
+                            debug!(
+                                target: "rpc::testing",
+                                tx_idx = idx,
+                                ?tx_hash,
+                                ?sender,
+                                tx_nonce,
+                                account_nonce = ?account_nonce,
+                                error = ?err,
+                                "Transaction execution failed"
+                            );
+                            continue
+                        }
+                        Ok(gas_used) => gas_used,
+                    };
+
+                    total_fees += U256::from(tip) * U256::from(gas_used);
+                }
+                let outcome = builder.finish(&state).map_err(Eth::Error::from_eth_err)?;
+
+                let requests = outcome
+                    .block
+                    .requests_hash()
+                    .is_some()
+                    .then_some(outcome.execution_result.requests);
+
+                let sealed_block = outcome.block.into_sealed_block();
+                info!(
+                    target: "rpc::testing",
+                    header = ?sealed_block.header(),
+                    block_hash = ?sealed_block.hash(),
+                    "testing_packBlock built block"
+                );
+
+                let envelope = EthBuiltPayload::new(
+                    alloy_rpc_types_engine::PayloadId::default(),
+                    Arc::new(sealed_block),
+                    total_fees,
+                    requests,
+                )
+                .try_into_v5()
+                .map_err(RethError::other)
+                .map_err(Eth::Error::from_eth_err)?;
+
+                info!(
+                    target: "rpc::testing",
+                    blobs_bundle = ?envelope.blobs_bundle,
+                    execution_requests = ?envelope.execution_requests,
+                    "testing_packBlock payload envelope sidecar"
+                );
+
+                Ok(envelope)
+            })
+            .await
+    }
 }
 
 #[async_trait]
@@ -124,4 +281,13 @@ where
     ) -> RpcResult<ExecutionPayloadEnvelopeV5> {
         self.build_block_v1(request).await.map_err(Into::into)
     }
+
+    /// Handles `testing_packBlock` by gating concurrency via a semaphore and offloading heavy
+    /// work to the blocking pool to avoid stalling the async runtime.
+    async fn pack_block(
+        &self,
+        request: TestingBuildBlockRequestV1,
+    ) -> RpcResult<ExecutionPayloadEnvelopeV5> {
+        self.pack_block(request).await.map_err(Into::into)
+    }
 }

crates/trie/common/Cargo.toml

@@ -28,6 +28,7 @@ bytes = { workspace = true, optional = true }
 derive_more.workspace = true
 itertools = { workspace = true, features = ["use_alloc"] }
 nybbles = { workspace = true, features = ["rlp"] }
+dashmap.workspace = true
 
 # reth
 revm-database.workspace = true

crates/trie/common/src/added_removed_keys.rs

@@ -1,14 +1,17 @@
 //! Tracking of keys having been added and removed from the tries.
 
+use std::sync::Arc;
+
 use crate::HashedPostState;
-use alloy_primitives::{map::B256Map, B256};
+use alloy_primitives::B256;
 use alloy_trie::proof::AddedRemovedKeys;
+use dashmap::{mapref::one::Ref, DashMap};
 
 /// Tracks added and removed keys across account and storage tries.
 #[derive(Debug, Clone)]
 pub struct MultiAddedRemovedKeys {
     account: AddedRemovedKeys,
-    storages: B256Map<AddedRemovedKeys>,
+    storages: Arc<DashMap<B256, AddedRemovedKeys, alloy_primitives::map::DefaultHashBuilder>>,
 }
 
 /// Returns [`AddedRemovedKeys`] with default parameters. This is necessary while we are not yet
@@ -50,7 +53,7 @@ impl MultiAddedRemovedKeys {
                 continue
             }
 
-            let storage_removed_keys =
+            let mut storage_removed_keys =
                 self.storages.entry(*hashed_address).or_insert_with(default_added_removed_keys);
 
             for (key, val) in &storage.storage {
@@ -68,7 +71,7 @@ impl MultiAddedRemovedKeys {
     }
 
     /// Returns a [`AddedRemovedKeys`] for the storage trie of a particular account, if any.
-    pub fn get_storage(&self, hashed_address: &B256) -> Option<&AddedRemovedKeys> {
+    pub fn get_storage(&self, hashed_address: &B256) -> Option<Ref<'_, B256, AddedRemovedKeys>> {
         self.storages.get(hashed_address)
     }
 

crates/trie/common/src/hashed_state.rs

@@ -215,7 +215,9 @@ impl HashedPostState {
                     let mut storage_not_in_targets = HashedStorage::default();
                     storage.storage.retain(|&slot, value| {
                         if storage_in_targets.contains(&slot) &&
-                            !storage_added_removed_keys.is_some_and(|k| k.is_removed(&slot))
+                            !storage_added_removed_keys
+                                .as_ref()
+                                .is_some_and(|k| k.is_removed(&slot))
                         {
                             return true
                         }

crates/trie/parallel/src/proof.rs

@@ -30,7 +30,7 @@ pub struct ParallelProof {
     /// Flag indicating whether to include branch node masks in the proof.
     collect_branch_node_masks: bool,
     /// Provided by the user to give the necessary context to retain extra proofs.
-    multi_added_removed_keys: Option<Arc<MultiAddedRemovedKeys>>,
+    multi_added_removed_keys: Option<MultiAddedRemovedKeys>,
     /// Handle to the proof worker pools.
     proof_worker_handle: ProofWorkerHandle,
     /// Whether to use V2 storage proofs.
@@ -72,7 +72,7 @@ impl ParallelProof {
     /// extra proofs needed to add and remove leaf nodes from the tries.
     pub fn with_multi_added_removed_keys(
         mut self,
-        multi_added_removed_keys: Option<Arc<MultiAddedRemovedKeys>>,
+        multi_added_removed_keys: Option<MultiAddedRemovedKeys>,
     ) -> Self {
         self.multi_added_removed_keys = multi_added_removed_keys;
         self

crates/trie/parallel/src/proof_task.rs

@@ -447,8 +447,9 @@ where
 
         // Get or create added/removed keys context
         let multi_added_removed_keys =
-            multi_added_removed_keys.unwrap_or_else(|| Arc::new(MultiAddedRemovedKeys::new()));
-        let added_removed_keys = multi_added_removed_keys.get_storage(&hashed_address);
+            multi_added_removed_keys.unwrap_or_else(MultiAddedRemovedKeys::new);
+        let added_removed_keys =
+            multi_added_removed_keys.get_storage(&hashed_address).map(|k| k.as_ref().clone());
 
         let span = debug_span!(
             target: "trie::proof_task",
@@ -1556,7 +1557,7 @@ fn dispatch_storage_proofs(
     targets: &MultiProofTargets,
     storage_prefix_sets: &mut B256Map<PrefixSet>,
     with_branch_node_masks: bool,
-    multi_added_removed_keys: Option<&Arc<MultiAddedRemovedKeys>>,
+    multi_added_removed_keys: Option<&MultiAddedRemovedKeys>,
     use_v2_proofs: bool,
 ) -> Result<B256Map<CrossbeamReceiver<StorageProofResultMessage>>, ParallelStateRootError> {
     let mut storage_proof_receivers =
@@ -1613,7 +1614,7 @@ pub enum StorageProofInput {
         /// Whether or not to collect branch node masks
         with_branch_node_masks: bool,
         /// Provided by the user to give the necessary context to retain extra proofs.
-        multi_added_removed_keys: Option<Arc<MultiAddedRemovedKeys>>,
+        multi_added_removed_keys: Option<MultiAddedRemovedKeys>,
     },
     /// V2 storage proof variant
     V2 {
@@ -1632,7 +1633,7 @@ impl StorageProofInput {
         prefix_set: PrefixSet,
         target_slots: B256Set,
         with_branch_node_masks: bool,
-        multi_added_removed_keys: Option<Arc<MultiAddedRemovedKeys>>,
+        multi_added_removed_keys: Option<MultiAddedRemovedKeys>,
     ) -> Self {
         Self::Legacy {
             hashed_address,
@@ -1668,7 +1669,7 @@ pub struct AccountMultiproofInput {
     /// Whether or not to collect branch node masks.
     pub collect_branch_node_masks: bool,
     /// Provided by the user to give the necessary context to retain extra proofs.
-    pub multi_added_removed_keys: Option<Arc<MultiAddedRemovedKeys>>,
+    pub multi_added_removed_keys: Option<MultiAddedRemovedKeys>,
     /// Context for sending the proof result.
     pub proof_result_sender: ProofResultContext,
     /// Whether to use V2 storage proofs.
@@ -1684,7 +1685,7 @@ struct AccountMultiproofParams<'a> {
     /// Whether or not to collect branch node masks.
     collect_branch_node_masks: bool,
     /// Provided by the user to give the necessary context to retain extra proofs.
-    multi_added_removed_keys: Option<&'a Arc<MultiAddedRemovedKeys>>,
+    multi_added_removed_keys: Option<&'a MultiAddedRemovedKeys>,
     /// Receivers for storage proofs being computed in parallel.
     storage_proof_receivers: B256Map<CrossbeamReceiver<StorageProofResultMessage>>,
     /// Cached storage roots. This will be used to read storage roots for missed leaves, as well as