fix: allow smaller header size

2026-04-30 03:01:58 -04:00 · 2025-12-16 17:08:53 +01:00
253 changed files with 2997 additions and 8016 deletions
--- a/.github/assets/hive/expected_failures.yaml
+++ b/.github/assets/hive/expected_failures.yaml
@@ -39,43 +39,6 @@ engine-cancun:
  # in hive or its dependencies
  - Blob Transaction Ordering, Multiple Clients (Cancun) (reth)

-  # Fork ID tests failing since mid-December 2025 - tests complete in <2s indicating peering issues
-  # These tests verify DevP2P peer connections with Fork ID validation
-  # TODO: investigate root cause (likely related to eth/69-70 protocol changes or hive test infrastructure)
-  - Fork ID: Genesis=0, Cancun=0 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=0, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=1, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=1, Shanghai=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=1, Shanghai=1, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=2 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=2, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=2, Shanghai=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=2, Shanghai=1, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=2, Shanghai=2 (Cancun) (reth)
-  - Fork ID: Genesis=0, Cancun=2, Shanghai=2, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=0 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=0, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=1, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=1, Shanghai=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=1, Shanghai=1, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=2 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=2, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=2, Shanghai=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=2, Shanghai=1, BlocksBeforePeering=1 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=2, Shanghai=2 (Cancun) (reth)
-  - Fork ID: Genesis=1, Cancun=2, Shanghai=2, BlocksBeforePeering=1 (Cancun) (reth)
-
-  # Request Blob Pooled Transactions tests - failing since mid-December 2025
-  - Request Blob Pooled Transactions Single (Cancun) (reth)
-  - Request Blob Pooled Transactions Multiple (Cancun) (reth)
-
-  # Invalid Missing Ancestor tests - timing-sensitive syncing tests
-  - Invalid Missing Ancestor ReOrg, StateRoot, EmptyTxs=False, Invalid P9 (Cancun) (reth)
-  - Invalid Missing Ancestor Syncing ReOrg, Timestamp, EmptyTxs=False, CanonicalReOrg=False, Invalid P8 (Cancun) (reth)
-  - Invalid Missing Ancestor Syncing ReOrg, Timestamp, EmptyTxs=False, CanonicalReOrg=True, Invalid P8 (Cancun) (reth)
-
 sync: [ ]

 engine-auth: [ ]
--- a/.gitignore
+++ b/.gitignore
@@ -12,9 +12,6 @@ target/
 # Generated by Intellij-based IDEs.
 .idea

-# ck-search metadata
-.ck
-
 # Generated by MacOS
 .DS_Store

--- a/Cargo.lock
+++ b/Cargo.lock
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -497,33 +497,33 @@ alloy-trie = { version = "0.9.1", default-features = false }

 alloy-hardforks = "0.4.5"

-alloy-consensus = { version = "1.2.1", default-features = false }
-alloy-contract = { version = "1.2.1", default-features = false }
-alloy-eips = { version = "1.2.1", default-features = false }
-alloy-genesis = { version = "1.2.1", default-features = false }
-alloy-json-rpc = { version = "1.2.1", default-features = false }
-alloy-network = { version = "1.2.1", default-features = false }
-alloy-network-primitives = { version = "1.2.1", default-features = false }
-alloy-provider = { version = "1.2.1", features = ["reqwest", "debug-api"], default-features = false }
-alloy-pubsub = { version = "1.2.1", default-features = false }
-alloy-rpc-client = { version = "1.2.1", default-features = false }
-alloy-rpc-types = { version = "1.2.1", features = ["eth"], default-features = false }
-alloy-rpc-types-admin = { version = "1.2.1", default-features = false }
-alloy-rpc-types-anvil = { version = "1.2.1", default-features = false }
-alloy-rpc-types-beacon = { version = "1.2.1", default-features = false }
-alloy-rpc-types-debug = { version = "1.2.1", default-features = false }
-alloy-rpc-types-engine = { version = "1.2.1", default-features = false }
-alloy-rpc-types-eth = { version = "1.2.1", default-features = false }
-alloy-rpc-types-mev = { version = "1.2.1", default-features = false }
-alloy-rpc-types-trace = { version = "1.2.1", default-features = false }
-alloy-rpc-types-txpool = { version = "1.2.1", default-features = false }
-alloy-serde = { version = "1.2.1", default-features = false }
-alloy-signer = { version = "1.2.1", default-features = false }
-alloy-signer-local = { version = "1.2.1", default-features = false }
-alloy-transport = { version = "1.2.1" }
-alloy-transport-http = { version = "1.2.1", features = ["reqwest-rustls-tls"], default-features = false }
-alloy-transport-ipc = { version = "1.2.1", default-features = false }
-alloy-transport-ws = { version = "1.2.1", default-features = false }
+alloy-consensus = { version = "1.1.3", default-features = false }
+alloy-contract = { version = "1.1.3", default-features = false }
+alloy-eips = { version = "1.1.3", default-features = false }
+alloy-genesis = { version = "1.1.3", default-features = false }
+alloy-json-rpc = { version = "1.1.3", default-features = false }
+alloy-network = { version = "1.1.3", default-features = false }
+alloy-network-primitives = { version = "1.1.3", default-features = false }
+alloy-provider = { version = "1.1.3", features = ["reqwest", "debug-api"], default-features = false }
+alloy-pubsub = { version = "1.1.3", default-features = false }
+alloy-rpc-client = { version = "1.1.3", default-features = false }
+alloy-rpc-types = { version = "1.1.3", features = ["eth"], default-features = false }
+alloy-rpc-types-admin = { version = "1.1.3", default-features = false }
+alloy-rpc-types-anvil = { version = "1.1.3", default-features = false }
+alloy-rpc-types-beacon = { version = "1.1.3", default-features = false }
+alloy-rpc-types-debug = { version = "1.1.3", default-features = false }
+alloy-rpc-types-engine = { version = "1.1.3", default-features = false }
+alloy-rpc-types-eth = { version = "1.1.3", default-features = false }
+alloy-rpc-types-mev = { version = "1.1.3", default-features = false }
+alloy-rpc-types-trace = { version = "1.1.3", default-features = false }
+alloy-rpc-types-txpool = { version = "1.1.3", default-features = false }
+alloy-serde = { version = "1.1.3", default-features = false }
+alloy-signer = { version = "1.1.3", default-features = false }
+alloy-signer-local = { version = "1.1.3", default-features = false }
+alloy-transport = { version = "1.1.3" }
+alloy-transport-http = { version = "1.1.3", features = ["reqwest-rustls-tls"], default-features = false }
+alloy-transport-ipc = { version = "1.1.3", default-features = false }
+alloy-transport-ws = { version = "1.1.3", default-features = false }

 # op
 alloy-op-evm = { version = "0.25.0", default-features = false }
@@ -548,7 +548,6 @@ bytes = { version = "1.5", default-features = false }
 brotli = "8"
 cfg-if = "1.0"
 clap = "4"
-color-eyre = "0.6"
 dashmap = "6.0"
 derive_more = { version = "2", default-features = false, features = ["full"] }
 dirs-next = "2.0.0"
@@ -588,7 +587,6 @@ url = { version = "2.3", default-features = false }
 zstd = "0.13"
 byteorder = "1"
 mini-moka = "0.10"
-moka = "0.12"
 tar-no-std = { version = "0.3.2", default-features = false }
 miniz_oxide = { version = "0.8.4", default-features = false }
 chrono = "0.4.41"
@@ -731,7 +729,6 @@ socket2 = { version = "0.5", default-features = false }
 sysinfo = { version = "0.33", default-features = false }
 tracing-journald = "0.3"
 tracing-logfmt = "0.3.3"
-tracing-samply = "0.1"
 tracing-subscriber = { version = "0.3", default-features = false }
 triehash = "0.8"
 typenum = "1.15.0"
--- a/2
+++ b/2
@@ -18,7 +18,7 @@ FROM chef AS builder
 COPY --from=planner /app/recipe.json recipe.json

 # Build profile, release by default
-ARG BUILD_PROFILE=maxperf
+ARG BUILD_PROFILE=release
 ENV BUILD_PROFILE=$BUILD_PROFILE

 # Extra Cargo flags
--- a/2
+++ b/2
@@ -14,7 +14,7 @@ RUN cargo chef prepare --recipe-path recipe.json
 FROM chef AS builder
 COPY --from=planner /app/recipe.json recipe.json

-ARG BUILD_PROFILE=maxperf
+ARG BUILD_PROFILE=release
 ENV BUILD_PROFILE=$BUILD_PROFILE

 ARG RUSTFLAGS=""
--- a/2
+++ b/2
@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
   same "printed page" as the copyright notice for easier
   identification within third-party archives.

-Copyright 2022-2026 Reth Contributors
+Copyright 2022-2025 Reth Contributors

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
--- a/2
+++ b/2
@@ -1,6 +1,6 @@
 The MIT License (MIT)

-Copyright (c) 2022-2026 Reth Contributors
+Copyright (c) 2022-2025 Reth Contributors

 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal
--- a/bin/reth-bench-compare/src/cli.rs
+++ b/bin/reth-bench-compare/src/cli.rs
@@ -164,42 +164,12 @@ pub(crate) struct Args {
    #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
    pub reth_args: Vec<String>,

-    /// Comma-separated list of features to enable during reth compilation (applied to both builds)
+    /// Comma-separated list of features to enable during reth compilation
    ///
    /// Example: `jemalloc,asm-keccak`
    #[arg(long, value_name = "FEATURES", default_value = "jemalloc,asm-keccak")]
    pub features: String,

-    /// Comma-separated list of features to enable only for baseline build (overrides --features)
-    ///
-    /// Example: `--baseline-features jemalloc`
-    #[arg(long, value_name = "FEATURES")]
-    pub baseline_features: Option<String>,
-
-    /// Comma-separated list of features to enable only for feature build (overrides --features)
-    ///
-    /// Example: `--feature-features jemalloc,asm-keccak`
-    #[arg(long, value_name = "FEATURES")]
-    pub feature_features: Option<String>,
-
-    /// RUSTFLAGS to use for both baseline and feature builds
-    ///
-    /// Example: `--rustflags "-C target-cpu=native"`
-    #[arg(long, value_name = "FLAGS", default_value = "-C target-cpu=native")]
-    pub rustflags: String,
-
-    /// RUSTFLAGS to use only for baseline build (overrides --rustflags)
-    ///
-    /// Example: `--baseline-rustflags "-C target-cpu=native -C lto"`
-    #[arg(long, value_name = "FLAGS")]
-    pub baseline_rustflags: Option<String>,
-
-    /// RUSTFLAGS to use only for feature build (overrides --rustflags)
-    ///
-    /// Example: `--feature-rustflags "-C target-cpu=native -C lto"`
-    #[arg(long, value_name = "FLAGS")]
-    pub feature_rustflags: Option<String>,
-
    /// Disable automatic --debug.startup-sync-state-idle flag for specific runs.
    /// Can be "baseline", "feature", or "all".
    /// By default, the flag is passed to warmup, baseline, and feature runs.
@@ -358,6 +328,7 @@ pub(crate) async fn run_comparison(args: Args, _ctx: CliContext) -> Result<()> {
        git_manager.repo_root().to_string(),
        output_dir.clone(),
        git_manager.clone(),
+        args.features.clone(),
    )?;
    // Initialize node manager
    let mut node_manager = NodeManager::new(&args);
@@ -477,18 +448,6 @@ async fn run_compilation_phase(
        let ref_type = ref_types[i];
        let commit = &ref_commits[git_ref];

-        // Get per-build features and rustflags
-        let features = match ref_type {
-            "baseline" => args.baseline_features.as_ref().unwrap_or(&args.features),
-            "feature" => args.feature_features.as_ref().unwrap_or(&args.features),
-            _ => &args.features,
-        };
-        let rustflags = match ref_type {
-            "baseline" => args.baseline_rustflags.as_ref().unwrap_or(&args.rustflags),
-            "feature" => args.feature_rustflags.as_ref().unwrap_or(&args.rustflags),
-            _ => &args.rustflags,
-        };
-
        info!(
            "Compiling {} binary for reference: {} (commit: {})",
            ref_type,
@@ -500,7 +459,7 @@ async fn run_compilation_phase(
        git_manager.switch_ref(git_ref)?;

        // Compile reth (with caching)
-        compilation_manager.compile_reth(commit, is_optimism, features, rustflags)?;
+        compilation_manager.compile_reth(commit, is_optimism)?;

        info!("Completed compilation for {} reference", ref_type);
    }
--- a/bin/reth-bench-compare/src/comparison.rs
+++ b/bin/reth-bench-compare/src/comparison.rs
@@ -39,8 +39,7 @@ pub(crate) struct BenchmarkResults {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct CombinedLatencyRow {
    pub block_number: u64,
-    #[serde(default)]
-    pub transaction_count: Option<u64>,
+    pub transaction_count: u64,
    pub gas_used: u64,
    pub new_payload_latency: u128,
 }
@@ -49,8 +48,7 @@ pub(crate) struct CombinedLatencyRow {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct TotalGasRow {
    pub block_number: u64,
-    #[serde(default)]
-    pub transaction_count: Option<u64>,
+    pub transaction_count: u64,
    pub gas_used: u64,
    pub time: u128,
 }
@@ -127,8 +125,7 @@ pub(crate) struct ComparisonSummary {
 #[derive(Debug, Serialize)]
 pub(crate) struct BlockComparison {
    pub block_number: u64,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub transaction_count: Option<u64>,
+    pub transaction_count: u64,
    pub gas_used: u64,
    pub baseline_new_payload_latency: u128,
    pub feature_new_payload_latency: u128,
--- a/bin/reth-bench-compare/src/compilation.rs
+++ b/bin/reth-bench-compare/src/compilation.rs
@@ -13,6 +13,7 @@ pub(crate) struct CompilationManager {
    repo_root: String,
    output_dir: PathBuf,
    git_manager: GitManager,
+    features: String,
 }

 impl CompilationManager {
@@ -21,8 +22,9 @@ impl CompilationManager {
        repo_root: String,
        output_dir: PathBuf,
        git_manager: GitManager,
+        features: String,
    ) -> Result<Self> {
-        Ok(Self { repo_root, output_dir, git_manager })
+        Ok(Self { repo_root, output_dir, git_manager, features })
    }

    /// Detect if the RPC endpoint is an Optimism chain
@@ -66,13 +68,7 @@ impl CompilationManager {
    }

    /// Compile reth using cargo build and cache the binary
-    pub(crate) fn compile_reth(
-        &self,
-        commit: &str,
-        is_optimism: bool,
-        features: &str,
-        rustflags: &str,
-    ) -> Result<()> {
+    pub(crate) fn compile_reth(&self, commit: &str, is_optimism: bool) -> Result<()> {
        // Validate that current git commit matches the expected commit
        let current_commit = self.git_manager.get_current_commit()?;
        if current_commit != commit {
@@ -104,8 +100,9 @@ impl CompilationManager {
        let mut cmd = Command::new("cargo");
        cmd.arg("build").arg("--profile").arg("profiling");

-        cmd.arg("--features").arg(features);
-        info!("Using features: {features}");
+        // Add features
+        cmd.arg("--features").arg(&self.features);
+        info!("Using features: {}", self.features);

        // Add bin-specific arguments for optimism
        if is_optimism {
@@ -117,9 +114,8 @@ impl CompilationManager {

        cmd.current_dir(&self.repo_root);

-        // Set RUSTFLAGS
-        cmd.env("RUSTFLAGS", rustflags);
-        info!("Using RUSTFLAGS: {rustflags}");
+        // Set RUSTFLAGS for native CPU optimization
+        cmd.env("RUSTFLAGS", "-C target-cpu=native");

        // Debug log the command
        debug!("Executing cargo command: {:?}", cmd);
--- a/bin/reth-bench-compare/src/node.rs
+++ b/bin/reth-bench-compare/src/node.rs
@@ -211,11 +211,6 @@ impl NodeManager {
        cmd.arg("--");
        cmd.args(reth_args);

-        // Enable tracing-samply
-        if supports_samply_flags(&reth_args[0]) {
-            cmd.arg("--log.samply");
-        }
-
        // Set environment variable to disable log styling
        cmd.env("RUST_LOG_STYLE", "never");

@@ -557,16 +552,3 @@ impl NodeManager {
        Ok(())
    }
 }
-
-fn supports_samply_flags(bin: &str) -> bool {
-    let mut cmd = std::process::Command::new(bin);
-    // NOTE: The flag to check must come before --help.
-    // We pass --help as a shortcut to not execute any command.
-    cmd.args(["--log.samply", "--help"]);
-    debug!(?cmd, "Checking samply flags support");
-    let Ok(output) = cmd.output() else {
-        return false;
-    };
-    debug!(?output, "Samply flags support check");
-    output.status.success()
-}
--- a/bin/reth-bench/Cargo.toml
+++ b/bin/reth-bench/Cargo.toml
@@ -58,7 +58,6 @@ tokio = { workspace = true, features = ["sync", "macros", "time", "rt-multi-thre
 # misc
 clap = { workspace = true, features = ["derive", "env"] }
 eyre.workspace = true
-color-eyre.workspace = true
 thiserror.workspace = true
 humantime.workspace = true

--- a/bin/reth-bench/src/bench/context.rs
+++ b/bin/reth-bench/src/bench/context.rs
@@ -103,15 +103,14 @@ impl BenchContext {
            (bench_args.from, bench_args.to)
        };

-        // If `--to` are not provided, we will run the benchmark continuously,
+        // If neither `--from` nor `--to` are provided, we will run the benchmark continuously,
        // starting at the latest block.
-        let latest_block =
-            block_provider.get_block_by_number(BlockNumberOrTag::Latest).full().await?.unwrap();
-        let mut benchmark_mode = BenchMode::new(from, to, latest_block.into_inner().number())?;
+        let mut benchmark_mode = BenchMode::new(from, to)?;

        let first_block = match benchmark_mode {
-            BenchMode::Continuous(start) => {
-                block_provider.get_block_by_number(start.into()).full().await?.unwrap()
+            BenchMode::Continuous => {
+                // fetch Latest block
+                block_provider.get_block_by_number(BlockNumberOrTag::Latest).full().await?.unwrap()
            }
            BenchMode::Range(ref mut range) => {
                match range.next() {
--- a/bin/reth-bench/src/bench_mode.rs
+++ b/bin/reth-bench/src/bench_mode.rs
@@ -5,8 +5,9 @@ use std::ops::RangeInclusive;
 /// Whether or not the benchmark should run as a continuous stream of payloads.
 #[derive(Debug, PartialEq, Eq)]
 pub enum BenchMode {
+    // TODO: just include the start block in `Continuous`
    /// Run the benchmark as a continuous stream of payloads, until the benchmark is interrupted.
-    Continuous(u64),
+    Continuous,
    /// Run the benchmark for a specific range of blocks.
    Range(RangeInclusive<u64>),
 }
@@ -15,19 +16,18 @@ impl BenchMode {
    /// Check if the block number is in the range
    pub fn contains(&self, block_number: u64) -> bool {
        match self {
-            Self::Continuous(start) => block_number >= *start,
+            Self::Continuous => true,
            Self::Range(range) => range.contains(&block_number),
        }
    }

    /// Create a [`BenchMode`] from optional `from` and `to` fields.
-    pub fn new(from: Option<u64>, to: Option<u64>, latest_block: u64) -> Result<Self, eyre::Error> {
+    pub fn new(from: Option<u64>, to: Option<u64>) -> Result<Self, eyre::Error> {
        // If neither `--from` nor `--to` are provided, we will run the benchmark continuously,
        // starting at the latest block.
        match (from, to) {
            (Some(from), Some(to)) => Ok(Self::Range(from..=to)),
-            (None, None) => Ok(Self::Continuous(latest_block)),
-            (Some(start), None) => Ok(Self::Continuous(start)),
+            (None, None) => Ok(Self::Continuous),
            _ => {
                // both or neither are allowed, everything else is ambiguous
                Err(eyre::eyre!("`from` and `to` must be provided together, or not at all."))
--- a/bin/reth-bench/src/main.rs
+++ b/bin/reth-bench/src/main.rs
@@ -23,7 +23,7 @@ use bench::BenchmarkCommand;
 use clap::Parser;
 use reth_cli_runner::CliRunner;

-fn main() -> eyre::Result<()> {
+fn main() {
    // Enable backtraces unless a RUST_BACKTRACE value has already been explicitly provided.
    if std::env::var_os("RUST_BACKTRACE").is_none() {
        unsafe {
@@ -31,11 +31,12 @@ fn main() -> eyre::Result<()> {
        }
    }

-    color_eyre::install()?;
-
    // Run until either exit or sigint or sigterm
-    let runner = CliRunner::try_default_runtime()?;
-    runner.run_command_until_exit(|ctx| BenchmarkCommand::parse().execute(ctx))?;
-
-    Ok(())
+    let runner = CliRunner::try_default_runtime().unwrap();
+    runner
+        .run_command_until_exit(|ctx| {
+            let command = BenchmarkCommand::parse();
+            command.execute(ctx)
+        })
+        .unwrap();
 }
--- a/bin/reth/Cargo.toml
+++ b/bin/reth/Cargo.toml
@@ -81,7 +81,7 @@ backon.workspace = true
 tempfile.workspace = true

 [features]
-default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
+default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global"]

 otlp = [
    "reth-ethereum-cli/otlp",
@@ -103,7 +103,6 @@ asm-keccak = [
    "reth-node-ethereum/asm-keccak",
 ]
 keccak-cache-global = [
-    "reth-node-core/keccak-cache-global",
    "reth-node-ethereum/keccak-cache-global",
 ]
 jemalloc = [
--- a/crates/chain-state/src/in_memory.rs
+++ b/crates/chain-state/src/in_memory.rs
@@ -86,20 +86,14 @@ impl<N: NodePrimitives> InMemoryState<N> {
    ///
    /// This tries to acquire a read lock. Drop any write locks before calling this.
    pub(crate) fn update_metrics(&self) {
-        let (count, earliest, latest) = {
-            let numbers = self.numbers.read();
-            let count = numbers.len();
-            let earliest = numbers.first_key_value().map(|(number, _)| *number);
-            let latest = numbers.last_key_value().map(|(number, _)| *number);
-            (count, earliest, latest)
-        };
-        if let Some(earliest_block_number) = earliest {
-            self.metrics.earliest_block.set(earliest_block_number as f64);
+        let numbers = self.numbers.read();
+        if let Some((earliest_block_number, _)) = numbers.first_key_value() {
+            self.metrics.earliest_block.set(*earliest_block_number as f64);
        }
-        if let Some(latest_block_number) = latest {
-            self.metrics.latest_block.set(latest_block_number as f64);
+        if let Some((latest_block_number, _)) = numbers.last_key_value() {
+            self.metrics.latest_block.set(*latest_block_number as f64);
        }
-        self.metrics.num_blocks.set(count as f64);
+        self.metrics.num_blocks.set(numbers.len() as f64);
    }

    /// Returns the state for a given block hash.
@@ -670,14 +664,22 @@ impl<N: NodePrimitives> BlockState<N> {
        receipts.first().map(|receipts| receipts.deref()).unwrap_or_default()
    }

-    /// Returns an iterator over __parent__ `BlockStates`.
+    /// Returns a vector of __parent__ `BlockStates`.
    ///
-    /// The block state order is newest to oldest (highest to lowest):
+    /// The block state order in the output vector is newest to oldest (highest to lowest):
    /// `[5,4,3,2,1]`
    ///
    /// Note: This does not include self.
-    pub fn parent_state_chain(&self) -> impl Iterator<Item = &Self> + '_ {
-        std::iter::successors(self.parent.as_deref(), |state| state.parent.as_deref())
+    pub fn parent_state_chain(&self) -> Vec<&Self> {
+        let mut parents = Vec::new();
+        let mut current = self.parent.as_deref();
+
+        while let Some(parent) = current {
+            parents.push(parent);
+            current = parent.parent.as_deref();
+        }
+
+        parents
    }

    /// Returns a vector of `BlockStates` representing the entire in memory chain.
@@ -688,11 +690,6 @@ impl<N: NodePrimitives> BlockState<N> {
    }

    /// Appends the parent chain of this [`BlockState`] to the given vector.
-    ///
-    /// Parents are appended in order from newest to oldest (highest to lowest).
-    /// This does not include self, only the parent states.
-    ///
-    /// This is a convenience method equivalent to `chain.extend(self.parent_state_chain())`.
    pub fn append_parent_chain<'a>(&'a self, chain: &mut Vec<&'a Self>) {
        chain.extend(self.parent_state_chain());
    }
@@ -1456,18 +1453,19 @@ mod tests {
        let mut test_block_builder: TestBlockBuilder = TestBlockBuilder::default();
        let chain = create_mock_state_chain(&mut test_block_builder, 4);

-        let parents: Vec<_> = chain[3].parent_state_chain().collect();
+        let parents = chain[3].parent_state_chain();
        assert_eq!(parents.len(), 3);
        assert_eq!(parents[0].block().recovered_block().number, 3);
        assert_eq!(parents[1].block().recovered_block().number, 2);
        assert_eq!(parents[2].block().recovered_block().number, 1);

-        let parents: Vec<_> = chain[2].parent_state_chain().collect();
+        let parents = chain[2].parent_state_chain();
        assert_eq!(parents.len(), 2);
        assert_eq!(parents[0].block().recovered_block().number, 2);
        assert_eq!(parents[1].block().recovered_block().number, 1);

-        assert_eq!(chain[0].parent_state_chain().count(), 0);
+        let parents = chain[0].parent_state_chain();
+        assert_eq!(parents.len(), 0);
    }

    #[test]
@@ -1478,7 +1476,7 @@ mod tests {
            create_mock_state(&mut test_block_builder, single_block_number, B256::random());
        let single_block_hash = single_block.block().recovered_block().hash();

-        let parents: Vec<_> = single_block.parent_state_chain().collect();
+        let parents = single_block.parent_state_chain();
        assert_eq!(parents.len(), 0);

        let block_state_chain = single_block.chain().collect::<Vec<_>>();
--- a/crates/chain-state/src/memory_overlay.rs
+++ b/crates/chain-state/src/memory_overlay.rs
@@ -5,14 +5,14 @@ use reth_errors::ProviderResult;
 use reth_primitives_traits::{Account, Bytecode, NodePrimitives};
 use reth_storage_api::{
    AccountReader, BlockHashReader, BytecodeReader, HashedPostStateProvider, StateProofProvider,
-    StateProvider, StateProviderBox, StateRootProvider, StorageRootProvider,
+    StateProvider, StateRootProvider, StorageRootProvider,
 };
 use reth_trie::{
    updates::TrieUpdates, AccountProof, HashedPostState, HashedStorage, MultiProof,
    MultiProofTargets, StorageMultiProof, TrieInput,
 };
 use revm_database::BundleState;
-use std::{borrow::Cow, sync::OnceLock};
+use std::sync::OnceLock;

 /// A state provider that stores references to in-memory blocks along with their state as well as a
 /// reference of the historical state provider for fallback lookups.
@@ -24,11 +24,15 @@ pub struct MemoryOverlayStateProviderRef<
    /// Historical state provider for state lookups that are not found in memory blocks.
    pub(crate) historical: Box<dyn StateProvider + 'a>,
    /// The collection of executed parent blocks. Expected order is newest to oldest.
-    pub(crate) in_memory: Cow<'a, [ExecutedBlock<N>]>,
+    pub(crate) in_memory: Vec<ExecutedBlock<N>>,
    /// Lazy-loaded in-memory trie data.
    pub(crate) trie_input: OnceLock<TrieInput>,
 }

+/// A state provider that stores references to in-memory blocks along with their state as well as
+/// the historical state provider for fallback lookups.
+pub type MemoryOverlayStateProvider<N> = MemoryOverlayStateProviderRef<'static, N>;
+
 impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
    /// Create new memory overlay state provider.
    ///
@@ -38,7 +42,7 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
    /// - `historical` - a historical state provider for the latest ancestor block stored in the
    ///   database.
    pub fn new(historical: Box<dyn StateProvider + 'a>, in_memory: Vec<ExecutedBlock<N>>) -> Self {
-        Self { historical, in_memory: Cow::Owned(in_memory), trie_input: OnceLock::new() }
+        Self { historical, in_memory, trie_input: OnceLock::new() }
    }

    /// Turn this state provider into a state provider
@@ -49,14 +53,11 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
    /// Return lazy-loaded trie state aggregated from in-memory blocks.
    fn trie_input(&self) -> &TrieInput {
        self.trie_input.get_or_init(|| {
-            let mut input = TrieInput::default();
-            // Iterate from oldest to newest
-            for block in self.in_memory.iter().rev() {
-                let data = block.trie_data();
-                input.nodes.extend_from_sorted(&data.trie_updates);
-                input.state.extend_from_sorted(&data.hashed_state);
-            }
-            input
+            let bundles: Vec<_> =
+                self.in_memory.iter().rev().map(|block| block.trie_data()).collect();
+            TrieInput::from_blocks_sorted(
+                bundles.iter().map(|data| (data.hashed_state.as_ref(), data.trie_updates.as_ref())),
+            )
        })
    }

@@ -70,7 +71,7 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {

 impl<N: NodePrimitives> BlockHashReader for MemoryOverlayStateProviderRef<'_, N> {
    fn block_hash(&self, number: BlockNumber) -> ProviderResult<Option<B256>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
            if block.recovered_block().number() == number {
                return Ok(Some(block.recovered_block().hash()));
            }
@@ -89,7 +90,7 @@ impl<N: NodePrimitives> BlockHashReader for MemoryOverlayStateProviderRef<'_, N>
        let mut in_memory_hashes = Vec::with_capacity(range.size_hint().0);

        // iterate in ascending order (oldest to newest = low to high)
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
            let block_num = block.recovered_block().number();
            if range.contains(&block_num) {
                in_memory_hashes.push(block.recovered_block().hash());
@@ -111,7 +112,7 @@ impl<N: NodePrimitives> BlockHashReader for MemoryOverlayStateProviderRef<'_, N>

 impl<N: NodePrimitives> AccountReader for MemoryOverlayStateProviderRef<'_, N> {
    fn basic_account(&self, address: &Address) -> ProviderResult<Option<Account>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
            if let Some(account) = block.execution_output.account(address) {
                return Ok(account);
            }
@@ -215,7 +216,7 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {
        address: Address,
        storage_key: StorageKey,
    ) -> ProviderResult<Option<StorageValue>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
            if let Some(value) = block.execution_output.storage(&address, storage_key.into()) {
                return Ok(Some(value));
            }
@@ -227,7 +228,7 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {

 impl<N: NodePrimitives> BytecodeReader for MemoryOverlayStateProviderRef<'_, N> {
    fn bytecode_by_hash(&self, code_hash: &B256) -> ProviderResult<Option<Bytecode>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
            if let Some(contract) = block.execution_output.bytecode(code_hash) {
                return Ok(Some(contract));
            }
@@ -236,46 +237,3 @@ impl<N: NodePrimitives> BytecodeReader for MemoryOverlayStateProviderRef<'_, N>
        self.historical.bytecode_by_hash(code_hash)
    }
 }
-
-/// An owned state provider that stores references to in-memory blocks along with their state as
-/// well as a reference of the historical state provider for fallback lookups.
-#[expect(missing_debug_implementations)]
-pub struct MemoryOverlayStateProvider<N: NodePrimitives = reth_ethereum_primitives::EthPrimitives> {
-    /// Historical state provider for state lookups that are not found in memory blocks.
-    pub(crate) historical: StateProviderBox,
-    /// The collection of executed parent blocks. Expected order is newest to oldest.
-    pub(crate) in_memory: Vec<ExecutedBlock<N>>,
-    /// Lazy-loaded in-memory trie data.
-    pub(crate) trie_input: OnceLock<TrieInput>,
-}
-
-impl<N: NodePrimitives> MemoryOverlayStateProvider<N> {
-    /// Create new memory overlay state provider.
-    ///
-    /// ## Arguments
-    ///
-    /// - `in_memory` - the collection of executed ancestor blocks in reverse.
-    /// - `historical` - a historical state provider for the latest ancestor block stored in the
-    ///   database.
-    pub fn new(historical: StateProviderBox, in_memory: Vec<ExecutedBlock<N>>) -> Self {
-        Self { historical, in_memory, trie_input: OnceLock::new() }
-    }
-
-    /// Returns a new provider that takes the `TX` as reference
-    #[inline(always)]
-    fn as_ref(&self) -> MemoryOverlayStateProviderRef<'_, N> {
-        MemoryOverlayStateProviderRef {
-            historical: Box::new(self.historical.as_ref()),
-            in_memory: Cow::Borrowed(&self.in_memory),
-            trie_input: self.trie_input.clone(),
-        }
-    }
-
-    /// Wraps the [`Self`] in a `Box`.
-    pub fn boxed(self) -> StateProviderBox {
-        Box::new(self)
-    }
-}
-
-// Delegates all provider impls to [`MemoryOverlayStateProviderRef`]
-reth_storage_api::macros::delegate_provider_impls!(MemoryOverlayStateProvider<N> where [N: NodePrimitives]);
--- a/crates/chain-state/src/test_utils.rs
+++ b/crates/chain-state/src/test_utils.rs
@@ -117,7 +117,7 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
            .map(|_| {
                let tx = mock_tx(self.signer_build_account_info.nonce);
                self.signer_build_account_info.nonce += 1;
-                self.signer_build_account_info.balance -= Self::single_tx_cost();
+                self.signer_build_account_info.balance -= signer_balance_decrease;
                tx
            })
            .collect();
--- a/crates/chainspec/src/spec.rs
+++ b/crates/chainspec/src/spec.rs
@@ -80,8 +80,6 @@ pub fn make_genesis_header(genesis: &Genesis, hardforks: &ChainHardforks) -> Hea
        .then_some(EMPTY_REQUESTS_HASH);

    Header {
-        number: genesis.number.unwrap_or_default(),
-        parent_hash: genesis.parent_hash.unwrap_or_default(),
        gas_limit: genesis.gas_limit,
        difficulty: genesis.difficulty,
        nonce: genesis.nonce.into(),
@@ -970,7 +968,7 @@ impl<H: BlockHeader> EthereumHardforks for ChainSpec<H> {

 /// A trait for reading the current chainspec.
 #[auto_impl::auto_impl(&, Arc)]
-pub trait ChainSpecProvider: Debug + Send {
+pub trait ChainSpecProvider: Debug + Send + Sync {
    /// The chain spec type.
    type ChainSpec: EthChainSpec + 'static;

--- a/crates/cli/commands/src/common.rs
+++ b/crates/cli/commands/src/common.rs
@@ -23,10 +23,7 @@ use reth_node_core::{
    dirs::{ChainPath, DataDirPath},
 };
 use reth_provider::{
-    providers::{
-        BlockchainProvider, NodeTypesForProvider, RocksDBProvider, StaticFileProvider,
-        StaticFileProviderBuilder,
-    },
+    providers::{BlockchainProvider, NodeTypesForProvider, RocksDBProvider, StaticFileProvider},
    ProviderFactory, StaticFileProviderFactory,
 };
 use reth_stages::{sets::DefaultStages, Pipeline, PipelineTarget};
@@ -103,23 +100,15 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
        }

        info!(target: "reth::cli", ?db_path, ?sf_path, "Opening storage");
-        let genesis_block_number = self.chain.genesis().number.unwrap_or_default();
        let (db, sfp) = match access {
            AccessRights::RW => (
                Arc::new(init_db(db_path, self.db.database_args())?),
-                StaticFileProviderBuilder::read_write(sf_path)?
-                    .with_genesis_block_number(genesis_block_number)
-                    .build()?,
+                StaticFileProvider::read_write(sf_path)?,
+            ),
+            AccessRights::RO | AccessRights::RoInconsistent => (
+                Arc::new(open_db_read_only(&db_path, self.db.database_args())?),
+                StaticFileProvider::read_only(sf_path, false)?,
            ),
-            AccessRights::RO | AccessRights::RoInconsistent => {
-                (Arc::new(open_db_read_only(&db_path, self.db.database_args())?), {
-                    let provider = StaticFileProviderBuilder::read_only(sf_path)?
-                        .with_genesis_block_number(genesis_block_number)
-                        .build()?;
-                    provider.watch_directory();
-                    provider
-                })
-            }
        };
        // TransactionDB only support read-write mode
        let rocksdb_provider = RocksDBProvider::builder(data_dir.rocksdb())
--- a/crates/cli/commands/src/db/repair_trie.rs
+++ b/crates/cli/commands/src/db/repair_trie.rs
@@ -301,8 +301,8 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
    if inconsistent_nodes == 0 {
        info!("No inconsistencies found");
    } else {
+        info!("Repaired {} inconsistencies, committing changes", inconsistent_nodes);
        provider_rw.commit()?;
-        info!("Repaired {} inconsistencies and committed changes", inconsistent_nodes);
    }

    Ok(())
--- a/crates/cli/commands/src/init_cmd.rs
+++ b/crates/cli/commands/src/init_cmd.rs
@@ -1,9 +1,8 @@
 //! Command that initializes the node from a genesis file.

 use crate::common::{AccessRights, CliNodeTypes, Environment, EnvironmentArgs};
-use alloy_consensus::BlockHeader;
 use clap::Parser;
-use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
+use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
 use reth_provider::BlockHashReader;
 use std::sync::Arc;
@@ -23,9 +22,8 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> InitComman

        let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RW)?;

-        let genesis_block_number = provider_factory.chain_spec().genesis_header().number();
        let hash = provider_factory
-            .block_hash(genesis_block_number)?
+            .block_hash(0)?
            .ok_or_else(|| eyre::eyre!("Genesis hash not found."))?;

        info!(target: "reth::cli", hash = ?hash, "Genesis block written");
--- a/crates/cli/commands/src/init_state/without_evm.rs
+++ b/crates/cli/commands/src/init_state/without_evm.rs
@@ -79,7 +79,7 @@ where
        + StaticFileProviderFactory<Primitives: NodePrimitives<BlockHeader: Compact>>,
 {
    provider_rw.insert_block(
-        &SealedBlock::<<Provider::Primitives as NodePrimitives>::Block>::from_sealed_parts(
+        SealedBlock::<<Provider::Primitives as NodePrimitives>::Block>::from_sealed_parts(
            header.clone(),
            Default::default(),
        )
--- a/crates/cli/commands/src/p2p/mod.rs
+++ b/crates/cli/commands/src/p2p/mod.rs
@@ -72,7 +72,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                .split();
                if result.len() != 1 {
                    eyre::bail!(
-                        "Invalid number of bodies received. Expected: 1. Received: {}",
+                        "Invalid number of headers received. Expected: 1. Received: {}",
                        result.len()
                    )
                }
--- a/crates/config/src/config.rs
+++ b/crates/config/src/config.rs
@@ -22,6 +22,7 @@ pub const DEFAULT_BLOCK_INTERVAL: usize = 5;
 #[cfg_attr(feature = "serde", serde(default))]
 pub struct Config {
    /// Configuration for each stage in the pipeline.
+    // TODO(onbjerg): Can we make this easier to maintain when we add/remove stages?
    pub stages: StageConfig,
    /// Configuration for pruning.
    #[cfg_attr(feature = "serde", serde(default))]
--- a/crates/e2e-test-utils/src/setup_builder.rs
+++ b/crates/e2e-test-utils/src/setup_builder.rs
@@ -11,7 +11,6 @@ use reth_node_builder::{
    PayloadTypes,
 };
 use reth_node_core::args::{DiscoveryArgs, NetworkArgs, RpcServerArgs};
-use reth_primitives_traits::AlloyBlockHeader;
 use reth_provider::providers::BlockchainProvider;
 use reth_rpc_server_types::RpcModuleSelection;
 use reth_tasks::TaskManager;
@@ -158,8 +157,8 @@ where
                    .await?;

                let node = NodeTestContext::new(node, self.attributes_generator).await?;
-                let genesis_number = self.chain_spec.genesis_header().number();
-                let genesis = node.block_hash(genesis_number);
+
+                let genesis = node.block_hash(0);
                node.update_forkchoice(genesis, genesis).await?;

                eyre::Ok(node)
--- a/crates/e2e-test-utils/src/testsuite/actions/engine_api.rs
+++ b/crates/e2e-test-utils/src/testsuite/actions/engine_api.rs
@@ -2,7 +2,9 @@

 use crate::testsuite::{Action, Environment};
 use alloy_primitives::B256;
-use alloy_rpc_types_engine::{ExecutionPayloadV3, PayloadStatusEnum};
+use alloy_rpc_types_engine::{
+    ExecutionPayloadV1, ExecutionPayloadV2, ExecutionPayloadV3, PayloadStatusEnum,
+};
 use alloy_rpc_types_eth::{Block, Header, Receipt, Transaction, TransactionRequest};
 use eyre::Result;
 use futures_util::future::BoxFuture;
@@ -129,10 +131,7 @@ where
            })?;

            // Convert block to ExecutionPayloadV3
-            let payload = ExecutionPayloadV3::from_block_unchecked(
-                block.hash(),
-                &block.map_transactions(|tx| tx.inner).into_consensus(),
-            );
+            let payload = block_to_payload_v3(block.clone());

            // Send the payload to the target node
            let target_engine = env.node_clients[self.node_idx].engine.http_client();
@@ -328,3 +327,32 @@ where
        })
    }
 }
+
+/// Helper function to convert a block to `ExecutionPayloadV3`
+fn block_to_payload_v3(block: Block) -> ExecutionPayloadV3 {
+    use alloy_primitives::U256;
+
+    ExecutionPayloadV3 {
+        payload_inner: ExecutionPayloadV2 {
+            payload_inner: ExecutionPayloadV1 {
+                parent_hash: block.header.inner.parent_hash,
+                fee_recipient: block.header.inner.beneficiary,
+                state_root: block.header.inner.state_root,
+                receipts_root: block.header.inner.receipts_root,
+                logs_bloom: block.header.inner.logs_bloom,
+                prev_randao: block.header.inner.mix_hash,
+                block_number: block.header.inner.number,
+                gas_limit: block.header.inner.gas_limit,
+                gas_used: block.header.inner.gas_used,
+                timestamp: block.header.inner.timestamp,
+                extra_data: block.header.inner.extra_data.clone(),
+                base_fee_per_gas: U256::from(block.header.inner.base_fee_per_gas.unwrap_or(0)),
+                block_hash: block.header.hash,
+                transactions: vec![], // No transactions needed for buffering tests
+            },
+            withdrawals: block.withdrawals.unwrap_or_default().to_vec(),
+        },
+        blob_gas_used: block.header.inner.blob_gas_used.unwrap_or(0),
+        excess_blob_gas: block.header.inner.excess_blob_gas.unwrap_or(0),
+    }
+}
--- a/crates/engine/invalid-block-hooks/src/witness.rs
+++ b/crates/engine/invalid-block-hooks/src/witness.rs
@@ -5,7 +5,7 @@ use pretty_assertions::Comparison;
 use reth_engine_primitives::InvalidBlockHook;
 use reth_evm::{execute::Executor, ConfigureEvm};
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedHeader};
-use reth_provider::{BlockExecutionOutput, StateProvider, StateProviderBox, StateProviderFactory};
+use reth_provider::{BlockExecutionOutput, StateProvider, StateProviderFactory};
 use reth_revm::{
    database::StateProviderDatabase,
    db::{BundleState, State},
@@ -80,13 +80,13 @@ fn sort_bundle_state_for_comparison(bundle_state: &BundleState) -> BundleStateSo
                    BundleAccountSorted {
                        info: acc.info.clone(),
                        original_info: acc.original_info.clone(),
-                        storage: acc.storage.iter().map(|(k, v)| (*k, *v)).collect(),
+                        storage: BTreeMap::from_iter(acc.storage.clone()),
                        status: acc.status,
                    },
                )
            })
            .collect(),
-        contracts: bundle_state.contracts.iter().map(|(k, v)| (*k, v.clone())).collect(),
+        contracts: BTreeMap::from_iter(bundle_state.contracts.clone()),
        reverts: bundle_state
            .reverts
            .iter()
@@ -98,7 +98,7 @@ fn sort_bundle_state_for_comparison(bundle_state: &BundleState) -> BundleStateSo
                            *addr,
                            AccountRevertSorted {
                                account: rev.account.clone(),
-                                storage: rev.storage.iter().map(|(k, v)| (*k, *v)).collect(),
+                                storage: BTreeMap::from_iter(rev.storage.clone()),
                                previous_status: rev.previous_status,
                                wipe_storage: rev.wipe_storage,
                            },
@@ -114,7 +114,7 @@ fn sort_bundle_state_for_comparison(bundle_state: &BundleState) -> BundleStateSo

 /// Extracts execution data including codes, preimages, and hashed state from database
 fn collect_execution_data(
-    mut db: State<StateProviderDatabase<StateProviderBox>>,
+    mut db: State<StateProviderDatabase<Box<dyn StateProvider>>>,
 ) -> eyre::Result<CollectionResult> {
    let bundle_state = db.take_bundle();
    let mut codes = BTreeMap::new();
@@ -530,7 +530,9 @@ mod tests {
        // Create a State with StateProviderTest
        let state_provider = StateProviderTest::default();
        let mut state = State::builder()
-            .with_database(StateProviderDatabase::new(Box::new(state_provider) as StateProviderBox))
+            .with_database(StateProviderDatabase::new(
+                Box::new(state_provider) as Box<dyn StateProvider>
+            ))
            .with_bundle_update()
            .build();

--- a/crates/engine/tree/Cargo.toml
+++ b/crates/engine/tree/Cargo.toml
@@ -29,7 +29,6 @@ reth-provider.workspace = true
 reth-prune.workspace = true
 reth-revm.workspace = true
 reth-stages-api.workspace = true
-reth-storage-errors.workspace = true
 reth-tasks.workspace = true
 reth-trie-parallel.workspace = true
 reth-trie-sparse = { workspace = true, features = ["std", "metrics"] }
@@ -53,7 +52,6 @@ futures.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "sync", "macros"] }
 mini-moka = { workspace = true, features = ["sync"] }
-moka = { workspace = true, features = ["sync"] }
 smallvec.workspace = true

 # metrics
--- a/crates/engine/tree/docs/root.md
+++ b/crates/engine/tree/docs/root.md
@@ -128,12 +128,12 @@ we send them along with the state updates to the [Sparse Trie Task](#sparse-trie

 ### Finishing the calculation

-Once all transactions are executed, the [Engine](#engine) sends a `StateRootMessage::FinishedStateUpdates` message
+Once all transactions are executed, the [Engine](#engine) sends a `StateRootMessage::FinishStateUpdates` message
 to the State Root Task, marking the end of receiving state updates.

 Every time we receive a new proof from the [MultiProof Manager](#multiproof-manager), we also check
 the following conditions:
-1. Are all updates received? (`StateRootMessage::FinishedStateUpdates` was sent)
+1. Are all updates received? (`StateRootMessage::FinishStateUpdates` was sent)
 2. Is `ProofSequencer` empty? (no proofs are pending for sequencing)
 3. Are all proofs that were sent to the [`MultiProofManager::spawn_or_queue`](#multiproof-manager) finished
 calculating and were sent to the [Sparse Trie Task](#sparse-trie-task)?
--- a/crates/engine/tree/src/backfill.rs
+++ b/crates/engine/tree/src/backfill.rs
@@ -47,7 +47,7 @@ impl BackfillSyncState {
 }

 /// Backfill sync mode functionality.
-pub trait BackfillSync: Send {
+pub trait BackfillSync: Send + Sync {
    /// Performs a backfill action.
    fn on_action(&mut self, action: BackfillAction);

--- a/crates/engine/tree/src/chain.rs
+++ b/crates/engine/tree/src/chain.rs
@@ -219,19 +219,10 @@ pub enum HandlerEvent<T> {
 }

 /// Internal events issued by the [`ChainOrchestrator`].
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub enum FromOrchestrator {
    /// Invoked when backfill sync finished
    BackfillSyncFinished(ControlFlow),
    /// Invoked when backfill sync started
    BackfillSyncStarted,
-    /// Gracefully terminate the engine service.
-    ///
-    /// When this variant is received, the engine will persist all remaining in-memory blocks
-    /// to disk before shutting down. Once persistence is complete, a signal is sent through
-    /// the oneshot channel to notify the caller.
-    Terminate {
-        /// Channel to signal termination completion.
-        tx: tokio::sync::oneshot::Sender<()>,
-    },
 }
--- a/crates/engine/tree/src/metrics.rs
+++ b/crates/engine/tree/src/metrics.rs
@@ -19,8 +19,6 @@ pub(crate) struct PersistenceMetrics {
    pub(crate) remove_blocks_above_duration_seconds: Histogram,
    /// How long it took for blocks to be saved
    pub(crate) save_blocks_duration_seconds: Histogram,
-    /// How many blocks we persist at once.
-    pub(crate) save_blocks_block_count: Histogram,
    /// How long it took for blocks to be pruned
    pub(crate) prune_before_duration_seconds: Histogram,
 }
--- a/crates/engine/tree/src/persistence.rs
+++ b/crates/engine/tree/src/persistence.rs
@@ -1,4 +1,5 @@
 use crate::metrics::PersistenceMetrics;
+use alloy_consensus::BlockHeader;
 use alloy_eips::BlockNumHash;
 use reth_chain_state::ExecutedBlock;
 use reth_errors::ProviderError;
@@ -141,25 +142,27 @@ where
        &self,
        blocks: Vec<ExecutedBlock<N::Primitives>>,
    ) -> Result<Option<BlockNumHash>, PersistenceError> {
-        let first_block = blocks.first().map(|b| b.recovered_block.num_hash());
-        let last_block = blocks.last().map(|b| b.recovered_block.num_hash());
-        let block_count = blocks.len();
-        debug!(target: "engine::persistence", ?block_count, first=?first_block, last=?last_block, "Saving range of blocks");
+        let first_block_hash = blocks.first().map(|b| b.recovered_block.num_hash());
+        let last_block_hash = blocks.last().map(|b| b.recovered_block.num_hash());
+        debug!(target: "engine::persistence", first=?first_block_hash, last=?last_block_hash, "Saving range of blocks");

        let start_time = Instant::now();
+        let last_block_hash_num = blocks.last().map(|block| BlockNumHash {
+            hash: block.recovered_block().hash(),
+            number: block.recovered_block().header().number(),
+        });

-        if last_block.is_some() {
+        if last_block_hash_num.is_some() {
            let provider_rw = self.provider.database_provider_rw()?;

            provider_rw.save_blocks(blocks)?;
            provider_rw.commit()?;
        }

-        debug!(target: "engine::persistence", first=?first_block, last=?last_block, "Saved range of blocks");
+        debug!(target: "engine::persistence", first=?first_block_hash, last=?last_block_hash, "Saved range of blocks");

-        self.metrics.save_blocks_block_count.record(block_count as f64);
        self.metrics.save_blocks_duration_seconds.record(start_time.elapsed());
-        Ok(last_block)
+        Ok(last_block_hash_num)
    }
 }

--- a/crates/engine/tree/src/tree/block_buffer.rs
+++ b/crates/engine/tree/src/tree/block_buffer.rs
@@ -14,7 +14,7 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 /// * [`BlockBuffer::remove_old_blocks`] to remove old blocks that precede the finalized number.
 ///
 /// Note: Buffer is limited by number of blocks that it can contain and eviction of the block
-/// is done in FIFO order (oldest inserted block is evicted first).
+/// is done by last recently used block.
 #[derive(Debug)]
 pub struct BlockBuffer<B: Block> {
    /// All blocks in the buffer stored by their block hash.
--- a/crates/engine/tree/src/tree/cached_state.rs
+++ b/crates/engine/tree/src/tree/cached_state.rs
@@ -31,9 +31,6 @@ pub(crate) struct CachedStateProvider<S> {

    /// Metrics for the cached state provider
    metrics: CachedStateMetrics,
-
-    /// If prewarm enabled we populate every cache miss
-    prewarm: bool,
 }

 impl<S> CachedStateProvider<S>
@@ -42,32 +39,12 @@ where
 {
    /// Creates a new [`CachedStateProvider`] from an [`ExecutionCache`], state provider, and
    /// [`CachedStateMetrics`].
-    pub(crate) const fn new(
+    pub(crate) const fn new_with_caches(
        state_provider: S,
        caches: ExecutionCache,
        metrics: CachedStateMetrics,
    ) -> Self {
-        Self { state_provider, caches, metrics, prewarm: false }
-    }
-}
-
-impl<S> CachedStateProvider<S> {
-    /// Enables pre-warm mode so that every cache miss is populated.
-    ///
-    /// This is only relevant for pre-warm transaction execution with the intention to pre-populate
-    /// the cache with data for regular block execution. During regular block execution the
-    /// cache doesn't need to be populated because the actual EVM database
-    /// [`State`](revm::database::State) also caches internally during block execution and the cache
-    /// is then updated after the block with the entire [`BundleState`] output of that block which
-    /// contains all accessed accounts,code,storage. See also [`ExecutionCache::insert_state`].
-    pub(crate) const fn prewarm(mut self) -> Self {
-        self.prewarm = true;
-        self
-    }
-
-    /// Returns whether this provider should pre-warm cache misses.
-    const fn is_prewarm(&self) -> bool {
-        self.prewarm
+        Self { state_provider, caches, metrics }
    }
 }

@@ -146,10 +123,7 @@ impl<S: AccountReader> AccountReader for CachedStateProvider<S> {
        self.metrics.account_cache_misses.increment(1);

        let res = self.state_provider.basic_account(address)?;
-
-        if self.is_prewarm() {
-            self.caches.account_cache.insert(*address, res);
-        }
+        self.caches.account_cache.insert(*address, res);
        Ok(res)
    }
 }
@@ -174,19 +148,15 @@ impl<S: StateProvider> StateProvider for CachedStateProvider<S> {
        match self.caches.get_storage(&account, &storage_key) {
            (SlotStatus::NotCached, maybe_cache) => {
                let final_res = self.state_provider.storage(account, storage_key)?;
-
-                if self.is_prewarm() {
-                    let account_cache = maybe_cache.unwrap_or_default();
-                    account_cache.insert_storage(storage_key, final_res);
-                    // we always need to insert the value to update the weights.
-                    // Note: there exists a race when the storage cache did not exist yet and two
-                    // consumers looking up the a storage value for this account for the first time,
-                    // however we can assume that this will only happen for the very first
-                    // (mostlikely the same) value, and don't expect that this
-                    // will accidentally replace an account storage cache with
-                    // additional values.
-                    self.caches.insert_storage_cache(account, account_cache);
-                }
+                let account_cache = maybe_cache.unwrap_or_default();
+                account_cache.insert_storage(storage_key, final_res);
+                // we always need to insert the value to update the weights.
+                // Note: there exists a race when the storage cache did not exist yet and two
+                // consumers looking up the a storage value for this account for the first time,
+                // however we can assume that this will only happen for the very first (mostlikely
+                // the same) value, and don't expect that this will accidentally
+                // replace an account storage cache with additional values.
+                self.caches.insert_storage_cache(account, account_cache);

                self.metrics.storage_cache_misses.increment(1);
                Ok(final_res)
@@ -213,11 +183,7 @@ impl<S: BytecodeReader> BytecodeReader for CachedStateProvider<S> {
        self.metrics.code_cache_misses.increment(1);

        let final_res = self.state_provider.bytecode_by_hash(code_hash)?;
-
-        if self.is_prewarm() {
-            self.caches.code_cache.insert(*code_hash, final_res.clone());
-        }
-
+        self.caches.code_cache.insert(*code_hash, final_res.clone());
        Ok(final_res)
    }
 }
@@ -819,7 +785,7 @@ mod tests {

        let caches = ExecutionCacheBuilder::default().build_caches(1000);
        let state_provider =
-            CachedStateProvider::new(provider, caches, CachedStateMetrics::zeroed());
+            CachedStateProvider::new_with_caches(provider, caches, CachedStateMetrics::zeroed());

        // check that the storage is empty
        let res = state_provider.storage(address, storage_key);
@@ -842,7 +808,7 @@ mod tests {

        let caches = ExecutionCacheBuilder::default().build_caches(1000);
        let state_provider =
-            CachedStateProvider::new(provider, caches, CachedStateMetrics::zeroed());
+            CachedStateProvider::new_with_caches(provider, caches, CachedStateMetrics::zeroed());

        // check that the storage returns the expected value
        let res = state_provider.storage(address, storage_key);
--- a/crates/engine/tree/src/tree/instrumented_state.rs
+++ b/crates/engine/tree/src/tree/instrumented_state.rs
@@ -83,7 +83,7 @@ where
 {
    /// Creates a new [`InstrumentedStateProvider`] from a state provider with the provided label
    /// for metrics.
-    pub fn new(state_provider: S, source: &'static str) -> Self {
+    pub fn from_state_provider(state_provider: S, source: &'static str) -> Self {
        Self {
            state_provider,
            metrics: StateProviderMetrics::new_with_labels(&[("source", source)]),
--- a/crates/engine/tree/src/tree/invalid_headers.rs
+++ b/crates/engine/tree/src/tree/invalid_headers.rs
@@ -48,7 +48,6 @@ impl InvalidHeaderCache {
        // if we get here, the entry has been hit too many times, so we evict it
        self.headers.remove(hash);
        self.metrics.hit_evictions.increment(1);
-        self.metrics.count.set(self.headers.len() as f64);
        None
    }

--- a/crates/engine/tree/src/tree/metrics.rs
+++ b/crates/engine/tree/src/tree/metrics.rs
@@ -64,7 +64,6 @@ impl EngineApiMetrics {
        &self,
        executor: E,
        mut transactions: impl Iterator<Item = Result<impl ExecutableTx<E>, BlockExecutionError>>,
-        transaction_count: usize,
        state_hook: Box<dyn OnStateHook>,
    ) -> Result<(BlockExecutionOutput<E::Receipt>, Vec<Address>), BlockExecutionError>
    where
@@ -76,7 +75,7 @@ impl EngineApiMetrics {
        // be accessible.
        let wrapper = MeteredStateHook { metrics: self.executor.clone(), inner_hook: state_hook };

-        let mut senders = Vec::with_capacity(transaction_count);
+        let mut senders = Vec::new();
        let mut executor = executor.with_state_hook(Some(Box::new(wrapper)));

        let f = || {
@@ -530,7 +529,6 @@ mod tests {
        let _result = metrics.execute_metered::<_, EmptyDB>(
            executor,
            input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
-            input.transaction_count(),
            state_hook,
        );

@@ -587,7 +585,6 @@ mod tests {
        let _result = metrics.execute_metered::<_, EmptyDB>(
            executor,
            input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
-            input.transaction_count(),
            state_hook,
        );

--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -39,7 +39,6 @@ use revm::state::EvmState;
 use state::TreeState;
 use std::{
    fmt::Debug,
-    ops,
    sync::{
        mpsc::{Receiver, RecvError, RecvTimeoutError, Sender},
        Arc,
@@ -64,6 +63,7 @@ mod persistence_state;
 pub mod precompile_cache;
 #[cfg(test)]
 mod tests;
+// TODO(alexey): compare trie updates in `insert_block_inner`
 #[expect(unused)]
 mod trie_updates;

@@ -426,13 +426,9 @@ where
            match self.try_recv_engine_message() {
                Ok(Some(msg)) => {
                    debug!(target: "engine::tree", %msg, "received new engine message");
-                    match self.on_engine_message(msg) {
-                        Ok(ops::ControlFlow::Break(())) => return,
-                        Ok(ops::ControlFlow::Continue(())) => {}
-                        Err(fatal) => {
-                            error!(target: "engine::tree", %fatal, "insert block fatal error");
-                            return
-                        }
+                    if let Err(fatal) = self.on_engine_message(msg) {
+                        error!(target: "engine::tree", %fatal, "insert block fatal error");
+                        return
                    }
                }
                Ok(None) => {
@@ -825,8 +821,7 @@ where
        new_head_number: u64,
        current_head_number: u64,
    ) -> Vec<ExecutedBlock<N>> {
-        let mut old_blocks =
-            Vec::with_capacity((current_head_number.saturating_sub(new_head_number)) as usize);
+        let mut old_blocks = Vec::new();

        for block_num in (new_head_number + 1)..=current_head_number {
            if let Some(block_state) = self.canonical_in_memory_state.state_by_number(block_num) {
@@ -931,6 +926,48 @@ where
        Ok(())
    }

+    /// Determines if the given block is part of a fork by checking that these
+    /// conditions are true:
+    /// * walking back from the target hash to verify that the target hash is not part of an
+    ///   extension of the canonical chain.
+    /// * walking back from the current head to verify that the target hash is not already part of
+    ///   the canonical chain.
+    ///
+    /// The header is required as an arg, because we might be checking that the header is a fork
+    /// block before it's in the tree state and before it's in the database.
+    fn is_fork(&self, target: BlockWithParent) -> ProviderResult<bool> {
+        let target_hash = target.block.hash;
+        // verify that the given hash is not part of an extension of the canon chain.
+        let canonical_head = self.state.tree_state.canonical_head();
+        let mut current_hash;
+        let mut current_block = target;
+        loop {
+            if current_block.block.hash == canonical_head.hash {
+                return Ok(false)
+            }
+            // We already passed the canonical head
+            if current_block.block.number <= canonical_head.number {
+                break
+            }
+            current_hash = current_block.parent;
+
+            let Some(next_block) = self.sealed_header_by_hash(current_hash)? else { break };
+            current_block = next_block.block_with_parent();
+        }
+
+        // verify that the given hash is not already part of canonical chain stored in memory
+        if self.canonical_in_memory_state.header_by_hash(target_hash).is_some() {
+            return Ok(false)
+        }
+
+        // verify that the given hash is not already part of persisted canonical chain
+        if self.provider.block_number(target_hash)?.is_some() {
+            return Ok(false)
+        }
+
+        Ok(true)
+    }
+
    /// Invoked when we receive a new forkchoice update message. Calls into the blockchain tree
    /// to resolve chain forks and ensure that the Execution Layer is working with the latest valid
    /// chain.
@@ -1265,7 +1302,22 @@ where
            // Check if persistence has complete
            match rx.try_recv() {
                Ok(last_persisted_hash_num) => {
-                    self.on_persistence_complete(last_persisted_hash_num, start_time)?;
+                    self.metrics.engine.persistence_duration.record(start_time.elapsed());
+                    let Some(BlockNumHash {
+                        hash: last_persisted_block_hash,
+                        number: last_persisted_block_number,
+                    }) = last_persisted_hash_num
+                    else {
+                        // if this happened, then we persisted no blocks because we sent an
+                        // empty vec of blocks
+                        warn!(target: "engine::tree", "Persistence task completed but did not persist any blocks");
+                        return Ok(())
+                    };
+
+                    debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, elapsed=?start_time.elapsed(), "Finished persisting, calling finish");
+                    self.persistence_state
+                        .finish(last_persisted_block_hash, last_persisted_block_number);
+                    self.on_new_persisted_block()?;
                }
                Err(TryRecvError::Closed) => return Err(TryRecvError::Closed.into()),
                Err(TryRecvError::Empty) => {
@@ -1278,8 +1330,7 @@ where
            if let Some(new_tip_num) = self.find_disk_reorg()? {
                self.remove_blocks(new_tip_num)
            } else if self.should_persist() {
-                let blocks_to_persist =
-                    self.get_canonical_blocks_to_persist(PersistTarget::Threshold)?;
+                let blocks_to_persist = self.get_canonical_blocks_to_persist()?;
                self.persist_blocks(blocks_to_persist);
            }
        }
@@ -1287,72 +1338,11 @@ where
        Ok(())
    }

-    /// Finishes termination by persisting all remaining blocks and signaling completion.
-    ///
-    /// This blocks until all persistence is complete. Always signals completion,
-    /// even if an error occurs.
-    fn finish_termination(
-        &mut self,
-        pending_termination: oneshot::Sender<()>,
-    ) -> Result<(), AdvancePersistenceError> {
-        trace!(target: "engine::tree", "finishing termination, persisting remaining blocks");
-        let result = self.persist_until_complete();
-        let _ = pending_termination.send(());
-        result
-    }
-
-    /// Persists all remaining blocks until none are left.
-    fn persist_until_complete(&mut self) -> Result<(), AdvancePersistenceError> {
-        loop {
-            // Wait for any in-progress persistence to complete (blocking)
-            if let Some((rx, start_time, _action)) = self.persistence_state.rx.take() {
-                let result = rx.blocking_recv().map_err(|_| TryRecvError::Closed)?;
-                self.on_persistence_complete(result, start_time)?;
-            }
-
-            let blocks_to_persist = self.get_canonical_blocks_to_persist(PersistTarget::Head)?;
-
-            if blocks_to_persist.is_empty() {
-                debug!(target: "engine::tree", "persistence complete, signaling termination");
-                return Ok(())
-            }
-
-            debug!(target: "engine::tree", count = blocks_to_persist.len(), "persisting remaining blocks before shutdown");
-            self.persist_blocks(blocks_to_persist);
-        }
-    }
-
-    /// Handles a completed persistence task.
-    fn on_persistence_complete(
-        &mut self,
-        last_persisted_hash_num: Option<BlockNumHash>,
-        start_time: Instant,
-    ) -> Result<(), AdvancePersistenceError> {
-        self.metrics.engine.persistence_duration.record(start_time.elapsed());
-
-        let Some(BlockNumHash {
-            hash: last_persisted_block_hash,
-            number: last_persisted_block_number,
-        }) = last_persisted_hash_num
-        else {
-            // if this happened, then we persisted no blocks because we sent an empty vec of blocks
-            warn!(target: "engine::tree", "Persistence task completed but did not persist any blocks");
-            return Ok(())
-        };
-
-        debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, elapsed=?start_time.elapsed(), "Finished persisting, calling finish");
-        self.persistence_state.finish(last_persisted_block_hash, last_persisted_block_number);
-        self.on_new_persisted_block()?;
-        Ok(())
-    }
-
    /// Handles a message from the engine.
-    ///
-    /// Returns `ControlFlow::Break(())` if the engine should terminate.
    fn on_engine_message(
        &mut self,
        msg: FromEngine<EngineApiRequest<T, N>, N::Block>,
-    ) -> Result<ops::ControlFlow<()>, InsertBlockFatalError> {
+    ) -> Result<(), InsertBlockFatalError> {
        match msg {
            FromEngine::Event(event) => match event {
                FromOrchestrator::BackfillSyncStarted => {
@@ -1362,13 +1352,6 @@ where
                FromOrchestrator::BackfillSyncFinished(ctrl) => {
                    self.on_backfill_sync_finished(ctrl)?;
                }
-                FromOrchestrator::Terminate { tx } => {
-                    debug!(target: "engine::tree", "received terminate request");
-                    if let Err(err) = self.finish_termination(tx) {
-                        error!(target: "engine::tree", %err, "Termination failed");
-                    }
-                    return Ok(ops::ControlFlow::Break(()))
-                }
            },
            FromEngine::Request(request) => {
                match request {
@@ -1376,7 +1359,7 @@ where
                        let block_num_hash = block.recovered_block().num_hash();
                        if block_num_hash.number <= self.state.tree_state.canonical_block_number() {
                            // outdated block that can be skipped
-                            return Ok(ops::ControlFlow::Continue(()))
+                            return Ok(())
                        }

                        debug!(target: "engine::tree", block=?block_num_hash, "inserting already executed block");
@@ -1484,7 +1467,7 @@ where
                }
            }
        }
-        Ok(ops::ControlFlow::Continue(()))
+        Ok(())
    }

    /// Invoked if the backfill sync has finished to target.
@@ -1718,10 +1701,10 @@ where
    }

    /// Returns a batch of consecutive canonical blocks to persist in the range
-    /// `(last_persisted_number .. target]`. The expected order is oldest -> newest.
+    /// `(last_persisted_number .. canonical_head - threshold]`. The expected
+    /// order is oldest -> newest.
    fn get_canonical_blocks_to_persist(
        &self,
-        target: PersistTarget,
    ) -> Result<Vec<ExecutedBlock<N>>, AdvancePersistenceError> {
        // We will calculate the state root using the database, so we need to be sure there are no
        // changes
@@ -1732,12 +1715,9 @@ where
        let last_persisted_number = self.persistence_state.last_persisted_block.number;
        let canonical_head_number = self.state.tree_state.canonical_block_number();

-        let target_number = match target {
-            PersistTarget::Head => canonical_head_number,
-            PersistTarget::Threshold => {
-                canonical_head_number.saturating_sub(self.config.memory_block_buffer_target())
-            }
-        };
+        // Persist only up to block buffer target
+        let target_number =
+            canonical_head_number.saturating_sub(self.config.memory_block_buffer_target());

        debug!(
            target: "engine::tree",
@@ -2527,11 +2507,14 @@ where
            Ok(Some(_)) => {}
        }

-        // determine whether we are on a fork chain by comparing the block number with the
-        // canonical head. This is a simple check that is sufficient for the event emission below.
-        // A block is considered a fork if its number is less than or equal to the canonical head,
-        // as this indicates there's already a canonical block at that height.
-        let is_fork = block_id.block.number <= self.state.tree_state.current_canonical_head.number;
+        // determine whether we are on a fork chain
+        let is_fork = match self.is_fork(block_id) {
+            Err(err) => {
+                let block = convert_to_block(self, input)?;
+                return Err(InsertBlockError::new(block, err.into()).into());
+            }
+            Ok(is_fork) => is_fork,
+        };

        let ctx = TreeCtx::new(&mut self.state, &self.canonical_in_memory_state);

@@ -2877,12 +2860,3 @@ pub enum InsertPayloadOk {
    /// The payload was valid and inserted into the tree.
    Inserted(BlockStatus),
 }
-
-/// Target for block persistence.
-#[derive(Debug, Clone, Copy)]
-enum PersistTarget {
-    /// Persist up to `canonical_head - memory_block_buffer_target`.
-    Threshold,
-    /// Persist all blocks up to and including the canonical head.
-    Head,
-}
--- a/crates/engine/tree/src/tree/payload_processor/bal.rs
+++ b/crates/engine/tree/src/tree/payload_processor/bal.rs
@@ -1,124 +1,17 @@
 //! BAL (Block Access List, EIP-7928) related functionality.

-use crate::tree::cached_state::CachedStateProvider;
 use alloy_consensus::constants::KECCAK_EMPTY;
 use alloy_eip7928::BlockAccessList;
-use alloy_primitives::{keccak256, Address, StorageKey, U256};
+use alloy_primitives::{keccak256, U256};
 use reth_primitives_traits::Account;
 use reth_provider::{AccountReader, ProviderError};
 use reth_trie::{HashedPostState, HashedStorage};
-use std::ops::Range;
-
-/// Returns the total number of storage slots (both changed and read-only) across all accounts in
-/// the BAL.
-pub fn total_slots(bal: &BlockAccessList) -> usize {
-    bal.iter().map(|account| account.storage_changes.len() + account.storage_reads.len()).sum()
-}
-
-/// Iterator over storage slots in a [`BlockAccessList`], with range-based filtering.
-///
-/// Iterates over all `(Address, StorageKey)` pairs representing both changed and read-only
-/// storage slots across all accounts in the BAL. For each account, changed slots are iterated
-/// first, followed by read-only slots. The iterator intelligently skips accounts and slots
-/// outside the specified range for efficient traversal.
-#[derive(Debug)]
-pub(crate) struct BALSlotIter<'a> {
-    bal: &'a BlockAccessList,
-    range: Range<usize>,
-    current_index: usize,
-    account_idx: usize,
-    /// Index within the current account's combined slots (changed + read-only).
-    /// If `slot_idx < storage_changes.len()`, we're in changed slots.
-    /// Otherwise, we're in read-only slots at index `slot_idx - storage_changes.len()`.
-    slot_idx: usize,
-}
-
-impl<'a> BALSlotIter<'a> {
-    /// Creates a new iterator over storage slots within the specified range.
-    pub(crate) fn new(bal: &'a BlockAccessList, range: Range<usize>) -> Self {
-        let mut iter = Self { bal, range, current_index: 0, account_idx: 0, slot_idx: 0 };
-        iter.skip_to_range_start();
-        iter
-    }
-
-    /// Skips to the first item within the range.
-    fn skip_to_range_start(&mut self) {
-        while self.account_idx < self.bal.len() {
-            let account = &self.bal[self.account_idx];
-            let slots_in_account = account.storage_changes.len() + account.storage_reads.len();
-
-            // Check if this account contains items in our range
-            let account_end = self.current_index + slots_in_account;
-
-            if account_end <= self.range.start {
-                // Entire account is before range, skip it
-                self.current_index = account_end;
-                self.account_idx += 1;
-                self.slot_idx = 0;
-            } else if self.current_index < self.range.start {
-                // Range starts somewhere in this account
-                let skip_slots = self.range.start - self.current_index;
-                self.slot_idx = skip_slots;
-                self.current_index = self.range.start;
-                break;
-            } else {
-                // We're at or past range start
-                break;
-            }
-        }
-    }
-}
-
-impl<'a> Iterator for BALSlotIter<'a> {
-    type Item = (Address, StorageKey);
-
-    fn next(&mut self) -> Option<Self::Item> {
-        // Check if we've exceeded the range
-        if self.current_index >= self.range.end {
-            return None;
-        }
-
-        // Find the next valid slot
-        while self.account_idx < self.bal.len() {
-            let account = &self.bal[self.account_idx];
-            let changed_len = account.storage_changes.len();
-            let total_len = changed_len + account.storage_reads.len();
-
-            if self.slot_idx < total_len {
-                let address = account.address;
-                let slot = if self.slot_idx < changed_len {
-                    // We're in changed slots
-                    account.storage_changes[self.slot_idx].slot
-                } else {
-                    // We're in read-only slots
-                    account.storage_reads[self.slot_idx - changed_len]
-                };
-
-                self.slot_idx += 1;
-                self.current_index += 1;
-
-                // Check if we've reached the end of range
-                if self.current_index > self.range.end {
-                    return None;
-                }
-
-                return Some((address, slot));
-            }
-
-            // Move to next account
-            self.account_idx += 1;
-            self.slot_idx = 0;
-        }
-
-        None
-    }
-}

 /// Converts a Block Access List into a [`HashedPostState`] by extracting the final state
 /// of modified accounts and storage slots.
-pub(crate) fn bal_to_hashed_post_state<P>(
+pub fn bal_to_hashed_post_state<P>(
    bal: &BlockAccessList,
-    provider: &CachedStateProvider<P>,
+    provider: &P,
 ) -> Result<HashedPostState, ProviderError>
 where
    P: AccountReader,
@@ -127,10 +20,7 @@ where

    for account_changes in bal {
        let address = account_changes.address;
-
-        // Always fetch the account; even if we don't need the db account to construct the final
-        // `Account`, doing this fills the cache.
-        let existing_account = provider.basic_account(&address)?;
+        let hashed_address = keccak256(address);

        // Get the latest balance (last balance change if any)
        let balance = account_changes.balance_changes.last().map(|change| change.post_balance);
@@ -149,14 +39,12 @@ where
            None
        };

-        // If the account was only read then don't add it to the HashedPostState
-        if balance.is_none() &&
-            nonce.is_none() &&
-            code_hash.is_none() &&
-            account_changes.storage_changes.is_empty()
-        {
-            continue
-        }
+        // Only fetch account from provider if we're missing any field
+        let existing_account = if balance.is_none() || nonce.is_none() || code_hash.is_none() {
+            provider.basic_account(&address)?
+        } else {
+            None
+        };

        // Build the final account state
        let account = Account {
@@ -170,7 +58,6 @@ where
            }),
        };

-        let hashed_address = keccak256(address);
        hashed_state.accounts.insert(hashed_address, Some(account));

        // Process storage changes
@@ -188,7 +75,9 @@ where
                }
            }

-            hashed_state.storages.insert(hashed_address, storage_map);
+            if !storage_map.storage.is_empty() {
+                hashed_state.storages.insert(hashed_address, storage_map);
+            }
        }
    }

@@ -198,17 +87,12 @@ where
 #[cfg(test)]
 mod tests {
    use super::*;
-    use crate::tree::cached_state::{ExecutionCache, ExecutionCacheBuilder};
    use alloy_eip7928::{
        AccountChanges, BalanceChange, CodeChange, NonceChange, SlotChanges, StorageChange,
    };
    use alloy_primitives::{Address, Bytes, StorageKey, B256};
    use reth_revm::test_utils::StateProviderTest;

-    fn new_cache() -> ExecutionCache {
-        ExecutionCacheBuilder::default().build_caches(1000)
-    }
-
    #[test]
    fn test_bal_to_hashed_post_state_basic() {
        let provider = StateProviderTest::default();
@@ -224,7 +108,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        assert_eq!(result.accounts.len(), 1);
@@ -259,7 +142,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        let hashed_address = keccak256(address);
@@ -289,7 +171,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        let hashed_address = keccak256(address);
@@ -317,7 +198,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        let hashed_address = keccak256(address);
@@ -352,7 +232,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        let hashed_address = keccak256(address);
@@ -385,7 +264,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        let hashed_address = keccak256(address);
@@ -426,7 +304,6 @@ mod tests {
        };

        let bal = vec![account_changes];
-        let provider = CachedStateProvider::new(provider, new_cache(), Default::default());
        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();

        let hashed_address = keccak256(address);
@@ -438,117 +315,4 @@ mod tests {
        // Should have the last value
        assert_eq!(*stored_value, U256::from(300));
    }
-
-    #[test]
-    fn test_bal_slot_iter() {
-        // Create test data with multiple accounts and slots (both changed and read-only)
-        let addr1 = Address::repeat_byte(0x01);
-        let addr2 = Address::repeat_byte(0x02);
-        let addr3 = Address::repeat_byte(0x03);
-
-        // Account 1: 2 changed slots + 1 read-only = 3 total slots (indices 0, 1, 2)
-        let account1 = AccountChanges {
-            address: addr1,
-            storage_changes: vec![
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(100)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(101)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-            ],
-            storage_reads: vec![StorageKey::from(U256::from(102))],
-            balance_changes: vec![],
-            nonce_changes: vec![],
-            code_changes: vec![],
-        };
-
-        // Account 2: 1 changed slot + 1 read-only = 2 total slots (indices 3, 4)
-        let account2 = AccountChanges {
-            address: addr2,
-            storage_changes: vec![SlotChanges {
-                slot: StorageKey::from(U256::from(200)),
-                changes: vec![StorageChange::new(0, B256::ZERO)],
-            }],
-            storage_reads: vec![StorageKey::from(U256::from(201))],
-            balance_changes: vec![],
-            nonce_changes: vec![],
-            code_changes: vec![],
-        };
-
-        // Account 3: 2 changed slots + 1 read-only = 3 total slots (indices 5, 6, 7)
-        let account3 = AccountChanges {
-            address: addr3,
-            storage_changes: vec![
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(300)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(301)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-            ],
-            storage_reads: vec![StorageKey::from(U256::from(302))],
-            balance_changes: vec![],
-            nonce_changes: vec![],
-            code_changes: vec![],
-        };
-
-        let bal = vec![account1, account2, account3];
-
-        // Test 1: Iterate over all slots (range 0..8)
-        let items: Vec<_> = BALSlotIter::new(&bal, 0..8).collect();
-        assert_eq!(items.len(), 8);
-        // Account 1: changed slots first (100, 101), then read-only (102)
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(100))));
-        assert_eq!(items[1], (addr1, StorageKey::from(U256::from(101))));
-        assert_eq!(items[2], (addr1, StorageKey::from(U256::from(102))));
-        // Account 2: changed slot (200), then read-only (201)
-        assert_eq!(items[3], (addr2, StorageKey::from(U256::from(200))));
-        assert_eq!(items[4], (addr2, StorageKey::from(U256::from(201))));
-        // Account 3: changed slots (300, 301), then read-only (302)
-        assert_eq!(items[5], (addr3, StorageKey::from(U256::from(300))));
-        assert_eq!(items[6], (addr3, StorageKey::from(U256::from(301))));
-        assert_eq!(items[7], (addr3, StorageKey::from(U256::from(302))));
-
-        // Test 2: Range that skips first account (range 3..6)
-        let items: Vec<_> = BALSlotIter::new(&bal, 3..6).collect();
-        assert_eq!(items.len(), 3);
-        assert_eq!(items[0], (addr2, StorageKey::from(U256::from(200))));
-        assert_eq!(items[1], (addr2, StorageKey::from(U256::from(201))));
-        assert_eq!(items[2], (addr3, StorageKey::from(U256::from(300))));
-
-        // Test 3: Range within first account (range 1..2)
-        let items: Vec<_> = BALSlotIter::new(&bal, 1..2).collect();
-        assert_eq!(items.len(), 1);
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(101))));
-
-        // Test 4: Range spanning multiple accounts (range 2..5)
-        let items: Vec<_> = BALSlotIter::new(&bal, 2..5).collect();
-        assert_eq!(items.len(), 3);
-        // Last slot from account 1 (read-only)
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(102))));
-        // Account 2 (changed + read-only)
-        assert_eq!(items[1], (addr2, StorageKey::from(U256::from(200))));
-        assert_eq!(items[2], (addr2, StorageKey::from(U256::from(201))));
-
-        // Test 5: Empty range
-        let items: Vec<_> = BALSlotIter::new(&bal, 5..5).collect();
-        assert_eq!(items.len(), 0);
-
-        // Test 6: Range beyond end (starts at index 6)
-        let items: Vec<_> = BALSlotIter::new(&bal, 6..100).collect();
-        assert_eq!(items.len(), 2);
-        assert_eq!(items[0], (addr3, StorageKey::from(U256::from(301))));
-        assert_eq!(items[1], (addr3, StorageKey::from(U256::from(302))));
-
-        // Test 7: Range that starts in read-only slots (index 2 is the read-only slot of account 1)
-        let items: Vec<_> = BALSlotIter::new(&bal, 2..4).collect();
-        assert_eq!(items.len(), 2);
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(102))));
-        assert_eq!(items[1], (addr2, StorageKey::from(U256::from(200))));
-    }
 }
--- a/crates/engine/tree/src/tree/payload_processor/mod.rs
+++ b/crates/engine/tree/src/tree/payload_processor/mod.rs
@@ -3,11 +3,11 @@
 use super::precompile_cache::PrecompileCacheMap;
 use crate::tree::{
    cached_state::{
-        CachedStateMetrics, CachedStateProvider, ExecutionCache as StateExecutionCache,
-        ExecutionCacheBuilder, SavedCache,
+        CachedStateMetrics, ExecutionCache as StateExecutionCache, ExecutionCacheBuilder,
+        SavedCache,
    },
    payload_processor::{
-        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, PrewarmTaskEvent},
+        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmTaskEvent},
        sparse_trie::StateRootComputeOutcome,
    },
    sparse_trie::SparseTrieTask,
@@ -23,12 +23,11 @@ use multiproof::{SparseTrieUpdate, *};
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
 use rayon::prelude::*;
+use reth_engine_primitives::ExecutableTxIterator;
 use reth_evm::{
    execute::{ExecutableTxFor, WithTxEnv},
-    ConfigureEvm, EvmEnvFor, ExecutableTxIterator, ExecutableTxTuple, OnStateHook, SpecFor,
-    TxEnvFor,
+    ConfigureEvm, EvmEnvFor, OnStateHook, SpecFor, TxEnvFor,
 };
-use reth_execution_types::ExecutionOutcome;
 use reth_primitives_traits::NodePrimitives;
 use reth_provider::{BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader};
 use reth_revm::{db::BundleState, state::EvmState};
@@ -51,7 +50,7 @@ use std::{
    },
    time::Instant,
 };
-use tracing::{debug, debug_span, instrument, warn, Span};
+use tracing::{debug, debug_span, error, instrument, warn, Span};

 pub mod bal;
 mod configured_sparse_trie;
@@ -93,13 +92,6 @@ pub const SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY: usize = 1_000_000;
 /// 144MB.
 pub const SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY: usize = 1_000_000;

-/// Type alias for [`PayloadHandle`] returned by payload processor spawn methods.
-type IteratorPayloadHandle<Evm, I, N> = PayloadHandle<
-    WithTxEnv<TxEnvFor<Evm>, <I as ExecutableTxTuple>::Tx>,
-    <I as ExecutableTxTuple>::Error,
-    <N as NodePrimitives>::Receipt,
->;
-
 /// Entrypoint for executing the payload.
 #[derive(Debug)]
 pub struct PayloadProcessor<Evm>
@@ -208,6 +200,7 @@ where
    ///
    /// This returns a handle to await the final state root and to interact with the tasks (e.g.
    /// canceling)
+    #[allow(clippy::type_complexity)]
    #[instrument(
        level = "debug",
        target = "engine::tree::payload_processor",
@@ -222,7 +215,7 @@ where
        multiproof_provider_factory: F,
        config: &TreeConfig,
        bal: Option<Arc<BlockAccessList>>,
-    ) -> IteratorPayloadHandle<Evm, I, N>
+    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
        F: DatabaseProviderROFactory<Provider: TrieCursorFactory + HashedCursorFactory>
@@ -230,44 +223,17 @@ where
            + Send
            + 'static,
    {
-        let parent_hash = env.parent_hash;
-
        // start preparing transactions immediately
        let (prewarm_rx, execution_rx, transaction_count_hint) =
            self.spawn_tx_iterator(transactions);

        let span = Span::current();
        let (to_sparse_trie, sparse_trie_rx) = channel();
-        let (to_multi_proof, from_multi_proof) = crossbeam_channel::unbounded();

-        // Handle BAL-based optimization if available
-        let prewarm_handle = if let Some(bal) = bal {
-            // When BAL is present, use BAL prewarming and send BAL to multiproof
-            debug!(target: "engine::tree::payload_processor", "BAL present, using BAL prewarming");
-
-            // Send BAL message immediately to MultiProofTask
-            let _ = to_multi_proof.send(MultiProofMessage::BlockAccessList(Arc::clone(&bal)));
-
-            // Spawn with BAL prewarming
-            self.spawn_caching_with(
-                env,
-                prewarm_rx,
-                transaction_count_hint,
-                provider_builder.clone(),
-                None, // Don't send proof targets when BAL is present
-                Some(bal),
-            )
-        } else {
-            // Normal path: spawn with transaction prewarming
-            self.spawn_caching_with(
-                env,
-                prewarm_rx,
-                transaction_count_hint,
-                provider_builder.clone(),
-                Some(to_multi_proof.clone()),
-                None,
-            )
-        };
+        // We rely on the cursor factory to provide whatever DB overlay is necessary to see a
+        // consistent view of the database, including the trie tables. Because of this there is no
+        // need for an overarching prefix set to invalidate any section of the trie tables, and so
+        // we use an empty prefix set.

        // Create and spawn the storage proof task
        let task_ctx = ProofTaskCtx::new(multiproof_provider_factory);
@@ -284,26 +250,50 @@ where
            proof_handle.clone(),
            to_sparse_trie,
            config.multiproof_chunking_enabled().then_some(config.multiproof_chunk_size()),
-            to_multi_proof,
-            from_multi_proof,
        );

        // wire the multiproof task to the prewarm task
        let to_multi_proof = Some(multi_proof_task.state_root_message_sender());

+        // Handle BAL-based optimization if available
+        let prewarm_handle = if let Some(bal) = bal {
+            // When BAL is present, skip spawning prewarm tasks entirely and send BAL to multiproof
+            debug!(target: "engine::tree::payload_processor", "BAL present, skipping prewarm tasks");
+
+            // Send BAL message immediately to MultiProofTask
+            if let Some(ref sender) = to_multi_proof &&
+                let Err(err) = sender.send(MultiProofMessage::BlockAccessList(bal))
+            {
+                // In this case state root validation will simply fail
+                error!(target: "engine::tree::payload_processor", ?err, "Failed to send BAL to MultiProofTask");
+            }
+
+            // Spawn minimal cache-only task without prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                transaction_count_hint,
+                provider_builder.clone(),
+                None, // Don't send proof targets when BAL is present
+            )
+        } else {
+            // Normal path: spawn with full prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                transaction_count_hint,
+                provider_builder.clone(),
+                to_multi_proof.clone(),
+            )
+        };
+
        // spawn multi-proof task
        let parent_span = span.clone();
-        self.executor.spawn_blocking({
-            let saved_cache = self.cache_for(parent_hash);
-            let cache = saved_cache.cache().clone();
-            let cache_metrics = saved_cache.metrics().clone();
-            move || {
-                let _enter = parent_span.entered();
-                // Build a state provider for the multiproof task
-                let provider = provider_builder.build().expect("failed to build provider");
-                let provider = CachedStateProvider::new(provider, cache, cache_metrics);
-                multi_proof_task.run(provider);
-            }
+        self.executor.spawn_blocking(move || {
+            let _enter = parent_span.entered();
+            // Build a state provider for the multiproof task
+            let provider = provider_builder.build().expect("failed to build provider");
+            multi_proof_task.run(provider);
        });

        // wire the sparse trie to the state root response receiver
@@ -330,14 +320,13 @@ where
        env: ExecutionEnv<Evm>,
        transactions: I,
        provider_builder: StateProviderBuilder<N, P>,
-        bal: Option<Arc<BlockAccessList>>,
-    ) -> IteratorPayloadHandle<Evm, I, N>
+    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
    {
        let (prewarm_rx, execution_rx, size_hint) = self.spawn_tx_iterator(transactions);
        let prewarm_handle =
-            self.spawn_caching_with(env, prewarm_rx, size_hint, provider_builder, None, bal);
+            self.spawn_caching_with(env, prewarm_rx, size_hint, provider_builder, None);
        PayloadHandle {
            to_multi_proof: None,
            prewarm_handle,
@@ -411,8 +400,7 @@ where
        transaction_count_hint: usize,
        provider_builder: StateProviderBuilder<N, P>,
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-        bal: Option<Arc<BlockAccessList>>,
-    ) -> CacheTaskHandle<N::Receipt>
+    ) -> CacheTaskHandle
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
    {
@@ -456,12 +444,7 @@ where
        {
            let to_prewarm_task = to_prewarm_task.clone();
            self.executor.spawn_blocking(move || {
-                let mode = if let Some(bal) = bal {
-                    PrewarmMode::BlockAccessList(bal)
-                } else {
-                    PrewarmMode::Transactions(transactions)
-                };
-                prewarm_task.run(mode, to_prewarm_task);
+                prewarm_task.run(transactions, to_prewarm_task);
            });
        }

@@ -598,15 +581,12 @@ where
 }

 /// Handle to all the spawned tasks.
-///
-/// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the
-/// caching task without cloning the expensive `BundleState`.
 #[derive(Debug)]
-pub struct PayloadHandle<Tx, Err, R> {
+pub struct PayloadHandle<Tx, Err> {
    /// Channel for evm state updates
    to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
    // must include the receiver of the state root wired to the sparse trie
-    prewarm_handle: CacheTaskHandle<R>,
+    prewarm_handle: CacheTaskHandle,
    /// Stream of block transactions
    transactions: mpsc::Receiver<Result<Tx, Err>>,
    /// Receiver for the state root
@@ -615,7 +595,7 @@ pub struct PayloadHandle<Tx, Err, R> {
    _span: Span,
 }

-impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
+impl<Tx, Err> PayloadHandle<Tx, Err> {
    /// Awaits the state root
    ///
    /// # Panics
@@ -668,14 +648,9 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {

    /// Terminates the entire caching task.
    ///
-    /// If the [`ExecutionOutcome`] is provided it will update the shared cache using its
-    /// bundle state. Using `Arc<ExecutionOutcome>` allows sharing with the main execution
-    /// path without cloning the expensive `BundleState`.
-    pub(super) fn terminate_caching(
-        &mut self,
-        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
-    ) {
-        self.prewarm_handle.terminate_caching(execution_outcome)
+    /// If the [`BundleState`] is provided it will update the shared cache.
+    pub(super) fn terminate_caching(&mut self, block_output: Option<&BundleState>) {
+        self.prewarm_handle.terminate_caching(block_output)
    }

    /// Returns iterator yielding transactions from the stream.
@@ -687,20 +662,17 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
 }

 /// Access to the spawned [`PrewarmCacheTask`].
-///
-/// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the
-/// prewarm task without cloning the expensive `BundleState`.
 #[derive(Debug)]
-pub(crate) struct CacheTaskHandle<R> {
+pub(crate) struct CacheTaskHandle {
    /// The shared cache the task operates with.
    cache: Option<StateExecutionCache>,
    /// Metrics for the caches
    cache_metrics: Option<CachedStateMetrics>,
    /// Channel to the spawned prewarm task if any
-    to_prewarm_task: Option<std::sync::mpsc::Sender<PrewarmTaskEvent<R>>>,
+    to_prewarm_task: Option<std::sync::mpsc::Sender<PrewarmTaskEvent>>,
 }

-impl<R: Send + Sync + 'static> CacheTaskHandle<R> {
+impl CacheTaskHandle {
    /// Terminates the pre-warming transaction processing.
    ///
    /// Note: This does not terminate the task yet.
@@ -712,25 +684,20 @@ impl<R: Send + Sync + 'static> CacheTaskHandle<R> {

    /// Terminates the entire pre-warming task.
    ///
-    /// If the [`ExecutionOutcome`] is provided it will update the shared cache using its
-    /// bundle state. Using `Arc<ExecutionOutcome>` avoids cloning the expensive `BundleState`.
-    pub(super) fn terminate_caching(
-        &mut self,
-        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
-    ) {
+    /// If the [`BundleState`] is provided it will update the shared cache.
+    pub(super) fn terminate_caching(&mut self, block_output: Option<&BundleState>) {
        if let Some(tx) = self.to_prewarm_task.take() {
-            let event = PrewarmTaskEvent::Terminate { execution_outcome };
+            // Only clone when we have an active task and a state to send
+            let event = PrewarmTaskEvent::Terminate { block_output: block_output.cloned() };
            let _ = tx.send(event);
        }
    }
 }

-impl<R> Drop for CacheTaskHandle<R> {
+impl Drop for CacheTaskHandle {
    fn drop(&mut self) {
-        // Ensure we always terminate on drop - send None without needing Send + Sync bounds
-        if let Some(tx) = self.to_prewarm_task.take() {
-            let _ = tx.send(PrewarmTaskEvent::Terminate { execution_outcome: None });
-        }
+        // Ensure we always terminate on drop
+        self.terminate_caching(None);
    }
 }

@@ -783,8 +750,6 @@ impl ExecutionCache {

        cache
            .as_ref()
-            // Check `is_available()` to ensure no other tasks (e.g., prewarming) currently hold
-            // a reference to this cache. We can only reuse it when we have exclusive access.
            .filter(|c| c.executed_block_hash() == parent_hash && c.is_available())
            .cloned()
    }
--- a/crates/engine/tree/src/tree/payload_processor/multiproof.rs
+++ b/crates/engine/tree/src/tree/payload_processor/multiproof.rs
@@ -1,16 +1,17 @@
 //! Multiproof task related functionality.

-use crate::tree::{
-    cached_state::CachedStateProvider, payload_processor::bal::bal_to_hashed_post_state,
-};
+use crate::tree::payload_processor::bal::bal_to_hashed_post_state;
 use alloy_eip7928::BlockAccessList;
 use alloy_evm::block::StateChangeSource;
-use alloy_primitives::{keccak256, map::HashSet, B256};
+use alloy_primitives::{
+    keccak256,
+    map::{B256Set, HashSet},
+    B256,
+};
 use crossbeam_channel::{unbounded, Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
 use dashmap::DashMap;
 use derive_more::derive::Deref;
 use metrics::{Gauge, Histogram};
-use rayon::prelude::*;
 use reth_metrics::Metrics;
 use reth_provider::AccountReader;
 use reth_revm::state::EvmState;
@@ -22,6 +23,7 @@ use reth_trie_parallel::{
    proof::ParallelProof,
    proof_task::{
        AccountMultiproofInput, ProofResultContext, ProofResultMessage, ProofWorkerHandle,
+        StorageProofInput,
    },
 };
 use std::{collections::BTreeMap, mem, ops::DerefMut, sync::Arc, time::Instant};
@@ -168,6 +170,11 @@ impl ProofSequencer {
        while let Some(pending) = self.pending_proofs.remove(&current_sequence) {
            consecutive_proofs.push(pending);
            current_sequence += 1;
+
+            // if we don't have the next number, stop collecting
+            if !self.pending_proofs.contains_key(&current_sequence) {
+                break;
+            }
        }

        self.next_to_deliver += consecutive_proofs.len() as u64;
@@ -203,38 +210,103 @@ impl Drop for StateHookSender {
 }

 pub(crate) fn evm_state_to_hashed_post_state(update: EvmState) -> HashedPostState {
-    update.into_par_iter()
-        .filter_map(|(address, account)| {
-            if !account.is_touched() {
-                return None;
-            }
+    let mut hashed_state = HashedPostState::with_capacity(update.len());

+    for (address, account) in update {
+        if account.is_touched() {
            let hashed_address = keccak256(address);
            trace!(target: "engine::tree::payload_processor::multiproof", ?address, ?hashed_address, "Adding account to state update");

            let destroyed = account.is_selfdestructed();
            let info = if destroyed { None } else { Some(account.info.into()) };
+            hashed_state.accounts.insert(hashed_address, info);

-            let hashed_storage = if destroyed {
-                Some(HashedStorage::new(true))
-            } else {
-                let storage: Vec<_> = account
-                    .storage
-                    .into_iter()
-                    .filter(|(_slot, value)| value.is_changed())
-                    .map(|(slot, value)| (keccak256(B256::from(slot)), value.present_value))
-                    .collect();
+            let mut changed_storage_iter = account
+                .storage
+                .into_iter()
+                .filter(|(_slot, value)| value.is_changed())
+                .map(|(slot, value)| (keccak256(B256::from(slot)), value.present_value))
+                .peekable();

-                if storage.is_empty() {
-                    None
-                } else {
-                    Some(HashedStorage::from_iter(false, storage))
-                }
-            };
+            if destroyed {
+                hashed_state.storages.insert(hashed_address, HashedStorage::new(true));
+            } else if changed_storage_iter.peek().is_some() {
+                hashed_state
+                    .storages
+                    .insert(hashed_address, HashedStorage::from_iter(false, changed_storage_iter));
+            }
+        }
+    }

-            Some((hashed_address, info, hashed_storage))
-        })
-        .collect()
+    hashed_state
+}
+
+/// A pending multiproof task, either [`StorageMultiproofInput`] or [`MultiproofInput`].
+#[derive(Debug)]
+enum PendingMultiproofTask {
+    /// A storage multiproof task input.
+    Storage(StorageMultiproofInput),
+    /// A regular multiproof task input.
+    Regular(MultiproofInput),
+}
+
+impl PendingMultiproofTask {
+    /// Returns the proof sequence number of the task.
+    const fn proof_sequence_number(&self) -> u64 {
+        match self {
+            Self::Storage(input) => input.proof_sequence_number,
+            Self::Regular(input) => input.proof_sequence_number,
+        }
+    }
+
+    /// Returns whether or not the proof targets are empty.
+    fn proof_targets_is_empty(&self) -> bool {
+        match self {
+            Self::Storage(input) => input.proof_targets.is_empty(),
+            Self::Regular(input) => input.proof_targets.is_empty(),
+        }
+    }
+
+    /// Destroys the input and sends a [`MultiProofMessage::EmptyProof`] message to the sender.
+    fn send_empty_proof(self) {
+        match self {
+            Self::Storage(input) => input.send_empty_proof(),
+            Self::Regular(input) => input.send_empty_proof(),
+        }
+    }
+}
+
+impl From<StorageMultiproofInput> for PendingMultiproofTask {
+    fn from(input: StorageMultiproofInput) -> Self {
+        Self::Storage(input)
+    }
+}
+
+impl From<MultiproofInput> for PendingMultiproofTask {
+    fn from(input: MultiproofInput) -> Self {
+        Self::Regular(input)
+    }
+}
+
+/// Input parameters for dispatching a dedicated storage multiproof calculation.
+#[derive(Debug)]
+struct StorageMultiproofInput {
+    hashed_state_update: HashedPostState,
+    hashed_address: B256,
+    proof_targets: B256Set,
+    proof_sequence_number: u64,
+    state_root_message_sender: CrossbeamSender<MultiProofMessage>,
+    multi_added_removed_keys: Arc<MultiAddedRemovedKeys>,
+}
+
+impl StorageMultiproofInput {
+    /// Destroys the input and sends a [`MultiProofMessage::EmptyProof`] message to the sender.
+    fn send_empty_proof(self) {
+        let _ = self.state_root_message_sender.send(MultiProofMessage::EmptyProof {
+            sequence_number: self.proof_sequence_number,
+            state: self.hashed_state_update,
+        });
+    }
 }

 /// Input parameters for dispatching a multiproof calculation.
@@ -311,18 +383,91 @@ impl MultiproofManager {
    }

    /// Dispatches a new multiproof calculation to worker pools.
-    fn dispatch(&self, input: MultiproofInput) {
+    fn dispatch(&self, input: PendingMultiproofTask) {
        // If there are no proof targets, we can just send an empty multiproof back immediately
-        if input.proof_targets.is_empty() {
+        if input.proof_targets_is_empty() {
            trace!(
-                sequence_number = input.proof_sequence_number,
+                sequence_number = input.proof_sequence_number(),
                "No proof targets, sending empty multiproof back immediately"
            );
            input.send_empty_proof();
            return;
        }

-        self.dispatch_multiproof(input);
+        match input {
+            PendingMultiproofTask::Storage(storage_input) => {
+                self.dispatch_storage_proof(storage_input);
+            }
+            PendingMultiproofTask::Regular(multiproof_input) => {
+                self.dispatch_multiproof(multiproof_input);
+            }
+        }
+    }
+
+    /// Dispatches a single storage proof calculation to worker pool.
+    fn dispatch_storage_proof(&self, storage_multiproof_input: StorageMultiproofInput) {
+        let StorageMultiproofInput {
+            hashed_state_update,
+            hashed_address,
+            proof_targets,
+            proof_sequence_number,
+            multi_added_removed_keys,
+            state_root_message_sender: _,
+        } = storage_multiproof_input;
+
+        let storage_targets = proof_targets.len();
+
+        trace!(
+            target: "engine::tree::payload_processor::multiproof",
+            proof_sequence_number,
+            ?proof_targets,
+            storage_targets,
+            "Dispatching storage proof to workers"
+        );
+
+        let start = Instant::now();
+
+        // Create prefix set from targets
+        let prefix_set = reth_trie::prefix_set::PrefixSetMut::from(
+            proof_targets.iter().map(reth_trie::Nibbles::unpack),
+        );
+        let prefix_set = prefix_set.freeze();
+
+        // Build computation input (data only)
+        let input = StorageProofInput::new(
+            hashed_address,
+            prefix_set,
+            proof_targets,
+            true, // with_branch_node_masks
+            Some(multi_added_removed_keys),
+        );
+
+        // Dispatch to storage worker
+        if let Err(e) = self.proof_worker_handle.dispatch_storage_proof(
+            input,
+            ProofResultContext::new(
+                self.proof_result_tx.clone(),
+                proof_sequence_number,
+                hashed_state_update,
+                start,
+            ),
+        ) {
+            error!(target: "engine::tree::payload_processor::multiproof", ?e, "Failed to dispatch storage proof");
+            return;
+        }
+
+        self.metrics
+            .active_storage_workers_histogram
+            .record(self.proof_worker_handle.active_storage_workers() as f64);
+        self.metrics
+            .active_account_workers_histogram
+            .record(self.proof_worker_handle.active_account_workers() as f64);
+        self.metrics
+            .pending_storage_multiproofs_histogram
+            .record(self.proof_worker_handle.pending_storage_tasks() as f64);
+        self.metrics
+            .pending_account_multiproofs_histogram
+            .record(self.proof_worker_handle.pending_account_tasks() as f64);
    }

    /// Signals that a multiproof calculation has finished.
@@ -600,9 +745,8 @@ impl MultiProofTask {
        proof_worker_handle: ProofWorkerHandle,
        to_sparse_trie: std::sync::mpsc::Sender<SparseTrieUpdate>,
        chunk_size: Option<usize>,
-        tx: CrossbeamSender<MultiProofMessage>,
-        rx: CrossbeamReceiver<MultiProofMessage>,
    ) -> Self {
+        let (tx, rx) = unbounded();
        let (proof_result_tx, proof_result_rx) = unbounded();
        let metrics = MultiProofTaskMetrics::default();

@@ -670,14 +814,17 @@ impl MultiProofTask {
            available_storage_workers,
            MultiProofTargets::chunks,
            |proof_targets| {
-                self.multiproof_manager.dispatch(MultiproofInput {
-                    source: None,
-                    hashed_state_update: Default::default(),
-                    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()),
-                });
+                self.multiproof_manager.dispatch(
+                    MultiproofInput {
+                        source: None,
+                        hashed_state_update: Default::default(),
+                        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()),
+                    }
+                    .into(),
+                );
            },
        );
        self.metrics.prefetch_proof_chunks_histogram.record(num_chunks as f64);
@@ -825,14 +972,17 @@ impl MultiProofTask {
                );
                spawned_proof_targets.extend_ref(&proof_targets);

-                self.multiproof_manager.dispatch(MultiproofInput {
-                    source: Some(source),
-                    hashed_state_update,
-                    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()),
-                });
+                self.multiproof_manager.dispatch(
+                    MultiproofInput {
+                        source: Some(source),
+                        hashed_state_update,
+                        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()),
+                    }
+                    .into(),
+                );
            },
        );
        self.metrics
@@ -879,7 +1029,7 @@ impl MultiProofTask {
        msg: MultiProofMessage,
        ctx: &mut MultiproofBatchCtx,
        batch_metrics: &mut MultiproofBatchMetrics,
-        provider: &CachedStateProvider<P>,
+        provider: &P,
    ) -> bool
    where
        P: AccountReader,
@@ -1055,7 +1205,7 @@ impl MultiProofTask {
                }

                // Convert BAL to HashedPostState and process it
-                match bal_to_hashed_post_state(&bal, provider) {
+                match bal_to_hashed_post_state(&bal, &provider) {
                    Ok(hashed_state) => {
                        debug!(
                            target: "engine::tree::payload_processor::multiproof",
@@ -1184,7 +1334,7 @@ impl MultiProofTask {
        target = "engine::tree::payload_processor::multiproof",
        skip_all
    )]
-    pub(crate) fn run<P>(mut self, provider: CachedStateProvider<P>)
+    pub(crate) fn run<P>(mut self, provider: P)
    where
        P: AccountReader,
    {
@@ -1308,9 +1458,6 @@ impl MultiProofTask {
 /// Context for multiproof message batching loop.
 ///
 /// Contains processing state that persists across loop iterations.
-///
-/// Used by `process_multiproof_message` to batch consecutive same-type messages received via
-/// `try_recv` for efficient processing.
 struct MultiproofBatchCtx {
    /// Buffers a non-matching message type encountered during batching.
    /// Processed first in next iteration to preserve ordering while allowing same-type
@@ -1501,13 +1648,12 @@ fn estimate_evm_state_targets(state: &EvmState) -> usize {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use crate::tree::cached_state::ExecutionCacheBuilder;
    use alloy_eip7928::{AccountChanges, BalanceChange};
    use alloy_primitives::{map::B256Set, Address};
    use reth_provider::{
        providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
-        BlockReader, DatabaseProviderFactory, LatestStateProvider, PruneCheckpointReader,
-        StageCheckpointReader, StateProviderBox, TrieReader,
+        BlockReader, DatabaseProviderFactory, PruneCheckpointReader, StageCheckpointReader,
+        TrieReader,
    };
    use reth_trie::MultiProof;
    use reth_trie_parallel::proof_task::{ProofTaskCtx, ProofWorkerHandle};
@@ -1539,23 +1685,8 @@ mod tests {
        let task_ctx = ProofTaskCtx::new(overlay_factory);
        let proof_handle = ProofWorkerHandle::new(rt_handle, task_ctx, 1, 1);
        let (to_sparse_trie, _receiver) = std::sync::mpsc::channel();
-        let (tx, rx) = crossbeam_channel::unbounded();

-        MultiProofTask::new(proof_handle, to_sparse_trie, Some(1), tx, rx)
-    }
-
-    fn create_cached_provider<F>(factory: F) -> CachedStateProvider<StateProviderBox>
-    where
-        F: DatabaseProviderFactory<
-                Provider: BlockReader + TrieReader + StageCheckpointReader + PruneCheckpointReader,
-            > + Clone
-            + Send
-            + 'static,
-    {
-        let db_provider = factory.database_provider_ro().unwrap();
-        let state_provider: StateProviderBox = Box::new(LatestStateProvider::new(db_provider));
-        let cache = ExecutionCacheBuilder::default().build_caches(1000);
-        CachedStateProvider::new(state_provider, cache, Default::default())
+        MultiProofTask::new(proof_handle, to_sparse_trie, Some(1))
    }

    #[test]
@@ -2487,7 +2618,7 @@ mod tests {
        use revm_state::Account;

        let test_provider_factory = create_test_provider_factory();
-        let test_provider = create_cached_provider(test_provider_factory.clone());
+        let test_provider = test_provider_factory.latest().unwrap();
        let mut task = create_test_state_root_task(test_provider_factory);

        // Queue: Prefetch1, StateUpdate, Prefetch2
@@ -2707,7 +2838,7 @@ mod tests {
    #[test]
    fn test_bal_message_processing() {
        let test_provider_factory = create_test_provider_factory();
-        let test_provider = create_cached_provider(test_provider_factory.clone());
+        let test_provider = test_provider_factory.latest().unwrap();
        let mut task = create_test_state_root_task(test_provider_factory);

        // Create a simple BAL with one account change
--- a/crates/engine/tree/src/tree/payload_processor/prewarm.rs
+++ b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
@@ -14,30 +14,25 @@
 use crate::tree::{
    cached_state::{CachedStateProvider, SavedCache},
    payload_processor::{
-        bal::{total_slots, BALSlotIter},
-        executor::WorkloadExecutor,
-        multiproof::MultiProofMessage,
+        executor::WorkloadExecutor, multiproof::MultiProofMessage,
        ExecutionCache as PayloadExecutionCache,
    },
    precompile_cache::{CachedPrecompile, PrecompileCacheMap},
    ExecutionEnv, StateProviderBuilder,
 };
 use alloy_consensus::transaction::TxHashRef;
-use alloy_eip7928::BlockAccessList;
 use alloy_eips::Typed2718;
 use alloy_evm::Database;
 use alloy_primitives::{keccak256, map::B256Set, B256};
 use crossbeam_channel::Sender as CrossbeamSender;
 use metrics::{Counter, Gauge, Histogram};
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
-use reth_execution_types::ExecutionOutcome;
 use reth_metrics::Metrics;
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{AccountReader, BlockReader, StateProvider, StateProviderFactory, StateReader};
-use reth_revm::{database::StateProviderDatabase, state::EvmState};
+use reth_provider::{BlockReader, StateProviderBox, StateProviderFactory, StateReader};
+use reth_revm::{database::StateProviderDatabase, db::BundleState, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
-    ops::Range,
    sync::{
        atomic::{AtomicBool, Ordering},
        mpsc::{self, channel, Receiver, Sender},
@@ -47,14 +42,6 @@ use std::{
 };
 use tracing::{debug, debug_span, instrument, trace, warn, Span};

-/// Determines the prewarming mode: transaction-based or BAL-based.
-pub(super) enum PrewarmMode<Tx> {
-    /// Prewarm by executing transactions from a stream.
-    Transactions(Receiver<Tx>),
-    /// Prewarm by prefetching slots from a Block Access List.
-    BlockAccessList(Arc<BlockAccessList>),
-}
-
 /// A wrapper for transactions that includes their index in the block.
 #[derive(Clone)]
 struct IndexedTransaction<Tx> {
@@ -99,7 +86,7 @@ where
    /// Sender to emit evm state outcome messages, if any.
    to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
    /// Receiver for events produced by tx execution
-    actions_rx: Receiver<PrewarmTaskEvent<N::Receipt>>,
+    actions_rx: Receiver<PrewarmTaskEvent>,
    /// Parent span for tracing
    parent_span: Span,
 }
@@ -118,7 +105,7 @@ where
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
        transaction_count_hint: usize,
        max_concurrency: usize,
-    ) -> (Self, Sender<PrewarmTaskEvent<N::Receipt>>) {
+    ) -> (Self, Sender<PrewarmTaskEvent>) {
        let (actions_tx, actions_rx) = channel();

        trace!(
@@ -148,11 +135,8 @@ where
    /// For Optimism chains, special handling is applied to the first transaction if it's a
    /// deposit transaction (type 0x7E/126) which sets critical metadata that affects all
    /// subsequent transactions in the block.
-    fn spawn_all<Tx>(
-        &self,
-        pending: mpsc::Receiver<Tx>,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-    ) where
+    fn spawn_all<Tx>(&self, pending: mpsc::Receiver<Tx>, actions_tx: Sender<PrewarmTaskEvent>)
+    where
        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
    {
        let executor = self.executor.clone();
@@ -176,7 +160,12 @@ where
            };

            // Initialize worker handles container
-            let handles = ctx.clone().spawn_workers(workers_needed, &executor, actions_tx.clone(), done_tx.clone());
+            let mut handles = Vec::with_capacity(workers_needed);
+
+            // Only spawn initial workers as needed
+            for i in 0..workers_needed {
+                handles.push(ctx.spawn_worker(i, &executor, actions_tx.clone(), done_tx.clone()));
+            }

            // Distribute transactions to workers
            let mut tx_index = 0usize;
@@ -259,7 +248,7 @@ where
    ///
    /// This method is called from `run()` only after all execution tasks are complete.
    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn save_cache(self, execution_outcome: Arc<ExecutionOutcome<N::Receipt>>) {
+    fn save_cache(self, state: BundleState) {
        let start = Instant::now();

        let Self { execution_cache, ctx: PrewarmContext { env, metrics, saved_cache, .. }, .. } =
@@ -276,8 +265,7 @@ where
                let new_cache = SavedCache::new(hash, caches, cache_metrics);

                // Insert state into cache while holding the lock
-                // Access the BundleState through the shared ExecutionOutcome
-                if new_cache.cache().insert_state(execution_outcome.state()).is_err() {
+                if new_cache.cache().insert_state(&state).is_err() {
                    // Clear the cache on error to prevent having a polluted cache
                    *cached = None;
                    debug!(target: "engine::caching", "cleared execution cache on update error");
@@ -298,86 +286,6 @@ where
        }
    }

-    /// Runs BAL-based prewarming by spawning workers to prefetch storage slots.
-    ///
-    /// Divides the total slots across `max_concurrency` workers, each responsible for
-    /// prefetching a range of slots from the BAL.
-    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn run_bal_prewarm(
-        &self,
-        bal: Arc<BlockAccessList>,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-    ) {
-        // Only prefetch if we have a cache to populate
-        if self.ctx.saved_cache.is_none() {
-            trace!(
-                target: "engine::tree::payload_processor::prewarm",
-                "Skipping BAL prewarm - no cache available"
-            );
-            let _ =
-                actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
-            return;
-        }
-
-        let total_slots = total_slots(&bal);
-
-        trace!(
-            target: "engine::tree::payload_processor::prewarm",
-            total_slots,
-            max_concurrency = self.max_concurrency,
-            "Starting BAL prewarm"
-        );
-
-        if total_slots == 0 {
-            // No slots to prefetch, signal completion immediately
-            let _ =
-                actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
-            return;
-        }
-
-        let (done_tx, done_rx) = mpsc::channel();
-
-        // Calculate number of workers needed (at most max_concurrency)
-        let workers_needed = total_slots.min(self.max_concurrency);
-
-        // Calculate slots per worker
-        let slots_per_worker = total_slots / workers_needed;
-        let remainder = total_slots % workers_needed;
-
-        // Spawn workers with their assigned ranges
-        for i in 0..workers_needed {
-            let start = i * slots_per_worker + i.min(remainder);
-            let extra = if i < remainder { 1 } else { 0 };
-            let end = start + slots_per_worker + extra;
-
-            self.ctx.spawn_bal_worker(
-                i,
-                &self.executor,
-                Arc::clone(&bal),
-                start..end,
-                done_tx.clone(),
-            );
-        }
-
-        // Drop our handle to done_tx so we can detect completion
-        drop(done_tx);
-
-        // Wait for all workers to complete
-        let mut completed_workers = 0;
-        while done_rx.recv().is_ok() {
-            completed_workers += 1;
-        }
-
-        trace!(
-            target: "engine::tree::payload_processor::prewarm",
-            completed_workers,
-            "All BAL prewarm workers completed"
-        );
-
-        // Signal that execution has finished
-        let _ = actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
-    }
-
    /// Executes the task.
    ///
    /// This will execute the transactions until all transactions have been processed or the task
@@ -389,24 +297,15 @@ where
        name = "prewarm and caching",
        skip_all
    )]
-    pub(super) fn run<Tx>(
+    pub(super) fn run(
        self,
-        mode: PrewarmMode<Tx>,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-    ) where
-        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
-    {
-        // Spawn execution tasks based on mode
-        match mode {
-            PrewarmMode::Transactions(pending) => {
-                self.spawn_all(pending, actions_tx);
-            }
-            PrewarmMode::BlockAccessList(bal) => {
-                self.run_bal_prewarm(bal, actions_tx);
-            }
-        }
+        pending: mpsc::Receiver<impl ExecutableTxFor<Evm> + Clone + Send + 'static>,
+        actions_tx: Sender<PrewarmTaskEvent>,
+    ) {
+        // spawn execution tasks.
+        self.spawn_all(pending, actions_tx);

-        let mut final_execution_outcome = None;
+        let mut final_block_output = None;
        let mut finished_execution = false;
        while let Ok(event) = self.actions_rx.recv() {
            match event {
@@ -419,9 +318,9 @@ where
                    // completed executing a set of transactions
                    self.send_multi_proof_targets(proof_targets);
                }
-                PrewarmTaskEvent::Terminate { execution_outcome } => {
+                PrewarmTaskEvent::Terminate { block_output } => {
                    trace!(target: "engine::tree::payload_processor::prewarm", "Received termination signal");
-                    final_execution_outcome = Some(execution_outcome);
+                    final_block_output = Some(block_output);

                    if finished_execution {
                        // all tasks are done, we can exit, which will save caches and exit
@@ -435,7 +334,7 @@ where

                    finished_execution = true;

-                    if final_execution_outcome.is_some() {
+                    if final_block_output.is_some() {
                        // all tasks are done, we can exit, which will save caches and exit
                        break
                    }
@@ -445,9 +344,9 @@ where

        debug!(target: "engine::tree::payload_processor::prewarm", "Completed prewarm execution");

-        // save caches and finish using the shared ExecutionOutcome
-        if let Some(Some(execution_outcome)) = final_execution_outcome {
-            self.save_cache(execution_outcome);
+        // save caches and finish
+        if let Some(Some(state)) = final_block_output {
+            self.save_cache(state);
        }
    }
 }
@@ -489,10 +388,10 @@ where
            metrics,
            terminate_execution,
            precompile_cache_disabled,
-            precompile_cache_map,
+            mut precompile_cache_map,
        } = self;

-        let mut state_provider = match provider.build() {
+        let state_provider = match provider.build() {
            Ok(provider) => provider,
            Err(err) => {
                trace!(
@@ -505,15 +404,13 @@ where
        };

        // Use the caches to create a new provider with caching
-        if let Some(saved_cache) = saved_cache {
+        let state_provider: StateProviderBox = if let Some(saved_cache) = saved_cache {
            let caches = saved_cache.cache().clone();
            let cache_metrics = saved_cache.metrics().clone();
-            state_provider = Box::new(
-                CachedStateProvider::new(state_provider, caches, cache_metrics)
-                    // ensure we pre-warm the cache
-                    .prewarm(),
-            );
-        }
+            Box::new(CachedStateProvider::new_with_caches(state_provider, caches, cache_metrics))
+        } else {
+            state_provider
+        };

        let state_provider = StateProviderDatabase::new(state_provider);

@@ -555,7 +452,7 @@ where
    fn transact_batch<Tx>(
        self,
        txs: mpsc::Receiver<IndexedTransaction<Tx>>,
-        sender: Sender<PrewarmTaskEvent<N::Receipt>>,
+        sender: Sender<PrewarmTaskEvent>,
        done_tx: Sender<()>,
    ) where
        Tx: ExecutableTxFor<Evm>,
@@ -632,134 +529,27 @@ where
    }

    /// Spawns a worker task for transaction execution and returns its sender channel.
-    fn spawn_workers<Tx>(
-        self,
-        workers_needed: usize,
-        task_executor: &WorkloadExecutor,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-        done_tx: Sender<()>,
-    ) -> Vec<mpsc::Sender<IndexedTransaction<Tx>>>
-    where
-        Tx: ExecutableTxFor<Evm> + Send + 'static,
-    {
-        let mut handles = Vec::with_capacity(workers_needed);
-        let mut receivers = Vec::with_capacity(workers_needed);
-
-        for _ in 0..workers_needed {
-            let (tx, rx) = mpsc::channel();
-            handles.push(tx);
-            receivers.push(rx);
-        }
-
-        // Spawn a separate task spawning workers in parallel.
-        let executor = task_executor.clone();
-        let span = Span::current();
-        task_executor.spawn_blocking(move || {
-            let _enter = span.entered();
-            for (idx, rx) in receivers.into_iter().enumerate() {
-                let ctx = self.clone();
-                let actions_tx = actions_tx.clone();
-                let done_tx = done_tx.clone();
-                let span = debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);
-                executor.spawn_blocking(move || {
-                    let _enter = span.entered();
-                    ctx.transact_batch(rx, actions_tx, done_tx);
-                });
-            }
-        });
-
-        handles
-    }
-
-    /// Spawns a worker task for BAL slot prefetching.
-    ///
-    /// The worker iterates over the specified range of slots in the BAL and ensures
-    /// each slot is loaded into the cache by accessing it through the state provider.
-    fn spawn_bal_worker(
+    fn spawn_worker<Tx>(
        &self,
        idx: usize,
        executor: &WorkloadExecutor,
-        bal: Arc<BlockAccessList>,
-        range: Range<usize>,
+        actions_tx: Sender<PrewarmTaskEvent>,
        done_tx: Sender<()>,
-    ) {
+    ) -> mpsc::Sender<IndexedTransaction<Tx>>
+    where
+        Tx: ExecutableTxFor<Evm> + Send + 'static,
+    {
+        let (tx, rx) = mpsc::channel();
        let ctx = self.clone();
-        let span = debug_span!(
-            target: "engine::tree::payload_processor::prewarm",
-            "bal prewarm worker",
-            idx,
-            range_start = range.start,
-            range_end = range.end
-        );
+        let span =
+            debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);

        executor.spawn_blocking(move || {
            let _enter = span.entered();
-            ctx.prefetch_bal_slots(bal, range, done_tx);
+            ctx.transact_batch(rx, actions_tx, done_tx);
        });
-    }

-    /// Prefetches storage slots from a BAL range into the cache.
-    ///
-    /// This iterates through the specified range of slots and accesses them via the state
-    /// provider to populate the cache.
-    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn prefetch_bal_slots(
-        self,
-        bal: Arc<BlockAccessList>,
-        range: Range<usize>,
-        done_tx: Sender<()>,
-    ) {
-        let Self { saved_cache, provider, metrics, .. } = self;
-
-        // Build state provider
-        let state_provider = match provider.build() {
-            Ok(provider) => provider,
-            Err(err) => {
-                trace!(
-                    target: "engine::tree::payload_processor::prewarm",
-                    %err,
-                    "Failed to build state provider in BAL prewarm thread"
-                );
-                let _ = done_tx.send(());
-                return;
-            }
-        };
-
-        // Wrap with cache (guaranteed to be Some since run_bal_prewarm checks)
-        let saved_cache = saved_cache.expect("BAL prewarm should only run with cache");
-        let caches = saved_cache.cache().clone();
-        let cache_metrics = saved_cache.metrics().clone();
-        let state_provider = CachedStateProvider::new(state_provider, caches, cache_metrics);
-
-        let start = Instant::now();
-
-        // Track last seen address to avoid fetching the same account multiple times.
-        let mut last_address = None;
-
-        // Iterate through the assigned range of slots
-        for (address, slot) in BALSlotIter::new(&bal, range.clone()) {
-            // Fetch the account if this is a different address than the last one
-            if last_address != Some(address) {
-                let _ = state_provider.basic_account(&address);
-                last_address = Some(address);
-            }
-
-            // Access the slot to populate the cache
-            let _ = state_provider.storage(address, slot);
-        }
-
-        let elapsed = start.elapsed();
-
-        trace!(
-            target: "engine::tree::payload_processor::prewarm",
-            ?range,
-            elapsed_ms = elapsed.as_millis(),
-            "BAL prewarm worker completed"
-        );
-
-        // Signal completion
-        let _ = done_tx.send(());
-        metrics.bal_slot_iteration_duration.record(elapsed.as_secs_f64());
+        tx
    }
 }

@@ -799,18 +589,14 @@ fn multiproof_targets_from_state(state: EvmState) -> (MultiProofTargets, usize)
 }

 /// The events the pre-warm task can handle.
-///
-/// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the main
-/// execution path without cloning the expensive `BundleState`.
-pub(super) enum PrewarmTaskEvent<R> {
+pub(super) enum PrewarmTaskEvent {
    /// Forcefully terminate all remaining transaction execution.
    TerminateTransactionExecution,
    /// Forcefully terminate the task on demand and update the shared cache with the given output
    /// before exiting.
    Terminate {
-        /// The final execution outcome. Using `Arc` allows sharing with the main execution
-        /// path without cloning the expensive `BundleState`.
-        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
+        /// The final block state output.
+        block_output: Option<BundleState>,
    },
    /// The outcome of a pre-warm task
    Outcome {
@@ -842,6 +628,4 @@ pub(crate) struct PrewarmMetrics {
    pub(crate) cache_saving_duration: Gauge,
    /// Counter for transaction execution errors during prewarming
    pub(crate) transaction_errors: Counter,
-    /// A histogram of BAL slot iteration duration during prefetching
-    pub(crate) bal_slot_iteration_duration: Histogram,
 }
--- a/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
+++ b/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
@@ -166,7 +166,8 @@ where

    // Update storage slots with new values and calculate storage roots.
    let span = tracing::Span::current();
-    let results: Vec<_> = state
+    let (tx, rx) = mpsc::channel();
+    state
        .storages
        .into_iter()
        .map(|(address, storage)| (address, storage, trie.take_storage_trie(&address)))
@@ -216,7 +217,13 @@ where

            SparseStateTrieResult::Ok((address, storage_trie))
        })
-        .collect();
+        .for_each_init(
+            || tx.clone(),
+            |tx, result| {
+                let _ = tx.send(result);
+            },
+        );
+    drop(tx);

    // Defer leaf removals until after updates/additions, so that we don't delete an intermediate
    // branch node during a removal and then re-add that branch back during a later leaf addition.
@@ -228,7 +235,7 @@ where
    let _enter =
        tracing::debug_span!(target: "engine::tree::payload_processor::sparse_trie", "account trie")
            .entered();
-    for result in results {
+    for result in rx {
        let (address, storage_trie) = result?;
        trie.insert_storage_trie(address, storage_trie);

--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/crates/engine/tree/src/tree/payload_validator.rs
@@ -35,13 +35,12 @@ use reth_primitives_traits::{
 };
 use reth_provider::{
    providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockReader,
-    DatabaseProviderFactory, DatabaseProviderROFactory, ExecutionOutcome, HashedPostStateProvider,
-    ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
-    StateProviderFactory, StateReader, TrieReader,
+    DatabaseProviderFactory, ExecutionOutcome, HashedPostStateProvider, ProviderError,
+    PruneCheckpointReader, StageCheckpointReader, StateProvider, StateProviderFactory, StateReader,
+    StateRootProvider, TrieReader,
 };
 use reth_revm::db::State;
-use reth_storage_errors::db::DatabaseError;
-use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot, TrieInputSorted};
+use reth_trie::{updates::TrieUpdates, HashedPostState, TrieInputSorted};
 use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
 use revm_primitives::Address;
 use std::{
@@ -375,8 +374,7 @@ where
        let mut state_provider = ensure_ok!(provider_builder.build());
        drop(_enter);

-        // Fetch parent block. This goes to memory most of the time unless the parent block is
-        // beyond the in-memory buffer.
+        // fetch parent block
        let Some(parent_block) = ensure_ok!(self.sealed_header_by_hash(parent_hash, ctx.state()))
        else {
            return Err(InsertBlockError::new(
@@ -401,7 +399,7 @@ where
            "Decided which state root algorithm to run"
        );

-        // Get an iterator over the transactions in the payload
+        // use prewarming background task
        let txs = self.tx_iterator_for(&input)?;

        // Extract the BAL, if valid and available
@@ -426,16 +424,21 @@ where
        // Use cached state provider before executing, used in execution after prewarming threads
        // complete
        if let Some((caches, cache_metrics)) = handle.caches().zip(handle.cache_metrics()) {
-            state_provider =
-                Box::new(CachedStateProvider::new(state_provider, caches, cache_metrics));
+            state_provider = Box::new(CachedStateProvider::new_with_caches(
+                state_provider,
+                caches,
+                cache_metrics,
+            ));
        };

-        if self.config.state_provider_metrics() {
-            state_provider = Box::new(InstrumentedStateProvider::new(state_provider, "engine"));
-        }
-
        // Execute the block and handle any execution errors
-        let (output, senders) = match self.execute_block(state_provider, env, &input, &mut handle) {
+        let (output, senders) = match if self.config.state_provider_metrics() {
+            let state_provider =
+                InstrumentedStateProvider::from_state_provider(&state_provider, "engine");
+            self.execute_block(&state_provider, env, &input, &mut handle)
+        } else {
+            self.execute_block(&state_provider, env, &input, &mut handle)
+        } {
            Ok(output) => output,
            Err(err) => return self.handle_execution_error(input, err, &parent_block),
        };
@@ -519,7 +522,7 @@ where
            }

            let (root, updates) = ensure_ok_post_block!(
-                self.compute_state_root_serial(block.parent_hash(), &hashed_state, ctx.state()),
+                state_provider.state_root_with_updates(hashed_state.clone()),
                block
            );
            (root, updates, root_time.elapsed())
@@ -549,14 +552,17 @@ where
            .into())
        }

-        // Create ExecutionOutcome and wrap in Arc for sharing with both the caching task
-        // and the deferred trie task. This avoids cloning the expensive BundleState.
-        let execution_outcome = Arc::new(ExecutionOutcome::from((output, block_num_hash.number)));
+        // terminate prewarming task with good state output
+        handle.terminate_caching(Some(&output.state));

-        // Terminate prewarming task with the shared execution outcome
-        handle.terminate_caching(Some(Arc::clone(&execution_outcome)));
-
-        Ok(self.spawn_deferred_trie_task(block, execution_outcome, &ctx, hashed_state, trie_output))
+        Ok(self.spawn_deferred_trie_task(
+            block,
+            output,
+            block_num_hash.number,
+            &ctx,
+            hashed_state,
+            trie_output,
+        ))
    }

    /// Return sealed block header from database or in-memory state by hash.
@@ -599,10 +605,10 @@ where
        state_provider: S,
        env: ExecutionEnv<Evm>,
        input: &BlockOrPayload<T>,
-        handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
+        handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err>,
    ) -> Result<(BlockExecutionOutput<N::Receipt>, Vec<Address>), InsertBlockErrorKind>
    where
-        S: StateProvider + Send,
+        S: StateProvider,
        Err: core::error::Error + Send + Sync + 'static,
        V: PayloadValidator<T, Block = N::Block>,
        T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
@@ -611,7 +617,7 @@ where
        debug!(target: "engine::tree::payload_validator", "Executing block");

        let mut db = State::builder()
-            .with_database(StateProviderDatabase::new(state_provider))
+            .with_database(StateProviderDatabase::new(&state_provider))
            .with_bundle_update()
            .without_state_clear()
            .build();
@@ -643,7 +649,6 @@ where
        let (output, senders) = self.metrics.execute_metered(
            executor,
            handle.iter_transactions().map(|res| res.map_err(BlockExecutionError::other)),
-            input.transaction_count(),
            state_hook,
        )?;
        let execution_finish = Instant::now();
@@ -658,6 +663,8 @@ where
    ///
    /// Returns `Ok(_)` if computed successfully.
    /// Returns `Err(_)` if error was encountered during computation.
+    /// `Err(ProviderError::ConsistentView(_))` can be safely ignored and fallback computation
+    /// should be used instead.
    #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
    fn compute_state_root_parallel(
        &self,
@@ -687,36 +694,6 @@ where
        ParallelStateRoot::new(factory, prefix_sets).incremental_root_with_updates()
    }

-    /// Compute state root for the given hashed post state in serial.
-    fn compute_state_root_serial(
-        &self,
-        parent_hash: B256,
-        hashed_state: &HashedPostState,
-        state: &EngineApiTreeState<N>,
-    ) -> ProviderResult<(B256, TrieUpdates)> {
-        let (mut input, block_hash) = self.compute_trie_input(parent_hash, state)?;
-
-        // Extend state overlay with current block's sorted state.
-        input.prefix_sets.extend(hashed_state.construct_prefix_sets());
-        let sorted_hashed_state = hashed_state.clone_into_sorted();
-        Arc::make_mut(&mut input.state).extend_ref(&sorted_hashed_state);
-
-        let TrieInputSorted { nodes, state, .. } = input;
-        let prefix_sets = hashed_state.construct_prefix_sets();
-
-        let factory = OverlayStateProviderFactory::new(self.provider.clone())
-            .with_block_hash(Some(block_hash))
-            .with_trie_overlay(Some(nodes))
-            .with_hashed_state_overlay(Some(state));
-
-        let provider = factory.database_provider_ro()?;
-
-        Ok(StateRoot::new(&provider, &provider)
-            .with_prefix_sets(prefix_sets.freeze())
-            .root_with_updates()
-            .map_err(Into::<DatabaseError>::into)?)
-    }
-
    /// Validates the block after execution.
    ///
    /// This performs:
@@ -816,7 +793,6 @@ where
        PayloadHandle<
            impl ExecutableTxFor<Evm> + use<N, P, Evm, V, T>,
            impl core::error::Error + Send + Sync + 'static + use<N, P, Evm, V, T>,
-            N::Receipt,
        >,
        InsertBlockErrorKind,
    > {
@@ -862,12 +838,8 @@ where
            }
            StateRootStrategy::Parallel | StateRootStrategy::Synchronous => {
                let start = Instant::now();
-                let handle = self.payload_processor.spawn_cache_exclusive(
-                    env,
-                    txs,
-                    provider_builder,
-                    block_access_list,
-                );
+                let handle =
+                    self.payload_processor.spawn_cache_exclusive(env, txs, provider_builder);

                // Record prewarming initialization duration
                self.metrics
@@ -1055,7 +1027,8 @@ where
    fn spawn_deferred_trie_task(
        &self,
        block: RecoveredBlock<N::Block>,
-        execution_outcome: Arc<ExecutionOutcome<N::Receipt>>,
+        output: BlockExecutionOutput<N::Receipt>,
+        block_number: u64,
        ctx: &TreeCtx<'_, N>,
        hashed_state: HashedPostState,
        trie_output: TrieUpdates,
@@ -1105,7 +1078,7 @@ where

        ExecutedBlock::with_deferred_trie_data(
            Arc::new(block),
-            execution_outcome,
+            Arc::new(ExecutionOutcome::from((output, block_number))),
            deferred_trie_data,
        )
    }
@@ -1292,15 +1265,4 @@ impl<T: PayloadTypes> BlockOrPayload<T> {
        // TODO decode and return `BlockAccessList`
        None
    }
-
-    /// Returns the number of transactions in the payload or block.
-    pub fn transaction_count(&self) -> usize
-    where
-        T::ExecutionData: ExecutionPayload,
-    {
-        match self {
-            Self::Payload(payload) => payload.transaction_count(),
-            Self::Block(block) => block.transaction_count(),
-        }
-    }
 }
--- a/crates/engine/tree/src/tree/precompile_cache.rs
+++ b/crates/engine/tree/src/tree/precompile_cache.rs
@@ -1,58 +1,50 @@
 //! Contains a precompile cache backed by `schnellru::LruMap` (LRU by length).

 use alloy_primitives::Bytes;
-use dashmap::DashMap;
-use moka::policy::EvictionPolicy;
+use parking_lot::Mutex;
 use reth_evm::precompiles::{DynPrecompile, Precompile, PrecompileInput};
 use revm::precompile::{PrecompileId, PrecompileOutput, PrecompileResult};
 use revm_primitives::Address;
-use std::{hash::Hash, sync::Arc};
+use schnellru::LruMap;
+use std::{
+    collections::HashMap,
+    hash::{Hash, Hasher},
+    sync::Arc,
+};

 /// Default max cache size for [`PrecompileCache`]
 const MAX_CACHE_SIZE: u32 = 10_000;

 /// Stores caches for each precompile.
 #[derive(Debug, Clone, Default)]
-pub struct PrecompileCacheMap<S>(Arc<DashMap<Address, PrecompileCache<S>>>)
+pub struct PrecompileCacheMap<S>(HashMap<Address, PrecompileCache<S>>)
 where
-    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static;
+    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone;

 impl<S> PrecompileCacheMap<S>
 where
    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
-    pub(crate) fn cache_for_address(&self, address: Address) -> PrecompileCache<S> {
-        // Try just using `.get` first to avoid acquiring a write lock.
-        if let Some(cache) = self.0.get(&address) {
-            return cache.clone();
-        }
-        // Otherwise, fallback to `.entry` and initialize the cache.
-        //
-        // This should be very rare as caches for all precompiles will be initialized as soon as
-        // first EVM is created.
+    pub(crate) fn cache_for_address(&mut self, address: Address) -> PrecompileCache<S> {
        self.0.entry(address).or_default().clone()
    }
 }

 /// Cache for precompiles, for each input stores the result.
+///
+/// [`LruMap`] requires a mutable reference on `get` since it updates the LRU order,
+/// so we use a [`Mutex`] instead of an `RwLock`.
 #[derive(Debug, Clone)]
-pub struct PrecompileCache<S>(
-    moka::sync::Cache<Bytes, CacheEntry<S>, alloy_primitives::map::DefaultHashBuilder>,
-)
+pub struct PrecompileCache<S>(Arc<Mutex<LruMap<CacheKey<S>, CacheEntry>>>)
 where
-    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static;
+    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone;

 impl<S> Default for PrecompileCache<S>
 where
    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
    fn default() -> Self {
-        Self(
-            moka::sync::CacheBuilder::new(MAX_CACHE_SIZE as u64)
-                .initial_capacity(MAX_CACHE_SIZE as usize)
-                .eviction_policy(EvictionPolicy::lru())
-                .build_with_hasher(Default::default()),
-        )
+        Self(Arc::new(Mutex::new(LruMap::new(schnellru::ByLength::new(MAX_CACHE_SIZE)))))
    }
 }

@@ -60,31 +52,63 @@ impl<S> PrecompileCache<S>
 where
    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
-    fn get(&self, input: &[u8], spec: S) -> Option<CacheEntry<S>> {
-        self.0.get(input).filter(|e| e.spec == spec)
+    fn get(&self, key: &CacheKeyRef<'_, S>) -> Option<CacheEntry> {
+        self.0.lock().get(key).cloned()
    }

    /// Inserts the given key and value into the cache, returning the new cache size.
-    fn insert(&self, input: Bytes, value: CacheEntry<S>) -> usize {
-        self.0.insert(input, value);
-        self.0.entry_count() as usize
+    fn insert(&self, key: CacheKey<S>, value: CacheEntry) -> usize {
+        let mut cache = self.0.lock();
+        cache.insert(key, value);
+        cache.len()
+    }
+}
+
+/// Cache key, spec id and precompile call input. spec id is included in the key to account for
+/// precompile repricing across fork activations.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CacheKey<S>((S, Bytes));
+
+impl<S> CacheKey<S> {
+    const fn new(spec_id: S, input: Bytes) -> Self {
+        Self((spec_id, input))
+    }
+}
+
+/// Cache key reference, used to avoid cloning the input bytes when looking up using a [`CacheKey`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CacheKeyRef<'a, S>((S, &'a [u8]));
+
+impl<'a, S> CacheKeyRef<'a, S> {
+    const fn new(spec_id: S, input: &'a [u8]) -> Self {
+        Self((spec_id, input))
+    }
+}
+
+impl<S: PartialEq> PartialEq<CacheKey<S>> for CacheKeyRef<'_, S> {
+    fn eq(&self, other: &CacheKey<S>) -> bool {
+        self.0 .0 == other.0 .0 && self.0 .1 == other.0 .1.as_ref()
+    }
+}
+
+impl<'a, S: Hash> Hash for CacheKeyRef<'a, S> {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.0 .0.hash(state);
+        self.0 .1.hash(state);
    }
 }

 /// Cache entry, precompile successful output.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CacheEntry<S> {
-    output: PrecompileOutput,
-    spec: S,
-}
+pub struct CacheEntry(PrecompileOutput);

-impl<S> CacheEntry<S> {
+impl CacheEntry {
    const fn gas_used(&self) -> u64 {
-        self.output.gas_used
+        self.0.gas_used
    }

    fn to_precompile_result(&self) -> PrecompileResult {
-        Ok(self.output.clone())
+        Ok(self.0.clone())
    }
 }

@@ -166,7 +190,9 @@ where
    }

    fn call(&self, input: PrecompileInput<'_>) -> PrecompileResult {
-        if let Some(entry) = &self.cache.get(input.data, self.spec_id.clone()) {
+        let key = CacheKeyRef::new(self.spec_id.clone(), input.data);
+
+        if let Some(entry) = &self.cache.get(&key) {
            self.increment_by_one_precompile_cache_hits();
            if input.gas >= entry.gas_used() {
                return entry.to_precompile_result()
@@ -178,10 +204,8 @@ where

        match &result {
            Ok(output) => {
-                let size = self.cache.insert(
-                    Bytes::copy_from_slice(calldata),
-                    CacheEntry { output: output.clone(), spec: self.spec_id.clone() },
-                );
+                let key = CacheKey::new(self.spec_id.clone(), Bytes::copy_from_slice(calldata));
+                let size = self.cache.insert(key, CacheEntry(output.clone()));
                self.set_precompile_cache_size_metric(size as f64);
                self.increment_by_one_precompile_cache_misses();
            }
@@ -222,12 +246,31 @@ impl CachedPrecompileMetrics {

 #[cfg(test)]
 mod tests {
+    use std::hash::DefaultHasher;
+
    use super::*;
    use reth_evm::{EthEvmFactory, Evm, EvmEnv, EvmFactory};
    use reth_revm::db::EmptyDB;
    use revm::{context::TxEnv, precompile::PrecompileOutput};
    use revm_primitives::hardfork::SpecId;

+    #[test]
+    fn test_cache_key_ref_hash() {
+        let key1 = CacheKey::new(SpecId::PRAGUE, b"test_input".into());
+        let key2 = CacheKeyRef::new(SpecId::PRAGUE, b"test_input");
+        assert!(PartialEq::eq(&key2, &key1));
+
+        let mut hasher = DefaultHasher::new();
+        key1.hash(&mut hasher);
+        let hash1 = hasher.finish();
+
+        let mut hasher = DefaultHasher::new();
+        key2.hash(&mut hasher);
+        let hash2 = hasher.finish();
+
+        assert_eq!(hash1, hash2);
+    }
+
    #[test]
    fn test_precompile_cache_basic() {
        let dyn_precompile: DynPrecompile = (|_input: PrecompileInput<'_>| -> PrecompileResult {
@@ -250,11 +293,12 @@ mod tests {
            reverted: false,
        };

-        let input = b"test_input";
-        let expected = CacheEntry { output, spec: SpecId::PRAGUE };
-        cache.cache.insert(input.into(), expected.clone());
+        let key = CacheKey::new(SpecId::PRAGUE, b"test_input".into());
+        let expected = CacheEntry(output);
+        cache.cache.insert(key, expected.clone());

-        let actual = cache.cache.get(input, SpecId::PRAGUE).unwrap();
+        let key = CacheKeyRef::new(SpecId::PRAGUE, b"test_input");
+        let actual = cache.cache.get(&key).unwrap();

        assert_eq!(actual, expected);
    }
@@ -268,7 +312,7 @@ mod tests {
        let address1 = Address::repeat_byte(1);
        let address2 = Address::repeat_byte(2);

-        let cache_map = PrecompileCacheMap::default();
+        let mut cache_map = PrecompileCacheMap::default();

        // create the first precompile with a specific output
        let precompile1: DynPrecompile = (PrecompileId::custom("custom"), {
--- a/crates/engine/tree/src/tree/tests.rs
+++ b/crates/engine/tree/src/tree/tests.rs
@@ -4,7 +4,7 @@ use crate::{
    tree::{
        payload_validator::{BasicEngineValidator, TreeCtx, ValidationOutcome},
        persistence_state::CurrentPersistenceAction,
-        PersistTarget, TreeConfig,
+        TreeConfig,
    },
 };

@@ -285,8 +285,7 @@ impl TestHarness {
        let fcu_state = self.fcu_state(block_hash);

        let (tx, rx) = oneshot::channel();
-        let _ = self
-            .tree
+        self.tree
            .on_engine_message(FromEngine::Request(
                BeaconEngineMessage::ForkchoiceUpdated {
                    state: fcu_state,
@@ -499,7 +498,7 @@ fn test_tree_persist_block_batch() {

    // process the message
    let msg = test_harness.tree.try_recv_engine_message().unwrap().unwrap();
-    let _ = test_harness.tree.on_engine_message(msg).unwrap();
+    test_harness.tree.on_engine_message(msg).unwrap();

    // we now should receive the other batch
    let msg = test_harness.tree.try_recv_engine_message().unwrap().unwrap();
@@ -578,7 +577,7 @@ async fn test_engine_request_during_backfill() {
        .with_backfill_state(BackfillSyncState::Active);

    let (tx, rx) = oneshot::channel();
-    let _ = test_harness
+    test_harness
        .tree
        .on_engine_message(FromEngine::Request(
            BeaconEngineMessage::ForkchoiceUpdated {
@@ -659,7 +658,7 @@ async fn test_holesky_payload() {
        TestHarness::new(HOLESKY.clone()).with_backfill_state(BackfillSyncState::Active);

    let (tx, rx) = oneshot::channel();
-    let _ = test_harness
+    test_harness
        .tree
        .on_engine_message(FromEngine::Request(
            BeaconEngineMessage::NewPayload {
@@ -884,8 +883,7 @@ async fn test_get_canonical_blocks_to_persist() {
        .with_persistence_threshold(persistence_threshold)
        .with_memory_block_buffer_target(memory_block_buffer_target);

-    let blocks_to_persist =
-        test_harness.tree.get_canonical_blocks_to_persist(PersistTarget::Threshold).unwrap();
+    let blocks_to_persist = test_harness.tree.get_canonical_blocks_to_persist().unwrap();

    let expected_blocks_to_persist_length: usize =
        (canonical_head_number - memory_block_buffer_target - last_persisted_block_number)
@@ -904,8 +902,7 @@ async fn test_get_canonical_blocks_to_persist() {

    assert!(test_harness.tree.state.tree_state.sealed_header_by_hash(&fork_block_hash).is_some());

-    let blocks_to_persist =
-        test_harness.tree.get_canonical_blocks_to_persist(PersistTarget::Threshold).unwrap();
+    let blocks_to_persist = test_harness.tree.get_canonical_blocks_to_persist().unwrap();
    assert_eq!(blocks_to_persist.len(), expected_blocks_to_persist_length);

    // check that the fork block is not included in the blocks to persist
@@ -984,7 +981,7 @@ async fn test_engine_tree_live_sync_transition_required_blocks_requested() {
    let backfill_tip_block = main_chain[(backfill_finished_block_number - 1) as usize].clone();
    // add block to mock provider to enable persistence clean up.
    test_harness.provider.add_block(backfill_tip_block.hash(), backfill_tip_block.into_block());
-    let _ = test_harness.tree.on_engine_message(FromEngine::Event(backfill_finished)).unwrap();
+    test_harness.tree.on_engine_message(FromEngine::Event(backfill_finished)).unwrap();

    let event = test_harness.from_tree_rx.recv().await.unwrap();
    match event {
@@ -994,7 +991,7 @@ async fn test_engine_tree_live_sync_transition_required_blocks_requested() {
        _ => panic!("Unexpected event: {event:#?}"),
    }

-    let _ = test_harness
+    test_harness
        .tree
        .on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain
            .last()
@@ -1050,7 +1047,7 @@ async fn test_fcu_with_canonical_ancestor_updates_latest_block() {

    // Send FCU to the canonical ancestor
    let (tx, rx) = oneshot::channel();
-    let _ = test_harness
+    test_harness
        .tree
        .on_engine_message(FromEngine::Request(
            BeaconEngineMessage::ForkchoiceUpdated {
@@ -1946,53 +1943,4 @@ mod forkchoice_updated_tests {
            .unwrap();
        assert!(result.is_some(), "OpStack should handle canonical head");
    }
-
-    /// Test that engine termination persists all blocks and signals completion.
-    #[test]
-    fn test_engine_termination_with_everything_persisted() {
-        let chain_spec = MAINNET.clone();
-        let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());
-
-        // Create 10 blocks to persist
-        let blocks: Vec<_> = test_block_builder.get_executed_blocks(1..11).collect();
-        let canonical_tip = blocks.last().unwrap().recovered_block().number;
-        let test_harness = TestHarness::new(chain_spec).with_blocks(blocks);
-
-        // Create termination channel
-        let (terminate_tx, mut terminate_rx) = oneshot::channel();
-
-        let to_tree_tx = test_harness.to_tree_tx.clone();
-        let action_rx = test_harness.action_rx;
-
-        // Spawn tree in background thread
-        std::thread::Builder::new()
-            .name("Engine Task".to_string())
-            .spawn(|| test_harness.tree.run())
-            .unwrap();
-
-        // Send terminate request
-        to_tree_tx
-            .send(FromEngine::Event(FromOrchestrator::Terminate { tx: terminate_tx }))
-            .unwrap();
-
-        // Handle persistence actions until termination completes
-        let mut last_persisted_number = 0;
-        loop {
-            if terminate_rx.try_recv().is_ok() {
-                break;
-            }
-
-            if let Ok(PersistenceAction::SaveBlocks(saved_blocks, sender)) =
-                action_rx.recv_timeout(std::time::Duration::from_millis(100))
-            {
-                if let Some(last) = saved_blocks.last() {
-                    last_persisted_number = last.recovered_block().number;
-                }
-                sender.send(saved_blocks.last().map(|b| b.recovered_block().num_hash())).unwrap();
-            }
-        }
-
-        // Ensure we persisted right to the tip
-        assert_eq!(last_persisted_number, canonical_tip);
-    }
 }
--- a/crates/era-utils/src/history.rs
+++ b/crates/era-utils/src/history.rs
@@ -116,7 +116,7 @@ where
 /// these stages that this work has already been done. Otherwise, there might be some conflict with
 /// database integrity.
 pub fn save_stage_checkpoints<P>(
-    provider: P,
+    provider: &P,
    from: BlockNumber,
    to: BlockNumber,
    processed: u64,
@@ -170,14 +170,18 @@ where
    <P as NodePrimitivesProvider>::Primitives: NodePrimitives<BlockHeader = BH, BlockBody = BB>,
 {
    let reader = open(meta)?;
-    let iter = reader.iter().map(decode as fn(_) -> _);
+    let iter =
+        reader
+            .iter()
+            .map(Box::new(decode)
+                as Box<dyn Fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>);
    let iter = ProcessIter { iter, era: meta };

    process_iter(iter, writer, provider, hash_collector, block_numbers)
 }

 type ProcessInnerIter<R, BH, BB> =
-    Map<BlockTupleIterator<R>, fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>;
+    Map<BlockTupleIterator<R>, Box<dyn Fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>>;

 /// An iterator that wraps era file extraction. After the final item [`EraMeta::mark_as_processed`]
 /// is called to ensure proper cleanup.
@@ -305,7 +309,7 @@ where
        writer.append_header(&header, &hash)?;

        // Write bodies to database.
-        provider.append_block_bodies(vec![(header.number(), Some(&body))])?;
+        provider.append_block_bodies(vec![(header.number(), Some(body))])?;

        hash_collector.insert(hash, number)?;
    }
--- a/crates/era/src/era/types/consensus.rs
+++ b/crates/era/src/era/types/consensus.rs
@@ -59,36 +59,11 @@
 //!     Ok(())
 //! }
 //! ```
+
 use crate::e2s::{error::E2sError, types::Entry};
 use snap::{read::FrameDecoder, write::FrameEncoder};
 use std::io::{Read, Write};

-/// Maximum allowed decompressed size for a signed beacon block SSZ payload.
-const MAX_DECOMPRESSED_SIGNED_BEACON_BLOCK_BYTES: usize = 256 * 1024 * 1024; // 256 MiB
-
-/// Maximum allowed decompressed size for a beacon state SSZ payload.
-const MAX_DECOMPRESSED_BEACON_STATE_BYTES: usize = 2 * 1024 * 1024 * 1024; // 2 GiB
-
-fn decompress_snappy_bounded(
-    compressed: &[u8],
-    max_decompressed_bytes: usize,
-    what: &str,
-) -> Result<Vec<u8>, E2sError> {
-    let mut decoder = FrameDecoder::new(compressed).take(max_decompressed_bytes as u64);
-    let mut decompressed = Vec::new();
-
-    Read::read_to_end(&mut decoder, &mut decompressed)
-        .map_err(|e| E2sError::SnappyDecompression(format!("Failed to decompress {what}: {e}")))?;
-
-    if decompressed.len() >= max_decompressed_bytes {
-        return Err(E2sError::SnappyDecompression(format!(
-            "Failed to decompress {what}: decompressed data exceeded limit of {max_decompressed_bytes} bytes"
-        )));
-    }
-
-    Ok(decompressed)
-}
-
 /// `CompressedSignedBeaconBlock` record type: [0x01, 0x00]
 pub const COMPRESSED_SIGNED_BEACON_BLOCK: [u8; 2] = [0x01, 0x00];

@@ -129,11 +104,13 @@ impl CompressedSignedBeaconBlock {

    /// Decompress to get the original ssz-encoded signed beacon block
    pub fn decompress(&self) -> Result<Vec<u8>, E2sError> {
-        decompress_snappy_bounded(
-            self.data.as_slice(),
-            MAX_DECOMPRESSED_SIGNED_BEACON_BLOCK_BYTES,
-            "signed beacon block",
-        )
+        let mut decoder = FrameDecoder::new(self.data.as_slice());
+        let mut decompressed = Vec::new();
+        Read::read_to_end(&mut decoder, &mut decompressed).map_err(|e| {
+            E2sError::SnappyDecompression(format!("Failed to decompress signed beacon block: {e}"))
+        })?;
+
+        Ok(decompressed)
    }

    /// Convert to an [`Entry`]
@@ -191,11 +168,13 @@ impl CompressedBeaconState {

    /// Decompress to get the original ssz-encoded beacon state
    pub fn decompress(&self) -> Result<Vec<u8>, E2sError> {
-        decompress_snappy_bounded(
-            self.data.as_slice(),
-            MAX_DECOMPRESSED_BEACON_STATE_BYTES,
-            "beacon state",
-        )
+        let mut decoder = FrameDecoder::new(self.data.as_slice());
+        let mut decompressed = Vec::new();
+        Read::read_to_end(&mut decoder, &mut decompressed).map_err(|e| {
+            E2sError::SnappyDecompression(format!("Failed to decompress beacon state: {e}"))
+        })?;
+
+        Ok(decompressed)
    }

    /// Convert to an [`Entry`]
@@ -281,15 +260,4 @@ mod tests {
        let result = CompressedBeaconState::from_entry(&invalid_entry);
        assert!(result.is_err());
    }
-
-    #[test]
-    fn test_bounded_decompression_rejects_oversized_output() {
-        let ssz_data = vec![42u8; 1024];
-        let compressed = CompressedBeaconState::from_ssz(&ssz_data).unwrap();
-
-        let err =
-            decompress_snappy_bounded(compressed.data.as_slice(), 100, "beacon state").unwrap_err();
-
-        assert!(format!("{err:?}").contains("exceeded limit"));
-    }
 }
--- a/crates/ethereum/cli/src/app.rs
+++ b/crates/ethereum/cli/src/app.rs
@@ -1,8 +1,4 @@
-use crate::{
-    interface::{Commands, NoSubCmd},
-    Cli,
-};
-use clap::Subcommand;
+use crate::{interface::Commands, Cli};
 use eyre::{eyre, Result};
 use reth_chainspec::{ChainSpec, EthChainSpec, Hardforks};
 use reth_cli::chainspec::ChainSpecParser;
@@ -22,26 +18,20 @@ use std::{fmt, sync::Arc};

 /// A wrapper around a parsed CLI that handles command execution.
 #[derive(Debug)]
-pub struct CliApp<
-    Spec: ChainSpecParser,
-    Ext: clap::Args + fmt::Debug,
-    Rpc: RpcModuleValidator,
-    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
-> {
-    cli: Cli<Spec, Ext, Rpc, SubCmd>,
+pub struct CliApp<Spec: ChainSpecParser, Ext: clap::Args + fmt::Debug, Rpc: RpcModuleValidator> {
+    cli: Cli<Spec, Ext, Rpc>,
    runner: Option<CliRunner>,
    layers: Option<Layers>,
    guard: Option<FileWorkerGuard>,
 }

-impl<C, Ext, Rpc, SubCmd> CliApp<C, Ext, Rpc, SubCmd>
+impl<C, Ext, Rpc> CliApp<C, Ext, Rpc>
 where
    C: ChainSpecParser,
    Ext: clap::Args + fmt::Debug,
    Rpc: RpcModuleValidator,
-    SubCmd: ExtendedCommand + Subcommand + fmt::Debug,
 {
-    pub(crate) fn new(cli: Cli<C, Ext, Rpc, SubCmd>) -> Self {
+    pub(crate) fn new(cli: Cli<C, Ext, Rpc>) -> Self {
        Self { cli, runner: None, layers: Some(Layers::new()), guard: None }
    }

@@ -108,9 +98,9 @@ where
        self.init_tracing(&runner)?;

        // Install the prometheus recorder to be sure to record all metrics
-        install_prometheus_recorder();
+        let _ = install_prometheus_recorder();

-        run_commands_with::<C, Ext, Rpc, N, SubCmd>(self.cli, runner, components, launcher)
+        run_commands_with::<C, Ext, Rpc, N>(self.cli, runner, components, launcher)
    }

    /// Initializes tracing with the configured options.
@@ -127,8 +117,8 @@ where

 /// Run CLI commands with the provided runner, components and launcher.
 /// This is the shared implementation used by both `CliApp` and Cli methods.
-pub(crate) fn run_commands_with<C, Ext, Rpc, N, SubCmd>(
-    cli: Cli<C, Ext, Rpc, SubCmd>,
+pub(crate) fn run_commands_with<C, Ext, Rpc, N>(
+    cli: Cli<C, Ext, Rpc>,
    runner: CliRunner,
    components: impl CliComponentsBuilder<N>,
    launcher: impl AsyncFnOnce(
@@ -141,7 +131,6 @@ where
    Ext: clap::Args + fmt::Debug,
    Rpc: RpcModuleValidator,
    N: CliNodeTypes<Primitives: NodePrimitives<BlockHeader: HeaderMut>, ChainSpec: Hardforks>,
-    SubCmd: ExtendedCommand + Subcommand + fmt::Debug,
 {
    match cli.command {
        Commands::Node(command) => {
@@ -178,19 +167,9 @@ where
        #[cfg(feature = "dev")]
        Commands::TestVectors(command) => runner.run_until_ctrl_c(command.execute()),
        Commands::ReExecute(command) => runner.run_until_ctrl_c(command.execute::<N>(components)),
-        Commands::Ext(command) => command.execute(runner),
    }
 }

-/// A trait for extension subcommands that can be added to the CLI.
-///
-/// Consumers implement this trait for their custom subcommands to define
-/// how they should be executed.
-pub trait ExtendedCommand {
-    /// Execute the extension command with the provided CLI runner.
-    fn execute(self, runner: CliRunner) -> Result<()>;
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
--- a/crates/ethereum/cli/src/interface.rs
+++ b/crates/ethereum/cli/src/interface.rs
@@ -37,11 +37,10 @@ pub struct Cli<
    C: ChainSpecParser = EthereumChainSpecParser,
    Ext: clap::Args + fmt::Debug = NoArgs,
    Rpc: RpcModuleValidator = DefaultRpcModuleValidator,
-    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
 > {
    /// The command to run
    #[command(subcommand)]
-    pub command: Commands<C, Ext, SubCmd>,
+    pub command: Commands<C, Ext>,

    /// The logging configuration for the CLI.
    #[command(flatten)]
@@ -72,18 +71,15 @@ impl Cli {
    }
 }

-impl<
-        C: ChainSpecParser,
-        Ext: clap::Args + fmt::Debug,
-        Rpc: RpcModuleValidator,
-        SubCmd: crate::app::ExtendedCommand + Subcommand + fmt::Debug,
-    > Cli<C, Ext, Rpc, SubCmd>
-{
+impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, Rpc: RpcModuleValidator> Cli<C, Ext, Rpc> {
    /// Configures the CLI and returns a [`CliApp`] instance.
    ///
    /// This method is used to prepare the CLI for execution by wrapping it in a
    /// [`CliApp`] that can be further configured before running.
-    pub fn configure(self) -> CliApp<C, Ext, Rpc, SubCmd> {
+    pub fn configure(self) -> CliApp<C, Ext, Rpc>
+    where
+        C: ChainSpecParser<ChainSpec = ChainSpec>,
+    {
        CliApp::new(self)
    }

@@ -212,10 +208,10 @@ impl<
        let _guard = self.init_tracing(&runner, Layers::new())?;

        // Install the prometheus recorder to be sure to record all metrics
-        install_prometheus_recorder();
+        let _ = install_prometheus_recorder();

        // Use the shared standalone function to avoid duplication
-        run_commands_with::<C, Ext, Rpc, N, SubCmd>(self, runner, components, launcher)
+        run_commands_with::<C, Ext, Rpc, N>(self, runner, components, launcher)
    }

    /// Initializes tracing with the configured options.
@@ -249,11 +245,7 @@ impl<

 /// Commands to be executed
 #[derive(Debug, Subcommand)]
-pub enum Commands<
-    C: ChainSpecParser,
-    Ext: clap::Args + fmt::Debug,
-    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
-> {
+pub enum Commands<C: ChainSpecParser, Ext: clap::Args + fmt::Debug> {
    /// Start the node
    #[command(name = "node")]
    Node(Box<node::NodeCommand<C, Ext>>),
@@ -299,27 +291,9 @@ pub enum Commands<
    /// Re-execute blocks in parallel to verify historical sync correctness.
    #[command(name = "re-execute")]
    ReExecute(re_execute::Command<C>),
-    /// Extension subcommands provided by consumers.
-    #[command(flatten)]
-    Ext(SubCmd),
 }

-/// A no-op subcommand type for when no extension subcommands are needed.
-///
-/// This is the default type parameter for `Commands` when consumers don't need
-/// to add custom subcommands.
-#[derive(Debug, Subcommand)]
-pub enum NoSubCmd {}
-
-impl crate::app::ExtendedCommand for NoSubCmd {
-    fn execute(self, _runner: CliRunner) -> eyre::Result<()> {
-        match self {}
-    }
-}
-
-impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, SubCmd: Subcommand + fmt::Debug>
-    Commands<C, Ext, SubCmd>
-{
+impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug> Commands<C, Ext> {
    /// Returns the underlying chain being used for commands
    pub fn chain_spec(&self) -> Option<&Arc<C::ChainSpec>> {
        match self {
@@ -339,7 +313,6 @@ impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, SubCmd: Subcommand + fmt:
            Self::Config(_) => None,
            Self::Prune(cmd) => cmd.chain_spec(),
            Self::ReExecute(cmd) => cmd.chain_spec(),
-            Self::Ext(_) => None,
        }
    }
 }
@@ -563,100 +536,4 @@ mod tests {
            _ => panic!("Expected Stage command"),
        };
    }
-
-    #[test]
-    fn test_extensible_subcommands() {
-        use crate::app::ExtendedCommand;
-        use reth_cli_runner::CliRunner;
-        use reth_rpc_server_types::DefaultRpcModuleValidator;
-        use std::sync::atomic::{AtomicBool, Ordering};
-
-        #[derive(Debug, Subcommand)]
-        enum CustomCommands {
-            /// A custom hello command
-            #[command(name = "hello")]
-            Hello {
-                /// Name to greet
-                #[arg(long)]
-                name: String,
-            },
-            /// Another custom command
-            #[command(name = "goodbye")]
-            Goodbye,
-        }
-
-        static EXECUTED: AtomicBool = AtomicBool::new(false);
-
-        impl ExtendedCommand for CustomCommands {
-            fn execute(self, _runner: CliRunner) -> eyre::Result<()> {
-                match self {
-                    Self::Hello { name } => {
-                        assert_eq!(name, "world");
-                        EXECUTED.store(true, Ordering::SeqCst);
-                        Ok(())
-                    }
-                    Self::Goodbye => Ok(()),
-                }
-            }
-        }
-
-        // Test parsing the custom "hello" command
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "hello", "--name", "world"])
-        .unwrap();
-
-        match &cli.command {
-            Commands::Ext(CustomCommands::Hello { name }) => {
-                assert_eq!(name, "world");
-            }
-            _ => panic!("Expected Ext(Hello) command"),
-        }
-
-        // Test parsing the custom "goodbye" command
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "goodbye"])
-        .unwrap();
-
-        match &cli.command {
-            Commands::Ext(CustomCommands::Goodbye) => {}
-            _ => panic!("Expected Ext(Goodbye) command"),
-        }
-
-        // Test that built-in commands still work alongside custom ones
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "node"])
-        .unwrap();
-
-        match &cli.command {
-            Commands::Node(_) => {}
-            _ => panic!("Expected Node command"),
-        }
-
-        // Test executing the custom command
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "hello", "--name", "world"])
-        .unwrap();
-
-        if let Commands::Ext(cmd) = cli.command {
-            let runner = CliRunner::try_default_runtime().unwrap();
-            cmd.execute(runner).unwrap();
-            assert!(EXECUTED.load(Ordering::SeqCst), "Custom command should have been executed");
-        }
-    }
 }
--- a/crates/ethereum/cli/src/lib.rs
+++ b/crates/ethereum/cli/src/lib.rs
@@ -14,8 +14,8 @@ pub mod app;
 pub mod chainspec;
 pub mod interface;

-pub use app::{CliApp, ExtendedCommand};
-pub use interface::{Cli, Commands, NoSubCmd};
+pub use app::CliApp;
+pub use interface::{Cli, Commands};

 #[cfg(test)]
 mod test {
--- a/crates/ethereum/consensus/src/validation.rs
+++ b/crates/ethereum/consensus/src/validation.rs
@@ -87,7 +87,9 @@ fn verify_receipts<R: Receipt>(
        logs_bloom,
        expected_receipts_root,
        expected_logs_bloom,
-    )
+    )?;
+
+    Ok(())
 }

 /// Compare the calculated receipts root with the expected receipts root, also compare
--- a/crates/ethereum/hardforks/src/display.rs
+++ b/crates/ethereum/hardforks/src/display.rs
@@ -170,7 +170,7 @@ impl DisplayHardforks {
        let mut post_merge = Vec::new();

        for (fork, condition, metadata) in hardforks {
-            let display_fork = DisplayFork {
+            let mut display_fork = DisplayFork {
                name: fork.name().to_string(),
                activated_at: condition,
                eip: None,
@@ -181,7 +181,12 @@ impl DisplayHardforks {
                ForkCondition::Block(_) => {
                    pre_merge.push(display_fork);
                }
-                ForkCondition::TTD { .. } => {
+                ForkCondition::TTD { activation_block_number, total_difficulty, fork_block } => {
+                    display_fork.activated_at = ForkCondition::TTD {
+                        activation_block_number,
+                        fork_block,
+                        total_difficulty,
+                    };
                    with_merge.push(display_fork);
                }
                ForkCondition::Timestamp(_) => {
--- a/crates/ethereum/node/Cargo.toml
+++ b/crates/ethereum/node/Cargo.toml
@@ -61,7 +61,6 @@ reth-node-core.workspace = true
 reth-e2e-test-utils.workspace = true
 reth-tasks.workspace = true
 reth-testing-utils.workspace = true
-reth-stages-types.workspace = true
 tempfile.workspace = true
 jsonrpsee-core.workspace = true

@@ -91,7 +90,6 @@ asm-keccak = [
 ]
 keccak-cache-global = [
    "alloy-primitives/keccak-cache-global",
-    "reth-node-core/keccak-cache-global",
 ]
 js-tracer = [
    "reth-node-builder/js-tracer",
@@ -111,5 +109,4 @@ test-utils = [
    "reth-evm/test-utils",
    "reth-primitives-traits/test-utils",
    "reth-evm-ethereum/test-utils",
-    "reth-stages-types/test-utils",
 ]
--- a/crates/ethereum/node/tests/e2e/custom_genesis.rs
+++ b/crates/ethereum/node/tests/e2e/custom_genesis.rs
@@ -1,100 +0,0 @@
-use crate::utils::eth_payload_attributes;
-use alloy_genesis::Genesis;
-use alloy_primitives::B256;
-use reth_chainspec::{ChainSpecBuilder, MAINNET};
-use reth_e2e_test_utils::{setup, transaction::TransactionTestContext};
-use reth_node_ethereum::EthereumNode;
-use reth_provider::{HeaderProvider, StageCheckpointReader};
-use reth_stages_types::StageId;
-use std::sync::Arc;
-
-/// Tests that a node can initialize and advance with a custom genesis block number.
-#[tokio::test]
-async fn can_run_eth_node_with_custom_genesis_number() -> eyre::Result<()> {
-    reth_tracing::init_test_tracing();
-
-    // Create genesis with custom block number (e.g., 1000)
-    let mut genesis: Genesis =
-        serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
-    genesis.number = Some(1000);
-    genesis.parent_hash = Some(B256::random());
-
-    let chain_spec = Arc::new(
-        ChainSpecBuilder::default()
-            .chain(MAINNET.chain)
-            .genesis(genesis)
-            .cancun_activated()
-            .build(),
-    );
-
-    let (mut nodes, _tasks, wallet) =
-        setup::<EthereumNode>(1, chain_spec, false, eth_payload_attributes).await?;
-
-    let mut node = nodes.pop().unwrap();
-
-    // Verify stage checkpoints are initialized to genesis block number (1000)
-    for stage in StageId::ALL {
-        let checkpoint = node.inner.provider.get_stage_checkpoint(stage)?;
-        assert!(checkpoint.is_some(), "Stage {:?} checkpoint should exist", stage);
-        assert_eq!(
-            checkpoint.unwrap().block_number,
-            1000,
-            "Stage {:?} checkpoint should be at genesis block 1000",
-            stage
-        );
-    }
-
-    // Advance the chain (block 1001)
-    let raw_tx = TransactionTestContext::transfer_tx_bytes(1, wallet.inner).await;
-    let tx_hash = node.rpc.inject_tx(raw_tx).await?;
-    let payload = node.advance_block().await?;
-
-    let block_hash = payload.block().hash();
-    let block_number = payload.block().number;
-
-    // Verify we're at block 1001 (genesis + 1)
-    assert_eq!(block_number, 1001, "Block number should be 1001 after advancing from genesis 1000");
-
-    // Assert the block has been committed
-    node.assert_new_block(tx_hash, block_hash, block_number).await?;
-
-    Ok(())
-}
-
-/// Tests that block queries respect custom genesis boundaries.
-#[tokio::test]
-async fn custom_genesis_block_query_boundaries() -> eyre::Result<()> {
-    reth_tracing::init_test_tracing();
-
-    let genesis_number = 5000u64;
-
-    let mut genesis: Genesis =
-        serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
-    genesis.number = Some(genesis_number);
-    genesis.parent_hash = Some(B256::random());
-
-    let chain_spec = Arc::new(
-        ChainSpecBuilder::default()
-            .chain(MAINNET.chain)
-            .genesis(genesis)
-            .cancun_activated()
-            .build(),
-    );
-
-    let (mut nodes, _tasks, _wallet) =
-        setup::<EthereumNode>(1, chain_spec, false, eth_payload_attributes).await?;
-
-    let node = nodes.pop().unwrap();
-
-    // Query genesis block should succeed
-    let genesis_header = node.inner.provider.header_by_number(genesis_number)?;
-    assert!(genesis_header.is_some(), "Genesis block at {} should exist", genesis_number);
-
-    // Query blocks before genesis should return None
-    for block_num in [0, 1, genesis_number - 1] {
-        let header = node.inner.provider.header_by_number(block_num)?;
-        assert!(header.is_none(), "Block {} before genesis should not exist", block_num);
-    }
-
-    Ok(())
-}
--- a/crates/ethereum/node/tests/e2e/eth.rs
+++ b/crates/ethereum/node/tests/e2e/eth.rs
@@ -7,7 +7,6 @@ use reth_e2e_test_utils::{
 use reth_node_builder::{NodeBuilder, NodeHandle};
 use reth_node_core::{args::RpcServerArgs, node_config::NodeConfig};
 use reth_node_ethereum::EthereumNode;
-use reth_provider::BlockNumReader;
 use reth_tasks::TaskManager;
 use std::sync::Arc;

@@ -128,55 +127,3 @@ async fn test_failed_run_eth_node_with_no_auth_engine_api_over_ipc_opts() -> eyr

    Ok(())
 }
-
-#[tokio::test]
-async fn test_engine_graceful_shutdown() -> eyre::Result<()> {
-    reth_tracing::init_test_tracing();
-
-    let (mut nodes, _tasks, wallet) = setup::<EthereumNode>(
-        1,
-        Arc::new(
-            ChainSpecBuilder::default()
-                .chain(MAINNET.chain)
-                .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
-                .cancun_activated()
-                .build(),
-        ),
-        false,
-        eth_payload_attributes,
-    )
-    .await?;
-
-    let mut node = nodes.pop().unwrap();
-
-    let raw_tx = TransactionTestContext::transfer_tx_bytes(1, wallet.inner).await;
-    let tx_hash = node.rpc.inject_tx(raw_tx).await?;
-    let payload = node.advance_block().await?;
-    node.assert_new_block(tx_hash, payload.block().hash(), payload.block().number).await?;
-
-    // Get block number before shutdown
-    let block_before = node.inner.provider.best_block_number()?;
-    assert_eq!(block_before, 1, "Expected 1 block before shutdown");
-
-    // Verify block is NOT yet persisted to database
-    let db_block_before = node.inner.provider.last_block_number()?;
-    assert_eq!(db_block_before, 0, "Block should not be persisted yet");
-
-    // Trigger graceful shutdown
-    let done_rx = node
-        .inner
-        .add_ons_handle
-        .engine_shutdown
-        .shutdown()
-        .expect("shutdown should return receiver");
-
-    tokio::time::timeout(std::time::Duration::from_secs(2), done_rx)
-        .await
-        .expect("shutdown timed out")
-        .expect("shutdown completion channel should not be closed");
-
-    let db_block = node.inner.provider.last_block_number()?;
-    assert_eq!(db_block, 1, "Database should have persisted block 1");
-
-    Ok(())
-}
--- a/crates/ethereum/node/tests/e2e/main.rs
+++ b/crates/ethereum/node/tests/e2e/main.rs
@@ -1,7 +1,6 @@
 #![allow(missing_docs)]

 mod blobs;
-mod custom_genesis;
 mod dev;
 mod eth;
 mod p2p;
--- a/crates/ethereum/payload/src/lib.rs
+++ b/crates/ethereum/payload/src/lib.rs
@@ -153,9 +153,9 @@ where
    let PayloadConfig { parent_header, attributes } = config;

    let state_provider = client.state_by_block_hash(parent_header.hash())?;
-    let state = StateProviderDatabase::new(state_provider.as_ref());
+    let state = StateProviderDatabase::new(&state_provider);
    let mut db =
-        State::builder().with_database_ref(cached_reads.as_db(state)).with_bundle_update().build();
+        State::builder().with_database(cached_reads.as_db_mut(state)).with_bundle_update().build();

    let mut builder = evm_config
        .builder_for_next_block(
@@ -211,8 +211,6 @@ where

    let is_osaka = chain_spec.is_osaka_active_at_timestamp(attributes.timestamp);

-    let withdrawals_rlp_length = attributes.withdrawals().length();
-
    while let Some(pool_tx) = best_txs.next() {
        // ensure we still have capacity for this transaction
        if cumulative_gas_used + pool_tx.gas_limit() > block_gas_limit {
@@ -234,10 +232,10 @@ where
        // convert tx to a signed transaction
        let tx = pool_tx.to_consensus();

-        let tx_rlp_len = tx.inner().length();
-
-        let estimated_block_size_with_tx =
-            block_transactions_rlp_length + tx_rlp_len + withdrawals_rlp_length + 1024; // 1Kb of overhead for the block header
+        let estimated_block_size_with_tx = block_transactions_rlp_length +
+            tx.inner().length() +
+            attributes.withdrawals().length() +
+            1024; // 1Kb of overhead for the block header

        if is_osaka && estimated_block_size_with_tx > MAX_RLP_BLOCK_SIZE {
            best_txs.mark_invalid(
@@ -338,7 +336,7 @@ where
            }
        }

-        block_transactions_rlp_length += tx_rlp_len;
+        block_transactions_rlp_length += tx.inner().length();

        // update and add to total fees
        let miner_fee =
@@ -360,8 +358,7 @@ where
        return Ok(BuildOutcome::Aborted { fees: total_fees, cached_reads })
    }

-    let BlockBuilderOutcome { execution_result, block, .. } =
-        builder.finish(state_provider.as_ref())?;
+    let BlockBuilderOutcome { execution_result, block, .. } = builder.finish(&state_provider)?;

    let requests = chain_spec
        .is_prague_active_at_timestamp(attributes.timestamp)
--- a/crates/ethereum/primitives/src/receipt.rs
+++ b/crates/ethereum/primitives/src/receipt.rs
@@ -258,7 +258,10 @@ impl<T: TxTy> IsTyped2718 for Receipt<T> {

 impl<T: TxTy> InMemorySize for Receipt<T> {
    fn size(&self) -> usize {
-        size_of::<Self>() + self.logs.iter().map(|log| log.size()).sum::<usize>()
+        self.tx_type.size() +
+            core::mem::size_of::<bool>() +
+            core::mem::size_of::<u64>() +
+            self.logs.iter().map(|log| log.size()).sum::<usize>()
    }
 }

--- a/crates/ethereum/reth/Cargo.toml
+++ b/crates/ethereum/reth/Cargo.toml
@@ -81,7 +81,6 @@ arbitrary = [
 ]
 keccak-cache-global = [
    "reth-node-ethereum?/keccak-cache-global",
-    "reth-node-core?/keccak-cache-global",
 ]
 test-utils = [
    "reth-chainspec/test-utils",
--- a/crates/evm/evm/src/lib.rs
+++ b/crates/evm/evm/src/lib.rs
@@ -61,6 +61,8 @@ pub use alloy_evm::{
    *,
 };

+pub use alloy_evm::block::state_changes as state_change;
+
 /// A complete configuration of EVM for Reth.
 ///
 /// This trait encapsulates complete configuration required for transaction execution and block
@@ -256,7 +258,7 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
        attributes: Self::NextBlockEnvCtx,
    ) -> Result<ExecutionCtxFor<'_, Self>, Self::Error>;

-    /// Returns a [`TxEnv`] from a transaction.
+    /// Returns a [`TxEnv`] from a transaction and [`Address`].
    fn tx_env(&self, transaction: impl IntoTxEnv<TxEnvFor<Self>>) -> TxEnvFor<Self> {
        transaction.into_tx_env()
    }
--- a/crates/evm/evm/src/metrics.rs
+++ b/crates/evm/evm/src/metrics.rs
@@ -6,6 +6,7 @@ use reth_primitives_traits::{Block, RecoveredBlock};
 use std::time::Instant;

 /// Executor metrics.
+// TODO(onbjerg): add sload/sstore
 #[derive(Metrics, Clone)]
 #[metrics(scope = "sync.execution")]
 pub struct ExecutorMetrics {
--- a/crates/evm/execution-types/src/chain.rs
+++ b/crates/evm/execution-types/src/chain.rs
@@ -7,8 +7,8 @@ use alloy_eips::{eip1898::ForkBlock, eip2718::Encodable2718, BlockNumHash};
 use alloy_primitives::{Address, BlockHash, BlockNumber, TxHash};
 use core::{fmt, ops::RangeInclusive};
 use reth_primitives_traits::{
-    transaction::signed::SignedTransaction, Block, BlockBody, IndexedTx, NodePrimitives,
-    RecoveredBlock, SealedHeader,
+    transaction::signed::SignedTransaction, Block, BlockBody, NodePrimitives, RecoveredBlock,
+    SealedHeader,
 };
 use reth_trie_common::updates::TrieUpdates;
 use revm::database::BundleState;
@@ -41,13 +41,6 @@ pub struct Chain<N: NodePrimitives = reth_ethereum_primitives::EthPrimitives> {
    trie_updates: Option<TrieUpdates>,
 }

-type ChainTxReceiptMeta<'a, N> = (
-    &'a RecoveredBlock<<N as NodePrimitives>::Block>,
-    IndexedTx<'a, <N as NodePrimitives>::Block>,
-    &'a <N as NodePrimitives>::Receipt,
-    &'a [<N as NodePrimitives>::Receipt],
-);
-
 impl<N: NodePrimitives> Default for Chain<N> {
    fn default() -> Self {
        Self {
@@ -192,24 +185,6 @@ impl<N: NodePrimitives> Chain<N> {
        self.blocks_iter().zip(self.block_receipts_iter())
    }

-    /// Finds a transaction by hash and returns it along with its corresponding receipt data.
-    ///
-    /// Returns `None` if the transaction is not found in this chain.
-    pub fn find_transaction_and_receipt_by_hash(
-        &self,
-        tx_hash: TxHash,
-    ) -> Option<ChainTxReceiptMeta<'_, N>> {
-        for (block, receipts) in self.blocks_and_receipts() {
-            let Some(indexed_tx) = block.find_indexed(tx_hash) else {
-                continue;
-            };
-            let receipt = receipts.get(indexed_tx.index())?;
-            return Some((block, indexed_tx, receipt, receipts.as_slice()));
-        }
-
-        None
-    }
-
    /// Get the block at which this chain forked.
    pub fn fork_block(&self) -> ForkBlock {
        let first = self.first();
--- a/crates/evm/execution-types/src/execution_outcome.rs
+++ b/crates/evm/execution-types/src/execution_outcome.rs
@@ -144,12 +144,7 @@ impl<T> ExecutionOutcome<T> {
        bundle: BundleState,
        results: Vec<BlockExecutionResult<T>>,
    ) -> Self {
-        let mut value = Self {
-            bundle,
-            first_block,
-            receipts: Vec::with_capacity(results.len()),
-            requests: Vec::with_capacity(results.len()),
-        };
+        let mut value = Self { bundle, first_block, receipts: Vec::new(), requests: Vec::new() };
        for result in results {
            value.receipts.push(result.receipts);
            value.requests.push(result.requests);
@@ -342,7 +337,7 @@ impl<T> ExecutionOutcome<T> {
    /// Prepends present the state with the given `BundleState`.
    /// It adds changes from the given state but does not override any existing changes.
    ///
-    /// Reverts and receipts are not updated.
+    /// Reverts  and receipts are not updated.
    pub fn prepend_state(&mut self, mut other: BundleState) {
        let other_len = other.reverts.len();
        // take this bundle
--- a/crates/exex/exex/src/backfill/job.rs
+++ b/crates/exex/exex/src/backfill/job.rs
@@ -118,6 +118,8 @@ where
            results.push(executor.execute_one(&block)?);
            execution_duration += execute_start.elapsed();

+            // TODO(alexey): report gas metrics using `block.header.gas_used`
+
            // Seal the block back and save it
            blocks.push(block);
            // Check if we should commit now
--- a/crates/exex/exex/src/backfill/test_utils.rs
+++ b/crates/exex/exex/src/backfill/test_utils.rs
@@ -13,7 +13,7 @@ use reth_evm_ethereum::EthEvmConfig;
 use reth_node_api::NodePrimitives;
 use reth_primitives_traits::{Block as _, RecoveredBlock};
 use reth_provider::{
-    providers::ProviderNodeTypes, BlockWriter as _, ExecutionOutcome, LatestStateProvider,
+    providers::ProviderNodeTypes, BlockWriter as _, ExecutionOutcome, LatestStateProviderRef,
    ProviderFactory,
 };
 use reth_revm::database::StateProviderDatabase;
@@ -69,7 +69,7 @@ where

    // Execute the block to produce a block execution output
    let mut block_execution_output = EthEvmConfig::ethereum(chain_spec)
-        .batch_executor(StateProviderDatabase::new(LatestStateProvider::new(provider)))
+        .batch_executor(StateProviderDatabase::new(LatestStateProviderRef::new(&provider)))
        .execute(block)?;
    block_execution_output.state.reverts.sort();

@@ -203,8 +203,8 @@ where
    let provider = provider_factory.provider()?;

    let evm_config = EthEvmConfig::new(chain_spec);
-    let executor =
-        evm_config.batch_executor(StateProviderDatabase::new(LatestStateProvider::new(provider)));
+    let executor = evm_config
+        .batch_executor(StateProviderDatabase::new(LatestStateProviderRef::new(&provider)));

    let mut execution_outcome = executor.execute_batch(vec![&block1, &block2])?;
    execution_outcome.state_mut().reverts.sort();
--- a/crates/exex/exex/src/manager.rs
+++ b/crates/exex/exex/src/manager.rs
@@ -1303,7 +1303,7 @@ mod tests {
        .try_recover()
        .unwrap();
        let provider_rw = provider_factory.database_provider_rw().unwrap();
-        provider_rw.insert_block(&block).unwrap();
+        provider_rw.insert_block(block.clone()).unwrap();
        provider_rw.commit().unwrap();

        let provider = BlockchainProvider::new(provider_factory).unwrap();
--- a/crates/exex/exex/src/notifications.rs
+++ b/crates/exex/exex/src/notifications.rs
@@ -354,10 +354,10 @@ where
    /// canonical chain.
    ///
    /// Possible situations are:
-    /// - ExEx is behind the node head (`exex_head.number < node_head.number`). Backfill from the
+    /// - ExEx is behind the node head (`node_head.number < exex_head.number`). Backfill from the
    ///   node database.
-    /// - ExEx is at the same block number as the node head (`exex_head.number ==
-    ///   node_head.number`). Nothing to do.
+    /// - ExEx is at the same block number as the node head (`node_head.number ==
+    ///   exex_head.number`). Nothing to do.
    fn check_backfill(&mut self) -> eyre::Result<()> {
        let backfill_job_factory =
            BackfillJobFactory::new(self.evm_config.clone(), self.provider.clone());
@@ -481,12 +481,12 @@ mod tests {
            &mut rng,
            genesis_block.number + 1,
            BlockParams { parent: Some(genesis_hash), tx_count: Some(0), ..Default::default() },
-        )
-        .try_recover()?;
-        let node_head = node_head_block.num_hash();
+        );
        let provider_rw = provider_factory.provider_rw()?;
-        provider_rw.insert_block(&node_head_block)?;
+        provider_rw.insert_block(node_head_block.clone().try_recover()?)?;
        provider_rw.commit()?;
+
+        let node_head = node_head_block.num_hash();
        let exex_head =
            ExExHead { block: BlockNumHash { number: genesis_block.number, hash: genesis_hash } };

@@ -613,7 +613,7 @@ mod tests {
        .try_recover()?;
        let node_head = node_head_block.num_hash();
        let provider_rw = provider.database_provider_rw()?;
-        provider_rw.insert_block(&node_head_block)?;
+        provider_rw.insert_block(node_head_block)?;
        provider_rw.commit()?;
        let node_head_notification = ExExNotification::ChainCommitted {
            new: Arc::new(
--- a/crates/exex/exex/src/wal/cache.rs
+++ b/crates/exex/exex/src/wal/cache.rs
@@ -116,11 +116,6 @@ impl BlockCache {
                self.committed_blocks.insert(block.hash(), (file_id, cached_block));
            }

-            let first_block_number = committed_chain.first().number();
-            self.lowest_committed_block_height = Some(
-                self.lowest_committed_block_height
-                    .map_or(first_block_number, |lowest| lowest.min(first_block_number)),
-            );
            self.highest_committed_block_height = Some(committed_chain.tip().number());
        }
    }
--- a/crates/net/discv4/src/config.rs
+++ b/crates/net/discv4/src/config.rs
@@ -212,13 +212,13 @@ impl Discv4ConfigBuilder {
        self
    }

-    /// Whether to enable EIP-868
+    /// Whether to enforce expiration timestamps in messages.
    pub const fn enable_eip868(&mut self, enable_eip868: bool) -> &mut Self {
        self.config.enable_eip868 = enable_eip868;
        self
    }

-    /// Whether to enforce expiration timestamps in messages.
+    /// Whether to enable EIP-868
    pub const fn enforce_expiration_timestamps(
        &mut self,
        enforce_expiration_timestamps: bool,
--- a/crates/net/discv5/src/enr.rs
+++ b/crates/net/discv5/src/enr.rs
@@ -24,7 +24,7 @@ pub fn discv4_id_to_discv5_id(peer_id: PeerId) -> Result<NodeId, secp256k1::Erro
    Ok(id2pk(peer_id)?.into())
 }

-/// Converts a [`PeerId`] to a [`discv5::libp2p_identity::PeerId`].
+/// Converts a [`PeerId`] to a [`reth_network_peers::PeerId`].
 pub fn discv4_id_to_multiaddr_id(
    peer_id: PeerId,
 ) -> Result<discv5::libp2p_identity::PeerId, secp256k1::Error> {
--- a/crates/net/discv5/src/lib.rs
+++ b/crates/net/discv5/src/lib.rs
@@ -77,7 +77,7 @@ pub struct Discv5 {
    metrics: Discv5Metrics,
    /// Returns the _local_ [`NodeRecord`] this service was started with.
    // Note: we must track this separately because the `discv5::Discv5` does not necessarily
-    // provide this via its [`local_enr`](discv5::Discv5::local_enr()). This is intended for
+    // provide this via it's [`local_enr`](discv5::Discv5::local_ner()) This is intended for
    // obtaining the port this service was launched at
    local_node_record: NodeRecord,
 }
--- a/crates/net/downloaders/src/headers/reverse_headers.rs
+++ b/crates/net/downloaders/src/headers/reverse_headers.rs
@@ -1218,7 +1218,9 @@ impl ReverseHeadersDownloaderBuilder {
            next_request_block_number: 0,
            next_chain_tip_block_number: 0,
            lowest_validated_header: None,
-            request_limit,
+            // TODO(mattsse): tmp hotfix to prevent issues with syncing from besu which has an upper
+            // limit of 512
+            request_limit: request_limit.min(512),
            min_concurrent_requests,
            max_concurrent_requests,
            stream_batch_size,
--- a/crates/net/ecies/src/algorithm.rs
+++ b/crates/net/ecies/src/algorithm.rs
@@ -312,6 +312,7 @@ impl ECIES {

    /// Create a new ECIES client with the given static secret key and remote peer ID.
    pub fn new_client(secret_key: SecretKey, remote_id: PeerId) -> Result<Self, ECIESError> {
+        // TODO(rand): use rng for nonce
        let mut rng = rng();
        let nonce = B256::random();
        let ephemeral_secret_key = SecretKey::new(&mut rng);
--- a/crates/net/eth-wire-types/src/broadcast.rs
+++ b/crates/net/eth-wire-types/src/broadcast.rs
@@ -409,13 +409,6 @@ impl NewPooledTransactionHashes68 {
        }
    }

-    /// Shrinks the capacity of the message vectors as much as possible.
-    pub fn shrink_to_fit(&mut self) {
-        self.hashes.shrink_to_fit();
-        self.sizes.shrink_to_fit();
-        self.types.shrink_to_fit()
-    }
-
    /// Consumes and appends a transaction
    pub fn with_transaction<T: SignedTransaction>(mut self, tx: &T) -> Self {
        self.push(tx);
--- a/crates/net/eth-wire/src/p2pstream.rs
+++ b/crates/net/eth-wire/src/p2pstream.rs
@@ -101,9 +101,8 @@ where
            .or(Err(P2PStreamError::HandshakeError(P2PHandshakeError::Timeout)))?
            .ok_or(P2PStreamError::HandshakeError(P2PHandshakeError::NoResponse))??;

-        // Check that the uncompressed message length does not exceed the max payload size.
-        // Note: The first message (Hello/Disconnect) is not snappy compressed. We will check the
-        // decompressed length again for subsequent messages after the handshake.
+        // let's check the compressed length first, we will need to check again once confirming
+        // that it contains snappy-compressed data (this will be the case for all non-p2p messages).
        if first_message_bytes.len() > MAX_PAYLOAD_SIZE {
            return Err(P2PStreamError::MessageTooBig {
                message_size: first_message_bytes.len(),
--- a/crates/net/network/src/discovery.rs
+++ b/crates/net/network/src/discovery.rs
@@ -151,16 +151,13 @@ impl Discovery {
        self.discovery_listeners.retain_mut(|listener| listener.send(event.clone()).is_ok());
    }

-    /// Updates the `eth:ForkId` field in discv4/discv5.
+    /// Updates the `eth:ForkId` field in discv4.
    pub(crate) fn update_fork_id(&self, fork_id: ForkId) {
        if let Some(discv4) = &self.discv4 {
            // use forward-compatible forkid entry
            discv4.set_eip868_rlp(b"eth".to_vec(), EnrForkIdEntry::from(fork_id))
        }
-        if let Some(discv5) = &self.discv5 {
-            discv5
-                .encode_and_set_eip868_in_local_enr(b"eth".to_vec(), EnrForkIdEntry::from(fork_id))
-        }
+        // todo: update discv5 enr
    }

    /// Bans the [`IpAddr`] in the discovery service.
--- a/crates/net/network/src/fetch/mod.rs
+++ b/crates/net/network/src/fetch/mod.rs
@@ -179,7 +179,7 @@ impl<N: NetworkPrimitives> StateFetcher<N> {

        let request = self.queued_requests.pop_front().expect("not empty");
        let Some(peer_id) = self.next_best_peer(request.best_peer_requirements()) else {
-            // need to put back the request
+            // need to put back the the request
            self.queued_requests.push_front(request);
            return PollAction::NoPeersAvailable
        };
@@ -397,7 +397,7 @@ impl Peer {
    /// A peer has a "better range" if:
    /// 1. It can fully cover the requested range while the other cannot
    /// 2. None can fully cover the range, but this peer has lower start value
-    /// 3. If a peer doesn't announce a range we assume it has full history, but check the other's
+    /// 3. If a peer doesnt announce a range we assume it has full history, but check the other's
    ///    range and treat that as better if it can cover the range
    fn has_better_range(&self, other: &Self, range: &RangeInclusive<u64>) -> bool {
        let self_range = self.range();
--- a/crates/net/network/src/transactions/mod.rs
+++ b/crates/net/network/src/transactions/mod.rs
@@ -860,8 +860,9 @@ where
        peer_id: PeerId,
        propagation_mode: PropagationMode,
    ) -> Option<PropagatedTransactions> {
-        let peer = self.peers.get_mut(&peer_id)?;
        trace!(target: "net::tx", ?peer_id, "Propagating transactions to peer");
+
+        let peer = self.peers.get_mut(&peer_id)?;
        let mut propagated = PropagatedTransactions::default();

        // filter all transactions unknown to the peer
@@ -1931,14 +1932,8 @@ impl PooledTransactionsHashesBuilder {

    fn build(self) -> NewPooledTransactionHashes {
        match self {
-            Self::Eth66(mut msg) => {
-                msg.0.shrink_to_fit();
-                msg.into()
-            }
-            Self::Eth68(mut msg) => {
-                msg.shrink_to_fit();
-                msg.into()
-            }
+            Self::Eth66(msg) => msg.into(),
+            Self::Eth68(msg) => msg.into(),
        }
    }
 }
--- a/crates/net/p2p/src/bodies/downloader.rs
+++ b/crates/net/p2p/src/bodies/downloader.rs
@@ -13,7 +13,9 @@ pub type BodyDownloaderResult<B> = DownloadResult<Vec<BlockResponse<B>>>;
 /// A downloader represents a distinct strategy for submitting requests to download block bodies,
 /// while a [`BodiesClient`][crate::bodies::client::BodiesClient] represents a client capable of
 /// fulfilling these requests.
-pub trait BodyDownloader: Send + Stream<Item = BodyDownloaderResult<Self::Block>> + Unpin {
+pub trait BodyDownloader:
+    Send + Sync + Stream<Item = BodyDownloaderResult<Self::Block>> + Unpin
+{
    /// The Block type this downloader supports
    type Block: Block + 'static;

--- a/crates/net/p2p/src/bodies/response.rs
+++ b/crates/net/p2p/src/bodies/response.rs
@@ -37,14 +37,6 @@ where
            Self::Empty(_) => None,
        }
    }
-
-    /// Return the reference to the response body
-    pub const fn body(&self) -> Option<&B::Body> {
-        match self {
-            Self::Full(block) => Some(block.body()),
-            Self::Empty(_) => None,
-        }
-    }
 }

 impl<B: Block> InMemorySize for BlockResponse<B> {
--- a/crates/net/p2p/src/headers/downloader.rs
+++ b/crates/net/p2p/src/headers/downloader.rs
@@ -100,7 +100,9 @@ impl<H: BlockHeader + Sealable> HeaderSyncGap<H> {
    }
 }

-/// Validate whether the header is valid in relation to its parent.
+/// Validate whether the header is valid in relation to it's parent
+///
+/// Returns Ok(false) if the
 pub fn validate_header_download<H: BlockHeader>(
    consensus: &dyn HeaderValidator<H>,
    header: &SealedHeader<H>,
--- a/crates/node/builder/Cargo.toml
+++ b/crates/node/builder/Cargo.toml
@@ -76,7 +76,6 @@ secp256k1 = { workspace = true, features = ["global-context", "std", "recovery"]
 ## misc
 aquamarine.workspace = true
 eyre.workspace = true
-parking_lot.workspace = true
 jsonrpsee.workspace = true
 fdlimit.workspace = true
 rayon.workspace = true
--- a/crates/node/builder/src/launch/common.rs
+++ b/crates/node/builder/src/launch/common.rs
@@ -66,9 +66,8 @@ use reth_node_metrics::{
 };
 use reth_provider::{
    providers::{NodeTypesForProvider, ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
-    BlockHashReader, BlockNumReader, DatabaseProviderFactory, ProviderError, ProviderFactory,
-    ProviderResult, RocksDBProviderFactory, StageCheckpointReader, StaticFileProviderBuilder,
-    StaticFileProviderFactory,
+    BlockHashReader, BlockNumReader, ProviderError, ProviderFactory, ProviderResult,
+    StageCheckpointReader, StaticFileProviderBuilder, StaticFileProviderFactory,
 };
 use reth_prune::{PruneModes, PrunerBuilder};
 use reth_rpc_builder::config::RethRpcServerConfig;
@@ -454,7 +453,11 @@ impl<R, ChainSpec: EthChainSpec> LaunchContextWith<Attached<WithConfigs<ChainSpe
    where
        Pool: TransactionPool + Unpin,
    {
-        self.node_config().dev_mining_mode(pool)
+        if let Some(interval) = self.node_config().dev.block_time {
+            MiningMode::interval(interval)
+        } else {
+            MiningMode::instant(pool, self.node_config().dev.block_max_transactions)
+        }
    }
 }

@@ -480,7 +483,6 @@ where
            StaticFileProviderBuilder::read_write(self.data_dir().static_files())?
                .with_metrics()
                .with_blocks_per_file_for_segments(static_files_config.as_blocks_per_file_map())
-                .with_genesis_block_number(self.chain_spec().genesis().number.unwrap_or_default())
                .build()?;

        // Initialize RocksDB provider with metrics, statistics, and default tables
@@ -498,54 +500,20 @@ where
        )?
        .with_prune_modes(self.prune_modes());

-        // Keep MDBX, static files, and RocksDB aligned. If any check fails, unwind to the
-        // earliest consistent block.
-        //
-        // Order matters:
-        // 1) heal static files (no pruning)
-        // 2) check RocksDB (needs static-file tx data)
-        // 3) check static-file checkpoints vs MDBX (may prune)
-        //
-        // Compute one unwind target and run a single unwind.
-
-        let provider_ro = factory.database_provider_ro()?;
-
-        // Step 1: heal file-level inconsistencies (no pruning)
-        factory.static_file_provider().check_file_consistency(&provider_ro)?;
-
-        // Step 2: RocksDB consistency check (needs static files tx data)
-        let rocksdb_unwind = factory.rocksdb_provider().check_consistency(&provider_ro)?;
-
-        // Step 3: Static file checkpoint consistency (may prune)
-        let static_file_unwind = factory
-            .static_file_provider()
-            .check_consistency(&provider_ro)?
-            .map(|target| match target {
-                PipelineTarget::Unwind(block) => block,
-                PipelineTarget::Sync(_) => unreachable!("check_consistency returns Unwind"),
-            });
-
-        // Take the minimum block number to ensure all storage layers are consistent.
-        let unwind_target = [rocksdb_unwind, static_file_unwind].into_iter().flatten().min();
-
-        if let Some(unwind_block) = unwind_target {
+        // Check for consistency between database and static files. If it fails, it unwinds to
+        // the first block that's consistent between database and static files.
+        if let Some(unwind_target) =
+            factory.static_file_provider().check_consistency(&factory.provider()?)?
+        {
            // Highly unlikely to happen, and given its destructive nature, it's better to panic
-            // instead. Unwinding to 0 would leave MDBX with a huge free list size.
-            let inconsistency_source = match (rocksdb_unwind, static_file_unwind) {
-                (Some(_), Some(_)) => "RocksDB and static file",
-                (Some(_), None) => "RocksDB",
-                (None, Some(_)) => "static file",
-                (None, None) => unreachable!(),
-            };
+            // instead.
            assert_ne!(
-                unwind_block, 0,
-                "A {} inconsistency was found that would trigger an unwind to block 0",
-                inconsistency_source
+                unwind_target,
+                PipelineTarget::Unwind(0),
+                "A static file <> database inconsistency was found that would trigger an unwind to block 0"
            );

-            let unwind_target = PipelineTarget::Unwind(unwind_block);
-
-            info!(target: "reth::cli", %unwind_target, %inconsistency_source, "Executing unwind after consistency check.");
+            info!(target: "reth::cli", unwind_target = %unwind_target, "Executing an unwind after a failed storage consistency check.");

            let (_tip_tx, tip_rx) = watch::channel(B256::ZERO);

@@ -579,7 +547,7 @@ where
                }),
            );
            rx.await?.inspect_err(|err| {
-                error!(target: "reth::cli", %unwind_target, %inconsistency_source, %err, "failed to run unwind")
+                error!(target: "reth::cli", unwind_target = %unwind_target, %err, "failed to run unwind")
            })?;
        }

@@ -648,7 +616,7 @@ where
                    target_triple: version_metadata().vergen_cargo_target_triple.as_ref(),
                    build_profile: version_metadata().build_profile_name.as_ref(),
                },
-                ChainSpecInfo { name: self.chain_id().to_string() },
+                ChainSpecInfo { name: self.left().config.chain.chain().to_string() },
                self.task_executor().clone(),
                Hooks::builder()
                    .with_hook({
@@ -969,44 +937,28 @@ where
    ///
    /// A target block hash if the pipeline is inconsistent, otherwise `None`.
    pub fn check_pipeline_consistency(&self) -> ProviderResult<Option<B256>> {
-        // We skip the era stage if it's not enabled
-        let era_enabled = self.era_import_source().is_some();
-        let mut all_stages =
-            StageId::ALL.into_iter().filter(|id| era_enabled || id != &StageId::Era);
-
-        // Get the expected first stage based on config.
-        let first_stage = all_stages.next().expect("there must be at least one stage");
-
        // If no target was provided, check if the stages are congruent - check if the
        // checkpoint of the last stage matches the checkpoint of the first.
        let first_stage_checkpoint = self
            .blockchain_db()
-            .get_stage_checkpoint(first_stage)?
+            .get_stage_checkpoint(*StageId::ALL.first().unwrap())?
            .unwrap_or_default()
            .block_number;

-        // Compare all other stages against the first
-        for stage_id in all_stages {
+        // Skip the first stage as we've already retrieved it and comparing all other checkpoints
+        // against it.
+        for stage_id in StageId::ALL.iter().skip(1) {
            let stage_checkpoint = self
                .blockchain_db()
-                .get_stage_checkpoint(stage_id)?
+                .get_stage_checkpoint(*stage_id)?
                .unwrap_or_default()
                .block_number;

            // If the checkpoint of any stage is less than the checkpoint of the first stage,
            // retrieve and return the block hash of the latest header and use it as the target.
-            debug!(
-                target: "consensus::engine",
-                first_stage_id = %first_stage,
-                first_stage_checkpoint,
-                stage_id = %stage_id,
-                stage_checkpoint = stage_checkpoint,
-                "Checking stage against first stage",
-            );
            if stage_checkpoint < first_stage_checkpoint {
                debug!(
                    target: "consensus::engine",
-                    first_stage_id = %first_stage,
                    first_stage_checkpoint,
                    inconsistent_stage_id = %stage_id,
                    inconsistent_stage_checkpoint = stage_checkpoint,
@@ -1091,13 +1043,7 @@ where
    }

    /// Spawns the [`EthStatsService`] service if configured.
-    pub async fn spawn_ethstats<St>(&self, mut engine_events: St) -> eyre::Result<()>
-    where
-        St: Stream<Item = reth_engine_primitives::ConsensusEngineEvent<PrimitivesTy<T::Types>>>
-            + Send
-            + Unpin
-            + 'static,
-    {
+    pub async fn spawn_ethstats(&self) -> eyre::Result<()> {
        let Some(url) = self.node_config().debug.ethstats.as_ref() else { return Ok(()) };

        let network = self.components().network().clone();
@@ -1107,37 +1053,7 @@ where
        info!(target: "reth::cli", "Starting EthStats service at {}", url);

        let ethstats = EthStatsService::new(url, network, provider, pool).await?;
-
-        // If engine events are provided, spawn listener for new payload reporting
-        let ethstats_for_events = ethstats.clone();
-        let task_executor = self.task_executor().clone();
-        task_executor.spawn(Box::pin(async move {
-            while let Some(event) = engine_events.next().await {
-                use reth_engine_primitives::ConsensusEngineEvent;
-                match event {
-                    ConsensusEngineEvent::ForkBlockAdded(executed, duration) |
-                    ConsensusEngineEvent::CanonicalBlockAdded(executed, duration) => {
-                        let block_hash = executed.recovered_block.num_hash().hash;
-                        let block_number = executed.recovered_block.num_hash().number;
-                        if let Err(e) = ethstats_for_events
-                            .report_new_payload(block_hash, block_number, duration)
-                            .await
-                        {
-                            debug!(
-                                target: "ethstats",
-                                "Failed to report new payload: {}", e
-                            );
-                        }
-                    }
-                    _ => {
-                        // Ignore other event types for ethstats reporting
-                    }
-                }
-            }
-        }));
-
-        // Spawn main ethstats service
-        task_executor.spawn(Box::pin(async move { ethstats.run().await }));
+        tokio::spawn(async move { ethstats.run().await });

        Ok(())
    }
--- a/crates/node/builder/src/launch/engine.rs
+++ b/crates/node/builder/src/launch/engine.rs
@@ -3,17 +3,16 @@
 use crate::{
    common::{Attached, LaunchContextWith, WithConfigs},
    hooks::NodeHooks,
-    rpc::{EngineShutdown, EngineValidatorAddOn, EngineValidatorBuilder, RethRpcAddOns, RpcHandle},
+    rpc::{EngineValidatorAddOn, EngineValidatorBuilder, RethRpcAddOns, RpcHandle},
    setup::build_networked_pipeline,
    AddOns, AddOnsContext, FullNode, LaunchContext, LaunchNode, NodeAdapter,
    NodeBuilderWithComponents, NodeComponents, NodeComponentsBuilder, NodeHandle, NodeTypesAdapter,
 };
 use alloy_consensus::BlockHeader;
-use futures::{stream_select, FutureExt, StreamExt};
+use futures::{stream_select, StreamExt};
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_engine_service::service::{ChainEvent, EngineService};
 use reth_engine_tree::{
-    chain::FromOrchestrator,
    engine::{EngineApiRequest, EngineRequestHandler},
    tree::TreeConfig,
 };
@@ -261,16 +260,8 @@ impl EngineNodeLauncher {
            )),
        );

-        let RpcHandle {
-            rpc_server_handles,
-            rpc_registry,
-            engine_events,
-            beacon_engine_handle,
-            engine_shutdown: _,
-        } = add_ons.launch_add_ons(add_ons_ctx).await?;
-
-        // Create engine shutdown handle
-        let (engine_shutdown, shutdown_rx) = EngineShutdown::new();
+        let RpcHandle { rpc_server_handles, rpc_registry, engine_events, beacon_engine_handle } =
+            add_ons.launch_add_ons(add_ons_ctx).await?;

        // Run consensus engine to completion
        let initial_target = ctx.initial_backfill_target()?;
@@ -290,7 +281,7 @@ impl EngineNodeLauncher {
        let startup_sync_state_idle = ctx.node_config().debug.startup_sync_state_idle;

        info!(target: "reth::cli", "Starting consensus engine");
-        let consensus_engine = async move {
+        ctx.task_executor().spawn_critical("consensus engine", Box::pin(async move {
            if let Some(initial_target) = initial_target {
                debug!(target: "reth::cli", %initial_target,  "start backfill sync");
                // network_handle's sync state is already initialized at Syncing
@@ -300,21 +291,10 @@ impl EngineNodeLauncher {
            }

            let mut res = Ok(());
-            let mut shutdown_rx = shutdown_rx.fuse();

-            // advance the chain and await payloads built locally to add into the engine api
-            // tree handler to prevent re-execution if that block is received as payload from
-            // the CL
+            // advance the chain and await payloads built locally to add into the engine api tree handler to prevent re-execution if that block is received as payload from the CL
            loop {
                tokio::select! {
-                    shutdown_req = &mut shutdown_rx => {
-                        if let Ok(req) = shutdown_req {
-                            debug!(target: "reth::cli", "received engine shutdown request");
-                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(
-                                FromOrchestrator::Terminate { tx: req.done_tx }.into()
-                            );
-                        }
-                    }
                    payload = built_payloads.select_next_some() => {
                        if let Some(executed_block) = payload.executed_block() {
                            debug!(target: "reth::cli", block=?executed_block.recovered_block.num_hash(),  "inserting built payload");
@@ -346,21 +326,19 @@ impl EngineNodeLauncher {
                                if let Some(head) = ev.canonical_header() {
                                    // Once we're progressing via live sync, we can consider the node is not syncing anymore
                                    network_handle.update_sync_state(SyncState::Idle);
-                                    let head_block = Head {
+                                                                        let head_block = Head {
                                        number: head.number(),
                                        hash: head.hash(),
                                        difficulty: head.difficulty(),
                                        timestamp: head.timestamp(),
-                                        total_difficulty: chainspec.final_paris_total_difficulty()
-                                            .filter(|_| chainspec.is_paris_active_at_block(head.number()))
-                                            .unwrap_or_default(),
+                                        total_difficulty: chainspec.final_paris_total_difficulty().filter(|_| chainspec.is_paris_active_at_block(head.number())).unwrap_or_default(),
                                    };
                                    network_handle.update_status(head_block);

                                    let updated = BlockRangeUpdate {
                                        earliest: provider.earliest_block_number().unwrap_or_default(),
-                                        latest: head.number(),
-                                        latest_hash: head.hash(),
+                                        latest:head.number(),
+                                        latest_hash:head.hash()
                                    };
                                    network_handle.update_block_range(updated);
                                }
@@ -372,10 +350,7 @@ impl EngineNodeLauncher {
            }

            let _ = exit.send(res);
-        };
-        ctx.task_executor().spawn_critical("consensus engine", Box::pin(consensus_engine));
-
-        let engine_events_for_ethstats = engine_events.new_listener();
+        }));

        let full_node = FullNode {
            evm_config: ctx.components().evm_config().clone(),
@@ -391,13 +366,12 @@ impl EngineNodeLauncher {
                rpc_registry,
                engine_events,
                beacon_engine_handle,
-                engine_shutdown,
            },
        };
        // Notify on node started
        on_node_started.on_event(FullNode::clone(&full_node))?;

-        ctx.spawn_ethstats(engine_events_for_ethstats).await?;
+        ctx.spawn_ethstats().await?;

        let handle = NodeHandle {
            node_exit_future: NodeExitFuture::new(
--- a/crates/node/builder/src/rpc.rs
+++ b/crates/node/builder/src/rpc.rs
@@ -11,7 +11,6 @@ use crate::{
 use alloy_rpc_types::engine::ClientVersionV1;
 use alloy_rpc_types_engine::ExecutionData;
 use jsonrpsee::{core::middleware::layer::Either, RpcModule};
-use parking_lot::Mutex;
 use reth_chain_state::CanonStateSubscriptions;
 use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks, Hardforks};
 use reth_node_api::{
@@ -42,9 +41,7 @@ use std::{
    fmt::{self, Debug},
    future::Future,
    ops::{Deref, DerefMut},
-    sync::Arc,
 };
-use tokio::sync::oneshot;

 /// Contains the handles to the spawned RPC servers.
 ///
@@ -335,8 +332,6 @@ pub struct RpcHandle<Node: FullNodeComponents, EthApi: EthApiTypes> {
    pub engine_events: EventSender<ConsensusEngineEvent<<Node::Types as NodeTypes>::Primitives>>,
    /// Handle to the beacon consensus engine.
    pub beacon_engine_handle: ConsensusEngineHandle<<Node::Types as NodeTypes>::Payload>,
-    /// Handle to trigger engine shutdown.
-    pub engine_shutdown: EngineShutdown,
 }

 impl<Node: FullNodeComponents, EthApi: EthApiTypes> Clone for RpcHandle<Node, EthApi> {
@@ -346,7 +341,6 @@ impl<Node: FullNodeComponents, EthApi: EthApiTypes> Clone for RpcHandle<Node, Et
            rpc_registry: self.rpc_registry.clone(),
            engine_events: self.engine_events.clone(),
            beacon_engine_handle: self.beacon_engine_handle.clone(),
-            engine_shutdown: self.engine_shutdown.clone(),
        }
    }
 }
@@ -367,7 +361,6 @@ where
        f.debug_struct("RpcHandle")
            .field("rpc_server_handles", &self.rpc_server_handles)
            .field("rpc_registry", &self.rpc_registry)
-            .field("engine_shutdown", &self.engine_shutdown)
            .finish()
    }
 }
@@ -963,7 +956,6 @@ where
            rpc_registry: registry,
            engine_events,
            beacon_engine_handle: engine_handle,
-            engine_shutdown: EngineShutdown::default(),
        })
    }

@@ -1389,7 +1381,6 @@ where
            version: version_metadata().cargo_pkg_version.to_string(),
            commit: version_metadata().vergen_git_sha.to_string(),
        };
-
        Ok(EngineApi::new(
            ctx.node.provider().clone(),
            ctx.config.chain.clone(),
@@ -1401,7 +1392,6 @@ where
            EngineCapabilities::default(),
            engine_validator,
            ctx.config.engine.accept_execution_requests_hash,
-            ctx.node.network().clone(),
        ))
    }
 }
@@ -1436,48 +1426,3 @@ impl IntoEngineApiRpcModule for NoopEngineApi {
        RpcModule::new(())
    }
 }
-
-/// Handle to trigger graceful engine shutdown.
-///
-/// This handle can be used to request a graceful shutdown of the engine,
-/// which will persist all remaining in-memory blocks before terminating.
-#[derive(Clone, Debug)]
-pub struct EngineShutdown {
-    /// Channel to send shutdown signal.
-    tx: Arc<Mutex<Option<oneshot::Sender<EngineShutdownRequest>>>>,
-}
-
-impl EngineShutdown {
-    /// Creates a new [`EngineShutdown`] handle and returns the receiver.
-    pub fn new() -> (Self, oneshot::Receiver<EngineShutdownRequest>) {
-        let (tx, rx) = oneshot::channel();
-        (Self { tx: Arc::new(Mutex::new(Some(tx))) }, rx)
-    }
-
-    /// Requests a graceful engine shutdown.
-    ///
-    /// All remaining in-memory blocks will be persisted before the engine terminates.
-    ///
-    /// Returns a receiver that resolves when shutdown is complete.
-    /// Returns `None` if shutdown was already triggered.
-    pub fn shutdown(&self) -> Option<oneshot::Receiver<()>> {
-        let mut guard = self.tx.lock();
-        let tx = guard.take()?;
-        let (done_tx, done_rx) = oneshot::channel();
-        let _ = tx.send(EngineShutdownRequest { done_tx });
-        Some(done_rx)
-    }
-}
-
-impl Default for EngineShutdown {
-    fn default() -> Self {
-        Self { tx: Arc::new(Mutex::new(None)) }
-    }
-}
-
-/// Request to shutdown the engine.
-#[derive(Debug)]
-pub struct EngineShutdownRequest {
-    /// Channel to signal shutdown completion.
-    pub done_tx: oneshot::Sender<()>,
-}
--- a/crates/node/core/Cargo.toml
+++ b/crates/node/core/Cargo.toml
@@ -80,7 +80,7 @@ tokio.workspace = true
 # Features for vergen to generate correct env vars
 jemalloc = ["reth-cli-util/jemalloc"]
 asm-keccak = ["alloy-primitives/asm-keccak"]
-keccak-cache-global = ["alloy-primitives/keccak-cache-global"]
+# Feature to enable opentelemetry export
 otlp = ["reth-tracing/otlp"]

 min-error-logs = ["tracing/release_max_level_error"]
--- a/crates/node/core/src/args/log.rs
+++ b/crates/node/core/src/args/log.rs
@@ -61,20 +61,6 @@ pub struct LogArgs {
    )]
    pub journald_filter: String,

-    /// Emit traces to samply. Only useful when profiling.
-    #[arg(long = "log.samply", global = true, hide = true)]
-    pub samply: bool,
-
-    /// The filter to use for traces emitted to samply.
-    #[arg(
-        long = "log.samply.filter",
-        value_name = "FILTER",
-        global = true,
-        default_value = "debug",
-        hide = true
-    )]
-    pub samply_filter: String,
-
    /// Sets whether or not the formatter emits ANSI terminal escape codes for colors and other
    /// text formatting.
    #[arg(
@@ -143,11 +129,6 @@ impl LogArgs {
            tracer = tracer.with_file(file, info);
        }

-        if self.samply {
-            let config = self.layer_info(LogFormat::Terminal, self.samply_filter.clone(), false);
-            tracer = tracer.with_samply(config);
-        }
-
        let guard = tracer.init_with_layers(layers)?;
        Ok(guard)
    }
--- a/crates/node/core/src/args/network.rs
+++ b/crates/node/core/src/args/network.rs
@@ -36,7 +36,7 @@ use reth_network::{
        DEFAULT_SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESP_ON_PACK_GET_POOLED_TRANSACTIONS_REQ,
        SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESPONSE,
    },
-    HelloMessageWithProtocols, NetworkConfigBuilder, NetworkPrimitives,
+    HelloMessageWithProtocols, NetworkConfigBuilder, NetworkPrimitives, SessionsConfig,
 };
 use reth_network_peers::{mainnet_nodes, TrustedPeer};
 use secp256k1::SecretKey;
@@ -339,7 +339,7 @@ impl NetworkArgs {
        NetworkConfigBuilder::<N>::new(secret_key)
            .external_ip_resolver(self.nat.clone())
            .sessions_config(
-                config.sessions.clone().with_upscaled_event_buffer(peers_config.max_peers()),
+                SessionsConfig::default().with_upscaled_event_buffer(peers_config.max_peers()),
            )
            .peer_config(peers_config)
            .boot_nodes(chain_bootnodes.clone())
--- a/crates/node/ethstats/src/ethstats.rs
+++ b/crates/node/ethstats/src/ethstats.rs
@@ -3,8 +3,8 @@ use crate::{
    credentials::EthstatsCredentials,
    error::EthStatsError,
    events::{
-        AuthMsg, BlockMsg, BlockStats, HistoryMsg, LatencyMsg, NodeInfo, NodeStats, PayloadMsg,
-        PayloadStats, PendingMsg, PendingStats, PingMsg, StatsMsg, TxStats, UncleStats,
+        AuthMsg, BlockMsg, BlockStats, HistoryMsg, LatencyMsg, NodeInfo, NodeStats, PendingMsg,
+        PendingStats, PingMsg, StatsMsg, TxStats, UncleStats,
    },
 };
 use alloy_consensus::{BlockHeader, Sealable};
@@ -50,7 +50,7 @@ const READ_TIMEOUT: Duration = Duration::from_secs(30);
 /// authentication, stats reporting, block notifications, and connection management.
 /// It maintains a persistent `WebSocket` connection and automatically reconnects
 /// when the connection is lost.
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct EthStatsService<Network, Provider, Pool> {
    /// Authentication credentials for the `EthStats` server
    credentials: EthstatsCredentials,
@@ -208,7 +208,7 @@ where
                active: true,
                syncing: self.network.is_syncing(),
                peers: self.network.num_connected_peers() as u64,
-                gas_price: self.pool.block_info().pending_basefee,
+                gas_price: 0, // TODO
                uptime: 100,
            },
        };
@@ -347,42 +347,6 @@ where
        Ok(())
    }

-    /// Report new payload information to the `EthStats` server
-    ///
-    /// Sends information about payload processing time and block details
-    /// to the server for monitoring purposes.
-    pub async fn report_new_payload(
-        &self,
-        block_hash: alloy_primitives::B256,
-        block_number: u64,
-        processing_time: Duration,
-    ) -> Result<(), EthStatsError> {
-        let conn = self.conn.read().await;
-        let conn = conn.as_ref().ok_or(EthStatsError::NotConnected)?;
-
-        let payload_stats = PayloadStats {
-            number: U256::from(block_number),
-            hash: block_hash,
-            processing_time: processing_time.as_millis() as u64,
-        };
-
-        let payload_msg =
-            PayloadMsg { id: self.credentials.node_id.clone(), payload: payload_stats };
-
-        debug!(
-            target: "ethstats",
-            "Reporting new payload: block={}, hash={:?}, processing_time={}ms",
-            block_number,
-            block_hash,
-            processing_time.as_millis()
-        );
-
-        let message = payload_msg.generate_new_payload_message();
-        conn.write_json(&message).await?;
-
-        Ok(())
-    }
-
    /// Convert a block to `EthStats` block statistics format
    ///
    /// Extracts relevant information from a block and formats it according
--- a/crates/node/ethstats/src/events.rs
+++ b/crates/node/ethstats/src/events.rs
@@ -281,66 +281,3 @@ impl PingMsg {
        .to_string()
    }
 }
-
-/// Information reported about a new payload processing event.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct PayloadStats {
-    /// Block number of the payload
-    pub number: U256,
-
-    /// Hash of the payload block
-    pub hash: B256,
-
-    /// Time taken to validate the payload in milliseconds
-    #[serde(rename = "processingTime")]
-    pub processing_time: u64,
-}
-
-/// Message containing new payload information to be reported to the ethstats monitoring server.
-#[derive(Debug, Serialize, Deserialize)]
-pub struct PayloadMsg {
-    /// The node's unique identifier
-    pub id: String,
-
-    /// The payload information to report
-    pub payload: PayloadStats,
-}
-
-impl PayloadMsg {
-    /// Generate a new payload message for the ethstats monitoring server.
-    pub fn generate_new_payload_message(&self) -> String {
-        serde_json::json!({
-            "emit": ["new-payload", self]
-        })
-        .to_string()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use alloy_primitives::{B256, U256};
-
-    #[test]
-    fn test_payload_msg_generation() {
-        let payload_stats = PayloadStats {
-            number: U256::from(12345),
-            hash: B256::from_slice(&[1u8; 32]),
-            processing_time: 150,
-        };
-
-        let payload_msg = PayloadMsg { id: "test-node".to_string(), payload: payload_stats };
-
-        let message = payload_msg.generate_new_payload_message();
-        let parsed: serde_json::Value = serde_json::from_str(&message).expect("Valid JSON");
-
-        assert_eq!(parsed["emit"][0], "new-payload");
-        assert_eq!(parsed["emit"][1]["id"], "test-node");
-        assert_eq!(parsed["emit"][1]["payload"]["number"], "0x3039"); // 12345 in hex
-        assert_eq!(parsed["emit"][1]["payload"]["processingTime"], 150);
-
-        // Verify the structure contains all expected fields
-        assert!(parsed["emit"][1]["payload"]["hash"].is_string());
-        assert!(parsed["emit"][1]["payload"]["processingTime"].is_number());
-    }
-}
--- a/crates/node/metrics/src/recorder.rs
+++ b/crates/node/metrics/src/recorder.rs
@@ -3,7 +3,7 @@
 use eyre::WrapErr;
 use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
 use metrics_util::layers::{PrefixLayer, Stack};
-use std::sync::{atomic::AtomicBool, OnceLock};
+use std::sync::{atomic::AtomicBool, LazyLock};

 /// Installs the Prometheus recorder as the global recorder.
 ///
@@ -11,43 +11,14 @@ use std::sync::{atomic::AtomicBool, OnceLock};
 ///
 /// Caution: This only configures the global recorder and does not spawn the exporter.
 /// Callers must run [`PrometheusRecorder::spawn_upkeep`] manually.
-///
-/// Use [`init_prometheus_recorder`] to install a custom recorder.
 pub fn install_prometheus_recorder() -> &'static PrometheusRecorder {
-    PROMETHEUS_RECORDER_HANDLE.get_or_init(|| {
-        PrometheusRecorder::install().expect("Failed to install Prometheus recorder")
-    })
-}
-
-/// Installs the provided recorder as the global recorder.
-///
-/// To customize the builder, first construct a recorder with
-/// [`PrometheusRecorder::install_with_builder`], then pass it here.
-///
-/// # Panics
-///
-/// Panics if a recorder has already been installed.
-pub fn init_prometheus_recorder(recorder: PrometheusRecorder) -> &'static PrometheusRecorder {
-    PROMETHEUS_RECORDER_HANDLE.set(recorder).expect("Prometheus recorder already installed");
-    PROMETHEUS_RECORDER_HANDLE.get().expect("Prometheus recorder is set")
+    &PROMETHEUS_RECORDER_HANDLE
 }

 /// The default Prometheus recorder handle. We use a global static to ensure that it is only
 /// installed once.
-static PROMETHEUS_RECORDER_HANDLE: OnceLock<PrometheusRecorder> = OnceLock::new();
-
-/// Installs the Prometheus recorder with a custom builder.
-///
-/// Returns an error if a recorder has already been installed.
-pub fn try_install_prometheus_recorder_with_builder(
-    builder: PrometheusBuilder,
-) -> eyre::Result<&'static PrometheusRecorder> {
-    let recorder = PrometheusRecorder::install_with_builder(builder)?;
-    PROMETHEUS_RECORDER_HANDLE
-        .set(recorder)
-        .map_err(|_| eyre::eyre!("Prometheus recorder already installed"))?;
-    Ok(PROMETHEUS_RECORDER_HANDLE.get().expect("recorder is set"))
-}
+static PROMETHEUS_RECORDER_HANDLE: LazyLock<PrometheusRecorder> =
+    LazyLock::new(|| PrometheusRecorder::install().unwrap());

 /// A handle to the Prometheus recorder.
 ///
@@ -104,15 +75,7 @@ impl PrometheusRecorder {
    /// Caution: This only configures the global recorder and does not spawn the exporter.
    /// Callers must run [`Self::spawn_upkeep`] manually.
    pub fn install() -> eyre::Result<Self> {
-        Self::install_with_builder(PrometheusBuilder::new())
-    }
-
-    /// Installs Prometheus as the metrics recorder with a custom builder.
-    ///
-    /// Caution: This only configures the global recorder and does not spawn the exporter.
-    /// Callers must run [`Self::spawn_upkeep`] manually.
-    pub fn install_with_builder(builder: PrometheusBuilder) -> eyre::Result<Self> {
-        let recorder = builder.build_recorder();
+        let recorder = PrometheusBuilder::new().build_recorder();
        let handle = recorder.handle();

        // Build metrics stack
@@ -135,13 +98,14 @@ mod tests {
    // `metrics-exporter-prometheus` dependency version.
    #[test]
    fn process_metrics() {
-        let recorder = install_prometheus_recorder();
+        // initialize the lazy handle
+        let _ = &*PROMETHEUS_RECORDER_HANDLE;

        let process = metrics_process::Collector::default();
        process.describe();
        process.collect();

-        let metrics = recorder.handle().render();
+        let metrics = PROMETHEUS_RECORDER_HANDLE.handle.render();
        assert!(metrics.contains("process_cpu_seconds_total"), "{metrics:?}");
    }
 }
--- a/crates/optimism/bin/Cargo.toml
+++ b/crates/optimism/bin/Cargo.toml
@@ -27,7 +27,7 @@ tracing.workspace = true
 workspace = true

 [features]
-default = ["jemalloc", "otlp", "reth-optimism-evm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
+default = ["jemalloc", "otlp", "reth-optimism-evm/portable", "js-tracer", "keccak-cache-global"]

 otlp = ["reth-optimism-cli/otlp"]

--- a/crates/optimism/chainspec/src/superchain/chain_spec_macro.rs
+++ b/crates/optimism/chainspec/src/superchain/chain_spec_macro.rs
@@ -6,7 +6,7 @@ macro_rules! create_chain_spec {
            /// The Optimism $name $environment spec
            pub static [<$name:upper _ $environment:upper>]: $crate::LazyLock<alloc::sync::Arc<$crate::OpChainSpec>> = $crate::LazyLock::new(|| {
                $crate::OpChainSpec::from_genesis($crate::superchain::configs::read_superchain_genesis($name, $environment)
-                    .unwrap_or_else(|e| panic!("Can't read {}-{} genesis: {}", $name, $environment, e)))
+                    .expect(&alloc::format!("Can't read {}-{} genesis", $name, $environment)))
                    .into()
            });
        }
--- a/crates/optimism/cli/src/app.rs
+++ b/crates/optimism/cli/src/app.rs
@@ -67,7 +67,7 @@ where
        self.init_tracing(&runner)?;

        // Install the prometheus recorder to be sure to record all metrics
-        install_prometheus_recorder();
+        let _ = install_prometheus_recorder();

        let components = |spec: Arc<OpChainSpec>| {
            (OpExecutorProvider::optimism(spec.clone()), Arc::new(OpBeaconConsensus::new(spec)))
--- a/crates/optimism/flashblocks/src/consensus.rs
+++ b/crates/optimism/flashblocks/src/consensus.rs
@@ -41,9 +41,9 @@ where
    /// Attempts to submit a new payload to the engine.
    ///
    /// The `TryFrom` conversion will fail if `execution_outcome.state_root` is `B256::ZERO`,
-    /// in which case this method uses the `parent_hash` instead to drive the chain forward.
+    /// in which case this returns the `parent_hash` instead to drive the chain forward.
    ///
-    /// Returns the block hash to use for FCU (either the new block's hash or the parent hash).
+    /// Returns the block hash to use for FCU (either the new block or parent).
    async fn submit_new_payload(&self, sequence: &FlashBlockCompleteSequence) -> B256 {
        let payload = match P::ExecutionData::try_from(sequence) {
            Ok(payload) => payload,
--- a/Show More
+++ b/Show More