fix: update gas_used/gas_refunded to ResultGas API

# Conflicts:
#	Cargo.lock
#	crates/optimism/evm/src/lib.rs

.changelog/fast-seals-play.md

@@ -1,5 +0,0 @@
----
-reth-transaction-pool: minor
----
-
-Added `consensus_ref` method to `PoolTransaction` trait for borrowing consensus transactions without cloning.

.changelog/fast-waves-smile.md

@@ -1,4 +0,0 @@
----
----
-
-Improved nightly Docker build failure Slack notification with more detailed formatting and context.

.changelog/lively-clouds-bake.md

@@ -1,4 +0,0 @@
----
----
-
-Improved documentation overview page with better structure and clarity.

.changelog/lively-foxes-play.md

@@ -1,5 +0,0 @@
----
-reth-node-events: patch
----
-
-Updated consensus engine log message to be more accurate about received updates.

.changelog/merry-koalas-nod.md

@@ -1,6 +0,0 @@
----
-reth-rpc-eth-api: minor
-reth-rpc-server-types: minor
----
-
-Added `eth_getStorageValues` RPC method for batch storage slot retrieval across multiple addresses.

.changelog/nice-trees-drink.md

@@ -1,9 +0,0 @@
----
-reth-network-api: minor
-reth-network-types: minor
-reth-network: minor
-reth-node-core: minor
-reth: minor
----
-
-Added optional ENR fork ID enforcement to filter out peers from incompatible networks during peer discovery, controlled by the `--enforce-enr-fork-id` CLI flag.

.changelog/proud-wolves-spin.md

@@ -1,5 +0,0 @@
----
-reth-storage-api: patch
----
-
-Added `Arc` to `auto_impl` derive for storage-api traits to support automatic `Arc` wrapper implementations.

.github/CODEOWNERS

@@ -38,7 +38,7 @@ crates/storage/libmdbx-rs/  @shekhirin
 crates/storage/nippy-jar/   @joshieDo @shekhirin
 crates/storage/provider/    @joshieDo @shekhirin @yongkangc
 crates/storage/storage-api/ @joshieDo
-crates/tasks/               @mattsse @DaniPopes
+crates/tasks/               @mattsse
 crates/tokio-util/          @mattsse 
 crates/tracing/             @mattsse @shekhirin
 crates/tracing-otlp/        @mattsse @Rjected 

.github/workflows/docker.yml

@@ -70,27 +70,18 @@ jobs:
             # Add 'latest' tag for non-RC releases
             if [[ ! "$VERSION" =~ -rc ]]; then
               echo "ethereum_tags=${REGISTRY}/reth:${VERSION},${REGISTRY}/reth:latest" >> "$GITHUB_OUTPUT"
-              {
-                echo "ethereum_set<<EOF"
-                echo "ethereum.tags=${REGISTRY}/reth:${VERSION}"
-                echo "ethereum.tags=${REGISTRY}/reth:latest"
-                echo "EOF"
-              } >> "$GITHUB_OUTPUT"
             else
               echo "ethereum_tags=${REGISTRY}/reth:${VERSION}" >> "$GITHUB_OUTPUT"
-              echo "ethereum_set=ethereum.tags=${REGISTRY}/reth:${VERSION}" >> "$GITHUB_OUTPUT"
             fi
 
           elif [[ "${{ github.event_name }}" == "schedule" ]] || [[ "${{ inputs.build_type }}" == "nightly" ]]; then
             echo "targets=nightly" >> "$GITHUB_OUTPUT"
             echo "ethereum_tags=${REGISTRY}/reth:nightly" >> "$GITHUB_OUTPUT"
-            echo "ethereum_set=ethereum.tags=${REGISTRY}/reth:nightly" >> "$GITHUB_OUTPUT"
 
           else
             # git-sha build
             echo "targets=ethereum" >> "$GITHUB_OUTPUT"
             echo "ethereum_tags=${REGISTRY}/reth:${{ github.sha }}" >> "$GITHUB_OUTPUT"
-            echo "ethereum_set=ethereum.tags=${REGISTRY}/reth:${{ github.sha }}" >> "$GITHUB_OUTPUT"
           fi
 
       - name: Build and push images
@@ -106,7 +97,7 @@ jobs:
           targets: ${{ steps.params.outputs.targets }}
           push: ${{ !(github.event_name == 'workflow_dispatch' && inputs.dry_run) }}
           set: |
-            ${{ steps.params.outputs.ethereum_set }}
+            ethereum.tags=${{ steps.params.outputs.ethereum_tags }}
 
       - name: Verify image architectures
         env:
@@ -126,18 +117,6 @@ jobs:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2
         env:
-          SLACK_COLOR: danger
-          SLACK_ICON_EMOJI: ":rotating_light:"
-          SLACK_USERNAME: "GitHub Actions"
-          SLACK_TITLE: ":rotating_light: Nightly Docker Build Failed"
-          SLACK_MESSAGE: |
-            The scheduled nightly Docker build failed.
-
-            *Commit:* `${{ github.sha }}`
-            *Branch:* `${{ github.ref_name }}`
-            *Run:* <https://github.com/paradigmxyz/reth/actions/runs/${{ github.run_id }}|View logs>
-
-            *Action required:* Re-run the workflow or investigate the build failure.
-          SLACK_FOOTER: "paradigmxyz/reth · docker.yml"
-          MSG_MINIMAL: true
+          SLACK_COLOR: ${{ job.status }}
+          SLACK_MESSAGE: "Failed run: https://github.com/paradigmxyz/reth/actions/runs/${{ github.run_id }}"
           SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK_URL }}

.github/workflows/lint.yml

@@ -124,7 +124,7 @@ jobs:
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@master
         with:
-          toolchain: "1.93" # MSRV
+          toolchain: "1.88" # MSRV
       - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:

Cargo.toml

@@ -1,7 +1,7 @@
 [workspace.package]
-version = "1.11.0"
+version = "1.10.2"
 edition = "2024"
-rust-version = "1.93"
+rust-version = "1.88"
 license = "MIT OR Apache-2.0"
 homepage = "https://paradigmxyz.github.io/reth"
 repository = "https://github.com/paradigmxyz/reth"
@@ -27,6 +27,7 @@ members = [
     "crates/engine/invalid-block-hooks/",
     "crates/engine/local",
     "crates/engine/primitives/",
+    "crates/engine/service",
     "crates/engine/tree/",
     "crates/engine/util/",
     "crates/era",
@@ -348,6 +349,7 @@ reth-ecies = { path = "crates/net/ecies" }
 reth-engine-local = { path = "crates/engine/local" }
 reth-engine-primitives = { path = "crates/engine/primitives", default-features = false }
 reth-engine-tree = { path = "crates/engine/tree" }
+reth-engine-service = { path = "crates/engine/service" }
 reth-engine-util = { path = "crates/engine/util" }
 reth-era = { path = "crates/era" }
 reth-era-downloader = { path = "crates/era-downloader" }
@@ -663,7 +665,6 @@ cipher = "0.4.3"
 comfy-table = "7.0"
 concat-kdf = "0.1.0"
 crossbeam-channel = "0.5.13"
-crossbeam-utils = "0.8"
 crossterm = "0.29.0"
 csv = "1.3.0"
 ctrlc = "3.4"
@@ -710,6 +711,28 @@ vergen-git2 = "9.1.0"
 ipnet = "2.11"
 
 [patch.crates-io]
+# revm staging patches
+revm = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+op-revm = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-bytecode = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-context = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-context-interface = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-database = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-database-interface = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-handler = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-inspector = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-interpreter = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-precompile = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-primitives = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+revm-state = { git = "https://github.com/bluealloy/revm", rev = "33330a285e621b9170c30a21cfea9ab32e2a2169" }
+
+# revm-inspectors staging patch
+revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "e80e2eab72dfa18011e6a99abd37027290a46e83" }
+
+# alloy-evm staging patches
+alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "742dc14749ea0279c03ca27b1c26f26ac19fbefb" }
+alloy-op-evm = { git = "https://github.com/alloy-rs/evm", rev = "742dc14749ea0279c03ca27b1c26f26ac19fbefb" }
+
 # alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
 # alloy-contract = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
 # alloy-eips = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }

Cross.toml

@@ -19,11 +19,10 @@ pre-build = [
 image = "ubuntu:24.04"
 pre-build = [
     "apt update",
-    "apt install --yes gcc gcc-riscv64-linux-gnu g++-riscv64-linux-gnu libclang-dev make",
+    "apt install --yes gcc gcc-riscv64-linux-gnu libclang-dev make",
 ]
 env.passthrough = [
     "CARGO_TARGET_RISCV64GC_UNKNOWN_LINUX_GNU_LINKER=riscv64-linux-gnu-gcc",
-    "CXX_riscv64gc_unknown_linux_gnu=riscv64-linux-gnu-g++",
 ]
 
 [build.env]

README.md

@@ -93,7 +93,7 @@ When updating this, also update:
 - .github/workflows/lint.yml
 -->
 
-The Minimum Supported Rust Version (MSRV) of this project is [1.93.0](https://blog.rust-lang.org/2026/01/22/Rust-1.93.0/).
+The Minimum Supported Rust Version (MSRV) of this project is [1.88.0](https://blog.rust-lang.org/2025/06/26/Rust-1.88.0/).
 
 See the docs for detailed instructions on how to [build from source](https://reth.rs/installation/source/).
 

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

@@ -192,15 +192,6 @@ impl Command {
             parent_header = block.header;
             parent_hash = block_hash;
             blocks_processed += 1;
-
-            let progress = match mode {
-                RampMode::Blocks(total) => format!("{blocks_processed}/{total}"),
-                RampMode::TargetGasLimit(target) => {
-                    let pct = (parent_header.gas_limit as f64 / target as f64 * 100.0).min(100.0);
-                    format!("{pct:.1}%")
-                }
-            };
-            info!(target: "reth-bench", progress, block_number = parent_header.number, gas_limit = parent_header.gas_limit, "Block processed");
         }
 
         let final_gas_limit = parent_header.gas_limit;

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

@@ -153,7 +153,6 @@ impl Command {
             ..
         } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
 
-        let total_blocks = benchmark_mode.total_blocks();
         let buffer_size = self.rpc_block_buffer_size;
 
         // Use a oneshot channel to propagate errors from the spawned task
@@ -207,7 +206,6 @@ impl Command {
         });
 
         let mut results = Vec::new();
-        let mut blocks_processed = 0u64;
         let total_benchmark_duration = Instant::now();
         let mut total_wait_time = Duration::ZERO;
 
@@ -251,13 +249,8 @@ impl Command {
 
             // Exclude time spent waiting on the block prefetch channel from the benchmark duration.
             // We want to measure engine throughput, not RPC fetch latency.
-            blocks_processed += 1;
             let current_duration = total_benchmark_duration.elapsed() - total_wait_time;
-            let progress = match total_blocks {
-                Some(total) => format!("{blocks_processed}/{total}"),
-                None => format!("{blocks_processed}"),
-            };
-            info!(target: "reth-bench", progress, %combined_result);
+            info!(target: "reth-bench", %combined_result);
 
             if let Some(w) = &mut waiter {
                 w.on_block(block_number).await?;

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

@@ -52,7 +52,6 @@ impl Command {
             ..
         } = BenchContext::new(&self.benchmark, self.rpc_url).await?;
 
-        let total_blocks = benchmark_mode.total_blocks();
         let buffer_size = self.rpc_block_buffer_size;
 
         // Use a oneshot channel to propagate errors from the spawned task
@@ -83,8 +82,8 @@ impl Command {
             }
         });
 
+        // put results in a summary vec so they can be printed at the end
         let mut results = Vec::new();
-        let mut blocks_processed = 0u64;
         let total_benchmark_duration = Instant::now();
         let mut total_wait_time = Duration::ZERO;
 
@@ -106,12 +105,7 @@ impl Command {
             call_new_payload(&auth_provider, version, params).await?;
 
             let new_payload_result = NewPayloadResult { gas_used, latency: start.elapsed() };
-            blocks_processed += 1;
-            let progress = match total_blocks {
-                Some(total) => format!("{blocks_processed}/{total}"),
-                None => format!("{blocks_processed}"),
-            };
-            info!(target: "reth-bench", progress, %new_payload_result);
+            info!(target: "reth-bench", %new_payload_result);
 
             // current duration since the start of the benchmark minus the time
             // waiting for blocks

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

@@ -341,8 +341,7 @@ impl Command {
             };
 
             let current_duration = total_benchmark_duration.elapsed();
-            let progress = format!("{}/{}", i + 1, payloads.len());
-            info!(target: "reth-bench", progress, %combined_result);
+            info!(target: "reth-bench", %combined_result);
 
             if let Some(w) = &mut waiter {
                 w.on_block(block_number).await?;

bin/reth-bench/src/bench_mode.rs

@@ -20,19 +20,6 @@ impl BenchMode {
         }
     }
 
-    /// Returns the total number of blocks in the benchmark, if known.
-    ///
-    /// For [`BenchMode::Range`] this is the length of the range.
-    /// For [`BenchMode::Continuous`] the total is unbounded, so `None` is returned.
-    pub const fn total_blocks(&self) -> Option<u64> {
-        match self {
-            Self::Continuous(_) => None,
-            Self::Range(range) => {
-                Some(range.end().saturating_sub(*range.start()).saturating_add(1))
-            }
-        }
-    }
-
     /// 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> {
         // If neither `--from` nor `--to` are provided, we will run the benchmark continuously,

crates/chain-state/src/in_memory.rs

@@ -1061,14 +1061,6 @@ mod tests {
         ) -> ProviderResult<Option<StorageValue>> {
             Ok(None)
         }
-
-        fn storage_by_hashed_key(
-            &self,
-            _address: Address,
-            _hashed_storage_key: StorageKey,
-        ) -> ProviderResult<Option<StorageValue>> {
-            Ok(None)
-        }
     }
 
     impl BytecodeReader for MockStateProvider {

crates/chain-state/src/memory_overlay.rs

@@ -223,26 +223,6 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {
 
         self.historical.storage(address, storage_key)
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        address: Address,
-        hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<StorageValue>> {
-        let hashed_address = keccak256(address);
-        let state = &self.trie_input().state;
-
-        if let Some(hs) = state.storages.get(&hashed_address) {
-            if let Some(value) = hs.storage.get(&hashed_storage_key) {
-                return Ok(Some(*value));
-            }
-            if hs.wiped {
-                return Ok(Some(StorageValue::ZERO));
-            }
-        }
-
-        self.historical.storage_by_hashed_key(address, hashed_storage_key)
-    }
 }
 
 impl<N: NodePrimitives> BytecodeReader for MemoryOverlayStateProviderRef<'_, N> {

crates/cli/commands/src/common.rs

@@ -19,7 +19,7 @@ use reth_node_builder::{
     Node, NodeComponents, NodeComponentsBuilder, NodeTypes, NodeTypesWithDBAdapter,
 };
 use reth_node_core::{
-    args::{DatabaseArgs, DatadirArgs, StaticFilesArgs, StorageArgs},
+    args::{DatabaseArgs, DatadirArgs, RocksDbArgs, StaticFilesArgs, StorageArgs},
     dirs::{ChainPath, DataDirPath},
 };
 use reth_provider::{
@@ -67,23 +67,62 @@ pub struct EnvironmentArgs<C: ChainSpecParser> {
     #[command(flatten)]
     pub static_files: StaticFilesArgs,
 
+    /// All `RocksDB` related arguments
+    #[command(flatten)]
+    pub rocksdb: RocksDbArgs,
+
     /// Storage mode configuration (v2 vs v1/legacy)
     #[command(flatten)]
     pub storage: StorageArgs,
 }
 
 impl<C: ChainSpecParser> EnvironmentArgs<C> {
-    /// Returns the effective storage settings derived from `--storage.v2`.
+    /// Returns the effective storage settings derived from `--storage.v2`, static-file, and
+    /// `RocksDB` CLI args.
     ///
     /// The base storage mode is determined by `--storage.v2`:
     /// - When `--storage.v2` is set: uses [`StorageSettings::v2()`] defaults
-    /// - Otherwise: uses [`StorageSettings::base()`] defaults
+    /// - Otherwise: uses [`StorageSettings::v1()`] defaults
+    ///
+    /// Individual `--static-files.*` and `--rocksdb.*` flags override the base when explicitly set.
     pub fn storage_settings(&self) -> StorageSettings {
-        if self.storage.v2 {
-            StorageSettings::v2()
-        } else {
-            StorageSettings::base()
+        let mut s = if self.storage.v2 { StorageSettings::v2() } else { StorageSettings::base() };
+
+        // Apply static files overrides (only when explicitly set)
+        if let Some(v) = self.static_files.receipts {
+            s = s.with_receipts_in_static_files(v);
         }
+        if let Some(v) = self.static_files.transaction_senders {
+            s = s.with_transaction_senders_in_static_files(v);
+        }
+        if let Some(v) = self.static_files.account_changesets {
+            s = s.with_account_changesets_in_static_files(v);
+        }
+        if let Some(v) = self.static_files.storage_changesets {
+            s = s.with_storage_changesets_in_static_files(v);
+        }
+
+        // Apply rocksdb overrides
+        // --rocksdb.all sets all rocksdb flags to true
+        if self.rocksdb.all {
+            s = s
+                .with_transaction_hash_numbers_in_rocksdb(true)
+                .with_storages_history_in_rocksdb(true)
+                .with_account_history_in_rocksdb(true);
+        }
+
+        // Individual rocksdb flags override --rocksdb.all when explicitly set
+        if let Some(v) = self.rocksdb.tx_hash {
+            s = s.with_transaction_hash_numbers_in_rocksdb(v);
+        }
+        if let Some(v) = self.rocksdb.storages_history {
+            s = s.with_storages_history_in_rocksdb(v);
+        }
+        if let Some(v) = self.rocksdb.account_history {
+            s = s.with_account_history_in_rocksdb(v);
+        }
+
+        s
     }
 
     /// Initializes environment according to [`AccessRights`] and returns an instance of

crates/cli/commands/src/db/account_storage.rs

@@ -5,7 +5,6 @@ use reth_codecs::Compact;
 use reth_db_api::{cursor::DbDupCursorRO, database::Database, tables, transaction::DbTx};
 use reth_db_common::DbTool;
 use reth_node_builder::NodeTypesWithDB;
-use reth_storage_api::StorageSettingsCache;
 use std::time::{Duration, Instant};
 use tracing::info;
 
@@ -23,94 +22,52 @@ impl Command {
     /// Execute `db account-storage` command
     pub fn execute<N: NodeTypesWithDB>(self, tool: &DbTool<N>) -> eyre::Result<()> {
         let address = self.address;
-        let use_hashed_state = tool.provider_factory.cached_storage_settings().use_hashed_state();
+        let (slot_count, plain_size) = tool.provider_factory.db_ref().view(|tx| {
+            let mut cursor = tx.cursor_dup_read::<tables::PlainStorageState>()?;
+            let mut count = 0usize;
+            let mut total_value_bytes = 0usize;
+            let mut last_log = Instant::now();
 
-        let (slot_count, storage_size) = if use_hashed_state {
-            let hashed_address = keccak256(address);
-            tool.provider_factory.db_ref().view(|tx| {
-                let mut cursor = tx.cursor_dup_read::<tables::HashedStorages>()?;
-                let mut count = 0usize;
-                let mut total_value_bytes = 0usize;
-                let mut last_log = Instant::now();
+            // Walk all storage entries for this address
+            let walker = cursor.walk_dup(Some(address), None)?;
+            for entry in walker {
+                let (_, storage_entry) = entry?;
+                count += 1;
+                // StorageEntry encodes as: 32 bytes (key/subkey uncompressed) + compressed U256
+                let mut buf = Vec::new();
+                let entry_len = storage_entry.to_compact(&mut buf);
+                total_value_bytes += entry_len;
 
-                let walker = cursor.walk_dup(Some(hashed_address), None)?;
-                for entry in walker {
-                    let (_, storage_entry) = entry?;
-                    count += 1;
-                    let mut buf = Vec::new();
-                    let entry_len = storage_entry.to_compact(&mut buf);
-                    total_value_bytes += entry_len;
-
-                    if last_log.elapsed() >= LOG_INTERVAL {
-                        info!(
-                            target: "reth::cli",
-                            address = %address,
-                            slots = count,
-                            key = %storage_entry.key,
-                            "Processing hashed storage slots"
-                        );
-                        last_log = Instant::now();
-                    }
+                if last_log.elapsed() >= LOG_INTERVAL {
+                    info!(
+                        target: "reth::cli",
+                        address = %address,
+                        slots = count,
+                        key = %storage_entry.key,
+                        "Processing storage slots"
+                    );
+                    last_log = Instant::now();
                 }
+            }
 
-                let total_size = if count > 0 { 32 + total_value_bytes } else { 0 };
+            // Add 20 bytes for the Address key (stored once per account in dupsort)
+            let total_size = if count > 0 { 20 + total_value_bytes } else { 0 };
 
-                Ok::<_, eyre::Report>((count, total_size))
-            })??
-        } else {
-            tool.provider_factory.db_ref().view(|tx| {
-                let mut cursor = tx.cursor_dup_read::<tables::PlainStorageState>()?;
-                let mut count = 0usize;
-                let mut total_value_bytes = 0usize;
-                let mut last_log = Instant::now();
+            Ok::<_, eyre::Report>((count, total_size))
+        })??;
 
-                // Walk all storage entries for this address
-                let walker = cursor.walk_dup(Some(address), None)?;
-                for entry in walker {
-                    let (_, storage_entry) = entry?;
-                    count += 1;
-                    let mut buf = Vec::new();
-                    // StorageEntry encodes as: 32 bytes (key/subkey uncompressed) + compressed U256
-                    let entry_len = storage_entry.to_compact(&mut buf);
-                    total_value_bytes += entry_len;
-
-                    if last_log.elapsed() >= LOG_INTERVAL {
-                        info!(
-                            target: "reth::cli",
-                            address = %address,
-                            slots = count,
-                            key = %storage_entry.key,
-                            "Processing storage slots"
-                        );
-                        last_log = Instant::now();
-                    }
-                }
-
-                // Add 20 bytes for the Address key (stored once per account in dupsort)
-                let total_size = if count > 0 { 20 + total_value_bytes } else { 0 };
-
-                Ok::<_, eyre::Report>((count, total_size))
-            })??
-        };
+        // Estimate hashed storage size: 32-byte B256 key instead of 20-byte Address
+        let hashed_size_estimate = if slot_count > 0 { plain_size + 12 } else { 0 };
+        let total_estimate = plain_size + hashed_size_estimate;
 
         let hashed_address = keccak256(address);
 
         println!("Account: {address}");
         println!("Hashed address: {hashed_address}");
         println!("Storage slots: {slot_count}");
-        if use_hashed_state {
-            println!("Hashed storage size: {} (estimated)", human_bytes(storage_size as f64));
-        } else {
-            // Estimate hashed storage size: 32-byte B256 key instead of 20-byte Address
-            let hashed_size_estimate = if slot_count > 0 { storage_size + 12 } else { 0 };
-            let total_estimate = storage_size + hashed_size_estimate;
-            println!("Plain storage size: {} (estimated)", human_bytes(storage_size as f64));
-            println!(
-                "Hashed storage size: {} (estimated)",
-                human_bytes(hashed_size_estimate as f64)
-            );
-            println!("Total estimated size: {}", human_bytes(total_estimate as f64));
-        }
+        println!("Plain storage size: {} (estimated)", human_bytes(plain_size as f64));
+        println!("Hashed storage size: {} (estimated)", human_bytes(hashed_size_estimate as f64));
+        println!("Total estimated size: {}", human_bytes(total_estimate as f64));
 
         Ok(())
     }

crates/cli/commands/src/db/copy.rs

@@ -1,61 +0,0 @@
-use clap::Parser;
-use reth_db::mdbx::{self, ffi};
-use std::path::PathBuf;
-
-/// Copies the MDBX database to a new location.
-///
-/// Equivalent to the standalone `mdbx_copy` tool but bundled into reth.
-#[derive(Parser, Debug)]
-pub struct Command {
-    /// Destination path for the database copy.
-    dest: PathBuf,
-
-    /// Compact the database while copying (reclaims free space).
-    #[arg(short, long)]
-    compact: bool,
-
-    /// Force dynamic size for the destination database.
-    #[arg(short = 'd', long)]
-    force_dynamic_size: bool,
-
-    /// Throttle to avoid MVCC pressure on writers.
-    #[arg(short = 'p', long)]
-    throttle_mvcc: bool,
-}
-
-impl Command {
-    /// Execute `db copy` command
-    pub fn execute(self, db: &mdbx::DatabaseEnv) -> eyre::Result<()> {
-        let mut flags: ffi::MDBX_copy_flags_t = ffi::MDBX_CP_DEFAULTS;
-        if self.compact {
-            flags |= ffi::MDBX_CP_COMPACT;
-        }
-        if self.force_dynamic_size {
-            flags |= ffi::MDBX_CP_FORCE_DYNAMIC_SIZE;
-        }
-        if self.throttle_mvcc {
-            flags |= ffi::MDBX_CP_THROTTLE_MVCC;
-        }
-
-        let dest = self
-            .dest
-            .to_str()
-            .ok_or_else(|| eyre::eyre!("destination path must be valid UTF-8"))?;
-        let dest_cstr = std::ffi::CString::new(dest)?;
-
-        println!("Copying database to {} ...", self.dest.display());
-
-        let rc = db.with_raw_env_ptr(|env_ptr| unsafe {
-            ffi::mdbx_env_copy(env_ptr, dest_cstr.as_ptr(), flags)
-        });
-
-        if rc != 0 {
-            eyre::bail!("mdbx_env_copy failed with error code {rc}: {}", unsafe {
-                std::ffi::CStr::from_ptr(ffi::mdbx_strerror(rc)).to_string_lossy()
-            });
-        }
-
-        println!("Done.");
-        Ok(())
-    }
-}

crates/cli/commands/src/db/get.rs

@@ -98,8 +98,7 @@ impl Command {
                         )?;
 
                         if let Some(entry) = entry {
-                            let se: reth_primitives_traits::StorageEntry = entry.into();
-                            println!("{}", serde_json::to_string_pretty(&se)?);
+                            println!("{}", serde_json::to_string_pretty(&entry)?);
                         } else {
                             error!(target: "reth::cli", "No content for the given table key.");
                         }
@@ -107,14 +106,7 @@ impl Command {
                     }
 
                     let changesets = provider.storage_changeset(key.block_number())?;
-                    let serializable: Vec<_> = changesets
-                        .into_iter()
-                        .map(|(addr, entry)| {
-                            let se: reth_primitives_traits::StorageEntry = entry.into();
-                            (addr, se)
-                        })
-                        .collect();
-                    println!("{}", serde_json::to_string_pretty(&serializable)?);
+                    println!("{}", serde_json::to_string_pretty(&changesets)?);
                     return Ok(());
                 }
 

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

@@ -12,7 +12,6 @@ use std::{
 mod account_storage;
 mod checksum;
 mod clear;
-mod copy;
 mod diff;
 mod get;
 mod list;
@@ -43,8 +42,6 @@ pub enum Subcommands {
     List(list::Command),
     /// Calculates the content checksum of a table or static file segment
     Checksum(checksum::Command),
-    /// Copies the MDBX database to a new location (bundled mdbx_copy)
-    Copy(copy::Command),
     /// Create a diff between two database tables or two entire databases.
     Diff(diff::Command),
     /// Gets the content of a table for the given key
@@ -127,11 +124,6 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
                     command.execute(&tool)?;
                 });
             }
-            Subcommands::Copy(command) => {
-                db_exec!(self.env, tool, N, AccessRights::RO, {
-                    command.execute(tool.provider_factory.db_ref())?;
-                });
-            }
             Subcommands::Diff(command) => {
                 db_exec!(self.env, tool, N, AccessRights::RO, {
                     command.execute(&tool)?;

crates/cli/commands/src/db/repair_trie.rs

@@ -285,6 +285,7 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
                 // (We can't just use `upsert` method with a dup cursor, it's not properly
                 // supported)
                 let nibbles = StoredNibblesSubKey(path);
+                let entry = StorageTrieEntry { nibbles: nibbles.clone(), node };
                 if storage_trie_cursor
                     .seek_by_key_subkey(account, nibbles.clone())?
                     .filter(|v| v.nibbles == nibbles)
@@ -292,7 +293,6 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
                 {
                     storage_trie_cursor.delete_current()?;
                 }
-                let entry = StorageTrieEntry { nibbles, node };
                 storage_trie_cursor.upsert(account, &entry)?;
             }
             Output::Progress(path) => {

crates/cli/commands/src/db/settings.rs

@@ -39,12 +39,50 @@ enum Subcommands {
 #[derive(Debug, Clone, Copy, Subcommand)]
 #[clap(rename_all = "snake_case")]
 pub enum SetCommand {
-    /// Enable or disable v2 storage layout
+    /// Store receipts in static files instead of the database
+    Receipts {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store transaction senders in static files instead of the database
+    TransactionSenders {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store account changesets in static files instead of the database
+    AccountChangesets {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store storage history in rocksdb instead of MDBX
+    StoragesHistory {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store transaction hash to number mapping in rocksdb instead of MDBX
+    TransactionHashNumbers {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store account history in rocksdb instead of MDBX
+    AccountHistory {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store storage changesets in static files instead of the database
+    StorageChangesets {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Use hashed state tables (HashedAccounts/HashedStorages) as canonical state
     ///
-    /// When enabled, uses static files for receipts/senders/changesets and RocksDB for
-    /// history indices and transaction hashes. When disabled, uses v1/legacy layout (everything in
-    /// MDBX).
-    V2 {
+    /// When enabled, execution writes directly to hashed tables, eliminating need for
+    /// separate hashing stages. State reads come from hashed tables.
+    ///
+    /// WARNING: Changing this setting in either direction requires re-syncing the database.
+    /// Enabling on an existing plain-state database leaves hashed tables empty.
+    /// Disabling on an existing hashed-state database leaves plain tables empty.
+    UseHashedState {
         #[clap(action(ArgAction::Set))]
         value: bool,
     },
@@ -87,18 +125,87 @@ impl Command {
             println!("No storage settings found, creating new settings.");
         }
 
-        let mut settings @ StorageSettings { storage_v2: _ } =
-            settings.unwrap_or_else(StorageSettings::v1);
+        let mut settings @ StorageSettings {
+            receipts_in_static_files: _,
+            transaction_senders_in_static_files: _,
+            storages_history_in_rocksdb: _,
+            transaction_hash_numbers_in_rocksdb: _,
+            account_history_in_rocksdb: _,
+            account_changesets_in_static_files: _,
+            storage_changesets_in_static_files: _,
+            use_hashed_state: _,
+        } = settings.unwrap_or_else(StorageSettings::v1);
 
         // Update the setting based on the key
         match cmd {
-            SetCommand::V2 { value } => {
-                if settings.storage_v2 == value {
-                    println!("storage_v2 is already set to {}", value);
+            SetCommand::Receipts { value } => {
+                if settings.receipts_in_static_files == value {
+                    println!("receipts_in_static_files is already set to {}", value);
                     return Ok(());
                 }
-                settings.storage_v2 = value;
-                println!("Set storage_v2 = {}", value);
+                settings.receipts_in_static_files = value;
+                println!("Set receipts_in_static_files = {}", value);
+            }
+            SetCommand::TransactionSenders { value } => {
+                if settings.transaction_senders_in_static_files == value {
+                    println!("transaction_senders_in_static_files is already set to {}", value);
+                    return Ok(());
+                }
+                settings.transaction_senders_in_static_files = value;
+                println!("Set transaction_senders_in_static_files = {}", value);
+            }
+            SetCommand::AccountChangesets { value } => {
+                if settings.account_changesets_in_static_files == value {
+                    println!("account_changesets_in_static_files is already set to {}", value);
+                    return Ok(());
+                }
+                settings.account_changesets_in_static_files = value;
+                println!("Set account_changesets_in_static_files = {}", value);
+            }
+            SetCommand::StoragesHistory { value } => {
+                if settings.storages_history_in_rocksdb == value {
+                    println!("storages_history_in_rocksdb is already set to {}", value);
+                    return Ok(());
+                }
+                settings.storages_history_in_rocksdb = value;
+                println!("Set storages_history_in_rocksdb = {}", value);
+            }
+            SetCommand::TransactionHashNumbers { value } => {
+                if settings.transaction_hash_numbers_in_rocksdb == value {
+                    println!("transaction_hash_numbers_in_rocksdb is already set to {}", value);
+                    return Ok(());
+                }
+                settings.transaction_hash_numbers_in_rocksdb = value;
+                println!("Set transaction_hash_numbers_in_rocksdb = {}", value);
+            }
+            SetCommand::AccountHistory { value } => {
+                if settings.account_history_in_rocksdb == value {
+                    println!("account_history_in_rocksdb is already set to {}", value);
+                    return Ok(());
+                }
+                settings.account_history_in_rocksdb = value;
+                println!("Set account_history_in_rocksdb = {}", value);
+            }
+            SetCommand::StorageChangesets { value } => {
+                if settings.storage_changesets_in_static_files == value {
+                    println!("storage_changesets_in_static_files is already set to {}", value);
+                    return Ok(());
+                }
+                settings.storage_changesets_in_static_files = value;
+                println!("Set storage_changesets_in_static_files = {}", value);
+            }
+            SetCommand::UseHashedState { value } => {
+                if settings.use_hashed_state == value {
+                    println!("use_hashed_state is already set to {}", value);
+                    return Ok(());
+                }
+                if settings.use_hashed_state && !value {
+                    println!("WARNING: Disabling use_hashed_state on an existing hashed-state database requires a full resync.");
+                } else {
+                    println!("WARNING: Enabling use_hashed_state on an existing plain-state database requires a full resync.");
+                }
+                settings.use_hashed_state = value;
+                println!("Set use_hashed_state = {}", value);
             }
         }
 

crates/cli/commands/src/db/state.rs

@@ -1,4 +1,4 @@
-use alloy_primitives::{keccak256, Address, BlockNumber, B256, U256};
+use alloy_primitives::{Address, BlockNumber, B256, U256};
 use clap::Parser;
 use parking_lot::Mutex;
 use reth_db_api::{
@@ -63,65 +63,39 @@ impl Command {
         address: Address,
         limit: usize,
     ) -> eyre::Result<()> {
-        let use_hashed_state = tool.provider_factory.cached_storage_settings().use_hashed_state();
-
         let entries = tool.provider_factory.db_ref().view(|tx| {
-            let (account, walker_entries) = if use_hashed_state {
-                let hashed_address = keccak256(address);
-                let account = tx.get::<tables::HashedAccounts>(hashed_address)?;
-                let mut cursor = tx.cursor_dup_read::<tables::HashedStorages>()?;
-                let walker = cursor.walk_dup(Some(hashed_address), None)?;
-                let mut entries = Vec::new();
-                let mut last_log = Instant::now();
-                for (idx, entry) in walker.enumerate() {
-                    let (_, storage_entry) = entry?;
-                    if storage_entry.value != U256::ZERO {
-                        entries.push((storage_entry.key, storage_entry.value));
-                    }
-                    if entries.len() >= limit {
-                        break;
-                    }
-                    if last_log.elapsed() >= LOG_INTERVAL {
-                        info!(
-                            target: "reth::cli",
-                            address = %address,
-                            slots_scanned = idx,
-                            "Scanning storage slots"
-                        );
-                        last_log = Instant::now();
-                    }
-                }
-                (account, entries)
-            } else {
-                // Get account info
-                let account = tx.get::<tables::PlainAccountState>(address)?;
-                // Get storage entries
-                let mut cursor = tx.cursor_dup_read::<tables::PlainStorageState>()?;
-                let walker = cursor.walk_dup(Some(address), None)?;
-                let mut entries = Vec::new();
-                let mut last_log = Instant::now();
-                for (idx, entry) in walker.enumerate() {
-                    let (_, storage_entry) = entry?;
-                    if storage_entry.value != U256::ZERO {
-                        entries.push((storage_entry.key, storage_entry.value));
-                    }
-                    if entries.len() >= limit {
-                        break;
-                    }
-                    if last_log.elapsed() >= LOG_INTERVAL {
-                        info!(
-                            target: "reth::cli",
-                            address = %address,
-                            slots_scanned = idx,
-                            "Scanning storage slots"
-                        );
-                        last_log = Instant::now();
-                    }
-                }
-                (account, entries)
-            };
+            // Get account info
+            let account = tx.get::<tables::PlainAccountState>(address)?;
 
-            Ok::<_, eyre::Report>((account, walker_entries))
+            // Get storage entries
+            let mut cursor = tx.cursor_dup_read::<tables::PlainStorageState>()?;
+            let mut entries = Vec::new();
+            let mut last_log = Instant::now();
+
+            let walker = cursor.walk_dup(Some(address), None)?;
+            for (idx, entry) in walker.enumerate() {
+                let (_, storage_entry) = entry?;
+
+                if storage_entry.value != U256::ZERO {
+                    entries.push((storage_entry.key, storage_entry.value));
+                }
+
+                if entries.len() >= limit {
+                    break;
+                }
+
+                if last_log.elapsed() >= LOG_INTERVAL {
+                    info!(
+                        target: "reth::cli",
+                        address = %address,
+                        slots_scanned = idx,
+                        "Scanning storage slots"
+                    );
+                    last_log = Instant::now();
+                }
+            }
+
+            Ok::<_, eyre::Report>((account, entries))
         })??;
 
         let (account, storage_entries) = entries;
@@ -145,7 +119,7 @@ impl Command {
 
         // Check storage settings to determine where history is stored
         let storage_settings = tool.provider_factory.cached_storage_settings();
-        let history_in_rocksdb = storage_settings.storage_v2;
+        let history_in_rocksdb = storage_settings.storages_history_in_rocksdb;
 
         // For historical queries, enumerate keys from history indices only
         // (not PlainStorageState, which reflects current state)

crates/cli/commands/src/download.rs

@@ -37,14 +37,6 @@ pub struct DownloadDefaults {
     pub available_snapshots: Vec<Cow<'static, str>>,
     /// Default base URL for snapshots
     pub default_base_url: Cow<'static, str>,
-    /// Default base URL for chain-aware snapshots.
-    ///
-    /// When set, the chain ID is appended to form the full URL: `{base_url}/{chain_id}`.
-    /// For example, given a base URL of `https://snapshots.example.com` and chain ID `1`,
-    /// the resulting URL would be `https://snapshots.example.com/1`.
-    ///
-    /// Falls back to [`default_base_url`](Self::default_base_url) when `None`.
-    pub default_chain_aware_base_url: Option<Cow<'static, str>>,
     /// Optional custom long help text that overrides the generated help
     pub long_help: Option<String>,
 }
@@ -68,7 +60,6 @@ impl DownloadDefaults {
                 Cow::Borrowed("https://publicnode.com/snapshots (full nodes & testnets)"),
             ],
             default_base_url: Cow::Borrowed(MERKLE_BASE_URL),
-            default_chain_aware_base_url: None,
             long_help: None,
         }
     }
@@ -93,11 +84,9 @@ impl DownloadDefaults {
         }
 
         help.push_str(
-            "\nIf no URL is provided, the latest archive snapshot for the selected chain\nwill be proposed for download from ",
-        );
-        help.push_str(
-            self.default_chain_aware_base_url.as_deref().unwrap_or(&self.default_base_url),
+            "\nIf no URL is provided, the latest mainnet archive snapshot\nwill be proposed for download from ",
         );
+        help.push_str(self.default_base_url.as_ref());
         help.push_str(
             ".\n\nLocal file:// URLs are also supported for extracting snapshots from disk.",
         );
@@ -122,12 +111,6 @@ impl DownloadDefaults {
         self
     }
 
-    /// Set the default chain-aware base URL.
-    pub fn with_chain_aware_base_url(mut self, url: impl Into<Cow<'static, str>>) -> Self {
-        self.default_chain_aware_base_url = Some(url.into());
-        self
-    }
-
     /// Builder: Set custom long help text, overriding the generated help
     pub fn with_long_help(mut self, help: impl Into<String>) -> Self {
         self.long_help = Some(help.into());
@@ -159,7 +142,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> DownloadCo
         let url = match self.url {
             Some(url) => url,
             None => {
-                let url = get_latest_snapshot_url(self.env.chain.chain().id()).await?;
+                let url = get_latest_snapshot_url().await?;
                 info!(target: "reth::cli", "Using default snapshot URL: {}", url);
                 url
             }
@@ -526,12 +509,8 @@ async fn stream_and_extract(url: &str, target_dir: &Path) -> Result<()> {
 }
 
 // Builds default URL for latest mainnet archive snapshot using configured defaults
-async fn get_latest_snapshot_url(chain_id: u64) -> Result<String> {
-    let defaults = DownloadDefaults::get_global();
-    let base_url = match &defaults.default_chain_aware_base_url {
-        Some(url) => format!("{url}/{chain_id}"),
-        None => defaults.default_base_url.to_string(),
-    };
+async fn get_latest_snapshot_url() -> Result<String> {
+    let base_url = &DownloadDefaults::get_global().default_base_url;
     let latest_url = format!("{base_url}/latest.txt");
     let filename = Client::new()
         .get(latest_url)

crates/cli/commands/src/node.rs

@@ -10,8 +10,8 @@ use reth_node_builder::NodeBuilder;
 use reth_node_core::{
     args::{
         DatabaseArgs, DatadirArgs, DebugArgs, DevArgs, EngineArgs, EraArgs, MetricArgs,
-        NetworkArgs, PayloadBuilderArgs, PruningArgs, RpcServerArgs, StaticFilesArgs, StorageArgs,
-        TxPoolArgs,
+        NetworkArgs, PayloadBuilderArgs, PruningArgs, RocksDbArgs, RpcServerArgs, StaticFilesArgs,
+        StorageArgs, TxPoolArgs,
     },
     node_config::NodeConfig,
     version,
@@ -103,6 +103,10 @@ pub struct NodeCommand<C: ChainSpecParser, Ext: clap::Args + fmt::Debug = NoArgs
     #[command(flatten)]
     pub pruning: PruningArgs,
 
+    /// All `RocksDB` table routing arguments
+    #[command(flatten)]
+    pub rocksdb: RocksDbArgs,
+
     /// Engine cli arguments
     #[command(flatten, next_help_heading = "Engine")]
     pub engine: EngineArgs,
@@ -115,8 +119,8 @@ pub struct NodeCommand<C: ChainSpecParser, Ext: clap::Args + fmt::Debug = NoArgs
     #[command(flatten, next_help_heading = "Static Files")]
     pub static_files: StaticFilesArgs,
 
-    /// All storage related arguments with --storage prefix
-    #[command(flatten, next_help_heading = "Storage")]
+    /// Storage mode configuration (v2 vs v1/legacy)
+    #[command(flatten)]
     pub storage: StorageArgs,
 
     /// Additional cli arguments
@@ -171,6 +175,7 @@ where
             db,
             dev,
             pruning,
+            rocksdb,
             engine,
             era,
             static_files,
@@ -178,6 +183,9 @@ where
             ext,
         } = self;
 
+        // Validate RocksDB arguments
+        rocksdb.validate()?;
+
         // set up node config
         let mut node_config = NodeConfig {
             datadir,
@@ -193,6 +201,7 @@ where
             db,
             dev,
             pruning,
+            rocksdb,
             engine,
             era,
             static_files,

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

@@ -45,16 +45,12 @@ impl<C: ChainSpecParser> Command<C> {
 
         let tool = DbTool::new(provider_factory)?;
 
-        let static_file_segments = match self.stage {
-            StageEnum::Headers => vec![StaticFileSegment::Headers],
-            StageEnum::Bodies => vec![StaticFileSegment::Transactions],
-            StageEnum::Execution => vec![
-                StaticFileSegment::Receipts,
-                StaticFileSegment::AccountChangeSets,
-                StaticFileSegment::StorageChangeSets,
-            ],
-            StageEnum::Senders => vec![StaticFileSegment::TransactionSenders],
-            _ => vec![],
+        let static_file_segment = match self.stage {
+            StageEnum::Headers => Some(StaticFileSegment::Headers),
+            StageEnum::Bodies => Some(StaticFileSegment::Transactions),
+            StageEnum::Execution => Some(StaticFileSegment::Receipts),
+            StageEnum::Senders => Some(StaticFileSegment::TransactionSenders),
+            _ => None,
         };
 
         // Calling `StaticFileProviderRW::prune_*` will instruct the writer to prune rows only
@@ -62,33 +58,35 @@ impl<C: ChainSpecParser> Command<C> {
         // deleting the jar files, otherwise if the task were to be interrupted after we
         // have deleted them, BUT before we have committed the checkpoints to the database, we'd
         // lose essential data.
-        let static_file_provider = tool.provider_factory.static_file_provider();
-        for segment in static_file_segments {
-            if let Some(highest_block) = static_file_provider.get_highest_static_file_block(segment)
+        if let Some(static_file_segment) = static_file_segment {
+            let static_file_provider = tool.provider_factory.static_file_provider();
+            if let Some(highest_block) =
+                static_file_provider.get_highest_static_file_block(static_file_segment)
             {
-                let mut writer = static_file_provider.latest_writer(segment)?;
+                let mut writer = static_file_provider.latest_writer(static_file_segment)?;
 
-                match segment {
+                match static_file_segment {
                     StaticFileSegment::Headers => {
+                        // Prune all headers leaving genesis intact.
                         writer.prune_headers(highest_block)?;
                     }
                     StaticFileSegment::Transactions => {
                         let to_delete = static_file_provider
-                            .get_highest_static_file_tx(segment)
+                            .get_highest_static_file_tx(static_file_segment)
                             .map(|tx_num| tx_num + 1)
                             .unwrap_or_default();
                         writer.prune_transactions(to_delete, 0)?;
                     }
                     StaticFileSegment::Receipts => {
                         let to_delete = static_file_provider
-                            .get_highest_static_file_tx(segment)
+                            .get_highest_static_file_tx(static_file_segment)
                             .map(|tx_num| tx_num + 1)
                             .unwrap_or_default();
                         writer.prune_receipts(to_delete, 0)?;
                     }
                     StaticFileSegment::TransactionSenders => {
                         let to_delete = static_file_provider
-                            .get_highest_static_file_tx(segment)
+                            .get_highest_static_file_tx(static_file_segment)
                             .map(|tx_num| tx_num + 1)
                             .unwrap_or_default();
                         writer.prune_transaction_senders(to_delete, 0)?;
@@ -133,15 +131,8 @@ impl<C: ChainSpecParser> Command<C> {
                 reset_stage_checkpoint(tx, StageId::SenderRecovery)?;
             }
             StageEnum::Execution => {
-                if provider_rw.cached_storage_settings().use_hashed_state() {
-                    tx.clear::<tables::HashedAccounts>()?;
-                    tx.clear::<tables::HashedStorages>()?;
-                    reset_stage_checkpoint(tx, StageId::AccountHashing)?;
-                    reset_stage_checkpoint(tx, StageId::StorageHashing)?;
-                } else {
-                    tx.clear::<tables::PlainAccountState>()?;
-                    tx.clear::<tables::PlainStorageState>()?;
-                }
+                tx.clear::<tables::PlainAccountState>()?;
+                tx.clear::<tables::PlainStorageState>()?;
                 tx.clear::<tables::AccountChangeSets>()?;
                 tx.clear::<tables::StorageChangeSets>()?;
                 tx.clear::<tables::Bytecodes>()?;
@@ -187,7 +178,7 @@ impl<C: ChainSpecParser> Command<C> {
                 let settings = provider_rw.cached_storage_settings();
                 let rocksdb = tool.provider_factory.rocksdb_provider();
 
-                if settings.storage_v2 {
+                if settings.account_history_in_rocksdb {
                     rocksdb.clear::<tables::AccountsHistory>()?;
                 } else {
                     tx.clear::<tables::AccountsHistory>()?;
@@ -204,7 +195,7 @@ impl<C: ChainSpecParser> Command<C> {
                 let settings = provider_rw.cached_storage_settings();
                 let rocksdb = tool.provider_factory.rocksdb_provider();
 
-                if settings.storage_v2 {
+                if settings.storages_history_in_rocksdb {
                     rocksdb.clear::<tables::StoragesHistory>()?;
                 } else {
                     tx.clear::<tables::StoragesHistory>()?;
@@ -218,7 +209,7 @@ impl<C: ChainSpecParser> Command<C> {
                 )?;
             }
             StageEnum::TxLookup => {
-                if provider_rw.cached_storage_settings().storage_v2 {
+                if provider_rw.cached_storage_settings().transaction_hash_numbers_in_rocksdb {
                     tool.provider_factory
                         .rocksdb_provider()
                         .clear::<tables::TransactionHashNumbers>()?;

crates/e2e-test-utils/src/testsuite/setup.rs

@@ -1,6 +1,6 @@
 //! Test setup utilities for configuring the initial state.
 
-use crate::{testsuite::Environment, E2ETestSetupBuilder, NodeBuilderHelper};
+use crate::{setup_engine_with_connection, testsuite::Environment, NodeBuilderHelper};
 use alloy_eips::BlockNumberOrTag;
 use alloy_primitives::B256;
 use alloy_rpc_types_engine::{ForkchoiceState, PayloadAttributes};
@@ -38,8 +38,6 @@ pub struct Setup<I> {
     shutdown_tx: Option<mpsc::Sender<()>>,
     /// Is this setup in dev mode
     pub is_dev: bool,
-    /// Whether to use v2 storage mode (hashed keys, static file changesets, rocksdb history)
-    pub storage_v2: bool,
     /// Tracks instance generic.
     _phantom: PhantomData<I>,
     /// Holds the import result to keep nodes alive when using imported chain
@@ -60,7 +58,6 @@ impl<I> Default for Setup<I> {
             tree_config: TreeConfig::default(),
             shutdown_tx: None,
             is_dev: true,
-            storage_v2: false,
             _phantom: Default::default(),
             import_result_holder: None,
             import_rlp_path: None,
@@ -129,12 +126,6 @@ where
         self
     }
 
-    /// Enable v2 storage mode (hashed keys, static file changesets, rocksdb history)
-    pub const fn with_storage_v2(mut self) -> Self {
-        self.storage_v2 = true;
-        self
-    }
-
     /// Apply setup using pre-imported chain data from RLP file
     pub async fn apply_with_import<N>(
         &mut self,
@@ -203,28 +194,19 @@ where
         self.shutdown_tx = Some(shutdown_tx);
 
         let is_dev = self.is_dev;
-        let storage_v2 = self.storage_v2;
         let node_count = self.network.node_count;
-        let tree_config = self.tree_config.clone();
 
         let attributes_generator = Self::create_static_attributes_generator::<N>();
 
-        let mut builder = E2ETestSetupBuilder::<N, _>::new(
+        let result = setup_engine_with_connection::<N>(
             node_count,
             Arc::<N::ChainSpec>::new((*chain_spec).clone().into()),
+            is_dev,
+            self.tree_config.clone(),
             attributes_generator,
+            self.network.connect_nodes,
         )
-        .with_tree_config_modifier(move |base| {
-            tree_config.clone().with_cross_block_cache_size(base.cross_block_cache_size())
-        })
-        .with_node_config_modifier(move |config| config.set_dev(is_dev))
-        .with_connect_nodes(self.network.connect_nodes);
-
-        if storage_v2 {
-            builder = builder.with_storage_v2();
-        }
-
-        let result = builder.build().await;
+        .await;
 
         let mut node_clients = Vec::new();
         match result {

crates/e2e-test-utils/tests/rocksdb/main.rs

@@ -10,6 +10,7 @@ use jsonrpsee::core::client::ClientT;
 use reth_chainspec::{ChainSpec, ChainSpecBuilder, MAINNET};
 use reth_db::tables;
 use reth_e2e_test_utils::{transaction::TransactionTestContext, wallet, E2ETestSetupBuilder};
+use reth_node_core::args::RocksDbArgs;
 use reth_node_ethereum::EthereumNode;
 use reth_payload_builder::EthPayloadBuilderAttributes;
 use reth_provider::RocksDBProviderFactory;
@@ -95,6 +96,22 @@ fn test_attributes_generator(timestamp: u64) -> EthPayloadBuilderAttributes {
     EthPayloadBuilderAttributes::new(B256::ZERO, attributes)
 }
 
+/// Verifies that `RocksDB` CLI defaults are `None` (deferred to storage mode).
+#[test]
+fn test_rocksdb_defaults_are_none() {
+    let args = RocksDbArgs::default();
+
+    assert!(args.tx_hash.is_none(), "tx_hash default should be None (deferred to --storage.v2)");
+    assert!(
+        args.storages_history.is_none(),
+        "storages_history default should be None (deferred to --storage.v2)"
+    );
+    assert!(
+        args.account_history.is_none(),
+        "account_history default should be None (deferred to --storage.v2)"
+    );
+}
+
 /// Smoke test: node boots with `RocksDB` routing enabled.
 #[tokio::test]
 async fn test_rocksdb_node_startup() -> Result<()> {
@@ -460,7 +477,7 @@ async fn test_rocksdb_pending_tx_not_in_storage() -> Result<()> {
 ///
 /// This test exercises `unwind_trie_state_from` which previously failed with
 /// `UnsortedInput` errors because it read changesets directly from MDBX tables
-/// instead of using storage-aware methods that check `is_v2()`.
+/// instead of using storage-aware methods that check `storage_changesets_in_static_files`.
 #[tokio::test]
 async fn test_rocksdb_reorg_unwind() -> Result<()> {
     reth_tracing::init_test_tracing();

crates/engine/primitives/src/config.rs

@@ -179,8 +179,6 @@ pub struct TreeConfig {
     sparse_trie_prune_depth: usize,
     /// Maximum number of storage tries to retain after pruning.
     sparse_trie_max_storage_tries: usize,
-    /// Whether to fully disable sparse trie cache pruning between blocks.
-    disable_sparse_trie_cache_pruning: bool,
     /// Timeout for the state root task before spawning a sequential fallback computation.
     /// If `Some`, after waiting this duration for the state root task, a sequential state root
     /// computation is spawned in parallel and whichever finishes first is used.
@@ -218,7 +216,6 @@ impl Default for TreeConfig {
             disable_trie_cache: false,
             sparse_trie_prune_depth: DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
             sparse_trie_max_storage_tries: DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
-            disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout: Some(DEFAULT_STATE_ROOT_TASK_TIMEOUT),
         }
     }
@@ -284,7 +281,6 @@ impl TreeConfig {
             disable_trie_cache: false,
             sparse_trie_prune_depth,
             sparse_trie_max_storage_tries,
-            disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout,
         }
     }
@@ -635,17 +631,6 @@ impl TreeConfig {
         self
     }
 
-    /// Returns whether sparse trie cache pruning is disabled.
-    pub const fn disable_sparse_trie_cache_pruning(&self) -> bool {
-        self.disable_sparse_trie_cache_pruning
-    }
-
-    /// Setter for whether to disable sparse trie cache pruning.
-    pub const fn with_disable_sparse_trie_cache_pruning(mut self, value: bool) -> Self {
-        self.disable_sparse_trie_cache_pruning = value;
-        self
-    }
-
     /// Returns the state root task timeout.
     pub const fn state_root_task_timeout(&self) -> Option<Duration> {
         self.state_root_task_timeout

crates/engine/service/Cargo.toml

@@ -0,0 +1,47 @@
+[package]
+name = "reth-engine-service"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+# reth
+reth-consensus.workspace = true
+reth-engine-tree.workspace = true
+reth-evm.workspace = true
+reth-network-p2p.workspace = true
+reth-payload-builder.workspace = true
+reth-provider.workspace = true
+reth-prune.workspace = true
+reth-stages-api.workspace = true
+reth-tasks.workspace = true
+reth-node-types.workspace = true
+reth-chainspec.workspace = true
+reth-engine-primitives.workspace = true
+reth-trie-db.workspace = true
+
+# async
+futures.workspace = true
+pin-project.workspace = true
+
+# misc
+
+[dev-dependencies]
+reth-engine-tree = { workspace = true, features = ["test-utils"] }
+reth-ethereum-consensus.workspace = true
+reth-ethereum-engine-primitives.workspace = true
+reth-evm-ethereum.workspace = true
+reth-exex-types.workspace = true
+reth-primitives-traits.workspace = true
+reth-node-ethereum.workspace = true
+reth-trie-db.workspace = true
+
+alloy-eips.workspace = true
+tokio = { workspace = true, features = ["sync"] }
+tokio-stream.workspace = true

crates/engine/service/src/lib.rs

@@ -0,0 +1,12 @@
+//! Engine service implementation.
+
+#![doc(
+    html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
+    html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
+    issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
+)]
+#![cfg_attr(docsrs, feature(doc_cfg))]
+#![cfg_attr(not(test), warn(unused_crate_dependencies))]
+
+/// Engine Service
+pub mod service;

crates/engine/service/src/service.rs

@@ -0,0 +1,227 @@
+use futures::{Stream, StreamExt};
+use pin_project::pin_project;
+use reth_chainspec::EthChainSpec;
+use reth_consensus::FullConsensus;
+use reth_engine_primitives::{BeaconEngineMessage, ConsensusEngineEvent};
+use reth_engine_tree::{
+    backfill::PipelineSync,
+    download::BasicBlockDownloader,
+    engine::{EngineApiKind, EngineApiRequest, EngineApiRequestHandler, EngineHandler},
+    persistence::PersistenceHandle,
+    tree::{EngineApiTreeHandler, EngineValidator, TreeConfig},
+};
+pub use reth_engine_tree::{
+    chain::{ChainEvent, ChainOrchestrator},
+    engine::EngineApiEvent,
+};
+use reth_evm::ConfigureEvm;
+use reth_network_p2p::BlockClient;
+use reth_node_types::{BlockTy, NodeTypes};
+use reth_payload_builder::PayloadBuilderHandle;
+use reth_provider::{
+    providers::{BlockchainProvider, ProviderNodeTypes},
+    ProviderFactory,
+};
+use reth_prune::PrunerWithFactory;
+use reth_stages_api::{MetricEventsSender, Pipeline};
+use reth_tasks::TaskSpawner;
+use reth_trie_db::ChangesetCache;
+use std::{
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+
+/// Alias for consensus engine stream.
+pub type EngineMessageStream<T> = Pin<Box<dyn Stream<Item = BeaconEngineMessage<T>> + Send + Sync>>;
+
+/// Alias for chain orchestrator.
+type EngineServiceType<N, Client> = ChainOrchestrator<
+    EngineHandler<
+        EngineApiRequestHandler<
+            EngineApiRequest<<N as NodeTypes>::Payload, <N as NodeTypes>::Primitives>,
+            <N as NodeTypes>::Primitives,
+        >,
+        EngineMessageStream<<N as NodeTypes>::Payload>,
+        BasicBlockDownloader<Client, BlockTy<N>>,
+    >,
+    PipelineSync<N>,
+>;
+
+/// The type that drives the chain forward and communicates progress.
+#[pin_project]
+#[expect(missing_debug_implementations)]
+// TODO(mattsse): remove hidden once fixed : <https://github.com/rust-lang/rust/issues/135363>
+//  otherwise rustdoc fails to resolve the alias
+#[doc(hidden)]
+pub struct EngineService<N, Client>
+where
+    N: ProviderNodeTypes,
+    Client: BlockClient<Block = BlockTy<N>> + 'static,
+{
+    orchestrator: EngineServiceType<N, Client>,
+}
+
+impl<N, Client> EngineService<N, Client>
+where
+    N: ProviderNodeTypes,
+    Client: BlockClient<Block = BlockTy<N>> + 'static,
+{
+    /// Constructor for `EngineService`.
+    #[expect(clippy::too_many_arguments)]
+    pub fn new<V, C>(
+        consensus: Arc<dyn FullConsensus<N::Primitives>>,
+        chain_spec: Arc<N::ChainSpec>,
+        client: Client,
+        incoming_requests: EngineMessageStream<N::Payload>,
+        pipeline: Pipeline<N>,
+        pipeline_task_spawner: Box<dyn TaskSpawner>,
+        provider: ProviderFactory<N>,
+        blockchain_db: BlockchainProvider<N>,
+        pruner: PrunerWithFactory<ProviderFactory<N>>,
+        payload_builder: PayloadBuilderHandle<N::Payload>,
+        payload_validator: V,
+        tree_config: TreeConfig,
+        sync_metrics_tx: MetricEventsSender,
+        evm_config: C,
+        changeset_cache: ChangesetCache,
+    ) -> Self
+    where
+        V: EngineValidator<N::Payload>,
+        C: ConfigureEvm<Primitives = N::Primitives> + 'static,
+    {
+        let engine_kind =
+            if chain_spec.is_optimism() { EngineApiKind::OpStack } else { EngineApiKind::Ethereum };
+
+        let downloader = BasicBlockDownloader::new(client, consensus.clone());
+
+        let persistence_handle =
+            PersistenceHandle::<N::Primitives>::spawn_service(provider, pruner, sync_metrics_tx);
+
+        let canonical_in_memory_state = blockchain_db.canonical_in_memory_state();
+
+        let (to_tree_tx, from_tree) = EngineApiTreeHandler::spawn_new(
+            blockchain_db,
+            consensus,
+            payload_validator,
+            persistence_handle,
+            payload_builder,
+            canonical_in_memory_state,
+            tree_config,
+            engine_kind,
+            evm_config,
+            changeset_cache,
+        );
+
+        let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
+        let handler = EngineHandler::new(engine_handler, downloader, incoming_requests);
+
+        let backfill_sync = PipelineSync::new(pipeline, pipeline_task_spawner);
+
+        Self { orchestrator: ChainOrchestrator::new(handler, backfill_sync) }
+    }
+
+    /// Returns a mutable reference to the orchestrator.
+    pub fn orchestrator_mut(&mut self) -> &mut EngineServiceType<N, Client> {
+        &mut self.orchestrator
+    }
+}
+
+impl<N, Client> Stream for EngineService<N, Client>
+where
+    N: ProviderNodeTypes,
+    Client: BlockClient<Block = BlockTy<N>> + 'static,
+{
+    type Item = ChainEvent<ConsensusEngineEvent<N::Primitives>>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut orchestrator = self.project().orchestrator;
+        StreamExt::poll_next_unpin(&mut orchestrator, cx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use reth_chainspec::{ChainSpecBuilder, MAINNET};
+    use reth_engine_primitives::{BeaconEngineMessage, NoopInvalidBlockHook};
+    use reth_engine_tree::{test_utils::TestPipelineBuilder, tree::BasicEngineValidator};
+    use reth_ethereum_consensus::EthBeaconConsensus;
+    use reth_ethereum_engine_primitives::EthEngineTypes;
+    use reth_evm_ethereum::EthEvmConfig;
+    use reth_exex_types::FinishedExExHeight;
+    use reth_network_p2p::test_utils::TestFullBlockClient;
+    use reth_node_ethereum::EthereumEngineValidator;
+    use reth_primitives_traits::SealedHeader;
+    use reth_provider::{
+        providers::BlockchainProvider, test_utils::create_test_provider_factory_with_chain_spec,
+    };
+    use reth_prune::Pruner;
+    use reth_tasks::TokioTaskExecutor;
+    use reth_trie_db::ChangesetCache;
+    use std::sync::Arc;
+    use tokio::sync::{mpsc::unbounded_channel, watch};
+    use tokio_stream::wrappers::UnboundedReceiverStream;
+
+    #[test]
+    fn eth_chain_orchestrator_build() {
+        let chain_spec = Arc::new(
+            ChainSpecBuilder::default()
+                .chain(MAINNET.chain)
+                .genesis(MAINNET.genesis.clone())
+                .paris_activated()
+                .build(),
+        );
+        let consensus = Arc::new(EthBeaconConsensus::new(chain_spec.clone()));
+
+        let client = TestFullBlockClient::default();
+
+        let (_tx, rx) = unbounded_channel::<BeaconEngineMessage<EthEngineTypes>>();
+        let incoming_requests = UnboundedReceiverStream::new(rx);
+
+        let pipeline = TestPipelineBuilder::new().build(chain_spec.clone());
+        let pipeline_task_spawner = Box::<TokioTaskExecutor>::default();
+        let provider_factory = create_test_provider_factory_with_chain_spec(chain_spec.clone());
+
+        let blockchain_db =
+            BlockchainProvider::with_latest(provider_factory.clone(), SealedHeader::default())
+                .unwrap();
+        let engine_payload_validator = EthereumEngineValidator::new(chain_spec.clone());
+        let (_tx, rx) = watch::channel(FinishedExExHeight::NoExExs);
+        let pruner = Pruner::new_with_factory(provider_factory.clone(), vec![], 0, 0, None, rx);
+        let evm_config = EthEvmConfig::new(chain_spec.clone());
+
+        let changeset_cache = ChangesetCache::new();
+
+        let engine_validator = BasicEngineValidator::new(
+            blockchain_db.clone(),
+            consensus.clone(),
+            evm_config.clone(),
+            engine_payload_validator,
+            TreeConfig::default(),
+            Box::new(NoopInvalidBlockHook::default()),
+            changeset_cache.clone(),
+            reth_tasks::Runtime::test(),
+        );
+
+        let (sync_metrics_tx, _sync_metrics_rx) = unbounded_channel();
+        let (tx, _rx) = unbounded_channel();
+        let _eth_service = EngineService::new(
+            consensus,
+            chain_spec,
+            client,
+            Box::pin(incoming_requests),
+            pipeline,
+            pipeline_task_spawner,
+            provider_factory,
+            blockchain_db,
+            pruner,
+            PayloadBuilderHandle::new(tx),
+            engine_validator,
+            TreeConfig::default(),
+            sync_metrics_tx,
+            evm_config,
+            changeset_cache,
+        );
+    }
+}

crates/engine/tree/Cargo.toml

@@ -29,7 +29,7 @@ reth-provider.workspace = true
 reth-prune.workspace = true
 reth-revm = { workspace = true, features = ["optional-balance-check"] }
 reth-stages-api.workspace = true
-reth-tasks = { workspace = true, features = ["rayon"] }
+reth-tasks.workspace = true
 reth-trie-parallel.workspace = true
 reth-trie-sparse = { workspace = true, features = ["std", "metrics"] }
 reth-trie.workspace = true
@@ -143,13 +143,6 @@ test-utils = [
     "reth-evm-ethereum/test-utils",
     "reth-tasks/test-utils",
 ]
-rocksdb = [
-    "reth-provider/rocksdb",
-    "reth-prune/rocksdb",
-    "reth-stages?/rocksdb",
-    "reth-e2e-test-utils/rocksdb",
-]
-edge = ["rocksdb"]
 
 [[test]]
 name = "e2e_testsuite"

crates/engine/tree/src/launch.rs

@@ -1,110 +0,0 @@
-//! Engine orchestrator launch helper.
-//!
-//! Provides [`build_engine_orchestrator`](crate::launch::build_engine_orchestrator) which wires
-//! together all engine components and returns a
-//! [`ChainOrchestrator`](crate::chain::ChainOrchestrator) ready to be polled as a `Stream`.
-
-use crate::{
-    backfill::PipelineSync,
-    chain::ChainOrchestrator,
-    download::BasicBlockDownloader,
-    engine::{EngineApiKind, EngineApiRequest, EngineApiRequestHandler, EngineHandler},
-    persistence::PersistenceHandle,
-    tree::{EngineApiTreeHandler, EngineValidator, TreeConfig},
-};
-use futures::Stream;
-use reth_consensus::FullConsensus;
-use reth_engine_primitives::BeaconEngineMessage;
-use reth_evm::ConfigureEvm;
-use reth_network_p2p::BlockClient;
-use reth_payload_builder::PayloadBuilderHandle;
-use reth_primitives_traits::NodePrimitives;
-use reth_provider::{
-    providers::{BlockchainProvider, ProviderNodeTypes},
-    ProviderFactory, StorageSettingsCache,
-};
-use reth_prune::PrunerWithFactory;
-use reth_stages_api::{MetricEventsSender, Pipeline};
-use reth_tasks::TaskSpawner;
-use reth_trie_db::ChangesetCache;
-use std::sync::Arc;
-
-/// Builds the engine [`ChainOrchestrator`] that drives the chain forward.
-///
-/// This spawns and wires together the following components:
-///
-/// - **[`BasicBlockDownloader`]** — downloads blocks on demand from the network during live sync.
-/// - **[`PersistenceHandle`]** — spawns the persistence service on a background thread for writing
-///   blocks and performing pruning outside the critical consensus path.
-/// - **[`EngineApiTreeHandler`]** — spawns the tree handler that processes engine API requests
-///   (`newPayload`, `forkchoiceUpdated`) and maintains the in-memory chain state.
-/// - **[`EngineApiRequestHandler`]** + **[`EngineHandler`]** — glue that routes incoming CL
-///   messages to the tree handler and manages download requests.
-/// - **[`PipelineSync`]** — wraps the staged sync [`Pipeline`] for backfill sync when the node
-///   needs to catch up over large block ranges.
-///
-/// The returned orchestrator implements [`Stream`] and yields
-/// [`ChainEvent`]s.
-///
-/// [`ChainEvent`]: crate::chain::ChainEvent
-#[expect(clippy::too_many_arguments, clippy::type_complexity)]
-pub fn build_engine_orchestrator<N, Client, S, V, C>(
-    engine_kind: EngineApiKind,
-    consensus: Arc<dyn FullConsensus<N::Primitives>>,
-    client: Client,
-    incoming_requests: S,
-    pipeline: Pipeline<N>,
-    pipeline_task_spawner: Box<dyn TaskSpawner>,
-    provider: ProviderFactory<N>,
-    blockchain_db: BlockchainProvider<N>,
-    pruner: PrunerWithFactory<ProviderFactory<N>>,
-    payload_builder: PayloadBuilderHandle<N::Payload>,
-    payload_validator: V,
-    tree_config: TreeConfig,
-    sync_metrics_tx: MetricEventsSender,
-    evm_config: C,
-    changeset_cache: ChangesetCache,
-) -> ChainOrchestrator<
-    EngineHandler<
-        EngineApiRequestHandler<EngineApiRequest<N::Payload, N::Primitives>, N::Primitives>,
-        S,
-        BasicBlockDownloader<Client, <N::Primitives as NodePrimitives>::Block>,
-    >,
-    PipelineSync<N>,
->
-where
-    N: ProviderNodeTypes,
-    Client: BlockClient<Block = <N::Primitives as NodePrimitives>::Block> + 'static,
-    S: Stream<Item = BeaconEngineMessage<N::Payload>> + Send + Sync + Unpin + 'static,
-    V: EngineValidator<N::Payload>,
-    C: ConfigureEvm<Primitives = N::Primitives> + 'static,
-{
-    let downloader = BasicBlockDownloader::new(client, consensus.clone());
-    let use_hashed_state = provider.cached_storage_settings().use_hashed_state();
-
-    let persistence_handle =
-        PersistenceHandle::<N::Primitives>::spawn_service(provider, pruner, sync_metrics_tx);
-
-    let canonical_in_memory_state = blockchain_db.canonical_in_memory_state();
-
-    let (to_tree_tx, from_tree) = EngineApiTreeHandler::spawn_new(
-        blockchain_db,
-        consensus,
-        payload_validator,
-        persistence_handle,
-        payload_builder,
-        canonical_in_memory_state,
-        tree_config,
-        engine_kind,
-        evm_config,
-        changeset_cache,
-        use_hashed_state,
-    );
-
-    let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
-    let handler = EngineHandler::new(engine_handler, downloader, incoming_requests);
-
-    let backfill_sync = PipelineSync::new(pipeline, pipeline_task_spawner);
-
-    ChainOrchestrator::new(handler, backfill_sync)
-}

crates/engine/tree/src/lib.rs

@@ -100,8 +100,6 @@ pub mod chain;
 pub mod download;
 /// Engine Api chain handler support.
 pub mod engine;
-/// Engine orchestrator launch helper.
-pub mod launch;
 /// Metrics support.
 pub mod metrics;
 /// The background writer service, coordinating write operations on static files and the database.

crates/engine/tree/src/persistence.rs

@@ -119,7 +119,7 @@ where
         Ok(())
     }
 
-    #[instrument(level = "debug", target = "engine::persistence", skip_all, fields(%new_tip_num))]
+    #[instrument(level = "debug", target = "engine::persistence", skip_all, fields(new_tip_num))]
     fn on_remove_blocks_above(
         &self,
         new_tip_num: u64,

crates/engine/tree/src/tree/cached_state.rs

@@ -351,14 +351,6 @@ impl<S: StateProvider, const PREWARM: bool> StateProvider for CachedStateProvide
             self.state_provider.storage(account, storage_key)
         }
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        address: Address,
-        hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<StorageValue>> {
-        self.state_provider.storage_by_hashed_key(address, hashed_storage_key)
-    }
 }
 
 impl<S: BytecodeReader, const PREWARM: bool> BytecodeReader for CachedStateProvider<S, PREWARM> {

crates/engine/tree/src/tree/instrumented_state.rs

@@ -199,17 +199,6 @@ impl<S: StateProvider> StateProvider for InstrumentedStateProvider<S> {
         self.record_storage_fetch(start.elapsed());
         res
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        address: Address,
-        hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<StorageValue>> {
-        let start = Instant::now();
-        let res = self.state_provider.storage_by_hashed_key(address, hashed_storage_key);
-        self.record_storage_fetch(start.elapsed());
-        res
-    }
 }
 
 impl<S: BytecodeReader> BytecodeReader for InstrumentedStateProvider<S> {

crates/engine/tree/src/tree/metrics.rs

@@ -12,13 +12,12 @@ use reth_primitives_traits::constants::gas_units::MEGAGAS;
 use reth_trie::updates::TrieUpdates;
 use std::time::{Duration, Instant};
 
-/// Upper bounds for each gas bucket. The last bucket is a catch-all for
-/// everything above the final threshold: <5M, 5-10M, 10-20M, 20-30M, 30-40M, >40M.
-const GAS_BUCKET_THRESHOLDS: [u64; 5] =
-    [5 * MEGAGAS, 10 * MEGAGAS, 20 * MEGAGAS, 30 * MEGAGAS, 40 * MEGAGAS];
+/// Width of each gas bucket in gas units (10 Mgas).
+const GAS_BUCKET_SIZE: u64 = 10 * MEGAGAS;
 
-/// Total number of gas buckets (thresholds + 1 catch-all).
-const NUM_GAS_BUCKETS: usize = GAS_BUCKET_THRESHOLDS.len() + 1;
+/// Number of gas buckets. The last bucket is a catch-all for everything above
+/// `(NUM_GAS_BUCKETS - 1) * GAS_BUCKET_SIZE`.
+const NUM_GAS_BUCKETS: usize = 5;
 
 /// Metrics for the `EngineApi`.
 #[derive(Debug, Default)]
@@ -281,23 +280,21 @@ impl GasBucketMetrics {
     }
 
     fn bucket_index(gas_used: u64) -> usize {
-        GAS_BUCKET_THRESHOLDS
-            .iter()
-            .position(|&threshold| gas_used < threshold)
-            .unwrap_or(GAS_BUCKET_THRESHOLDS.len())
+        let idx = gas_used / GAS_BUCKET_SIZE;
+        (idx as usize).min(NUM_GAS_BUCKETS - 1)
     }
 
-    /// Returns a human-readable label like `<5M`, `5-10M`, … `>40M`.
+    /// Returns a human-readable label like `<10M`, `10-20M`, … `>40M`.
     fn bucket_label(index: usize) -> String {
+        let m = GAS_BUCKET_SIZE / 1_000_000;
         if index == 0 {
-            let hi = GAS_BUCKET_THRESHOLDS[0] / MEGAGAS;
-            format!("<{hi}M")
-        } else if index < GAS_BUCKET_THRESHOLDS.len() {
-            let lo = GAS_BUCKET_THRESHOLDS[index - 1] / MEGAGAS;
-            let hi = GAS_BUCKET_THRESHOLDS[index] / MEGAGAS;
+            format!("<{m}M")
+        } else if index < NUM_GAS_BUCKETS - 1 {
+            let lo = m * index as u64;
+            let hi = lo + m;
             format!("{lo}-{hi}M")
         } else {
-            let lo = GAS_BUCKET_THRESHOLDS[GAS_BUCKET_THRESHOLDS.len() - 1] / MEGAGAS;
+            let lo = m * index as u64;
             format!(">{lo}M")
         }
     }

crates/engine/tree/src/tree/mod.rs

@@ -32,7 +32,7 @@ use reth_provider::{
     BlockExecutionOutput, BlockExecutionResult, BlockReader, ChangeSetReader,
     DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StageCheckpointReader,
     StateProviderBox, StateProviderFactory, StateReader, StorageChangeSetReader,
-    StorageSettingsCache, TransactionVariant,
+    TransactionVariant,
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_stages_api::ControlFlow;
@@ -271,9 +271,6 @@ where
     evm_config: C,
     /// Changeset cache for in-memory trie changesets
     changeset_cache: ChangesetCache,
-    /// Whether the node uses hashed state as canonical storage (v2 mode).
-    /// Cached at construction to avoid threading `StorageSettingsCache` bounds everywhere.
-    use_hashed_state: bool,
 }
 
 impl<N, P: Debug, T: PayloadTypes + Debug, V: Debug, C> std::fmt::Debug
@@ -299,7 +296,6 @@ where
             .field("engine_kind", &self.engine_kind)
             .field("evm_config", &self.evm_config)
             .field("changeset_cache", &self.changeset_cache)
-            .field("use_hashed_state", &self.use_hashed_state)
             .finish()
     }
 }
@@ -317,8 +313,7 @@ where
     P::Provider: BlockReader<Block = N::Block, Header = N::BlockHeader>
         + StageCheckpointReader
         + ChangeSetReader
-        + StorageChangeSetReader
-        + StorageSettingsCache,
+        + StorageChangeSetReader,
     C: ConfigureEvm<Primitives = N> + 'static,
     T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
     V: EngineValidator<T>,
@@ -339,7 +334,6 @@ where
         engine_kind: EngineApiKind,
         evm_config: C,
         changeset_cache: ChangesetCache,
-        use_hashed_state: bool,
     ) -> Self {
         let (incoming_tx, incoming) = crossbeam_channel::unbounded();
 
@@ -361,7 +355,6 @@ where
             engine_kind,
             evm_config,
             changeset_cache,
-            use_hashed_state,
         }
     }
 
@@ -382,7 +375,6 @@ where
         kind: EngineApiKind,
         evm_config: C,
         changeset_cache: ChangesetCache,
-        use_hashed_state: bool,
     ) -> (Sender<FromEngine<EngineApiRequest<T, N>, N::Block>>, UnboundedReceiver<EngineApiEvent<N>>)
     {
         let best_block_number = provider.best_block_number().unwrap_or(0);
@@ -415,7 +407,6 @@ where
             kind,
             evm_config,
             changeset_cache,
-            use_hashed_state,
         );
         let incoming = task.incoming_tx.clone();
         spawn_os_thread("engine", || task.run());
@@ -2388,12 +2379,7 @@ where
 
             self.update_reorg_metrics(old.len(), old_first);
             self.reinsert_reorged_blocks(new.clone());
-
-            // When use_hashed_state is enabled, skip reinserting the old chain — the
-            // bundle state references plain state reverts which don't exist.
-            if !self.use_hashed_state {
-                self.reinsert_reorged_blocks(old.clone());
-            }
+            self.reinsert_reorged_blocks(old.clone());
         }
 
         // update the tracked in-memory state with the new chain
@@ -2602,7 +2588,7 @@ where
     /// Returns `InsertPayloadOk::Inserted(BlockStatus::Valid)` on successful execution,
     /// `InsertPayloadOk::AlreadySeen` if the block already exists, or
     /// `InsertPayloadOk::Inserted(BlockStatus::Disconnected)` if parent state is missing.
-    #[instrument(level = "debug", target = "engine::tree", skip_all, fields(?block_id))]
+    #[instrument(level = "debug", target = "engine::tree", skip_all, fields(block_id))]
     fn insert_block_or_payload<Input, Err>(
         &mut self,
         block_id: BlockWithParent,

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

@@ -33,7 +33,7 @@ use reth_provider::{
     StateProviderFactory, StateReader,
 };
 use reth_revm::{db::BundleState, state::EvmState};
-use reth_tasks::{ForEachOrdered, Runtime};
+use reth_tasks::Runtime;
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
 use reth_trie_parallel::{
     proof_task::{ProofTaskCtx, ProofWorkerHandle},
@@ -43,6 +43,7 @@ use reth_trie_sparse::{
     ParallelSparseTrie, ParallelismThresholds, RevealableSparseTrie, SparseStateTrie,
 };
 use std::{
+    collections::BTreeMap,
     ops::Not,
     sync::{
         atomic::AtomicBool,
@@ -93,10 +94,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;
 
-/// Blocks with fewer transactions than this skip prewarming, since the fixed overhead of spawning
-/// prewarm workers exceeds the execution time saved.
-pub const SMALL_BLOCK_TX_THRESHOLD: usize = 5;
-
 /// Type alias for [`PayloadHandle`] returned by payload processor spawn methods.
 type IteratorPayloadHandle<Evm, I, N> = PayloadHandle<
     WithTxEnv<TxEnvFor<Evm>, <I as ExecutableTxIterator<Evm>>::Recovered>,
@@ -138,8 +135,6 @@ where
     sparse_trie_prune_depth: usize,
     /// Maximum storage tries to retain after pruning.
     sparse_trie_max_storage_tries: usize,
-    /// Whether sparse trie cache pruning is fully disabled.
-    disable_sparse_trie_cache_pruning: bool,
     /// Whether to disable cache metrics recording.
     disable_cache_metrics: bool,
 }
@@ -175,7 +170,6 @@ where
             prewarm_max_concurrency: config.prewarm_max_concurrency(),
             sparse_trie_prune_depth: config.sparse_trie_prune_depth(),
             sparse_trie_max_storage_tries: config.sparse_trie_max_storage_tries(),
-            disable_sparse_trie_cache_pruning: config.disable_sparse_trie_cache_pruning(),
             disable_cache_metrics: config.disable_cache_metrics(),
         }
     }
@@ -248,23 +242,42 @@ where
         let (to_sparse_trie, sparse_trie_rx) = channel();
         let (to_multi_proof, from_multi_proof) = crossbeam_channel::unbounded();
 
+        // Extract V2 proofs flag early so we can pass it to prewarm
         let v2_proofs_enabled = !config.disable_proof_v2();
-        let parent_state_root = env.parent_state_root;
-        let transaction_count = env.transaction_count;
-        let prewarm_handle = self.spawn_caching_with(
-            env,
-            prewarm_rx,
-            provider_builder.clone(),
-            Some(to_multi_proof.clone()),
-            bal,
-            v2_proofs_enabled,
-        );
 
-        // Create and spawn the storage proof task.
+        // Capture parent_state_root before env is moved into spawn_caching_with
+        let parent_state_root = env.parent_state_root;
+
+        // 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");
+
+            // The prewarm task converts the BAL to HashedPostState and sends it on
+            // to_multi_proof after slot prefetching completes.
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                provider_builder.clone(),
+                Some(to_multi_proof.clone()),
+                Some(bal),
+                v2_proofs_enabled,
+            )
+        } else {
+            // Normal path: spawn with transaction prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                provider_builder.clone(),
+                Some(to_multi_proof.clone()),
+                None,
+                v2_proofs_enabled,
+            )
+        };
+
+        // Create and spawn the storage proof task
         let task_ctx = ProofTaskCtx::new(multiproof_provider_factory);
-        let halve_workers = transaction_count <= Self::SMALL_BLOCK_PROOF_WORKER_TX_THRESHOLD;
-        let proof_handle =
-            ProofWorkerHandle::new(&self.executor, task_ctx, halve_workers, v2_proofs_enabled);
+        let proof_handle = ProofWorkerHandle::new(&self.executor, task_ctx, v2_proofs_enabled);
 
         if config.disable_trie_cache() {
             let multi_proof_task = MultiProofTask::new(
@@ -344,10 +357,6 @@ where
         }
     }
 
-    /// Transaction count threshold below which proof workers are halved, since fewer transactions
-    /// produce fewer state changes and most workers would be idle overhead.
-    const SMALL_BLOCK_PROOF_WORKER_TX_THRESHOLD: usize = 30;
-
     /// Transaction count threshold below which sequential signature recovery is used.
     ///
     /// For blocks with fewer than this many transactions, the rayon parallel iterator overhead
@@ -359,11 +368,8 @@ where
     /// Spawns a task advancing transaction env iterator and streaming updates through a channel.
     ///
     /// For blocks with fewer than [`Self::SMALL_BLOCK_TX_THRESHOLD`] transactions, uses
-    /// sequential iteration to avoid rayon overhead. For larger blocks, uses rayon parallel
-    /// iteration with [`ForEachOrdered`] to recover signatures in parallel while streaming
-    /// results to execution in the original transaction order.
+    /// sequential iteration to avoid rayon overhead.
     #[expect(clippy::type_complexity)]
-    #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
     fn spawn_tx_iterator<I: ExecutableTxIterator<Evm>>(
         &self,
         transactions: I,
@@ -372,8 +378,9 @@ where
         mpsc::Receiver<WithTxEnv<TxEnvFor<Evm>, I::Recovered>>,
         mpsc::Receiver<Result<WithTxEnv<TxEnvFor<Evm>, I::Recovered>, I::Error>>,
     ) {
-        let (prewarm_tx, prewarm_rx) = mpsc::sync_channel(transaction_count);
-        let (execute_tx, execute_rx) = mpsc::sync_channel(transaction_count);
+        let (ooo_tx, ooo_rx) = mpsc::channel();
+        let (prewarm_tx, prewarm_rx) = mpsc::channel();
+        let (execute_tx, execute_rx) = mpsc::channel();
 
         if transaction_count == 0 {
             // Empty block — nothing to do.
@@ -387,7 +394,7 @@ where
             );
             self.executor.spawn_blocking(move || {
                 let (transactions, convert) = transactions.into_parts();
-                for tx in transactions {
+                for (idx, tx) in transactions.into_iter().enumerate() {
                     let tx = convert.convert(tx);
                     let tx = tx.map(|tx| {
                         let (tx_env, tx) = tx.into_parts();
@@ -396,42 +403,57 @@ where
                     if let Ok(tx) = &tx {
                         let _ = prewarm_tx.send(tx.clone());
                     }
-                    let _ = execute_tx.send(tx);
+                    let _ = ooo_tx.send((idx, tx));
                 }
             });
         } else {
-            // Parallel path — recover signatures in parallel on rayon, stream results
-            // to execution in order via `for_each_ordered`.
+            // Parallel path — spawn on rayon for parallel signature recovery.
             rayon::spawn(move || {
                 let (transactions, convert) = transactions.into_parts();
-                transactions
-                    .into_par_iter()
-                    .map(|tx| {
+                transactions.into_par_iter().enumerate().for_each_with(
+                    ooo_tx,
+                    |ooo_tx, (idx, tx)| {
                         let tx = convert.convert(tx);
-                        tx.map(|tx| {
+                        let tx = tx.map(|tx| {
                             let (tx_env, tx) = tx.into_parts();
-                            let tx = WithTxEnv { tx_env, tx: Arc::new(tx) };
-                            // Send to prewarming out of order — order doesn't matter there.
+                            WithTxEnv { tx_env, tx: Arc::new(tx) }
+                        });
+                        // Only send Ok(_) variants to prewarming task.
+                        if let Ok(tx) = &tx {
                             let _ = prewarm_tx.send(tx.clone());
-                            tx
-                        })
-                    })
-                    .for_each_ordered(|tx| {
-                        let _ = execute_tx.send(tx);
-                    });
+                        }
+                        let _ = ooo_tx.send((idx, tx));
+                    },
+                );
             });
         }
 
+        // Spawn a task that processes out-of-order transactions from the task above and sends them
+        // to the execution task in order.
+        self.executor.spawn_blocking(move || {
+            let mut next_for_execution = 0;
+            let mut queue = BTreeMap::new();
+            while let Ok((idx, tx)) = ooo_rx.recv() {
+                if next_for_execution == idx {
+                    let _ = execute_tx.send(tx);
+                    next_for_execution += 1;
+
+                    while let Some(entry) = queue.first_entry() &&
+                        *entry.key() == next_for_execution
+                    {
+                        let _ = execute_tx.send(entry.remove());
+                        next_for_execution += 1;
+                    }
+                } else {
+                    queue.insert(idx, tx);
+                }
+            }
+        });
+
         (prewarm_rx, execute_rx)
     }
 
     /// Spawn prewarming optionally wired to the multiproof task for target updates.
-    #[instrument(
-        level = "debug",
-        target = "engine::tree::payload_processor",
-        skip_all,
-        fields(bal=%bal.is_some(), %v2_proofs_enabled)
-    )]
     fn spawn_caching_with<P>(
         &self,
         env: ExecutionEnv<Evm>,
@@ -444,8 +466,7 @@ where
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
-        let skip_prewarm =
-            self.disable_transaction_prewarming || env.transaction_count < SMALL_BLOCK_TX_THRESHOLD;
+        let skip_prewarm = self.disable_transaction_prewarming;
 
         let saved_cache = self.disable_state_cache.not().then(|| self.cache_for(env.parent_hash));
 
@@ -470,6 +491,7 @@ where
             self.prewarm_max_concurrency,
         );
 
+        // spawn pre-warm task
         {
             let to_prewarm_task = to_prewarm_task.clone();
             self.executor.spawn_blocking(move || {
@@ -524,7 +546,6 @@ where
         let disable_trie_cache = config.disable_trie_cache();
         let prune_depth = self.sparse_trie_prune_depth;
         let max_storage_tries = self.sparse_trie_max_storage_tries;
-        let disable_cache_pruning = self.disable_sparse_trie_cache_pruning;
         let chunk_size =
             config.multiproof_chunking_enabled().then_some(config.multiproof_chunk_size());
         let executor = self.executor.clone();
@@ -621,7 +642,6 @@ where
                     max_storage_tries,
                     SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                     SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
-                    disable_cache_pruning,
                 );
                 trie_metrics
                     .into_trie_for_reuse_duration_histogram

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

@@ -771,11 +771,6 @@ impl MultiProofTask {
     fn on_prefetch_proof(&mut self, mut targets: VersionedMultiProofTargets) -> u64 {
         // Remove already fetched proof targets to avoid redundant work.
         targets.retain_difference(&self.fetched_proof_targets);
-
-        if targets.is_empty() {
-            return 0;
-        }
-
         extend_multiproof_targets(&mut self.fetched_proof_targets, &targets);
 
         // For Legacy multiproofs, make sure all target accounts have an `AddedRemovedKeySet` in the
@@ -894,10 +889,6 @@ impl MultiProofTask {
             state_updates += 1;
         }
 
-        if not_fetched_state_update.is_empty() {
-            return state_updates;
-        }
-
         // Clone+Arc MultiAddedRemovedKeys for sharing with the dispatched multiproof tasks
         let multi_added_removed_keys = Arc::new(MultiAddedRemovedKeys {
             account: self.multi_added_removed_keys.account.clone(),
@@ -1550,7 +1541,6 @@ mod tests {
         providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
         BlockNumReader, BlockReader, ChangeSetReader, DatabaseProviderFactory, LatestStateProvider,
         PruneCheckpointReader, StageCheckpointReader, StateProviderBox, StorageChangeSetReader,
-        StorageSettingsCache,
     };
     use reth_trie::MultiProof;
     use reth_trie_db::ChangesetCache;
@@ -1572,7 +1562,6 @@ mod tests {
                               + PruneCheckpointReader
                               + ChangeSetReader
                               + StorageChangeSetReader
-                              + StorageSettingsCache
                               + BlockNumReader,
             > + Clone
             + Send
@@ -1582,7 +1571,7 @@ mod tests {
         let changeset_cache = ChangesetCache::new();
         let overlay_factory = OverlayStateProviderFactory::new(factory, changeset_cache);
         let task_ctx = ProofTaskCtx::new(overlay_factory);
-        let proof_handle = ProofWorkerHandle::new(runtime, task_ctx, false, false);
+        let proof_handle = ProofWorkerHandle::new(runtime, task_ctx, false);
         let (to_sparse_trie, _receiver) = std::sync::mpsc::channel();
         let (tx, rx) = crossbeam_channel::unbounded();
 
@@ -1592,10 +1581,7 @@ mod tests {
     fn create_cached_provider<F>(factory: F) -> CachedStateProvider<StateProviderBox>
     where
         F: DatabaseProviderFactory<
-                Provider: BlockReader
-                              + StageCheckpointReader
-                              + PruneCheckpointReader
-                              + reth_provider::StorageSettingsCache,
+                Provider: BlockReader + StageCheckpointReader + PruneCheckpointReader,
             > + Clone
             + Send
             + 'static,
@@ -2065,7 +2051,7 @@ mod tests {
                 panic!("Expected PrefetchProofs message");
             };
 
-        assert!(proofs_requested >= 1);
+        assert_eq!(proofs_requested, 1);
     }
 
     /// Verifies that different message types arriving mid-batch are not lost and preserve order.

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

@@ -42,7 +42,7 @@ use std::{
     ops::Range,
     sync::{
         atomic::{AtomicBool, Ordering},
-        mpsc::{self, channel, Receiver, Sender, SyncSender},
+        mpsc::{self, channel, Receiver, Sender},
         Arc,
     },
     time::Instant,
@@ -154,6 +154,8 @@ where
         self.executor.spawn_blocking(move || {
             let _enter = debug_span!(target: "engine::tree::payload_processor::prewarm", parent: span, "spawn_all").entered();
 
+            let (done_tx, done_rx) = mpsc::channel();
+
             // When transaction_count is 0, it means the count is unknown. In this case, spawn
             // max workers to handle potentially many transactions in parallel rather
             // than bottlenecking on a single worker.
@@ -164,8 +166,6 @@ where
                 transaction_count.min(max_concurrency)
             };
 
-            let (done_tx, done_rx) = mpsc::sync_channel(workers_needed);
-
             // Spawn workers
             let tx_sender = ctx.clone().spawn_workers(workers_needed, &executor,  to_multi_proof.clone(), done_tx.clone());
 
@@ -236,33 +236,36 @@ where
 
         if let Some(saved_cache) = saved_cache {
             debug!(target: "engine::caching", parent_hash=?hash, "Updating execution cache");
-
-            // Detach the published cache so readers see None during the update.
-            // This is necessary because ExecutionCache is Arc-shared: mutating
-            // it via insert_state would be visible through the old SavedCache.
+            // Perform all cache operations atomically under the lock
             execution_cache.update_with_guard(|cached| {
-                cached.take();
-            });
+                // consumes the `SavedCache` held by the prewarming task, which releases its usage
+                // guard
+                let (caches, cache_metrics, disable_cache_metrics) = saved_cache.split();
+                let new_cache = SavedCache::new(hash, caches, cache_metrics)
+                    .with_disable_cache_metrics(disable_cache_metrics);
 
-            let (caches, cache_metrics, disable_cache_metrics) = saved_cache.split();
-            let new_cache = SavedCache::new(hash, caches, cache_metrics)
-                .with_disable_cache_metrics(disable_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() {
+                    // Clear the cache on error to prevent having a polluted cache
+                    *cached = None;
+                    debug!(target: "engine::caching", "cleared execution cache on update error");
+                    return;
+                }
 
-            if new_cache.cache().insert_state(&execution_outcome.state).is_err() {
-                debug!(target: "engine::caching", "cleared execution cache on update error");
-            } else {
                 new_cache.update_metrics();
 
-                let valid = valid_block_rx.recv().is_ok();
-
-                execution_cache.update_with_guard(|cached| {
-                    if valid {
-                        *cached = Some(new_cache);
-                    } else {
-                        debug!(target: "engine::caching", "cleared execution cache on invalid block");
-                    }
-                });
-            }
+                if valid_block_rx.recv().is_ok() {
+                    // Replace the shared cache with the new one; the previous cache (if any) is
+                    // dropped.
+                    *cached = Some(new_cache);
+                } else {
+                    // Block was invalid; caches were already mutated by insert_state above,
+                    // so we must clear to prevent using polluted state
+                    *cached = None;
+                    debug!(target: "engine::caching", "cleared execution cache on invalid block");
+                }
+            });
 
             let elapsed = start.elapsed();
             debug!(target: "engine::caching", parent_hash=?hash, elapsed=?elapsed, "Updated execution cache");
@@ -309,11 +312,11 @@ where
             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);
 
-        let (done_tx, done_rx) = mpsc::sync_channel(workers_needed);
-
         // Calculate slots per worker
         let slots_per_worker = total_slots / workers_needed;
         let remainder = total_slots % workers_needed;
@@ -582,7 +585,7 @@ where
         self,
         txs: CrossbeamReceiver<IndexedTransaction<Tx>>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-        done_tx: SyncSender<()>,
+        done_tx: Sender<()>,
     ) where
         Tx: ExecutableTxFor<Evm>,
     {
@@ -657,7 +660,7 @@ where
         workers_needed: usize,
         task_executor: &Runtime,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-        done_tx: SyncSender<()>,
+        done_tx: Sender<()>,
     ) -> CrossbeamSender<IndexedTransaction<Tx>>
     where
         Tx: ExecutableTxFor<Evm> + Send + 'static,
@@ -695,7 +698,7 @@ where
         executor: &Runtime,
         bal: Arc<BlockAccessList>,
         range: Range<usize>,
-        done_tx: SyncSender<()>,
+        done_tx: Sender<()>,
     ) {
         let ctx = self.clone();
         let span = debug_span!(
@@ -721,7 +724,7 @@ where
         self,
         bal: Arc<BlockAccessList>,
         range: Range<usize>,
-        done_tx: SyncSender<()>,
+        done_tx: Sender<()>,
     ) {
         let Self { saved_cache, provider, metrics, .. } = self;
 

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

@@ -72,7 +72,6 @@ where
         max_storage_tries: usize,
         max_nodes_capacity: usize,
         max_values_capacity: usize,
-        disable_pruning: bool,
     ) -> (SparseStateTrie<A, S>, DeferredDrops) {
         match self {
             Self::Cleared(task) => task.into_cleared_trie(max_nodes_capacity, max_values_capacity),
@@ -81,7 +80,6 @@ where
                 max_storage_tries,
                 max_nodes_capacity,
                 max_values_capacity,
-                disable_pruning,
             ),
         }
     }
@@ -358,23 +356,16 @@ where
     /// Prunes and shrinks the trie for reuse in the next payload built on top of this one.
     ///
     /// Should be called after the state root result has been sent.
-    ///
-    /// When `disable_pruning` is true, the trie is preserved without any node pruning,
-    /// storage trie eviction, or capacity shrinking, keeping the full cache intact for
-    /// benchmarking purposes.
     pub(super) fn into_trie_for_reuse(
         self,
         prune_depth: usize,
         max_storage_tries: usize,
         max_nodes_capacity: usize,
         max_values_capacity: usize,
-        disable_pruning: bool,
     ) -> (SparseStateTrie<A, S>, DeferredDrops) {
         let Self { mut trie, .. } = self;
-        if !disable_pruning {
-            trie.prune(prune_depth, max_storage_tries);
-            trie.shrink_to(max_nodes_capacity, max_values_capacity);
-        }
+        trie.prune(prune_depth, max_storage_tries);
+        trie.shrink_to(max_nodes_capacity, max_values_capacity);
         let deferred = trie.take_deferred_drops();
         (trie, deferred)
     }
@@ -416,9 +407,7 @@ where
                     let update = match message {
                         Ok(m) => m,
                         Err(_) => {
-                            return Err(ParallelStateRootError::Other(
-                                "updates channel disconnected before state root calculation".to_string(),
-                            ))
+                            break
                         }
                     };
 
@@ -593,24 +582,18 @@ where
         self.process_leaf_updates(true)?;
 
         for (address, mut new) in self.new_storage_updates.drain() {
-            match self.storage_updates.entry(address) {
-                Entry::Vacant(entry) => {
-                    entry.insert(new); // insert the whole map at once, no per-slot loop
-                }
-                Entry::Occupied(mut entry) => {
-                    let updates = entry.get_mut();
-                    for (slot, new) in new.drain() {
-                        match updates.entry(slot) {
-                            Entry::Occupied(mut slot_entry) => {
-                                if new.is_changed() {
-                                    slot_entry.insert(new);
-                                }
-                            }
-                            Entry::Vacant(slot_entry) => {
-                                slot_entry.insert(new);
-                            }
+            let updates = self.storage_updates.entry(address).or_default();
+            for (slot, new) in new.drain() {
+                match updates.entry(slot) {
+                    Entry::Occupied(mut entry) => {
+                        // Only overwrite existing entries with new values
+                        if new.is_changed() {
+                            entry.insert(new);
                         }
                     }
+                    Entry::Vacant(entry) => {
+                        entry.insert(new);
+                    }
                 }
             }
         }

crates/engine/tree/src/tree/payload_validator.rs

@@ -10,11 +10,11 @@ use crate::tree::{
     EngineApiMetrics, EngineApiTreeState, ExecutionEnv, PayloadHandle, StateProviderBuilder,
     StateProviderDatabase, TreeConfig,
 };
-use alloy_consensus::{proofs::calculate_receipt_root, transaction::Either, TxHashRef, TxReceipt};
+use alloy_consensus::transaction::{Either, TxHashRef};
 use alloy_eip7928::BlockAccessList;
 use alloy_eips::{eip1898::BlockWithParent, eip4895::Withdrawal, NumHash};
 use alloy_evm::Evm;
-use alloy_primitives::{Bloom, B256};
+use alloy_primitives::B256;
 
 use crate::tree::payload_processor::receipt_root_task::{IndexedReceipt, ReceiptRootTaskHandle};
 use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock, LazyOverlay};
@@ -38,7 +38,7 @@ use reth_provider::{
     providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockNumReader, BlockReader,
     ChangeSetReader, DatabaseProviderFactory, DatabaseProviderROFactory, HashedPostStateProvider,
     ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
-    StateProviderFactory, StateReader, StorageChangeSetReader, StorageSettingsCache,
+    StateProviderFactory, StateReader, StorageChangeSetReader,
 };
 use reth_revm::db::{states::bundle_state::BundleRetention, State};
 use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot};
@@ -53,16 +53,6 @@ use std::{
 };
 use tracing::{debug, debug_span, error, info, instrument, trace, warn};
 
-/// Blocks with at most this many transactions compute the receipt root inline to avoid
-/// background task overhead.
-const SMALL_BLOCK_RECEIPT_ROOT_TX_THRESHOLD: usize = 50;
-const SMALL_BLOCK_STATE_ROOT_TX_THRESHOLD: usize = 50;
-
-enum ReceiptRootResult {
-    Precomputed(ReceiptRootBloom),
-    Pending(tokio::sync::oneshot::Receiver<ReceiptRootBloom>),
-}
-
 /// Context providing access to tree state during validation.
 ///
 /// This context is provided to the [`EngineValidator`] and includes the state of the tree's
@@ -156,8 +146,7 @@ where
                           + PruneCheckpointReader
                           + ChangeSetReader
                           + StorageChangeSetReader
-                          + BlockNumReader
-                          + StorageSettingsCache,
+                          + BlockNumReader,
         > + BlockReader<Header = N::BlockHeader>
         + ChangeSetReader
         + BlockNumReader
@@ -300,7 +289,7 @@ where
         // Validate block consensus rules which includes header validation
         if let Err(consensus_err) = self.validate_block_inner(&block) {
             // Header validation error takes precedence over execution error
-            return Err(InsertBlockError::new(block, consensus_err.into()).into());
+            return Err(InsertBlockError::new(block, consensus_err.into()).into())
         }
 
         // Also validate against the parent
@@ -308,7 +297,7 @@ where
             self.consensus.validate_header_against_parent(block.sealed_header(), parent_block)
         {
             // Parent validation error takes precedence over execution error
-            return Err(InsertBlockError::new(block, consensus_err.into()).into());
+            return Err(InsertBlockError::new(block, consensus_err.into()).into())
         }
 
         // No header validation errors, return the original execution error
@@ -348,7 +337,7 @@ where
                     Ok(val) => val,
                     Err(e) => {
                         let block = self.convert_to_block(input)?;
-                        return Err(InsertBlockError::new(block, e.into()).into());
+                        return Err(InsertBlockError::new(block, e.into()).into())
                     }
                 }
             };
@@ -381,7 +370,7 @@ where
                 self.convert_to_block(input)?,
                 ProviderError::HeaderNotFound(parent_hash.into()).into(),
             )
-            .into());
+            .into())
         };
         let mut state_provider = ensure_ok!(provider_builder.build());
         drop(_enter);
@@ -394,7 +383,7 @@ where
                 self.convert_to_block(input)?,
                 ProviderError::HeaderNotFound(parent_hash.into()).into(),
             )
-            .into());
+            .into())
         };
 
         let evm_env = debug_span!(target: "engine::tree::payload_validator", "evm env")
@@ -411,7 +400,7 @@ where
         };
 
         // Plan the strategy used for state root computation.
-        let strategy = self.plan_state_root_computation(input.transaction_count());
+        let strategy = self.plan_state_root_computation();
 
         debug!(
             target: "engine::tree::payload_validator",
@@ -465,7 +454,7 @@ where
         // Execute the block and handle any execution errors.
         // The receipt root task is spawned before execution and receives receipts incrementally
         // as transactions complete, allowing parallel computation during execution.
-        let (output, senders, receipt_root_result) =
+        let (output, senders, receipt_root_rx) =
             match self.execute_block(state_provider, env, &input, &mut handle) {
                 Ok(output) => output,
                 Err(err) => return self.handle_execution_error(input, err, &parent_block),
@@ -486,18 +475,15 @@ where
         let block = self.convert_to_block(input)?.with_senders(senders);
 
         // Wait for the receipt root computation to complete.
-        let receipt_root_bloom = match receipt_root_result {
-            ReceiptRootResult::Precomputed(receipt_root_bloom) => Some(receipt_root_bloom),
-            ReceiptRootResult::Pending(receipt_root_rx) => receipt_root_rx
-                .blocking_recv()
-                .inspect_err(|_| {
-                    tracing::error!(
-                        target: "engine::tree::payload_validator",
-                        "Receipt root task dropped sender without result, receipt root calculation likely aborted"
-                    );
-                })
-                .ok(),
-        };
+        let receipt_root_bloom = receipt_root_rx
+            .blocking_recv()
+            .inspect_err(|_| {
+                tracing::error!(
+                    target: "engine::tree::payload_validator",
+                    "Receipt root task dropped sender without result, receipt root calculation likely aborted"
+                );
+            })
+            .ok();
 
         let hashed_state = ensure_ok_post_block!(
             self.validate_post_execution(
@@ -630,7 +616,7 @@ where
                 )
                 .into(),
             )
-            .into());
+            .into())
         }
 
         if let Some(valid_block_tx) = valid_block_tx {
@@ -669,12 +655,12 @@ where
     fn validate_block_inner(&self, block: &SealedBlock<N::Block>) -> Result<(), ConsensusError> {
         if let Err(e) = self.consensus.validate_header(block.sealed_header()) {
             error!(target: "engine::tree::payload_validator", ?block, "Failed to validate header {}: {e}", block.hash());
-            return Err(e);
+            return Err(e)
         }
 
         if let Err(e) = self.consensus.validate_block_pre_execution(block) {
             error!(target: "engine::tree::payload_validator", ?block, "Failed to validate block {}: {e}", block.hash());
-            return Err(e);
+            return Err(e)
         }
 
         Ok(())
@@ -684,7 +670,7 @@ where
     ///
     /// This method orchestrates block execution:
     /// 1. Sets up the EVM with state database and precompile caching
-    /// 2. Spawns a background task for incremental receipt root computation (if needed)
+    /// 2. Spawns a background task for incremental receipt root computation
     /// 3. Executes transactions with metrics collection via state hooks
     /// 4. Merges state transitions and records execution metrics
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
@@ -696,7 +682,11 @@ where
         input: &BlockOrPayload<T>,
         handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
     ) -> Result<
-        (BlockExecutionOutput<N::Receipt>, Vec<Address>, ReceiptRootResult),
+        (
+            BlockExecutionOutput<N::Receipt>,
+            Vec<Address>,
+            tokio::sync::oneshot::Receiver<(B256, alloy_primitives::Bloom)>,
+        ),
         InsertBlockErrorKind,
     >
     where
@@ -740,16 +730,10 @@ where
         // Spawn background task to compute receipt root and logs bloom incrementally.
         // Unbounded channel is used since tx count bounds capacity anyway (max ~30k txs per block).
         let receipts_len = input.transaction_count();
-        let compute_receipt_root_inline = receipts_len <= SMALL_BLOCK_RECEIPT_ROOT_TX_THRESHOLD;
-        let (receipt_tx, result_rx) = if compute_receipt_root_inline {
-            (None, None)
-        } else {
-            let (receipt_tx, receipt_rx) = crossbeam_channel::unbounded();
-            let (result_tx, result_rx) = tokio::sync::oneshot::channel();
-            let task_handle = ReceiptRootTaskHandle::new(receipt_rx, result_tx);
-            self.payload_processor.executor().spawn_blocking(move || task_handle.run(receipts_len));
-            (Some(receipt_tx), Some(result_rx))
-        };
+        let (receipt_tx, receipt_rx) = crossbeam_channel::unbounded();
+        let (result_tx, result_rx) = tokio::sync::oneshot::channel();
+        let task_handle = ReceiptRootTaskHandle::new(receipt_rx, result_tx);
+        self.payload_processor.executor().spawn_blocking(move || task_handle.run(receipts_len));
 
         let transaction_count = input.transaction_count();
         let executor = executor.with_state_hook(Some(Box::new(handle.state_hook())));
@@ -761,7 +745,7 @@ where
             executor,
             transaction_count,
             handle.iter_transactions(),
-            receipt_tx.as_ref(),
+            &receipt_tx,
         )?;
         drop(receipt_tx);
 
@@ -778,21 +762,11 @@ where
 
         let output = BlockExecutionOutput { result, state: db.take_bundle() };
 
-        let receipt_root_result = if compute_receipt_root_inline {
-            ReceiptRootResult::Precomputed(Self::compute_receipt_root_bloom(
-                &output.result.receipts,
-            ))
-        } else {
-            ReceiptRootResult::Pending(
-                result_rx.expect("receipt root receiver missing when task spawned"),
-            )
-        };
-
         let execution_duration = execution_start.elapsed();
         self.metrics.record_block_execution(&output, execution_duration);
 
         debug!(target: "engine::tree::payload_validator", elapsed = ?execution_duration, "Executed block");
-        Ok((output, senders, receipt_root_result))
+        Ok((output, senders, result_rx))
     }
 
     /// Executes transactions and collects senders, streaming receipts to a background task.
@@ -809,7 +783,7 @@ where
         mut executor: E,
         transaction_count: usize,
         transactions: impl Iterator<Item = Result<Tx, Err>>,
-        receipt_tx: Option<&crossbeam_channel::Sender<IndexedReceipt<N::Receipt>>>,
+        receipt_tx: &crossbeam_channel::Sender<IndexedReceipt<N::Receipt>>,
     ) -> Result<(E, Vec<Address>), BlockExecutionError>
     where
         E: BlockExecutor<Receipt = N::Receipt>,
@@ -856,15 +830,13 @@ where
             executor.execute_transaction(tx)?;
             self.metrics.record_transaction_execution(tx_start.elapsed());
 
-            if let Some(receipt_tx) = receipt_tx {
-                let current_len = executor.receipts().len();
-                if current_len > last_sent_len {
-                    last_sent_len = current_len;
-                    // Send the latest receipt to the background task for incremental root computation.
-                    if let Some(receipt) = executor.receipts().last() {
-                        let tx_index = current_len - 1;
-                        let _ = receipt_tx.send(IndexedReceipt::new(tx_index, receipt.clone()));
-                    }
+            let current_len = executor.receipts().len();
+            if current_len > last_sent_len {
+                last_sent_len = current_len;
+                // Send the latest receipt to the background task for incremental root computation.
+                if let Some(receipt) = executor.receipts().last() {
+                    let tx_index = current_len - 1;
+                    let _ = receipt_tx.send(IndexedReceipt::new(tx_index, receipt.clone()));
                 }
             }
         }
@@ -1100,7 +1072,7 @@ where
         trace!(target: "engine::tree::payload_validator", block=?block.num_hash(), "Validating block consensus");
         // validate block consensus rules
         if let Err(e) = self.validate_block_inner(block) {
-            return Err(e.into());
+            return Err(e.into())
         }
 
         // now validate against the parent
@@ -1109,7 +1081,7 @@ where
             self.consensus.validate_header_against_parent(block.sealed_header(), parent_block)
         {
             warn!(target: "engine::tree::payload_validator", ?block, "Failed to validate header {} against parent: {e}", block.hash());
-            return Err(e.into());
+            return Err(e.into())
         }
         drop(_enter);
 
@@ -1122,7 +1094,7 @@ where
         {
             // call post-block hook
             self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
-            return Err(err.into());
+            return Err(err.into())
         }
         drop(_enter);
 
@@ -1137,7 +1109,7 @@ where
         {
             // call post-block hook
             self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
-            return Err(err.into());
+            return Err(err.into())
         }
 
         // record post-execution validation duration
@@ -1149,16 +1121,6 @@ where
         Ok(hashed_state)
     }
 
-    fn compute_receipt_root_bloom(receipts: &[N::Receipt]) -> ReceiptRootBloom {
-        let receipts_with_bloom =
-            receipts.iter().map(TxReceipt::with_bloom_ref).collect::<Vec<_>>();
-        let receipts_root = calculate_receipt_root(&receipts_with_bloom);
-        let logs_bloom = receipts_with_bloom
-            .iter()
-            .fold(Bloom::ZERO, |bloom, receipt| bloom | receipt.bloom_ref());
-        (receipts_root, logs_bloom)
-    }
-
     /// Spawns a payload processor task based on the state root strategy.
     ///
     /// This method determines how to execute the block and compute its state root based on
@@ -1179,7 +1141,7 @@ where
         level = "debug",
         target = "engine::tree::payload_validator",
         skip_all,
-        fields(?strategy)
+        fields(strategy)
     )]
     fn spawn_payload_processor<T: ExecutableTxIterator<Evm>>(
         &mut self,
@@ -1255,7 +1217,7 @@ where
                 self.provider.clone(),
                 historical,
                 Some(blocks),
-            )));
+            )))
         }
 
         // Check if the block is persisted
@@ -1263,7 +1225,7 @@ where
             debug!(target: "engine::tree::payload_validator", %hash, number = %header.number(), "found canonical state for block in database, creating provider builder");
             // For persisted blocks, we create a builder that will fetch state directly from the
             // database
-            return Ok(Some(StateProviderBuilder::new(self.provider.clone(), hash, None)));
+            return Ok(Some(StateProviderBuilder::new(self.provider.clone(), hash, None)))
         }
 
         debug!(target: "engine::tree::payload_validator", %hash, "no canonical state found for block");
@@ -1274,12 +1236,7 @@ where
     ///
     /// Note: Use state root task only if prefix sets are empty, otherwise proof generation is
     /// too expensive because it requires walking all paths in every proof.
-    fn plan_state_root_computation(&self, transaction_count: usize) -> StateRootStrategy {
-        // Small blocks are faster without spawning parallel state root tasks.
-        if transaction_count > 0 && transaction_count <= SMALL_BLOCK_STATE_ROOT_TX_THRESHOLD {
-            return StateRootStrategy::Synchronous;
-        }
-
+    const fn plan_state_root_computation(&self) -> StateRootStrategy {
         if self.config.state_root_fallback() {
             StateRootStrategy::Synchronous
         } else if self.config.use_state_root_task() {
@@ -1300,7 +1257,7 @@ where
     ) {
         if state.invalid_headers.get(&block.hash()).is_some() {
             // we already marked this block as invalid
-            return;
+            return
         }
         self.invalid_block_hook.on_invalid_block(parent_header, block, output, trie_updates);
     }
@@ -1569,8 +1526,7 @@ where
                           + PruneCheckpointReader
                           + ChangeSetReader
                           + StorageChangeSetReader
-                          + BlockNumReader
-                          + StorageSettingsCache,
+                          + BlockNumReader,
         > + BlockReader<Header = N::BlockHeader>
         + StateProviderFactory
         + StateReader

crates/engine/tree/src/tree/tests.rs

@@ -221,7 +221,6 @@ impl TestHarness {
             EngineApiKind::Ethereum,
             evm_config,
             changeset_cache,
-            provider.cached_storage_settings().use_hashed_state(),
         );
 
         let block_builder = TestBlockBuilder::default().with_chain_spec((*chain_spec).clone());

crates/engine/tree/tests/e2e-testsuite/main.rs

@@ -2,15 +2,13 @@
 
 mod fcu_finalized_blocks;
 
-use alloy_rpc_types_engine::PayloadStatusEnum;
 use eyre::Result;
 use reth_chainspec::{ChainSpecBuilder, MAINNET};
 use reth_e2e_test_utils::testsuite::{
     actions::{
-        BlockReference, CaptureBlock, CompareNodeChainTips, CreateFork, ExpectFcuStatus,
-        MakeCanonical, ProduceBlocks, ProduceBlocksLocally, ProduceInvalidBlocks, ReorgTo,
-        SelectActiveNode, SendForkchoiceUpdate, SendNewPayloads, SetForkBase, UpdateBlockInfo,
-        ValidateCanonicalTag, WaitForSync,
+        CaptureBlock, CompareNodeChainTips, CreateFork, ExpectFcuStatus, MakeCanonical,
+        ProduceBlocks, ProduceBlocksLocally, ProduceInvalidBlocks, ReorgTo, SelectActiveNode,
+        SendNewPayloads, UpdateBlockInfo, ValidateCanonicalTag, WaitForSync,
     },
     setup::{NetworkSetup, Setup},
     TestBuilder,
@@ -41,14 +39,6 @@ fn default_engine_tree_setup() -> Setup<EthEngineTypes> {
         )
 }
 
-/// Creates a v2 storage mode setup for engine tree e2e tests.
-///
-/// v2 mode uses keccak256-hashed slot keys in static file changesets and rocksdb history
-/// instead of plain keys in MDBX.
-fn v2_engine_tree_setup() -> Setup<EthEngineTypes> {
-    default_engine_tree_setup().with_storage_v2()
-}
-
 /// Test that verifies forkchoice update and canonical chain insertion functionality.
 #[tokio::test]
 async fn test_engine_tree_fcu_canon_chain_insertion_e2e() -> Result<()> {
@@ -344,152 +334,3 @@ async fn test_engine_tree_live_sync_transition_eventually_canonical_e2e() -> Res
 
     Ok(())
 }
-
-// ==================== v2 storage mode variants ====================
-
-/// v2 variant: Verifies forkchoice update and canonical chain insertion in v2 storage mode.
-///
-/// Exercises the full `save_blocks` → `write_state` → static file changeset path with hashed keys.
-#[tokio::test]
-async fn test_engine_tree_fcu_canon_chain_insertion_v2_e2e() -> Result<()> {
-    reth_tracing::init_test_tracing();
-
-    let test = TestBuilder::new()
-        .with_setup(v2_engine_tree_setup())
-        .with_action(ProduceBlocks::<EthEngineTypes>::new(1))
-        .with_action(MakeCanonical::new())
-        .with_action(ProduceBlocks::<EthEngineTypes>::new(3))
-        .with_action(MakeCanonical::new());
-
-    test.run::<EthereumNode>().await?;
-
-    Ok(())
-}
-
-/// v2 variant: Verifies forkchoice update with a reorg where all blocks are already available.
-///
-/// Exercises `write_state_reverts` path with hashed changeset keys during CL-driven reorgs.
-#[tokio::test]
-async fn test_engine_tree_fcu_reorg_with_all_blocks_v2_e2e() -> Result<()> {
-    reth_tracing::init_test_tracing();
-
-    let test = TestBuilder::new()
-        .with_setup(v2_engine_tree_setup())
-        .with_action(ProduceBlocks::<EthEngineTypes>::new(5))
-        .with_action(MakeCanonical::new())
-        .with_action(CreateFork::<EthEngineTypes>::new(2, 3))
-        .with_action(CaptureBlock::new("fork_tip"))
-        .with_action(ReorgTo::<EthEngineTypes>::new_from_tag("fork_tip"));
-
-    test.run::<EthereumNode>().await?;
-
-    Ok(())
-}
-
-/// v2 variant: Verifies progressive canonical chain extension in v2 storage mode.
-#[tokio::test]
-async fn test_engine_tree_fcu_extends_canon_chain_v2_e2e() -> Result<()> {
-    reth_tracing::init_test_tracing();
-
-    let test = TestBuilder::new()
-        .with_setup(v2_engine_tree_setup())
-        .with_action(ProduceBlocks::<EthEngineTypes>::new(1))
-        .with_action(MakeCanonical::new())
-        .with_action(ProduceBlocks::<EthEngineTypes>::new(10))
-        .with_action(CaptureBlock::new("target_block"))
-        .with_action(ReorgTo::<EthEngineTypes>::new_from_tag("target_block"))
-        .with_action(MakeCanonical::new());
-
-    test.run::<EthereumNode>().await?;
-
-    Ok(())
-}
-
-/// Creates a 2-node setup for disk-level reorg testing.
-///
-/// Uses unconnected nodes so fork blocks can be produced independently on Node 1 and then
-/// sent to Node 0 via newPayload only (no FCU), keeping Node 0's persisted chain intact
-/// until the final `ReorgTo` triggers `find_disk_reorg`.
-fn disk_reorg_setup(storage_v2: bool) -> Setup<EthEngineTypes> {
-    let mut setup = Setup::default()
-        .with_chain_spec(Arc::new(
-            ChainSpecBuilder::default()
-                .chain(MAINNET.chain)
-                .genesis(
-                    serde_json::from_str(include_str!(
-                        "../../../../e2e-test-utils/src/testsuite/assets/genesis.json"
-                    ))
-                    .unwrap(),
-                )
-                .cancun_activated()
-                .build(),
-        ))
-        .with_network(NetworkSetup::multi_node_unconnected(2))
-        .with_tree_config(
-            TreeConfig::default().with_legacy_state_root(false).with_has_enough_parallelism(true),
-        );
-    if storage_v2 {
-        setup = setup.with_storage_v2();
-    }
-    setup
-}
-
-/// Builds a disk-level reorg test scenario.
-///
-/// 1. Both nodes receive 3 shared blocks
-/// 2. Node 0 extends to 10 blocks locally (persisted to disk)
-/// 3. Node 1 builds an 8-block fork from block 3 (its canonical head)
-/// 4. Fork blocks are sent to Node 0 via newPayload (no FCU, old chain stays on disk)
-/// 5. FCU to fork tip on Node 0 triggers `find_disk_reorg` → `RemoveBlocksAbove(3)`
-fn disk_reorg_test(storage_v2: bool) -> TestBuilder<EthEngineTypes> {
-    TestBuilder::new()
-        .with_setup(disk_reorg_setup(storage_v2))
-        .with_action(SelectActiveNode::new(0))
-        .with_action(ProduceBlocks::<EthEngineTypes>::new(3))
-        .with_action(MakeCanonical::new())
-        .with_action(ProduceBlocksLocally::<EthEngineTypes>::new(7))
-        .with_action(MakeCanonical::with_active_node())
-        .with_action(SelectActiveNode::new(1))
-        .with_action(SetForkBase::new(3))
-        .with_action(ProduceBlocksLocally::<EthEngineTypes>::new(8))
-        .with_action(MakeCanonical::with_active_node())
-        .with_action(CaptureBlock::new("fork_tip"))
-        .with_action(
-            SendNewPayloads::<EthEngineTypes>::new()
-                .with_source_node(1)
-                .with_target_node(0)
-                .with_start_block(4)
-                .with_total_blocks(8),
-        )
-        .with_action(
-            SendForkchoiceUpdate::<EthEngineTypes>::new(
-                BlockReference::Tag("fork_tip".into()),
-                BlockReference::Tag("fork_tip".into()),
-                BlockReference::Tag("fork_tip".into()),
-            )
-            .with_expected_status(PayloadStatusEnum::Valid)
-            .with_node_idx(0),
-        )
-}
-
-/// Verifies disk-level reorg in v1 (plain key) storage mode.
-///
-/// Confirms `find_disk_reorg()` detects persisted blocks on the wrong fork and calls
-/// `RemoveBlocksAbove` to truncate, then re-persists the correct fork chain.
-#[tokio::test]
-async fn test_engine_tree_disk_reorg_v1_e2e() -> Result<()> {
-    reth_tracing::init_test_tracing();
-    disk_reorg_test(false).run::<EthereumNode>().await?;
-    Ok(())
-}
-
-/// v2 variant: Verifies disk-level reorg in v2 storage mode.
-///
-/// Same scenario as v1 but with hashed changeset keys in static files and rocksdb history.
-/// Exercises `find_disk_reorg()` → `RemoveBlocksAbove` with v2 hashed key format.
-#[tokio::test]
-async fn test_engine_tree_disk_reorg_v2_e2e() -> Result<()> {
-    reth_tracing::init_test_tracing();
-    disk_reorg_test(true).run::<EthereumNode>().await?;
-    Ok(())
-}

crates/era/src/era1/types/execution.rs

@@ -425,9 +425,17 @@ impl TotalDifficulty {
 
     /// Convert to an [`Entry`]
     pub fn to_entry(&self) -> Entry {
-        // era1 spec: `total-difficulty = { type: 0x0600, data: SSZ uint256 }` (little-endian)
-        let data = self.value.to_le_bytes::<32>().to_vec();
-        Entry::new(TOTAL_DIFFICULTY, data)
+        let mut data = [0u8; 32];
+
+        let be_bytes = self.value.to_be_bytes_vec();
+
+        if be_bytes.len() <= 32 {
+            data[32 - be_bytes.len()..].copy_from_slice(&be_bytes);
+        } else {
+            data.copy_from_slice(&be_bytes[be_bytes.len() - 32..]);
+        }
+
+        Entry::new(TOTAL_DIFFICULTY, data.to_vec())
     }
 
     /// Create from an [`Entry`]
@@ -446,8 +454,8 @@ impl TotalDifficulty {
             )));
         }
 
-        // era1 spec: `total-difficulty = { type: 0x0600, data: SSZ uint256 }` (little-endian)
-        let value = U256::from_le_slice(&entry.data);
+        // Convert 32-byte array to U256
+        let value = U256::from_be_slice(&entry.data);
 
         Ok(Self { value })
     }
@@ -600,19 +608,6 @@ mod tests {
         assert_eq!(recovered.value, value);
     }
 
-    #[test]
-    fn test_total_difficulty_ssz_le_encoding() {
-        // Verify that total-difficulty is encoded as SSZ uint256 (little-endian).
-        // See https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md
-        let value = U256::from(1u64);
-        let td = TotalDifficulty::new(value);
-        let entry = td.to_entry();
-
-        // Little-endian: least significant byte first [1, 0, 0, ..., 0]
-        assert_eq!(entry.data[0], 1, "First byte must be 1 (little-endian)");
-        assert_eq!(entry.data[31], 0, "Last byte must be 0 (little-endian)");
-    }
-
     #[test]
     fn test_compression_roundtrip() {
         let rlp_data = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10];

crates/ethereum/evm/src/lib.rs

@@ -273,6 +273,7 @@ where
             gas_limit: payload.payload.gas_limit(),
             basefee: payload.payload.saturated_base_fee_per_gas(),
             blob_excess_gas_and_price,
+            slot_num: 0,
         };
 
         Ok(EvmEnv { cfg_env, block_env })

crates/ethereum/evm/src/test_utils.rs

@@ -108,8 +108,7 @@ impl<'a, DB: Database, I: Inspector<EthEvmContext<&'a mut State<DB>>>> BlockExec
             result: ResultAndState::new(
                 ExecutionResult::Success {
                     reason: SuccessReason::Return,
-                    gas_used: 0,
-                    gas_refunded: 0,
+                    gas: Default::default(),
                     logs: vec![],
                     output: Output::Call(Bytes::from(vec![])),
                 },

crates/evm/execution-types/src/execute.rs

@@ -1,6 +1,6 @@
 use alloy_primitives::{Address, B256, U256};
 use reth_primitives_traits::{Account, Bytecode};
-use revm::database::{states::BundleState, BundleAccount};
+use revm::database::BundleState;
 
 pub use alloy_evm::block::BlockExecutionResult;
 
@@ -37,11 +37,6 @@ impl<T> BlockExecutionOutput<T> {
         self.state.account(address).map(|a| a.info.as_ref().map(Into::into))
     }
 
-    /// Returns the state [`BundleAccount`] for the given address.
-    pub fn account_state(&self, address: &Address) -> Option<&BundleAccount> {
-        self.state.account(address)
-    }
-
     /// Get storage if value is known.
     ///
     /// This means that depending on status we can potentially return `U256::ZERO`.

crates/exex/exex/src/backfill/test_utils.rs

@@ -18,7 +18,6 @@ use reth_provider::{
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_testing_utils::generators::sign_tx_with_key_pair;
-use reth_trie_common::KeccakKeyHasher;
 use secp256k1::Keypair;
 
 pub(crate) fn to_execution_outcome(
@@ -78,9 +77,12 @@ where
     let execution_outcome = to_execution_outcome(block.number(), &block_execution_output);
 
     // Commit the block's execution outcome to the database
-    let hashed_state = execution_outcome.hash_state_slow::<KeccakKeyHasher>().into_sorted();
     let provider_rw = provider_factory.provider_rw()?;
-    provider_rw.append_blocks_with_state(vec![block.clone()], &execution_outcome, hashed_state)?;
+    provider_rw.append_blocks_with_state(
+        vec![block.clone()],
+        &execution_outcome,
+        Default::default(),
+    )?;
     provider_rw.commit()?;
 
     Ok(block_execution_output)
@@ -208,12 +210,11 @@ where
     execution_outcome.state_mut().reverts.sort();
 
     // Commit the block's execution outcome to the database
-    let hashed_state = execution_outcome.hash_state_slow::<KeccakKeyHasher>().into_sorted();
     let provider_rw = provider_factory.provider_rw()?;
     provider_rw.append_blocks_with_state(
         vec![block1.clone(), block2.clone()],
         &execution_outcome,
-        hashed_state,
+        Default::default(),
     )?;
     provider_rw.commit()?;
 

crates/exex/exex/src/notifications.rs

@@ -10,7 +10,6 @@ use reth_provider::{BlockReader, Chain, HeaderProvider, StateProviderFactory};
 use reth_stages_api::ExecutionStageThresholds;
 use reth_tracing::tracing::debug;
 use std::{
-    collections::VecDeque,
     fmt::Debug,
     pin::Pin,
     sync::Arc,
@@ -287,9 +286,6 @@ where
     backfill_job: Option<StreamBackfillJob<E, P, Chain<E::Primitives>>>,
     /// Custom thresholds for the backfill job, if set.
     backfill_thresholds: Option<ExecutionStageThresholds>,
-    /// Notifications that arrived during backfill and need to be delivered after it completes.
-    /// These are notifications for blocks beyond the backfill range that we must not drop.
-    pending_notifications: VecDeque<ExExNotification<E::Primitives>>,
 }
 
 impl<P, E> ExExNotificationsWithHead<P, E>
@@ -316,7 +312,6 @@ where
             pending_check_backfill: true,
             backfill_job: None,
             backfill_thresholds: None,
-            pending_notifications: VecDeque::new(),
         }
     }
 
@@ -453,34 +448,6 @@ where
         // 3. If backfill is in progress yield new notifications
         if let Some(backfill_job) = &mut this.backfill_job {
             debug!(target: "exex::notifications", "Polling backfill job");
-
-            // Drain the notification channel to prevent backpressure from stalling the
-            // ExExManager. During backfill, the ExEx is not consuming from the channel,
-            // so the capacity-1 channel fills up, which blocks the manager's PollSender,
-            // which fills the manager's 1024-entry buffer, which blocks all upstream
-            // senders. Notifications for blocks covered by the backfill range are
-            // discarded (they'll be re-delivered by the backfill job), while
-            // notifications beyond the backfill range are buffered for delivery after the
-            // backfill completes.
-            while let Poll::Ready(Some(notification)) = this.notifications.poll_recv(cx) {
-                // Always buffer revert-containing notifications (ChainReverted,
-                // ChainReorged) because the backfill job only re-delivers
-                // ChainCommitted from the database. Discarding a reorg here would
-                // leave the ExEx unaware of the fork switch.
-                if notification.reverted_chain().is_some() {
-                    this.pending_notifications.push_back(notification);
-                    continue;
-                }
-                if let Some(committed) = notification.committed_chain() &&
-                    committed.tip().number() <= this.initial_local_head.number
-                {
-                    // Covered by backfill range, safe to discard
-                    continue;
-                }
-                // Beyond the backfill range — buffer for delivery after backfill
-                this.pending_notifications.push_back(notification);
-            }
-
             if let Some(chain) = ready!(backfill_job.poll_next_unpin(cx)).transpose()? {
                 debug!(target: "exex::notifications", range = ?chain.range(), "Backfill job returned a chain");
                 return Poll::Ready(Some(Ok(ExExNotification::ChainCommitted {
@@ -492,18 +459,13 @@ where
             this.backfill_job = None;
         }
 
-        // 4. Deliver any notifications that were buffered during backfill
-        if let Some(notification) = this.pending_notifications.pop_front() {
-            return Poll::Ready(Some(Ok(notification)))
-        }
-
-        // 5. Otherwise advance the regular event stream
+        // 4. Otherwise advance the regular event stream
         loop {
             let Some(notification) = ready!(this.notifications.poll_recv(cx)) else {
                 return Poll::Ready(None)
             };
 
-            // 6. In case the exex is ahead of the new tip, we must skip it
+            // 5. In case the exex is ahead of the new tip, we must skip it
             if let Some(committed) = notification.committed_chain() {
                 // inclusive check because we should start with `exex.head + 1`
                 if this.initial_exex_head.block.number >= committed.tip().number() {
@@ -827,135 +789,4 @@ mod tests {
 
         Ok(())
     }
-
-    /// Regression test for <https://github.com/paradigmxyz/reth/issues/19665>.
-    ///
-    /// During backfill, `poll_next` must drain the notification channel so that
-    /// the upstream `ExExManager` is never blocked by a full channel. Without
-    /// the drain loop the capacity-1 channel stays full for the entire backfill
-    /// duration, which stalls the manager's `PollSender` and eventually blocks
-    /// all upstream senders once the 1024-entry buffer fills up.
-    ///
-    /// The key assertion is the `try_send` after the first `poll_next`: it
-    /// proves the channel was drained during the backfill poll. Without the
-    /// fix this `try_send` fails because the notification is still sitting in
-    /// the channel.
-    #[tokio::test]
-    async fn exex_notifications_backfill_drains_channel() -> eyre::Result<()> {
-        let mut rng = generators::rng();
-
-        let temp_dir = tempfile::tempdir().unwrap();
-        let wal = Wal::new(temp_dir.path()).unwrap();
-
-        let provider_factory = create_test_provider_factory();
-        let genesis_hash = init_genesis(&provider_factory)?;
-        let genesis_block = provider_factory
-            .block(genesis_hash.into())?
-            .ok_or_else(|| eyre::eyre!("genesis block not found"))?;
-
-        let provider = BlockchainProvider::new(provider_factory.clone())?;
-
-        // Insert block 1 into the DB so there's something to backfill
-        let node_head_block = random_block(
-            &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.commit()?;
-
-        // ExEx head is at genesis — backfill will run for block 1
-        let exex_head =
-            ExExHead { block: BlockNumHash { number: genesis_block.number, hash: genesis_hash } };
-
-        // Notification for a block AFTER the backfill range (block 2).
-        let post_backfill_notification = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(
-                vec![random_block(
-                    &mut rng,
-                    node_head.number + 1,
-                    BlockParams { parent: Some(node_head.hash), ..Default::default() },
-                )
-                .try_recover()?],
-                Default::default(),
-                BTreeMap::new(),
-            )),
-        };
-
-        // Another notification (block 3) used to probe channel capacity.
-        let probe_notification = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(
-                vec![random_block(
-                    &mut rng,
-                    node_head.number + 2,
-                    BlockParams { parent: None, ..Default::default() },
-                )
-                .try_recover()?],
-                Default::default(),
-                BTreeMap::new(),
-            )),
-        };
-
-        let (notifications_tx, notifications_rx) = mpsc::channel(1);
-
-        // Fill the capacity-1 channel.
-        notifications_tx.send(post_backfill_notification.clone()).await?;
-
-        // Confirm the channel is full — this is the precondition that causes the
-        // stall in production: the ExExManager's PollSender would block here.
-        assert!(
-            notifications_tx.try_send(probe_notification.clone()).is_err(),
-            "channel should be full before backfill poll"
-        );
-
-        let mut notifications = ExExNotificationsWithoutHead::new(
-            node_head,
-            provider,
-            EthEvmConfig::mainnet(),
-            notifications_rx,
-            wal.handle(),
-        )
-        .with_head(exex_head);
-
-        // Poll once — this returns the backfill result for block 1. Crucially,
-        // the drain loop in poll_next runs in this same call, consuming the
-        // notification from the channel and buffering it.
-        let backfill_result = notifications.next().await.transpose()?;
-        assert_eq!(
-            backfill_result,
-            Some(ExExNotification::ChainCommitted {
-                new: Arc::new(
-                    BackfillJobFactory::new(
-                        notifications.evm_config.clone(),
-                        notifications.provider.clone()
-                    )
-                    .backfill(1..=1)
-                    .next()
-                    .ok_or_eyre("failed to backfill")??
-                )
-            })
-        );
-
-        // KEY ASSERTION: the channel was drained during the backfill poll above.
-        // Without the drain loop this try_send fails because the original
-        // notification is still occupying the capacity-1 channel.
-        assert!(
-            notifications_tx.try_send(probe_notification.clone()).is_ok(),
-            "channel should have been drained during backfill poll"
-        );
-
-        // The first buffered notification (block 2) was drained from the channel
-        // during backfill and is delivered now.
-        let buffered = notifications.next().await.transpose()?;
-        assert_eq!(buffered, Some(post_backfill_notification));
-
-        // The probe notification (block 3) that we just sent is delivered next.
-        let probe = notifications.next().await.transpose()?;
-        assert_eq!(probe, Some(probe_notification));
-
-        Ok(())
-    }
 }

crates/net/eth-wire-types/src/block_access_lists.rs

@@ -1,27 +0,0 @@
-//! Implements the `GetBlockAccessLists` and `BlockAccessLists` message types.
-
-use alloc::vec::Vec;
-use alloy_primitives::{Bytes, B256};
-use alloy_rlp::{RlpDecodableWrapper, RlpEncodableWrapper};
-use reth_codecs_derive::add_arbitrary_tests;
-
-/// A request for block access lists from the given block hashes.
-#[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Default)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
-#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
-#[add_arbitrary_tests(rlp)]
-pub struct GetBlockAccessLists(
-    /// The block hashes to request block access lists for.
-    pub Vec<B256>,
-);
-
-/// Response for [`GetBlockAccessLists`] containing one BAL per requested block hash.
-#[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Default)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
-#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
-#[add_arbitrary_tests(rlp)]
-pub struct BlockAccessLists(
-    /// The requested block access lists as opaque bytes. Unavailable entries are represented by
-    /// empty byte slices.
-    pub Vec<Bytes>,
-);

crates/net/eth-wire-types/src/broadcast.rs

@@ -169,10 +169,7 @@ impl NewPooledTransactionHashes {
                 matches!(version, EthVersion::Eth67 | EthVersion::Eth66)
             }
             Self::Eth68(_) => {
-                matches!(
-                    version,
-                    EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 | EthVersion::Eth71
-                )
+                matches!(version, EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70)
             }
         }
     }

crates/net/eth-wire-types/src/capability.rs

@@ -110,11 +110,6 @@ impl Capability {
         Self::eth(EthVersion::Eth70)
     }
 
-    /// Returns the [`EthVersion::Eth71`] capability.
-    pub const fn eth_71() -> Self {
-        Self::eth(EthVersion::Eth71)
-    }
-
     /// Whether this is eth v66 protocol.
     #[inline]
     pub fn is_eth_v66(&self) -> bool {
@@ -145,12 +140,6 @@ impl Capability {
         self.name == "eth" && self.version == 70
     }
 
-    /// Whether this is eth v71.
-    #[inline]
-    pub fn is_eth_v71(&self) -> bool {
-        self.name == "eth" && self.version == 71
-    }
-
     /// Whether this is any eth version.
     #[inline]
     pub fn is_eth(&self) -> bool {
@@ -158,8 +147,7 @@ impl Capability {
             self.is_eth_v67() ||
             self.is_eth_v68() ||
             self.is_eth_v69() ||
-            self.is_eth_v70() ||
-            self.is_eth_v71()
+            self.is_eth_v70()
     }
 }
 
@@ -179,7 +167,7 @@ impl From<EthVersion> for Capability {
 #[cfg(any(test, feature = "arbitrary"))]
 impl<'a> arbitrary::Arbitrary<'a> for Capability {
     fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
-        let version = u.int_in_range(66..=71)?; // Valid eth protocol versions are 66-71
+        let version = u.int_in_range(66..=70)?; // Valid eth protocol versions are 66-70
                                                 // Only generate valid eth protocol name for now since it's the only supported protocol
         Ok(Self::new_static("eth", version))
     }
@@ -195,7 +183,6 @@ pub struct Capabilities {
     eth_68: bool,
     eth_69: bool,
     eth_70: bool,
-    eth_71: bool,
 }
 
 impl Capabilities {
@@ -207,7 +194,6 @@ impl Capabilities {
             eth_68: value.iter().any(Capability::is_eth_v68),
             eth_69: value.iter().any(Capability::is_eth_v69),
             eth_70: value.iter().any(Capability::is_eth_v70),
-            eth_71: value.iter().any(Capability::is_eth_v71),
             inner: value,
         }
     }
@@ -226,7 +212,7 @@ impl Capabilities {
     /// Whether the peer supports `eth` sub-protocol.
     #[inline]
     pub const fn supports_eth(&self) -> bool {
-        self.eth_71 || self.eth_70 || self.eth_69 || self.eth_68 || self.eth_67 || self.eth_66
+        self.eth_70 || self.eth_69 || self.eth_68 || self.eth_67 || self.eth_66
     }
 
     /// Whether this peer supports eth v66 protocol.
@@ -258,12 +244,6 @@ impl Capabilities {
     pub const fn supports_eth_v70(&self) -> bool {
         self.eth_70
     }
-
-    /// Whether this peer supports eth v71 protocol.
-    #[inline]
-    pub const fn supports_eth_v71(&self) -> bool {
-        self.eth_71
-    }
 }
 
 impl From<Vec<Capability>> for Capabilities {
@@ -288,7 +268,6 @@ impl Decodable for Capabilities {
             eth_68: inner.iter().any(Capability::is_eth_v68),
             eth_69: inner.iter().any(Capability::is_eth_v69),
             eth_70: inner.iter().any(Capability::is_eth_v70),
-            eth_71: inner.iter().any(Capability::is_eth_v71),
             inner,
         })
     }

crates/net/eth-wire-types/src/lib.rs

@@ -38,9 +38,6 @@ pub use state::*;
 pub mod receipts;
 pub use receipts::*;
 
-pub mod block_access_lists;
-pub use block_access_lists::*;
-
 pub mod disconnect_reason;
 pub use disconnect_reason::*;
 

crates/net/eth-wire-types/src/message.rs

@@ -1,4 +1,4 @@
-//! Implements Ethereum wire protocol for versions 66 through 71.
+//! Implements Ethereum wire protocol for versions 66 through 70.
 //! Defines structs/enums for messages, request-response pairs, and broadcasts.
 //! Handles compatibility with [`EthVersion`].
 //!
@@ -7,10 +7,10 @@
 //! Reference: [Ethereum Wire Protocol](https://github.com/ethereum/devp2p/blob/master/caps/eth.md).
 
 use super::{
-    broadcast::NewBlockHashes, BlockAccessLists, BlockBodies, BlockHeaders, GetBlockAccessLists,
-    GetBlockBodies, GetBlockHeaders, GetNodeData, GetPooledTransactions, GetReceipts,
-    GetReceipts70, NewPooledTransactionHashes66, NewPooledTransactionHashes68, NodeData,
-    PooledTransactions, Receipts, Status, StatusEth69, Transactions,
+    broadcast::NewBlockHashes, BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders,
+    GetNodeData, GetPooledTransactions, GetReceipts, GetReceipts70, NewPooledTransactionHashes66,
+    NewPooledTransactionHashes68, NodeData, PooledTransactions, Receipts, Status, StatusEth69,
+    Transactions,
 };
 use crate::{
     status::StatusMessage, BlockRangeUpdate, EthNetworkPrimitives, EthVersion, NetworkPrimitives,
@@ -168,18 +168,6 @@ impl<N: NetworkPrimitives> ProtocolMessage<N> {
                 }
                 EthMessage::BlockRangeUpdate(BlockRangeUpdate::decode(buf)?)
             }
-            EthMessageID::GetBlockAccessLists => {
-                if version < EthVersion::Eth71 {
-                    return Err(MessageError::Invalid(version, EthMessageID::GetBlockAccessLists))
-                }
-                EthMessage::GetBlockAccessLists(RequestPair::decode(buf)?)
-            }
-            EthMessageID::BlockAccessLists => {
-                if version < EthVersion::Eth71 {
-                    return Err(MessageError::Invalid(version, EthMessageID::BlockAccessLists))
-                }
-                EthMessage::BlockAccessLists(RequestPair::decode(buf)?)
-            }
             EthMessageID::Other(_) => {
                 let raw_payload = Bytes::copy_from_slice(buf);
                 buf.advance(raw_payload.len());
@@ -262,8 +250,6 @@ impl<N: NetworkPrimitives> From<EthBroadcastMessage<N>> for ProtocolBroadcastMes
 ///
 /// The `eth/70` (EIP-7975) keeps the eth/69 status format and introduces partial receipts.
 /// requests/responses.
-///
-/// The `eth/71` draft extends eth/70 with block access list request/response messages.
 #[derive(Clone, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
@@ -324,8 +310,6 @@ pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
     /// `GetReceipts` in EIP-7975 inlines the request id. The type still wraps
     /// a [`RequestPair`], but with a custom inline encoding.
     GetReceipts70(RequestPair<GetReceipts70>),
-    /// Represents a `GetBlockAccessLists` request-response pair for eth/71.
-    GetBlockAccessLists(RequestPair<GetBlockAccessLists>),
     /// Represents a Receipts request-response pair.
     #[cfg_attr(
         feature = "serde",
@@ -348,8 +332,6 @@ pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
     /// request id. The type still wraps a [`RequestPair`], but with a custom
     /// inline encoding.
     Receipts70(RequestPair<Receipts70<N::Receipt>>),
-    /// Represents a `BlockAccessLists` request-response pair for eth/71.
-    BlockAccessLists(RequestPair<BlockAccessLists>),
     /// Represents a `BlockRangeUpdate` message broadcast to the network.
     #[cfg_attr(
         feature = "serde",
@@ -382,8 +364,6 @@ impl<N: NetworkPrimitives> EthMessage<N> {
             Self::GetReceipts(_) | Self::GetReceipts70(_) => EthMessageID::GetReceipts,
             Self::Receipts(_) | Self::Receipts69(_) | Self::Receipts70(_) => EthMessageID::Receipts,
             Self::BlockRangeUpdate(_) => EthMessageID::BlockRangeUpdate,
-            Self::GetBlockAccessLists(_) => EthMessageID::GetBlockAccessLists,
-            Self::BlockAccessLists(_) => EthMessageID::BlockAccessLists,
             Self::Other(msg) => EthMessageID::Other(msg.id as u8),
         }
     }
@@ -396,7 +376,6 @@ impl<N: NetworkPrimitives> EthMessage<N> {
                 Self::GetBlockHeaders(_) |
                 Self::GetReceipts(_) |
                 Self::GetReceipts70(_) |
-                Self::GetBlockAccessLists(_) |
                 Self::GetPooledTransactions(_) |
                 Self::GetNodeData(_)
         )
@@ -410,7 +389,6 @@ impl<N: NetworkPrimitives> EthMessage<N> {
                 Self::Receipts(_) |
                 Self::Receipts69(_) |
                 Self::Receipts70(_) |
-                Self::BlockAccessLists(_) |
                 Self::BlockHeaders(_) |
                 Self::BlockBodies(_) |
                 Self::NodeData(_)
@@ -465,11 +443,9 @@ impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
             Self::NodeData(data) => data.encode(out),
             Self::GetReceipts(request) => request.encode(out),
             Self::GetReceipts70(request) => request.encode(out),
-            Self::GetBlockAccessLists(request) => request.encode(out),
             Self::Receipts(receipts) => receipts.encode(out),
             Self::Receipts69(receipt69) => receipt69.encode(out),
             Self::Receipts70(receipt70) => receipt70.encode(out),
-            Self::BlockAccessLists(block_access_lists) => block_access_lists.encode(out),
             Self::BlockRangeUpdate(block_range_update) => block_range_update.encode(out),
             Self::Other(unknown) => out.put_slice(&unknown.payload),
         }
@@ -492,11 +468,9 @@ impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
             Self::NodeData(data) => data.length(),
             Self::GetReceipts(request) => request.length(),
             Self::GetReceipts70(request) => request.length(),
-            Self::GetBlockAccessLists(request) => request.length(),
             Self::Receipts(receipts) => receipts.length(),
             Self::Receipts69(receipt69) => receipt69.length(),
             Self::Receipts70(receipt70) => receipt70.length(),
-            Self::BlockAccessLists(block_access_lists) => block_access_lists.length(),
             Self::BlockRangeUpdate(block_range_update) => block_range_update.length(),
             Self::Other(unknown) => unknown.length(),
         }
@@ -585,14 +559,6 @@ pub enum EthMessageID {
     ///
     /// Introduced in Eth69
     BlockRangeUpdate = 0x11,
-    /// Requests block access lists.
-    ///
-    /// Introduced in Eth71
-    GetBlockAccessLists = 0x12,
-    /// Represents block access lists.
-    ///
-    /// Introduced in Eth71
-    BlockAccessLists = 0x13,
     /// Represents unknown message types.
     Other(u8),
 }
@@ -617,17 +583,13 @@ impl EthMessageID {
             Self::GetReceipts => 0x0f,
             Self::Receipts => 0x10,
             Self::BlockRangeUpdate => 0x11,
-            Self::GetBlockAccessLists => 0x12,
-            Self::BlockAccessLists => 0x13,
             Self::Other(value) => *value, // Return the stored `u8`
         }
     }
 
     /// Returns the max value for the given version.
     pub const fn max(version: EthVersion) -> u8 {
-        if version.is_eth71() {
-            Self::BlockAccessLists.to_u8()
-        } else if version.is_eth69_or_newer() {
+        if version.is_eth69() {
             Self::BlockRangeUpdate.to_u8()
         } else {
             Self::Receipts.to_u8()
@@ -672,8 +634,6 @@ impl Decodable for EthMessageID {
             0x0f => Self::GetReceipts,
             0x10 => Self::Receipts,
             0x11 => Self::BlockRangeUpdate,
-            0x12 => Self::GetBlockAccessLists,
-            0x13 => Self::BlockAccessLists,
             unknown => Self::Other(*unknown),
         };
         buf.advance(1);
@@ -702,8 +662,6 @@ impl TryFrom<usize> for EthMessageID {
             0x0f => Ok(Self::GetReceipts),
             0x10 => Ok(Self::Receipts),
             0x11 => Ok(Self::BlockRangeUpdate),
-            0x12 => Ok(Self::GetBlockAccessLists),
-            0x13 => Ok(Self::BlockAccessLists),
             _ => Err("Invalid message ID"),
         }
     }
@@ -784,9 +742,8 @@ where
 mod tests {
     use super::MessageError;
     use crate::{
-        message::RequestPair, BlockAccessLists, EthMessage, EthMessageID, EthNetworkPrimitives,
-        EthVersion, GetBlockAccessLists, GetNodeData, NodeData, ProtocolMessage,
-        RawCapabilityMessage,
+        message::RequestPair, EthMessage, EthMessageID, EthNetworkPrimitives, EthVersion,
+        GetNodeData, NodeData, ProtocolMessage, RawCapabilityMessage,
     };
     use alloy_primitives::hex;
     use alloy_rlp::{Decodable, Encodable, Error};
@@ -827,60 +784,6 @@ mod tests {
         assert!(matches!(msg, Err(MessageError::Invalid(..))));
     }
 
-    #[test]
-    fn test_bal_message_version_gating() {
-        let get_block_access_lists =
-            EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(RequestPair {
-                request_id: 1337,
-                message: GetBlockAccessLists(vec![]),
-            });
-        let buf = encode(ProtocolMessage {
-            message_type: EthMessageID::GetBlockAccessLists,
-            message: get_block_access_lists,
-        });
-        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
-            EthVersion::Eth70,
-            &mut &buf[..],
-        );
-        assert!(matches!(
-            msg,
-            Err(MessageError::Invalid(EthVersion::Eth70, EthMessageID::GetBlockAccessLists))
-        ));
-
-        let block_access_lists =
-            EthMessage::<EthNetworkPrimitives>::BlockAccessLists(RequestPair {
-                request_id: 1337,
-                message: BlockAccessLists(vec![]),
-            });
-        let buf = encode(ProtocolMessage {
-            message_type: EthMessageID::BlockAccessLists,
-            message: block_access_lists,
-        });
-        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
-            EthVersion::Eth70,
-            &mut &buf[..],
-        );
-        assert!(matches!(
-            msg,
-            Err(MessageError::Invalid(EthVersion::Eth70, EthMessageID::BlockAccessLists))
-        ));
-    }
-
-    #[test]
-    fn test_bal_message_eth71_roundtrip() {
-        let msg = ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(
-            RequestPair { request_id: 42, message: GetBlockAccessLists(vec![]) },
-        ));
-        let encoded = encode(msg.clone());
-        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
-            EthVersion::Eth71,
-            &mut &encoded[..],
-        )
-        .unwrap();
-
-        assert_eq!(decoded, msg);
-    }
-
     #[test]
     fn request_pair_encode() {
         let request_pair = RequestPair { request_id: 1337, message: vec![5u8] };
@@ -1034,13 +937,6 @@ mod tests {
         assert!(matches!(decoded, StatusMessage::Legacy(s) if s == status));
     }
 
-    #[test]
-    fn eth_message_id_max_includes_block_range_update() {
-        assert_eq!(EthMessageID::max(EthVersion::Eth69), EthMessageID::BlockRangeUpdate.to_u8(),);
-        assert_eq!(EthMessageID::max(EthVersion::Eth70), EthMessageID::BlockRangeUpdate.to_u8(),);
-        assert_eq!(EthMessageID::max(EthVersion::Eth68), EthMessageID::Receipts.to_u8());
-    }
-
     #[test]
     fn decode_status_rejects_non_status() {
         let msg = EthMessage::<EthNetworkPrimitives>::GetBlockBodies(RequestPair {

crates/net/eth-wire-types/src/version.rs

@@ -29,8 +29,6 @@ pub enum EthVersion {
     Eth69 = 69,
     /// The `eth` protocol version 70.
     Eth70 = 70,
-    /// The `eth` protocol version 71.
-    Eth71 = 71,
 }
 
 impl EthVersion {
@@ -64,19 +62,9 @@ impl EthVersion {
     pub const fn is_eth70(&self) -> bool {
         matches!(self, Self::Eth70)
     }
-
-    /// Returns true if the version is eth/71
-    pub const fn is_eth71(&self) -> bool {
-        matches!(self, Self::Eth71)
-    }
-
-    /// Returns true if the version is eth/69 or newer.
-    pub const fn is_eth69_or_newer(&self) -> bool {
-        matches!(self, Self::Eth69 | Self::Eth70 | Self::Eth71)
-    }
 }
 
-/// RLP encodes `EthVersion` as a single byte (66-71).
+/// RLP encodes `EthVersion` as a single byte (66-69).
 impl Encodable for EthVersion {
     fn encode(&self, out: &mut dyn BufMut) {
         (*self as u8).encode(out)
@@ -88,7 +76,7 @@ impl Encodable for EthVersion {
 }
 
 /// RLP decodes a single byte into `EthVersion`.
-/// Returns error if byte is not a valid version (66-71).
+/// Returns error if byte is not a valid version (66-69).
 impl Decodable for EthVersion {
     fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
         let version = u8::decode(buf)?;
@@ -116,7 +104,6 @@ impl TryFrom<&str> for EthVersion {
             "68" => Ok(Self::Eth68),
             "69" => Ok(Self::Eth69),
             "70" => Ok(Self::Eth70),
-            "71" => Ok(Self::Eth71),
             _ => Err(ParseVersionError(s.to_string())),
         }
     }
@@ -142,7 +129,6 @@ impl TryFrom<u8> for EthVersion {
             68 => Ok(Self::Eth68),
             69 => Ok(Self::Eth69),
             70 => Ok(Self::Eth70),
-            71 => Ok(Self::Eth71),
             _ => Err(ParseVersionError(u.to_string())),
         }
     }
@@ -173,7 +159,6 @@ impl From<EthVersion> for &'static str {
             EthVersion::Eth68 => "68",
             EthVersion::Eth69 => "69",
             EthVersion::Eth70 => "70",
-            EthVersion::Eth71 => "71",
         }
     }
 }
@@ -231,7 +216,6 @@ mod tests {
         assert_eq!(EthVersion::Eth68, EthVersion::try_from("68").unwrap());
         assert_eq!(EthVersion::Eth69, EthVersion::try_from("69").unwrap());
         assert_eq!(EthVersion::Eth70, EthVersion::try_from("70").unwrap());
-        assert_eq!(EthVersion::Eth71, EthVersion::try_from("71").unwrap());
     }
 
     #[test]
@@ -241,7 +225,6 @@ mod tests {
         assert_eq!(EthVersion::Eth68, "68".parse().unwrap());
         assert_eq!(EthVersion::Eth69, "69".parse().unwrap());
         assert_eq!(EthVersion::Eth70, "70".parse().unwrap());
-        assert_eq!(EthVersion::Eth71, "71".parse().unwrap());
     }
 
     #[test]
@@ -252,7 +235,6 @@ mod tests {
             EthVersion::Eth68,
             EthVersion::Eth69,
             EthVersion::Eth70,
-            EthVersion::Eth71,
         ];
 
         for version in versions {
@@ -271,7 +253,6 @@ mod tests {
             (68_u8, Ok(EthVersion::Eth68)),
             (69_u8, Ok(EthVersion::Eth69)),
             (70_u8, Ok(EthVersion::Eth70)),
-            (71_u8, Ok(EthVersion::Eth71)),
             (65_u8, Err(RlpError::Custom("invalid eth version"))),
         ];
 

crates/net/eth-wire/src/eth_snap_stream.rs

@@ -294,8 +294,7 @@ mod tests {
     use alloy_primitives::B256;
     use alloy_rlp::Encodable;
     use reth_eth_wire_types::{
-        message::RequestPair, GetAccountRangeMessage, GetBlockAccessLists, GetBlockHeaders,
-        HeadersDirection,
+        message::RequestPair, GetAccountRangeMessage, GetBlockHeaders, HeadersDirection,
     };
 
     // Helper to create eth message and its bytes
@@ -420,40 +419,4 @@ mod tests {
         let snap_boundary_result = inner.decode_message(snap_boundary_bytes);
         assert!(snap_boundary_result.is_err());
     }
-
-    #[test]
-    fn test_eth70_message_id_0x12_is_snap() {
-        let inner = EthSnapStreamInner::<EthNetworkPrimitives>::new(EthVersion::Eth70);
-        let snap_msg = SnapProtocolMessage::GetAccountRange(GetAccountRangeMessage {
-            request_id: 1,
-            root_hash: B256::default(),
-            starting_hash: B256::default(),
-            limit_hash: B256::default(),
-            response_bytes: 1000,
-        });
-
-        let encoded = inner.encode_snap_message(snap_msg);
-        assert_eq!(encoded[0], EthMessageID::message_count(EthVersion::Eth70));
-
-        let decoded = inner.decode_message(BytesMut::from(&encoded[..])).unwrap();
-        assert!(matches!(decoded, EthSnapMessage::Snap(_)));
-    }
-
-    #[test]
-    fn test_eth71_message_id_0x12_is_eth() {
-        let inner = EthSnapStreamInner::<EthNetworkPrimitives>::new(EthVersion::Eth71);
-        let eth_msg = EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(RequestPair {
-            request_id: 1,
-            message: GetBlockAccessLists(vec![B256::ZERO]),
-        });
-        let protocol_msg = ProtocolMessage::from(eth_msg.clone());
-        let mut buf = Vec::new();
-        protocol_msg.encode(&mut buf);
-
-        let decoded = inner.decode_message(BytesMut::from(&buf[..])).unwrap();
-        let EthSnapMessage::Eth(decoded_eth) = decoded else {
-            panic!("expected eth message");
-        };
-        assert_eq!(decoded_eth, eth_msg);
-    }
 }

crates/net/eth-wire/src/protocol.rs

@@ -84,7 +84,5 @@ mod tests {
         assert_eq!(Protocol::eth(EthVersion::Eth67).messages(), 17);
         assert_eq!(Protocol::eth(EthVersion::Eth68).messages(), 17);
         assert_eq!(Protocol::eth(EthVersion::Eth69).messages(), 18);
-        assert_eq!(Protocol::eth(EthVersion::Eth70).messages(), 18);
-        assert_eq!(Protocol::eth(EthVersion::Eth71).messages(), 20);
     }
 }

crates/net/network-api/src/events.rs

@@ -1,15 +1,14 @@
 //! API related to listening for network events.
 
 use reth_eth_wire_types::{
-    message::RequestPair, BlockAccessLists, BlockBodies, BlockHeaders, Capabilities,
-    DisconnectReason, EthMessage, EthNetworkPrimitives, EthVersion, GetBlockAccessLists,
-    GetBlockBodies, GetBlockHeaders, GetNodeData, GetPooledTransactions, GetReceipts,
-    GetReceipts70, NetworkPrimitives, NodeData, PooledTransactions, Receipts, Receipts69,
-    Receipts70, UnifiedStatus,
+    message::RequestPair, BlockBodies, BlockHeaders, Capabilities, DisconnectReason, EthMessage,
+    EthNetworkPrimitives, EthVersion, GetBlockBodies, GetBlockHeaders, GetNodeData,
+    GetPooledTransactions, GetReceipts, GetReceipts70, NetworkPrimitives, NodeData,
+    PooledTransactions, Receipts, Receipts69, Receipts70, UnifiedStatus,
 };
 use reth_ethereum_forks::ForkId;
 use reth_network_p2p::error::{RequestError, RequestResult};
-use reth_network_peers::{NodeRecord, PeerId};
+use reth_network_peers::PeerId;
 use reth_network_types::{PeerAddr, PeerKind};
 use reth_tokio_util::EventStream;
 use std::{
@@ -153,13 +152,8 @@ pub trait NetworkEventListenerProvider: NetworkPeersEvents {
 pub enum DiscoveryEvent {
     /// Discovered a node
     NewNode(DiscoveredEvent),
-    /// Retrieved a [`ForkId`] from the peer via ENR request.
-    ///
-    /// Contains the full [`NodeRecord`] (peer ID + address) and the reported [`ForkId`].
-    /// Used to verify fork compatibility before admitting the peer.
-    ///
-    /// See also <https://eips.ethereum.org/EIPS/eip-868>
-    EnrForkId(NodeRecord, ForkId),
+    /// Retrieved a [`ForkId`] from the peer via ENR request, See <https://eips.ethereum.org/EIPS/eip-868>
+    EnrForkId(PeerId, ForkId),
 }
 
 /// Represents events related to peer discovery in the network.
@@ -253,15 +247,6 @@ pub enum PeerRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
         /// The channel to send the response for receipts.
         response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
     },
-    /// Requests block access lists from the peer.
-    ///
-    /// The response should be sent through the channel.
-    GetBlockAccessLists {
-        /// The request for block access lists.
-        request: GetBlockAccessLists,
-        /// The channel to send the response for block access lists.
-        response: oneshot::Sender<RequestResult<BlockAccessLists>>,
-    },
 }
 
 // === impl PeerRequest ===
@@ -282,19 +267,9 @@ impl<N: NetworkPrimitives> PeerRequest<N> {
             Self::GetReceipts { response, .. } => response.send(Err(err)).ok(),
             Self::GetReceipts69 { response, .. } => response.send(Err(err)).ok(),
             Self::GetReceipts70 { response, .. } => response.send(Err(err)).ok(),
-            Self::GetBlockAccessLists { response, .. } => response.send(Err(err)).ok(),
         };
     }
 
-    /// Returns true if this request is supported for the negotiated eth protocol version.
-    #[inline]
-    pub fn is_supported_by_eth_version(&self, version: EthVersion) -> bool {
-        match self {
-            Self::GetBlockAccessLists { .. } => version >= EthVersion::Eth71,
-            _ => true,
-        }
-    }
-
     /// Returns the [`EthMessage`] for this type
     pub fn create_request_message(&self, request_id: u64) -> EthMessage<N> {
         match self {
@@ -319,12 +294,6 @@ impl<N: NetworkPrimitives> PeerRequest<N> {
             Self::GetReceipts70 { request, .. } => {
                 EthMessage::GetReceipts70(RequestPair { request_id, message: request.clone() })
             }
-            Self::GetBlockAccessLists { request, .. } => {
-                EthMessage::GetBlockAccessLists(RequestPair {
-                    request_id,
-                    message: request.clone(),
-                })
-            }
         }
     }
 
@@ -375,18 +344,3 @@ impl<R> fmt::Debug for PeerRequestSender<R> {
         f.debug_struct("PeerRequestSender").field("peer_id", &self.peer_id).finish_non_exhaustive()
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_get_block_access_lists_version_support() {
-        let (tx, _rx) = oneshot::channel();
-        let req: PeerRequest<EthNetworkPrimitives> =
-            PeerRequest::GetBlockAccessLists { request: GetBlockAccessLists(vec![]), response: tx };
-
-        assert!(!req.is_supported_by_eth_version(EthVersion::Eth70));
-        assert!(req.is_supported_by_eth_version(EthVersion::Eth71));
-    }
-}

crates/net/network-types/src/peers/config.rs

@@ -172,11 +172,6 @@ pub struct PeersConfig {
     /// IPs within the specified CIDR ranges will be allowed.
     #[cfg_attr(feature = "serde", serde(skip))]
     pub ip_filter: IpFilter,
-    /// If true, discovered peers without a confirmed ENR [`ForkId`](alloy_eip2124::ForkId)
-    /// (EIP-868) will not be added to the peer set until their fork ID is verified.
-    ///
-    /// This filters out peers from other networks that pollute the discovery table.
-    pub enforce_enr_fork_id: bool,
 }
 
 impl Default for PeersConfig {
@@ -196,7 +191,6 @@ impl Default for PeersConfig {
             max_backoff_count: 5,
             incoming_ip_throttle_duration: INBOUND_IP_THROTTLE_DURATION,
             ip_filter: IpFilter::default(),
-            enforce_enr_fork_id: false,
         }
     }
 }
@@ -320,13 +314,6 @@ impl PeersConfig {
         self
     }
 
-    /// If set, discovered peers without a confirmed ENR [`ForkId`](alloy_eip2124::ForkId) will not
-    /// be added to the peer set until their fork ID is verified via EIP-868.
-    pub const fn with_enforce_enr_fork_id(mut self, enforce: bool) -> Self {
-        self.enforce_enr_fork_id = enforce;
-        self
-    }
-
     /// Returns settings for testing
     #[cfg(any(test, feature = "test-utils"))]
     pub fn test() -> Self {

crates/net/network/src/discovery.rs

@@ -240,7 +240,7 @@ impl Discovery {
                 self.on_node_record_update(record, None);
             }
             DiscoveryUpdate::EnrForkId(node, fork_id) => {
-                self.queued_events.push_back(DiscoveryEvent::EnrForkId(node, fork_id))
+                self.queued_events.push_back(DiscoveryEvent::EnrForkId(node.id, fork_id))
             }
             DiscoveryUpdate::Removed(peer_id) => {
                 self.discovered_nodes.remove(&peer_id);

crates/net/network/src/eth_requests.rs

@@ -6,13 +6,12 @@ use crate::{
 };
 use alloy_consensus::{BlockHeader, ReceiptWithBloom};
 use alloy_eips::BlockHashOrNumber;
-use alloy_primitives::Bytes;
 use alloy_rlp::Encodable;
 use futures::StreamExt;
 use reth_eth_wire::{
-    BlockAccessLists, BlockBodies, BlockHeaders, EthNetworkPrimitives, GetBlockAccessLists,
-    GetBlockBodies, GetBlockHeaders, GetNodeData, GetReceipts, GetReceipts70, HeadersDirection,
-    NetworkPrimitives, NodeData, Receipts, Receipts69, Receipts70,
+    BlockBodies, BlockHeaders, EthNetworkPrimitives, GetBlockBodies, GetBlockHeaders, GetNodeData,
+    GetReceipts, GetReceipts70, HeadersDirection, NetworkPrimitives, NodeData, Receipts,
+    Receipts69, Receipts70,
 };
 use reth_network_api::test_utils::PeersHandle;
 use reth_network_p2p::error::RequestResult;
@@ -282,19 +281,6 @@ where
         let _ = response.send(Ok(Receipts70 { last_block_incomplete, receipts }));
     }
 
-    /// Handles [`GetBlockAccessLists`] queries.
-    ///
-    /// For now this returns one empty BAL per requested hash.
-    fn on_block_access_lists_request(
-        &self,
-        _peer_id: PeerId,
-        request: GetBlockAccessLists,
-        response: oneshot::Sender<RequestResult<BlockAccessLists>>,
-    ) {
-        let access_lists = request.0.into_iter().map(|_| Bytes::new()).collect();
-        let _ = response.send(Ok(BlockAccessLists(access_lists)));
-    }
-
     #[inline]
     fn get_receipts_response<T, F>(&self, request: GetReceipts, transform_fn: F) -> Vec<Vec<T>>
     where
@@ -366,9 +352,6 @@ where
                     IncomingEthRequest::GetReceipts70 { peer_id, request, response } => {
                         this.on_receipts70_request(peer_id, request, response)
                     }
-                    IncomingEthRequest::GetBlockAccessLists { peer_id, request, response } => {
-                        this.on_block_access_lists_request(peer_id, request, response)
-                    }
                 }
             },
         );
@@ -454,15 +437,4 @@ pub enum IncomingEthRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
         /// The channel sender for the response containing Receipts70.
         response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
     },
-    /// Request Block Access Lists from the peer.
-    ///
-    /// The response should be sent through the channel.
-    GetBlockAccessLists {
-        /// The ID of the peer to request block access lists from.
-        peer_id: PeerId,
-        /// The requested block hashes.
-        request: GetBlockAccessLists,
-        /// The channel sender for the response containing block access lists.
-        response: oneshot::Sender<RequestResult<BlockAccessLists>>,
-    },
 }

crates/net/network/src/manager.rs

@@ -25,7 +25,7 @@ use crate::{
     listener::ConnectionListener,
     message::{NewBlockMessage, PeerMessage},
     metrics::{
-        BackedOffPeersMetrics, ClosedSessionsMetrics, DirectionalDisconnectMetrics, NetworkMetrics,
+        BackedOffPeersMetrics, ClosedSessionsMetrics, DisconnectMetrics, NetworkMetrics,
         PendingSessionFailureMetrics, NETWORK_POOL_TRANSACTIONS_SCOPE,
     },
     network::{NetworkHandle, NetworkHandleMessage},
@@ -140,8 +140,8 @@ pub struct NetworkManager<N: NetworkPrimitives = EthNetworkPrimitives> {
     num_active_peers: Arc<AtomicUsize>,
     /// Metrics for the Network
     metrics: NetworkMetrics,
-    /// Disconnect metrics for the Network, split by connection direction.
-    disconnect_metrics: DirectionalDisconnectMetrics,
+    /// Disconnect metrics for the Network
+    disconnect_metrics: DisconnectMetrics,
     /// Closed sessions metrics, split by direction.
     closed_sessions_metrics: ClosedSessionsMetrics,
     /// Pending session failure metrics, split by direction.
@@ -551,13 +551,6 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                     response,
                 })
             }
-            PeerRequest::GetBlockAccessLists { request, response } => {
-                self.delegate_eth_request(IncomingEthRequest::GetBlockAccessLists {
-                    peer_id,
-                    request,
-                    response,
-                })
-            }
             PeerRequest::GetPooledTransactions { request, response } => {
                 self.notify_tx_manager(NetworkTransactionEvent::GetPooledTransactions {
                     peer_id,
@@ -871,9 +864,6 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                     "Session disconnected"
                 );
 
-                // Capture direction before state is reset to Idle
-                let is_inbound = self.swarm.state().peers().is_inbound_peer(&peer_id);
-
                 let reason = if let Some(ref err) = error {
                     // If the connection was closed due to an error, we report
                     // the peer
@@ -897,11 +887,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                 self.update_active_connection_metrics();
 
                 if let Some(reason) = reason {
-                    if is_inbound {
-                        self.disconnect_metrics.increment_inbound(reason);
-                    } else {
-                        self.disconnect_metrics.increment_outbound(reason);
-                    }
+                    self.disconnect_metrics.increment(reason);
                 }
                 self.metrics
                     .backed_off_peers
@@ -924,7 +910,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                         .on_incoming_pending_session_dropped(remote_addr, err);
                     self.pending_session_failure_metrics.inbound.increment(1);
                     if let Some(reason) = err.as_disconnected() {
-                        self.disconnect_metrics.increment_inbound(reason);
+                        self.disconnect_metrics.increment(reason);
                     }
                 } else {
                     self.swarm
@@ -957,7 +943,7 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                         BackoffReason::from_disconnect(err.as_disconnected()),
                     );
                     if let Some(reason) = err.as_disconnected() {
-                        self.disconnect_metrics.increment_outbound(reason);
+                        self.disconnect_metrics.increment(reason);
                     }
                 } else {
                     self.swarm

crates/net/network/src/message.rs

@@ -3,7 +3,7 @@
 //! An `RLPx` stream is multiplexed via the prepended message-id of a framed message.
 //! Capabilities are exchanged via the `RLPx` `Hello` message as pairs of `(id, version)`, <https://github.com/ethereum/devp2p/blob/master/rlpx.md#capability-messaging>
 
-use crate::types::{BlockAccessLists, Receipts69, Receipts70};
+use crate::types::{Receipts69, Receipts70};
 use alloy_consensus::{BlockHeader, ReceiptWithBloom};
 use alloy_primitives::{Bytes, B256};
 use futures::FutureExt;
@@ -121,11 +121,6 @@ pub enum PeerResponse<N: NetworkPrimitives = EthNetworkPrimitives> {
         /// The receiver channel for the response to a receipts request.
         response: oneshot::Receiver<RequestResult<Receipts70<N::Receipt>>>,
     },
-    /// Represents a response to a request for block access lists.
-    BlockAccessLists {
-        /// The receiver channel for the response to a block access lists request.
-        response: oneshot::Receiver<RequestResult<BlockAccessLists>>,
-    },
 }
 
 // === impl PeerResponse ===
@@ -165,10 +160,6 @@ impl<N: NetworkPrimitives> PeerResponse<N> {
                 Ok(res) => PeerResponseResult::Receipts70(res),
                 Err(err) => PeerResponseResult::Receipts70(Err(err.into())),
             },
-            Self::BlockAccessLists { response } => match ready!(response.poll_unpin(cx)) {
-                Ok(res) => PeerResponseResult::BlockAccessLists(res),
-                Err(err) => PeerResponseResult::BlockAccessLists(Err(err.into())),
-            },
         };
         Poll::Ready(res)
     }
@@ -191,8 +182,6 @@ pub enum PeerResponseResult<N: NetworkPrimitives = EthNetworkPrimitives> {
     Receipts69(RequestResult<Vec<Vec<N::Receipt>>>),
     /// Represents a result containing receipts or an error for eth/70.
     Receipts70(RequestResult<Receipts70<N::Receipt>>),
-    /// Represents a result containing block access lists or an error.
-    BlockAccessLists(RequestResult<BlockAccessLists>),
 }
 
 // === impl PeerResponseResult ===
@@ -237,13 +226,6 @@ impl<N: NetworkPrimitives> PeerResponseResult<N> {
                 }
                 Err(err) => Err(err),
             },
-            Self::BlockAccessLists(resp) => match resp {
-                Ok(res) => {
-                    let request = RequestPair { request_id: id, message: res };
-                    Ok(EthMessage::BlockAccessLists(request))
-                }
-                Err(err) => Err(err),
-            },
         }
     }
 
@@ -257,7 +239,6 @@ impl<N: NetworkPrimitives> PeerResponseResult<N> {
             Self::Receipts(res) => res.as_ref().err(),
             Self::Receipts69(res) => res.as_ref().err(),
             Self::Receipts70(res) => res.as_ref().err(),
-            Self::BlockAccessLists(res) => res.as_ref().err(),
         }
     }
 

crates/net/network/src/metrics.rs

@@ -301,34 +301,6 @@ macro_rules! duration_metered_exec {
     }};
 }
 
-/// Direction-aware wrapper for disconnect metrics.
-///
-/// Tracks disconnect reasons for inbound and outbound connections separately, in addition to
-/// the combined (legacy) counters.
-#[derive(Debug, Default)]
-pub(crate) struct DirectionalDisconnectMetrics {
-    /// Combined disconnect metrics (all directions).
-    pub(crate) total: DisconnectMetrics,
-    /// Disconnect metrics for inbound connections only.
-    pub(crate) inbound: InboundDisconnectMetrics,
-    /// Disconnect metrics for outbound connections only.
-    pub(crate) outbound: OutboundDisconnectMetrics,
-}
-
-impl DirectionalDisconnectMetrics {
-    /// Increments disconnect counters for an inbound connection.
-    pub(crate) fn increment_inbound(&self, reason: DisconnectReason) {
-        self.total.increment(reason);
-        self.inbound.increment(reason);
-    }
-
-    /// Increments disconnect counters for an outbound connection.
-    pub(crate) fn increment_outbound(&self, reason: DisconnectReason) {
-        self.total.increment(reason);
-        self.outbound.increment(reason);
-    }
-}
-
 /// Metrics for Disconnection types
 ///
 /// These are just counters, and ideally we would implement these metrics on a peer-by-peer basis,
@@ -398,144 +370,6 @@ impl DisconnectMetrics {
     }
 }
 
-/// Disconnect metrics scoped to inbound connections only.
-///
-/// These counters track disconnect reasons exclusively for sessions that were initiated by
-/// remote peers connecting to this node. This helps operators distinguish between being rejected
-/// by remote peers (outbound) vs rejecting incoming peers (inbound).
-#[derive(Metrics)]
-#[metrics(scope = "network.inbound")]
-pub struct InboundDisconnectMetrics {
-    /// Number of inbound peer disconnects due to `DisconnectRequested` (0x00)
-    pub(crate) disconnect_requested: Counter,
-
-    /// Number of inbound peer disconnects due to `TcpSubsystemError` (0x01)
-    pub(crate) tcp_subsystem_error: Counter,
-
-    /// Number of inbound peer disconnects due to `ProtocolBreach` (0x02)
-    pub(crate) protocol_breach: Counter,
-
-    /// Number of inbound peer disconnects due to `UselessPeer` (0x03)
-    pub(crate) useless_peer: Counter,
-
-    /// Number of inbound peer disconnects due to `TooManyPeers` (0x04)
-    pub(crate) too_many_peers: Counter,
-
-    /// Number of inbound peer disconnects due to `AlreadyConnected` (0x05)
-    pub(crate) already_connected: Counter,
-
-    /// Number of inbound peer disconnects due to `IncompatibleP2PProtocolVersion` (0x06)
-    pub(crate) incompatible: Counter,
-
-    /// Number of inbound peer disconnects due to `NullNodeIdentity` (0x07)
-    pub(crate) null_node_identity: Counter,
-
-    /// Number of inbound peer disconnects due to `ClientQuitting` (0x08)
-    pub(crate) client_quitting: Counter,
-
-    /// Number of inbound peer disconnects due to `UnexpectedHandshakeIdentity` (0x09)
-    pub(crate) unexpected_identity: Counter,
-
-    /// Number of inbound peer disconnects due to `ConnectedToSelf` (0x0a)
-    pub(crate) connected_to_self: Counter,
-
-    /// Number of inbound peer disconnects due to `PingTimeout` (0x0b)
-    pub(crate) ping_timeout: Counter,
-
-    /// Number of inbound peer disconnects due to `SubprotocolSpecific` (0x10)
-    pub(crate) subprotocol_specific: Counter,
-}
-
-impl InboundDisconnectMetrics {
-    /// Increments the proper counter for the given disconnect reason
-    pub(crate) fn increment(&self, reason: DisconnectReason) {
-        match reason {
-            DisconnectReason::DisconnectRequested => self.disconnect_requested.increment(1),
-            DisconnectReason::TcpSubsystemError => self.tcp_subsystem_error.increment(1),
-            DisconnectReason::ProtocolBreach => self.protocol_breach.increment(1),
-            DisconnectReason::UselessPeer => self.useless_peer.increment(1),
-            DisconnectReason::TooManyPeers => self.too_many_peers.increment(1),
-            DisconnectReason::AlreadyConnected => self.already_connected.increment(1),
-            DisconnectReason::IncompatibleP2PProtocolVersion => self.incompatible.increment(1),
-            DisconnectReason::NullNodeIdentity => self.null_node_identity.increment(1),
-            DisconnectReason::ClientQuitting => self.client_quitting.increment(1),
-            DisconnectReason::UnexpectedHandshakeIdentity => self.unexpected_identity.increment(1),
-            DisconnectReason::ConnectedToSelf => self.connected_to_self.increment(1),
-            DisconnectReason::PingTimeout => self.ping_timeout.increment(1),
-            DisconnectReason::SubprotocolSpecific => self.subprotocol_specific.increment(1),
-        }
-    }
-}
-
-/// Disconnect metrics scoped to outbound connections only.
-///
-/// These counters track disconnect reasons exclusively for sessions that this node initiated
-/// by dialing out to remote peers. A high `too_many_peers` count here indicates remote peers
-/// are rejecting our connection attempts because they are full.
-#[derive(Metrics)]
-#[metrics(scope = "network.outbound")]
-pub struct OutboundDisconnectMetrics {
-    /// Number of outbound peer disconnects due to `DisconnectRequested` (0x00)
-    pub(crate) disconnect_requested: Counter,
-
-    /// Number of outbound peer disconnects due to `TcpSubsystemError` (0x01)
-    pub(crate) tcp_subsystem_error: Counter,
-
-    /// Number of outbound peer disconnects due to `ProtocolBreach` (0x02)
-    pub(crate) protocol_breach: Counter,
-
-    /// Number of outbound peer disconnects due to `UselessPeer` (0x03)
-    pub(crate) useless_peer: Counter,
-
-    /// Number of outbound peer disconnects due to `TooManyPeers` (0x04)
-    pub(crate) too_many_peers: Counter,
-
-    /// Number of outbound peer disconnects due to `AlreadyConnected` (0x05)
-    pub(crate) already_connected: Counter,
-
-    /// Number of outbound peer disconnects due to `IncompatibleP2PProtocolVersion` (0x06)
-    pub(crate) incompatible: Counter,
-
-    /// Number of outbound peer disconnects due to `NullNodeIdentity` (0x07)
-    pub(crate) null_node_identity: Counter,
-
-    /// Number of outbound peer disconnects due to `ClientQuitting` (0x08)
-    pub(crate) client_quitting: Counter,
-
-    /// Number of outbound peer disconnects due to `UnexpectedHandshakeIdentity` (0x09)
-    pub(crate) unexpected_identity: Counter,
-
-    /// Number of outbound peer disconnects due to `ConnectedToSelf` (0x0a)
-    pub(crate) connected_to_self: Counter,
-
-    /// Number of outbound peer disconnects due to `PingTimeout` (0x0b)
-    pub(crate) ping_timeout: Counter,
-
-    /// Number of outbound peer disconnects due to `SubprotocolSpecific` (0x10)
-    pub(crate) subprotocol_specific: Counter,
-}
-
-impl OutboundDisconnectMetrics {
-    /// Increments the proper counter for the given disconnect reason
-    pub(crate) fn increment(&self, reason: DisconnectReason) {
-        match reason {
-            DisconnectReason::DisconnectRequested => self.disconnect_requested.increment(1),
-            DisconnectReason::TcpSubsystemError => self.tcp_subsystem_error.increment(1),
-            DisconnectReason::ProtocolBreach => self.protocol_breach.increment(1),
-            DisconnectReason::UselessPeer => self.useless_peer.increment(1),
-            DisconnectReason::TooManyPeers => self.too_many_peers.increment(1),
-            DisconnectReason::AlreadyConnected => self.already_connected.increment(1),
-            DisconnectReason::IncompatibleP2PProtocolVersion => self.incompatible.increment(1),
-            DisconnectReason::NullNodeIdentity => self.null_node_identity.increment(1),
-            DisconnectReason::ClientQuitting => self.client_quitting.increment(1),
-            DisconnectReason::UnexpectedHandshakeIdentity => self.unexpected_identity.increment(1),
-            DisconnectReason::ConnectedToSelf => self.connected_to_self.increment(1),
-            DisconnectReason::PingTimeout => self.ping_timeout.increment(1),
-            DisconnectReason::SubprotocolSpecific => self.subprotocol_specific.increment(1),
-        }
-    }
-}
-
 /// Metrics for the `EthRequestHandler`
 #[derive(Metrics)]
 #[metrics(scope = "network")]

crates/net/network/src/peers.rs

@@ -92,9 +92,6 @@ pub struct PeersManager {
     incoming_ip_throttle_duration: Duration,
     /// IP address filter for restricting network connections to specific IP ranges.
     ip_filter: reth_net_banlist::IpFilter,
-    /// If true, discovered peers without a confirmed ENR fork ID will not be added until their
-    /// fork ID is verified via EIP-868.
-    enforce_enr_fork_id: bool,
 }
 
 impl PeersManager {
@@ -114,7 +111,6 @@ impl PeersManager {
             max_backoff_count,
             incoming_ip_throttle_duration,
             ip_filter,
-            enforce_enr_fork_id,
         } = config;
         let (manager_tx, handle_rx) = mpsc::unbounded_channel();
         let now = Instant::now();
@@ -171,7 +167,6 @@ impl PeersManager {
             net_connection_state: NetworkConnectionState::default(),
             incoming_ip_throttle_duration,
             ip_filter,
-            enforce_enr_fork_id,
         }
     }
 
@@ -180,11 +175,6 @@ impl PeersManager {
         PeersHandle::new(self.manager_tx.clone())
     }
 
-    /// Returns `true` if discovered peers must have a confirmed ENR fork ID before being added.
-    pub(crate) const fn enforce_enr_fork_id(&self) -> bool {
-        self.enforce_enr_fork_id
-    }
-
     /// Returns the number of peers in the peer set
     #[inline]
     pub(crate) fn num_known_peers(&self) -> usize {
@@ -218,13 +208,6 @@ impl PeersManager {
         })
     }
 
-    /// Returns `true` if the given peer is connected via an inbound session.
-    pub(crate) fn is_inbound_peer(&self, peer_id: &PeerId) -> bool {
-        self.peers.get(peer_id).is_some_and(|p| {
-            matches!(p.state, PeerConnectionState::In | PeerConnectionState::DisconnectingIn)
-        })
-    }
-
     /// Returns an iterator over all peer ids for peers with the given kind
     pub(crate) fn peers_by_kind(&self, kind: PeerKind) -> impl Iterator<Item = PeerId> + '_ {
         self.peers.iter().filter_map(move |(peer_id, peer)| (peer.kind == kind).then_some(*peer_id))
@@ -755,6 +738,17 @@ impl PeersManager {
         }
     }
 
+    /// Called as follow-up for a discovered peer.
+    ///
+    /// The [`ForkId`] is retrieved from an ENR record that the peer announces over the discovery
+    /// protocol
+    pub(crate) fn set_discovered_fork_id(&mut self, peer_id: PeerId, fork_id: ForkId) {
+        if let Some(peer) = self.peers.get_mut(&peer_id) {
+            trace!(target: "net::peers", ?peer_id, ?fork_id, "set discovered fork id");
+            peer.fork_id = Some(Box::new(fork_id));
+        }
+    }
+
     /// Called for a newly discovered peer.
     ///
     /// If the peer already exists, then the address, kind and `fork_id` will be updated.

crates/net/network/src/session/active.rs

@@ -282,12 +282,6 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
             EthMessage::Receipts70(resp) => {
                 on_response!(resp, GetReceipts70)
             }
-            EthMessage::GetBlockAccessLists(req) => {
-                on_request!(req, BlockAccessLists, GetBlockAccessLists)
-            }
-            EthMessage::BlockAccessLists(resp) => {
-                on_response!(resp, GetBlockAccessLists)
-            }
             EthMessage::BlockRangeUpdate(msg) => {
                 // Validate that earliest <= latest according to the spec
                 if msg.earliest > msg.latest {
@@ -322,22 +316,9 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
 
     /// Handle an internal peer request that will be sent to the remote.
     fn on_internal_peer_request(&mut self, request: PeerRequest<N>, deadline: Instant) {
-        let version = self.conn.version();
-        if !Self::is_request_supported_for_version(&request, version) {
-            debug!(
-                target: "net",
-                ?request,
-                peer_id=?self.remote_peer_id,
-                ?version,
-                "Request not supported for negotiated eth version",
-            );
-            request.send_err_response(RequestError::UnsupportedCapability);
-            return;
-        }
-
         let request_id = self.next_id();
         trace!(?request, peer_id=?self.remote_peer_id, ?request_id, "sending request to peer");
-        let msg = request.create_request_message(request_id).map_versioned(version);
+        let msg = request.create_request_message(request_id).map_versioned(self.conn.version());
 
         self.queued_outgoing.push_back(msg.into());
         let req = InflightRequest {
@@ -348,11 +329,6 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
         self.inflight_requests.insert(request_id, req);
     }
 
-    #[inline]
-    fn is_request_supported_for_version(request: &PeerRequest<N>, version: EthVersion) -> bool {
-        request.is_supported_by_eth_version(version)
-    }
-
     /// Handle a message received from the internal network
     fn on_internal_peer_message(&mut self, msg: PeerMessage<N>) {
         match msg {
@@ -962,9 +938,9 @@ mod tests {
     use reth_chainspec::MAINNET;
     use reth_ecies::stream::ECIESStream;
     use reth_eth_wire::{
-        handshake::EthHandshake, EthNetworkPrimitives, EthStream, GetBlockAccessLists,
-        GetBlockBodies, HelloMessageWithProtocols, P2PStream, StatusBuilder, UnauthedEthStream,
-        UnauthedP2PStream, UnifiedStatus,
+        handshake::EthHandshake, EthNetworkPrimitives, EthStream, GetBlockBodies,
+        HelloMessageWithProtocols, P2PStream, StatusBuilder, UnauthedEthStream, UnauthedP2PStream,
+        UnifiedStatus,
     };
     use reth_ethereum_forks::EthereumHardfork;
     use reth_network_peers::pk2id;
@@ -1264,22 +1240,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_reject_bal_request_for_eth70() {
-        let (tx, _rx) = oneshot::channel();
-        let request: PeerRequest<EthNetworkPrimitives> =
-            PeerRequest::GetBlockAccessLists { request: GetBlockAccessLists(vec![]), response: tx };
-
-        assert!(!ActiveSession::<EthNetworkPrimitives>::is_request_supported_for_version(
-            &request,
-            EthVersion::Eth70
-        ));
-        assert!(ActiveSession::<EthNetworkPrimitives>::is_request_supported_for_version(
-            &request,
-            EthVersion::Eth71
-        ));
-    }
-
     #[tokio::test(flavor = "multi_thread")]
     async fn test_keep_alive() {
         let mut builder = SessionBuilder::default();

crates/net/network/src/session/mod.rs

@@ -908,7 +908,7 @@ pub(crate) async fn start_pending_incoming_session<N: NetworkPrimitives>(
 }
 
 /// Starts the authentication process for a connection initiated by a remote peer.
-#[instrument(level = "trace", target = "net::network", skip_all, fields(%remote_addr, peer_id = ?remote_peer_id))]
+#[instrument(level = "trace", target = "net::network", skip_all, fields(%remote_addr, peer_id))]
 #[expect(clippy::too_many_arguments)]
 async fn start_pending_outbound_session<N: NetworkPrimitives>(
     handshake: Arc<dyn EthRlpxHandshake>,

crates/net/network/src/state.rs

@@ -332,19 +332,9 @@ impl<N: NetworkPrimitives> NetworkState<N> {
                     fork_id,
                 });
             }
-            DiscoveryEvent::EnrForkId(record, fork_id) => {
-                let peer_id = record.id;
-                let tcp_addr = record.tcp_addr();
-                if tcp_addr.port() == 0 {
-                    return
-                }
-                let udp_addr = record.udp_addr();
-                let addr = PeerAddr::new(tcp_addr, Some(udp_addr));
-                self.queued_messages.push_back(StateAction::DiscoveredEnrForkId {
-                    peer_id,
-                    addr,
-                    fork_id,
-                });
+            DiscoveryEvent::EnrForkId(peer_id, fork_id) => {
+                self.queued_messages
+                    .push_back(StateAction::DiscoveredEnrForkId { peer_id, fork_id });
             }
         }
     }
@@ -562,8 +552,6 @@ pub(crate) enum StateAction<N: NetworkPrimitives> {
     /// Retrieved a [`ForkId`] from the peer via ENR request, See <https://eips.ethereum.org/EIPS/eip-868>
     DiscoveredEnrForkId {
         peer_id: PeerId,
-        /// The address of the peer.
-        addr: PeerAddr,
         /// The reported [`ForkId`] by this peer.
         fork_id: ForkId,
     },

crates/net/network/src/swarm.rs

@@ -246,28 +246,18 @@ impl<N: NetworkPrimitives> Swarm<N> {
             StateAction::PeerAdded(peer_id) => return Some(SwarmEvent::PeerAdded(peer_id)),
             StateAction::PeerRemoved(peer_id) => return Some(SwarmEvent::PeerRemoved(peer_id)),
             StateAction::DiscoveredNode { peer_id, addr, fork_id } => {
+                // Don't try to connect to peer if node is shutting down
                 if self.is_shutting_down() {
                     return None
                 }
-
-                // When `enforce_enr_fork_id` is enabled, peers discovered without a confirmed
-                // fork ID (via EIP-868 ENR) are deferred — they'll only be added once a
-                // `DiscoveredEnrForkId` event arrives with a validated fork ID.
-                //
-                // When disabled (default), peers without a fork ID are admitted immediately.
-                // Peers that *do* carry a fork ID are always validated against ours.
-                let enforce = self.state().peers().enforce_enr_fork_id();
-                let allow = match fork_id {
-                    Some(f) => self.sessions.is_valid_fork_id(f),
-                    None => !enforce,
-                };
-                if allow {
+                // Insert peer only if no fork id or a valid fork id
+                if fork_id.map_or_else(|| true, |f| self.sessions.is_valid_fork_id(f)) {
                     self.state_mut().peers_mut().add_peer(peer_id, addr, fork_id);
                 }
             }
-            StateAction::DiscoveredEnrForkId { peer_id, addr, fork_id } => {
+            StateAction::DiscoveredEnrForkId { peer_id, fork_id } => {
                 if self.sessions.is_valid_fork_id(fork_id) {
-                    self.state_mut().peers_mut().add_peer(peer_id, addr, Some(fork_id));
+                    self.state_mut().peers_mut().set_discovered_fork_id(peer_id, fork_id);
                 } else {
                     trace!(target: "net", ?peer_id, remote_fork_id=?fork_id, our_fork_id=?self.sessions.fork_id(), "fork id mismatch, removing peer");
                     self.state_mut().peers_mut().remove_peer(peer_id);

crates/net/network/src/transactions/mod.rs

@@ -1949,7 +1949,7 @@ impl PooledTransactionsHashesBuilder {
     fn new(version: EthVersion) -> Self {
         match version {
             EthVersion::Eth66 | EthVersion::Eth67 => Self::Eth66(Default::default()),
-            EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 | EthVersion::Eth71 => {
+            EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 => {
                 Self::Eth68(Default::default())
             }
         }

crates/node/builder/Cargo.toml

@@ -24,6 +24,7 @@ reth-db-common.workspace = true
 reth-downloaders.workspace = true
 reth-engine-local.workspace = true
 reth-engine-primitives.workspace = true
+reth-engine-service.workspace = true
 reth-engine-tree.workspace = true
 reth-engine-util.workspace = true
 reth-evm.workspace = true

crates/node/builder/src/launch/engine.rs

@@ -11,10 +11,10 @@ use crate::{
 use alloy_consensus::BlockHeader;
 use futures::{stream_select, FutureExt, StreamExt};
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
+use reth_engine_service::service::{ChainEvent, EngineService};
 use reth_engine_tree::{
-    chain::{ChainEvent, FromOrchestrator},
-    engine::{EngineApiKind, EngineApiRequest, EngineRequestHandler},
-    launch::build_engine_orchestrator,
+    chain::FromOrchestrator,
+    engine::{EngineApiRequest, EngineRequestHandler},
     tree::TreeConfig,
 };
 use reth_engine_util::EngineMessageStreamExt;
@@ -219,15 +219,9 @@ impl EngineNodeLauncher {
             // during this run.
             .maybe_store_messages(node_config.debug.engine_api_store.clone());
 
-        let engine_kind = if ctx.chain_spec().is_optimism() {
-            EngineApiKind::OpStack
-        } else {
-            EngineApiKind::Ethereum
-        };
-
-        let mut orchestrator = build_engine_orchestrator(
-            engine_kind,
+        let mut engine_service = EngineService::new(
             consensus.clone(),
+            ctx.chain_spec(),
             network_client.clone(),
             Box::pin(consensus_engine_stream),
             pipeline,
@@ -296,7 +290,7 @@ impl EngineNodeLauncher {
             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
-                orchestrator.start_backfill_sync(initial_target);
+                engine_service.orchestrator_mut().start_backfill_sync(initial_target);
             } else if startup_sync_state_idle {
                 network_handle.update_sync_state(SyncState::Idle);
             }
@@ -309,7 +303,7 @@ impl EngineNodeLauncher {
             // the CL
             loop {
                 tokio::select! {
-                    event = orchestrator.next() => {
+                    event = engine_service.next() => {
                         let Some(event) = event else { break };
                         debug!(target: "reth::cli", "Event: {event}");
                         match event {
@@ -359,13 +353,13 @@ impl EngineNodeLauncher {
                     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");
-                            orchestrator.handler_mut().handler_mut().on_event(EngineApiRequest::InsertExecutedBlock(executed_block.into_executed_payload()).into());
+                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(EngineApiRequest::InsertExecutedBlock(executed_block.into_executed_payload()).into());
                         }
                     }
                     shutdown_req = &mut shutdown_rx => {
                         if let Ok(req) = shutdown_req {
                             debug!(target: "reth::cli", "received engine shutdown request");
-                            orchestrator.handler_mut().handler_mut().on_event(
+                            engine_service.orchestrator_mut().handler_mut().handler_mut().on_event(
                                 FromOrchestrator::Terminate { tx: req.done_tx }.into()
                             );
                         }

crates/node/core/src/args/engine.rs

@@ -43,7 +43,6 @@ pub struct DefaultEngineValues {
     disable_trie_cache: bool,
     sparse_trie_prune_depth: usize,
     sparse_trie_max_storage_tries: usize,
-    disable_sparse_trie_cache_pruning: bool,
     state_root_task_timeout: Option<String>,
 }
 
@@ -199,12 +198,6 @@ impl DefaultEngineValues {
         self
     }
 
-    /// Set whether to disable sparse trie cache pruning by default
-    pub const fn with_disable_sparse_trie_cache_pruning(mut self, v: bool) -> Self {
-        self.disable_sparse_trie_cache_pruning = v;
-        self
-    }
-
     /// Set the default state root task timeout
     pub fn with_state_root_task_timeout(mut self, v: Option<String>) -> Self {
         self.state_root_task_timeout = v;
@@ -238,7 +231,6 @@ impl Default for DefaultEngineValues {
             disable_trie_cache: false,
             sparse_trie_prune_depth: DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
             sparse_trie_max_storage_tries: DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
-            disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout: Some("1s".to_string()),
         }
     }
@@ -380,12 +372,6 @@ pub struct EngineArgs {
     #[arg(long = "engine.sparse-trie-max-storage-tries", default_value_t = DefaultEngineValues::get_global().sparse_trie_max_storage_tries)]
     pub sparse_trie_max_storage_tries: usize,
 
-    /// Fully disable sparse trie cache pruning. When set, the cached sparse trie is preserved
-    /// without any node pruning or storage trie eviction between blocks. Useful for benchmarking
-    /// the effects of retaining the full trie cache.
-    #[arg(long = "engine.disable-sparse-trie-cache-pruning", default_value_t = DefaultEngineValues::get_global().disable_sparse_trie_cache_pruning)]
-    pub disable_sparse_trie_cache_pruning: bool,
-
     /// Configure the timeout for the state root task before spawning a sequential fallback.
     /// If the state root task takes longer than this, a sequential computation starts in
     /// parallel and whichever finishes first is used.
@@ -429,7 +415,6 @@ impl Default for EngineArgs {
             disable_trie_cache,
             sparse_trie_prune_depth,
             sparse_trie_max_storage_tries,
-            disable_sparse_trie_cache_pruning,
             state_root_task_timeout,
         } = DefaultEngineValues::get_global().clone();
         Self {
@@ -460,7 +445,6 @@ impl Default for EngineArgs {
             disable_trie_cache,
             sparse_trie_prune_depth,
             sparse_trie_max_storage_tries,
-            disable_sparse_trie_cache_pruning,
             state_root_task_timeout: state_root_task_timeout
                 .as_deref()
                 .map(|s| humantime::parse_duration(s).expect("valid default duration")),
@@ -496,7 +480,6 @@ impl EngineArgs {
             .with_disable_trie_cache(self.disable_trie_cache)
             .with_sparse_trie_prune_depth(self.sparse_trie_prune_depth)
             .with_sparse_trie_max_storage_tries(self.sparse_trie_max_storage_tries)
-            .with_disable_sparse_trie_cache_pruning(self.disable_sparse_trie_cache_pruning)
             .with_state_root_task_timeout(self.state_root_task_timeout.filter(|d| !d.is_zero()))
     }
 }
@@ -551,7 +534,6 @@ mod tests {
             disable_trie_cache: true,
             sparse_trie_prune_depth: 10,
             sparse_trie_max_storage_tries: 100,
-            disable_sparse_trie_cache_pruning: true,
             state_root_task_timeout: Some(Duration::from_secs(2)),
         };
 
@@ -588,7 +570,6 @@ mod tests {
             "10",
             "--engine.sparse-trie-max-storage-tries",
             "100",
-            "--engine.disable-sparse-trie-cache-pruning",
             "--engine.state-root-task-timeout",
             "2s",
         ])

crates/node/core/src/args/mod.rs

@@ -76,7 +76,11 @@ pub use era::{DefaultEraHost, EraArgs, EraSourceArgs};
 mod static_files;
 pub use static_files::{StaticFilesArgs, MINIMAL_BLOCKS_PER_FILE};
 
-/// `StorageArgs` for configuring storage settings.
+/// `RocksDbArgs` for configuring RocksDB table routing.
+mod rocksdb;
+pub use rocksdb::{RocksDbArgs, RocksDbArgsError};
+
+/// `StorageArgs` for configuring storage mode (v2 vs v1/legacy).
 mod storage;
 pub use storage::StorageArgs;
 

crates/node/core/src/args/network.rs

@@ -227,14 +227,6 @@ pub struct NetworkArgs {
     /// Example: --netrestrict "192.168.0.0/16,10.0.0.0/8"
     #[arg(long, value_name = "NETRESTRICT")]
     pub netrestrict: Option<String>,
-
-    /// Enforce EIP-868 ENR fork ID validation for discovered peers.
-    ///
-    /// When enabled, peers discovered without a confirmed fork ID are not added to the peer set
-    /// until their fork ID is verified via EIP-868 ENR request. This filters out peers from other
-    /// networks that pollute the discovery table.
-    #[arg(long)]
-    pub enforce_enr_fork_id: bool,
 }
 
 impl NetworkArgs {
@@ -341,8 +333,7 @@ impl NetworkArgs {
             )
             .with_max_inbound_opt(self.resolved_max_inbound_peers())
             .with_max_outbound_opt(self.resolved_max_outbound_peers())
-            .with_ip_filter(ip_filter)
-            .with_enforce_enr_fork_id(self.enforce_enr_fork_id);
+            .with_ip_filter(ip_filter);
 
         // Configure basic network stack
         NetworkConfigBuilder::<N>::new(secret_key)
@@ -500,7 +491,6 @@ impl Default for NetworkArgs {
             required_block_hashes: vec![],
             network_id: None,
             netrestrict: None,
-            enforce_enr_fork_id: false,
         }
     }
 }

crates/node/core/src/args/pruning.rs

@@ -354,7 +354,7 @@ pub(crate) fn parse_receipts_log_filter(
 ) -> Result<ReceiptsLogPruneConfig, ReceiptsLogError> {
     let mut config = BTreeMap::new();
     // Split out each of the filters.
-    let filters = value.split(',').map(str::trim);
+    let filters = value.split(',');
     for filter in filters {
         let parts: Vec<&str> = filter.split(':').collect();
         if parts.len() < 2 {
@@ -450,23 +450,6 @@ mod tests {
         assert_eq!(config.0.get(&addr3), Some(&PruneMode::Before(5000000)));
     }
 
-    #[test]
-    fn test_parse_receipts_log_filter_with_spaces() {
-        // Verify that spaces after commas are handled correctly
-        let filters = "0x0000000000000000000000000000000000000001:full, 0x0000000000000000000000000000000000000002:distance:1000";
-
-        let result = parse_receipts_log_filter(filters);
-        assert!(result.is_ok());
-        let config = result.unwrap();
-        assert_eq!(config.0.len(), 2);
-
-        let addr1: Address = "0x0000000000000000000000000000000000000001".parse().unwrap();
-        let addr2: Address = "0x0000000000000000000000000000000000000002".parse().unwrap();
-
-        assert_eq!(config.0.get(&addr1), Some(&PruneMode::Full));
-        assert_eq!(config.0.get(&addr2), Some(&PruneMode::Distance(1000)));
-    }
-
     #[test]
     fn test_parse_receipts_log_filter_invalid_filter_format() {
         let result = parse_receipts_log_filter("invalid_format");

crates/node/core/src/args/rocksdb.rs

@@ -0,0 +1,160 @@
+//! clap [Args](clap::Args) for `RocksDB` table routing configuration
+
+use clap::{ArgAction, Args};
+
+/// Parameters for `RocksDB` table routing configuration.
+///
+/// These flags control which database tables are stored in `RocksDB` instead of MDBX.
+/// All flags are genesis-initialization-only: changing them after genesis requires a re-sync.
+///
+/// When `--storage.v2` is used, the defaults for these flags change to enable `RocksDB` routing.
+/// Individual flags can still override those defaults when explicitly set.
+#[derive(Debug, Args, PartialEq, Eq, Clone, Copy, Default)]
+#[command(next_help_heading = "RocksDB")]
+pub struct RocksDbArgs {
+    /// Route all supported tables to `RocksDB` instead of MDBX.
+    ///
+    /// This enables `RocksDB` for `tx-hash`, `storages-history`, and `account-history` tables.
+    /// Cannot be combined with individual flags set to false.
+    #[arg(long = "rocksdb.all", action = ArgAction::SetTrue)]
+    pub all: bool,
+
+    /// Route tx hash -> number table to `RocksDB` instead of MDBX.
+    ///
+    /// This is a genesis-initialization-only flag: changing it after genesis requires a re-sync.
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "rocksdb.tx-hash", action = ArgAction::Set)]
+    pub tx_hash: Option<bool>,
+
+    /// Route storages history tables to `RocksDB` instead of MDBX.
+    ///
+    /// This is a genesis-initialization-only flag: changing it after genesis requires a re-sync.
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "rocksdb.storages-history", action = ArgAction::Set)]
+    pub storages_history: Option<bool>,
+
+    /// Route account history tables to `RocksDB` instead of MDBX.
+    ///
+    /// This is a genesis-initialization-only flag: changing it after genesis requires a re-sync.
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "rocksdb.account-history", action = ArgAction::Set)]
+    pub account_history: Option<bool>,
+}
+
+impl RocksDbArgs {
+    /// Validates the `RocksDB` arguments.
+    ///
+    /// Returns an error if `--rocksdb.all` is used with any individual flag explicitly set to
+    /// `false`.
+    pub const fn validate(&self) -> Result<(), RocksDbArgsError> {
+        if self.all {
+            if matches!(self.tx_hash, Some(false)) {
+                return Err(RocksDbArgsError::ConflictingFlags("tx-hash"));
+            }
+            if matches!(self.storages_history, Some(false)) {
+                return Err(RocksDbArgsError::ConflictingFlags("storages-history"));
+            }
+            if matches!(self.account_history, Some(false)) {
+                return Err(RocksDbArgsError::ConflictingFlags("account-history"));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Error type for `RocksDB` argument validation.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum RocksDbArgsError {
+    /// `--rocksdb.all` cannot be combined with an individual flag set to false.
+    #[error("--rocksdb.all cannot be combined with --rocksdb.{0}=false")]
+    ConflictingFlags(&'static str),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::Parser;
+
+    #[derive(Parser)]
+    struct CommandParser<T: Args> {
+        #[command(flatten)]
+        args: T,
+    }
+
+    #[test]
+    fn test_default_rocksdb_args() {
+        let args = CommandParser::<RocksDbArgs>::parse_from(["reth"]).args;
+        assert_eq!(args, RocksDbArgs::default());
+        assert!(!args.all);
+        assert!(args.tx_hash.is_none());
+        assert!(args.storages_history.is_none());
+        assert!(args.account_history.is_none());
+    }
+
+    #[test]
+    fn test_parse_all_flag() {
+        let args = CommandParser::<RocksDbArgs>::parse_from(["reth", "--rocksdb.all"]).args;
+        assert!(args.all);
+        assert!(args.tx_hash.is_none());
+    }
+
+    #[test]
+    fn test_parse_individual_flags() {
+        let args = CommandParser::<RocksDbArgs>::parse_from([
+            "reth",
+            "--rocksdb.tx-hash=true",
+            "--rocksdb.storages-history=false",
+            "--rocksdb.account-history=true",
+        ])
+        .args;
+        assert!(!args.all);
+        assert_eq!(args.tx_hash, Some(true));
+        assert_eq!(args.storages_history, Some(false));
+        assert_eq!(args.account_history, Some(true));
+    }
+
+    #[test]
+    fn test_validate_all_with_none_ok() {
+        let args =
+            RocksDbArgs { all: true, tx_hash: None, storages_history: None, account_history: None };
+        assert!(args.validate().is_ok());
+    }
+
+    #[test]
+    fn test_validate_all_with_true_ok() {
+        let args = RocksDbArgs {
+            all: true,
+            tx_hash: Some(true),
+            storages_history: Some(true),
+            account_history: Some(true),
+        };
+        assert!(args.validate().is_ok());
+    }
+
+    #[test]
+    fn test_validate_all_with_false_errors() {
+        let args = RocksDbArgs {
+            all: true,
+            tx_hash: Some(false),
+            storages_history: None,
+            account_history: None,
+        };
+        assert_eq!(args.validate(), Err(RocksDbArgsError::ConflictingFlags("tx-hash")));
+
+        let args = RocksDbArgs {
+            all: true,
+            tx_hash: None,
+            storages_history: Some(false),
+            account_history: None,
+        };
+        assert_eq!(args.validate(), Err(RocksDbArgsError::ConflictingFlags("storages-history")));
+
+        let args = RocksDbArgs {
+            all: true,
+            tx_hash: None,
+            storages_history: None,
+            account_history: Some(false),
+        };
+        assert_eq!(args.validate(), Err(RocksDbArgsError::ConflictingFlags("account-history")));
+    }
+}

crates/node/core/src/args/static_files.rs

@@ -9,6 +9,9 @@ use reth_config::config::{BlocksPerFileConfig, StaticFilesConfig};
 pub const MINIMAL_BLOCKS_PER_FILE: u64 = 10000;
 
 /// Parameters for static files configuration
+///
+/// When `--storage.v2` is used, the defaults for the storage flags change to enable static file
+/// storage. Individual flags can still override those defaults when explicitly set.
 #[derive(Debug, Args, PartialEq, Eq, Clone, Copy, Default)]
 #[command(next_help_heading = "Static Files")]
 pub struct StaticFilesArgs {
@@ -35,6 +38,53 @@ pub struct StaticFilesArgs {
     /// Number of blocks per file for the storage changesets segment.
     #[arg(long = "static-files.blocks-per-file.storage-change-sets")]
     pub blocks_per_file_storage_change_sets: Option<u64>,
+
+    /// Store receipts in static files instead of the database.
+    ///
+    /// When enabled, receipts will be written to static files on disk instead of the database.
+    ///
+    /// Note: This setting can only be configured at genesis initialization. Once
+    /// the node has been initialized, changing this flag requires re-syncing from scratch.
+    ///
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "static-files.receipts", action = clap::ArgAction::Set)]
+    pub receipts: Option<bool>,
+
+    /// Store transaction senders in static files instead of the database.
+    ///
+    /// When enabled, transaction senders will be written to static files on disk instead of the
+    /// database.
+    ///
+    /// Note: This setting can only be configured at genesis initialization. Once
+    /// the node has been initialized, changing this flag requires re-syncing from scratch.
+    ///
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "static-files.transaction-senders", action = clap::ArgAction::Set)]
+    pub transaction_senders: Option<bool>,
+
+    /// Store account changesets in static files.
+    ///
+    /// When enabled, account changesets will be written to static files on disk instead of the
+    /// database.
+    ///
+    /// Note: This setting can only be configured at genesis initialization. Once
+    /// the node has been initialized, changing this flag requires re-syncing from scratch.
+    ///
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "static-files.account-change-sets", action = clap::ArgAction::Set)]
+    pub account_changesets: Option<bool>,
+
+    /// Store storage changesets in static files.
+    ///
+    /// When enabled, storage changesets will be written to static files on disk instead of the
+    /// database.
+    ///
+    /// Note: This setting can only be configured at genesis initialization. Once
+    /// the node has been initialized, changing this flag requires re-syncing from scratch.
+    ///
+    /// Defaults to the base storage mode (v1: false, v2: true).
+    #[arg(long = "static-files.storage-change-sets", action = clap::ArgAction::Set)]
+    pub storage_changesets: Option<bool>,
 }
 
 impl StaticFilesArgs {

crates/node/core/src/args/storage.rs

@@ -1,13 +1,11 @@
-//! clap [Args](clap::Args) for storage configuration
+//! clap [Args](clap::Args) for storage mode configuration
 
 use clap::{ArgAction, Args};
 
-/// Parameters for storage configuration.
+/// Parameters for storage mode configuration.
 ///
 /// This controls whether the node uses v2 storage defaults (with `RocksDB` and static file
 /// optimizations) or v1/legacy storage defaults.
-///
-/// Individual storage settings can be overridden with `--static-files.*` and `--rocksdb.*` flags.
 #[derive(Debug, Args, PartialEq, Eq, Clone, Copy, Default)]
 #[command(next_help_heading = "Storage")]
 pub struct StorageArgs {
@@ -25,6 +23,16 @@ pub struct StorageArgs {
     /// flags.
     #[arg(long = "storage.v2", action = ArgAction::SetTrue)]
     pub v2: bool,
+
+    /// Use hashed state tables (`HashedAccounts`/`HashedStorages`) as canonical state
+    /// representation instead of plain state tables.
+    ///
+    /// When enabled, execution writes directly to hashed tables, eliminating the need for
+    /// separate hashing stages. This should only be enabled for new databases.
+    ///
+    /// WARNING: Changing this setting on an existing database requires a full resync.
+    #[arg(long = "storage.use-hashed-state", default_value_t = false)]
+    pub use_hashed_state: bool,
 }
 
 #[cfg(test)]
@@ -32,24 +40,21 @@ mod tests {
     use super::*;
     use clap::Parser;
 
-    /// A helper type to parse Args more easily
     #[derive(Parser)]
-    struct CommandParser<T: Args> {
+    struct CommandParser {
         #[command(flatten)]
-        args: T,
+        args: StorageArgs,
     }
 
     #[test]
     fn test_default_storage_args() {
-        let default_args = StorageArgs::default();
-        let args = CommandParser::<StorageArgs>::parse_from(["reth"]).args;
-        assert_eq!(args, default_args);
+        let args = CommandParser::parse_from(["reth"]).args;
         assert!(!args.v2);
     }
 
     #[test]
     fn test_parse_v2_flag() {
-        let args = CommandParser::<StorageArgs>::parse_from(["reth", "--storage.v2"]).args;
+        let args = CommandParser::parse_from(["reth", "--storage.v2"]).args;
         assert!(args.v2);
     }
 }

crates/node/core/src/node_config.rs

@@ -3,7 +3,7 @@
 use crate::{
     args::{
         DatabaseArgs, DatadirArgs, DebugArgs, DevArgs, EngineArgs, NetworkArgs, PayloadBuilderArgs,
-        PruningArgs, RpcServerArgs, StaticFilesArgs, StorageArgs, TxPoolArgs,
+        PruningArgs, RocksDbArgs, RpcServerArgs, StaticFilesArgs, StorageArgs, TxPoolArgs,
     },
     dirs::{ChainPath, DataDirPath},
     utils::get_single_header,
@@ -152,7 +152,10 @@ pub struct NodeConfig<ChainSpec> {
     /// All static files related arguments
     pub static_files: StaticFilesArgs,
 
-    /// All storage related arguments with --storage prefix
+    /// All `RocksDB` table routing arguments
+    pub rocksdb: RocksDbArgs,
+
+    /// Storage mode configuration (v2 vs v1/legacy)
     pub storage: StorageArgs,
 }
 
@@ -185,6 +188,7 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
             engine: EngineArgs::default(),
             era: EraArgs::default(),
             static_files: StaticFilesArgs::default(),
+            rocksdb: RocksDbArgs::default(),
             storage: StorageArgs::default(),
         }
     }
@@ -260,6 +264,7 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
             engine,
             era,
             static_files,
+            rocksdb,
             storage,
             ..
         } = self;
@@ -280,6 +285,7 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
             engine,
             era,
             static_files,
+            rocksdb,
             storage,
         }
     }
@@ -349,12 +355,6 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
         self
     }
 
-    /// Set the storage args for the node
-    pub const fn with_storage(mut self, storage: StorageArgs) -> Self {
-        self.storage = storage;
-        self
-    }
-
     /// Returns pruning configuration.
     pub fn prune_config(&self) -> Option<PruneConfig>
     where
@@ -363,17 +363,42 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
         self.pruning.prune_config(&self.chain)
     }
 
-    /// Returns the effective storage settings derived from `--storage.v2`.
+    /// Returns the effective storage settings derived from `--storage.v2`, static-file, and
+    /// `RocksDB` CLI args.
     ///
     /// The base storage mode is determined by `--storage.v2`:
     /// - When `--storage.v2` is set: uses [`StorageSettings::v2()`] defaults
-    /// - Otherwise: uses [`StorageSettings::base()`] defaults
+    /// - Otherwise: uses [`StorageSettings::v1()`] defaults
+    ///
+    /// Individual `--static-files.*` and `--rocksdb.*` flags override the base when explicitly set.
     pub const fn storage_settings(&self) -> StorageSettings {
-        if self.storage.v2 {
-            StorageSettings::v2()
-        } else {
-            StorageSettings::base()
+        let mut s = if self.storage.v2 { StorageSettings::v2() } else { StorageSettings::base() };
+
+        // Apply static files overrides (only when explicitly set)
+        s = s
+            .with_receipts_in_static_files_opt(self.static_files.receipts)
+            .with_transaction_senders_in_static_files_opt(self.static_files.transaction_senders)
+            .with_account_changesets_in_static_files_opt(self.static_files.account_changesets)
+            .with_storage_changesets_in_static_files_opt(self.static_files.storage_changesets);
+
+        // Apply rocksdb overrides
+        // --rocksdb.all sets all rocksdb flags to true
+        if self.rocksdb.all {
+            s = s
+                .with_transaction_hash_numbers_in_rocksdb(true)
+                .with_storages_history_in_rocksdb(true)
+                .with_account_history_in_rocksdb(true);
         }
+
+        // Individual rocksdb flags override --rocksdb.all when explicitly set
+        s = s
+            .with_transaction_hash_numbers_in_rocksdb_opt(self.rocksdb.tx_hash)
+            .with_storages_history_in_rocksdb_opt(self.rocksdb.storages_history)
+            .with_account_history_in_rocksdb_opt(self.rocksdb.account_history);
+
+        s = s.with_use_hashed_state(self.storage.use_hashed_state);
+
+        s
     }
 
     /// Returns the max block that the node should run to, looking it up from the network if
@@ -570,6 +595,7 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
             engine: self.engine,
             era: self.era,
             static_files: self.static_files,
+            rocksdb: self.rocksdb,
             storage: self.storage,
         }
     }
@@ -612,6 +638,7 @@ impl<ChainSpec> Clone for NodeConfig<ChainSpec> {
             engine: self.engine.clone(),
             era: self.era.clone(),
             static_files: self.static_files,
+            rocksdb: self.rocksdb,
             storage: self.storage,
         }
     }

crates/node/events/src/node.rs

@@ -265,7 +265,7 @@ impl NodeState {
                 warn!(number=block.number(), hash=?block.hash(), "Encountered invalid block");
             }
             ConsensusEngineEvent::BlockReceived(num_hash) => {
-                info!(number=num_hash.number, hash=?num_hash.hash, "Received new payload from consensus engine");
+                info!(number=num_hash.number, hash=?num_hash.hash, "Received block from consensus engine");
             }
         }
     }

crates/primitives-traits/src/account.rs

@@ -4,7 +4,7 @@ use alloy_genesis::GenesisAccount;
 use alloy_primitives::{keccak256, Bytes, B256, U256};
 use alloy_trie::TrieAccount;
 use derive_more::Deref;
-use revm_bytecode::{Bytecode as RevmBytecode, BytecodeDecodeError};
+use revm_bytecode::{Bytecode as RevmBytecode, BytecodeDecodeError, BytecodeKind};
 use revm_state::AccountInfo;
 
 #[cfg(any(test, feature = "reth-codec"))]
@@ -16,10 +16,10 @@ pub mod compact_ids {
     /// Identifier for removed bytecode variant.
     pub const REMOVED_BYTECODE_ID: u8 = 1;
 
-    /// Identifier for [`LegacyAnalyzed`](revm_bytecode::Bytecode::LegacyAnalyzed).
+    /// Identifier for [`LegacyAnalyzed`](revm_bytecode::BytecodeKind::LegacyAnalyzed).
     pub const LEGACY_ANALYZED_BYTECODE_ID: u8 = 2;
 
-    /// Identifier for [`Eip7702`](revm_bytecode::Bytecode::Eip7702).
+    /// Identifier for [`Eip7702`](revm_bytecode::BytecodeKind::Eip7702).
     pub const EIP7702_BYTECODE_ID: u8 = 4;
 }
 
@@ -137,22 +137,19 @@ impl reth_codecs::Compact for Bytecode {
     {
         use compact_ids::{EIP7702_BYTECODE_ID, LEGACY_ANALYZED_BYTECODE_ID};
 
-        let bytecode = match &self.0 {
-            RevmBytecode::LegacyAnalyzed(analyzed) => analyzed.bytecode(),
-            RevmBytecode::Eip7702(eip7702) => eip7702.raw(),
-        };
+        let bytecode = self.0.bytecode();
         buf.put_u32(bytecode.len() as u32);
         buf.put_slice(bytecode.as_ref());
-        let len = match &self.0 {
+        let len = match self.0.kind() {
             // [`REMOVED_BYTECODE_ID`] has been removed.
-            RevmBytecode::LegacyAnalyzed(analyzed) => {
+            BytecodeKind::LegacyAnalyzed => {
                 buf.put_u8(LEGACY_ANALYZED_BYTECODE_ID);
-                buf.put_u64(analyzed.original_len() as u64);
-                let map = analyzed.jump_table().as_slice();
+                buf.put_u64(self.0.len() as u64);
+                let map = self.0.legacy_jump_table().unwrap().as_slice();
                 buf.put_slice(map);
                 1 + 8 + map.len()
             }
-            RevmBytecode::Eip7702(_) => {
+            BytecodeKind::Eip7702 => {
                 buf.put_u8(EIP7702_BYTECODE_ID);
                 1
             }
@@ -251,12 +248,10 @@ impl From<Account> for AccountInfo {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
     use super::*;
     use alloy_primitives::{hex_literal::hex, B256, U256};
     use reth_codecs::Compact;
-    use revm_bytecode::{JumpTable, LegacyAnalyzedBytecode};
+    use revm_bytecode::JumpTable;
 
     #[test]
     fn test_account() {
@@ -313,12 +308,11 @@ mod tests {
         assert_eq!(len, 17);
 
         let mut buf = vec![];
-        let bytecode =
-            Bytecode(RevmBytecode::LegacyAnalyzed(Arc::new(LegacyAnalyzedBytecode::new(
-                Bytes::from(&hex!("ff00")),
-                2,
-                JumpTable::from_slice(&[0], 2),
-            ))));
+        let bytecode = Bytecode(RevmBytecode::new_analyzed(
+            Bytes::from(&hex!("ff00")),
+            2,
+            JumpTable::from_slice(&[0], 2),
+        ));
         let len = bytecode.to_compact(&mut buf);
         assert_eq!(len, 16);
 

crates/primitives-traits/src/lib.rs

@@ -164,7 +164,7 @@ pub use alloy_primitives::{logs_bloom, Log, LogData};
 pub mod proofs;
 
 mod storage;
-pub use storage::{StorageEntry, StorageSlotKey, ValueWithSubKey};
+pub use storage::{StorageEntry, ValueWithSubKey};
 
 pub mod sync;
 

crates/primitives-traits/src/storage.rs

@@ -1,4 +1,4 @@
-use alloy_primitives::{keccak256, B256, U256};
+use alloy_primitives::{B256, U256};
 
 /// Trait for `DupSort` table values that contain a subkey.
 ///
@@ -12,117 +12,6 @@ pub trait ValueWithSubKey {
     fn get_subkey(&self) -> Self::SubKey;
 }
 
-/// A storage slot key that tracks whether it holds a plain (unhashed) EVM slot
-/// or a keccak256-hashed slot.
-///
-/// This enum replaces the `use_hashed_state: bool` parameter pattern by carrying
-/// provenance with the key itself. Once tagged at a read/write boundary, downstream
-/// code can call [`Self::to_hashed`] without risk of double-hashing — hashing a
-/// [`StorageSlotKey::Hashed`] is a no-op.
-///
-/// The on-disk encoding is unchanged (raw 32-byte [`B256`]). The variant is set
-/// by the code that knows the context (which table, which storage mode).
-#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub enum StorageSlotKey {
-    /// An unhashed EVM storage slot, as produced by REVM execution.
-    Plain(B256),
-    /// A keccak256-hashed storage slot, as stored in `HashedStorages` and
-    /// in v2-mode `StorageChangeSets`.
-    Hashed(B256),
-}
-
-impl Default for StorageSlotKey {
-    fn default() -> Self {
-        Self::Plain(B256::ZERO)
-    }
-}
-
-impl StorageSlotKey {
-    /// Create a plain slot key from a REVM [`U256`] storage index.
-    pub const fn from_u256(slot: U256) -> Self {
-        Self::Plain(B256::new(slot.to_be_bytes()))
-    }
-
-    /// Create a plain slot key from a raw [`B256`].
-    pub const fn plain(key: B256) -> Self {
-        Self::Plain(key)
-    }
-
-    /// Create a hashed slot key from a raw [`B256`].
-    pub const fn hashed(key: B256) -> Self {
-        Self::Hashed(key)
-    }
-
-    /// Tag a raw [`B256`] based on the storage mode.
-    ///
-    /// When `use_hashed_state` is true the key is assumed already hashed.
-    /// When false it is assumed to be a plain slot.
-    pub const fn from_raw(key: B256, use_hashed_state: bool) -> Self {
-        if use_hashed_state {
-            Self::Hashed(key)
-        } else {
-            Self::Plain(key)
-        }
-    }
-
-    /// Returns the raw [`B256`] regardless of variant.
-    pub const fn as_b256(&self) -> B256 {
-        match *self {
-            Self::Plain(b) | Self::Hashed(b) => b,
-        }
-    }
-
-    /// Returns `true` if this key is already hashed.
-    pub const fn is_hashed(&self) -> bool {
-        matches!(self, Self::Hashed(_))
-    }
-
-    /// Returns `true` if this key is plain (unhashed).
-    pub const fn is_plain(&self) -> bool {
-        matches!(self, Self::Plain(_))
-    }
-
-    /// Produce the keccak256-hashed form of this slot key.
-    ///
-    /// - If already [`Hashed`](Self::Hashed), returns the inner value as-is (no double-hash).
-    /// - If [`Plain`](Self::Plain), applies keccak256 and returns the result.
-    pub fn to_hashed(&self) -> B256 {
-        match *self {
-            Self::Hashed(b) => b,
-            Self::Plain(b) => keccak256(b),
-        }
-    }
-
-    /// Convert a plain slot to its changeset representation.
-    ///
-    /// In v2 mode (`use_hashed_state = true`), the changeset stores hashed keys,
-    /// so the plain key is hashed. In v1 mode, the plain key is stored as-is.
-    ///
-    /// Panics (debug) if called on an already-hashed key.
-    pub fn to_changeset_key(self, use_hashed_state: bool) -> B256 {
-        debug_assert!(self.is_plain(), "to_changeset_key called on already-hashed key");
-        if use_hashed_state {
-            self.to_hashed()
-        } else {
-            self.as_b256()
-        }
-    }
-
-    /// Like [`to_changeset_key`](Self::to_changeset_key) but returns a tagged
-    /// [`StorageSlotKey`] instead of a raw [`B256`].
-    ///
-    /// Panics (debug) if called on an already-hashed key.
-    pub fn to_changeset(self, use_hashed_state: bool) -> Self {
-        Self::from_raw(self.to_changeset_key(use_hashed_state), use_hashed_state)
-    }
-}
-
-impl From<StorageSlotKey> for B256 {
-    fn from(key: StorageSlotKey) -> Self {
-        key.as_b256()
-    }
-}
-
 /// Account storage entry.
 ///
 /// `key` is the subkey when used as a value in the `StorageChangeSets` table.
@@ -142,14 +31,6 @@ impl StorageEntry {
     pub const fn new(key: B256, value: U256) -> Self {
         Self { key, value }
     }
-
-    /// Tag this entry's key as a [`StorageSlotKey`] based on the storage mode.
-    ///
-    /// When `use_hashed_state` is true, the key is tagged as already-hashed.
-    /// When false, it is tagged as plain.
-    pub const fn slot_key(&self, use_hashed_state: bool) -> StorageSlotKey {
-        StorageSlotKey::from_raw(self.key, use_hashed_state)
-    }
 }
 
 impl ValueWithSubKey for StorageEntry {

crates/prune/prune/src/segments/user/account_history.rs

@@ -75,7 +75,7 @@ where
 
         // Check where account history indices are stored
         #[cfg(all(unix, feature = "rocksdb"))]
-        if provider.cached_storage_settings().storage_v2 {
+        if provider.cached_storage_settings().account_history_in_rocksdb {
             return self.prune_rocksdb(provider, input, range, range_end);
         }
 
@@ -405,7 +405,9 @@ mod tests {
                 let segment = AccountHistory::new(prune_mode);
 
                 let provider = db.factory.database_provider_rw().unwrap();
-                provider.set_storage_settings_cache(StorageSettings::v1());
+                provider.set_storage_settings_cache(
+                    StorageSettings::default().with_account_changesets_in_static_files(false),
+                );
                 let result = segment.prune(&provider, input).unwrap();
                 limiter.increment_deleted_entries_count_by(result.pruned);
 
@@ -506,11 +508,7 @@ mod tests {
         test_prune(1400, 3, (PruneProgress::Finished, 804));
     }
 
-    /// Tests the `prune_static_files` code path. On unix with rocksdb feature, v2 storage
-    /// routes to `prune_rocksdb` instead, so this test only runs without rocksdb (the
-    /// `prune_rocksdb_path` test covers that configuration).
     #[test]
-    #[cfg(not(all(unix, feature = "rocksdb")))]
     fn prune_static_file() {
         let db = TestStageDB::default();
         let mut rng = generators::rng();
@@ -566,7 +564,9 @@ mod tests {
                 let segment = AccountHistory::new(prune_mode);
 
                 let provider = db.factory.database_provider_rw().unwrap();
-                provider.set_storage_settings_cache(StorageSettings::v2());
+                provider.set_storage_settings_cache(
+                    StorageSettings::default().with_account_changesets_in_static_files(true),
+                );
                 let result = segment.prune(&provider, input).unwrap();
                 limiter.increment_deleted_entries_count_by(result.pruned);
 
@@ -714,7 +714,11 @@ mod tests {
             PruneInput { previous_checkpoint: None, to_block, limiter: PruneLimiter::default() };
         let segment = AccountHistory::new(prune_mode);
 
-        db.factory.set_storage_settings_cache(StorageSettings::v2());
+        db.factory.set_storage_settings_cache(
+            StorageSettings::default()
+                .with_account_changesets_in_static_files(true)
+                .with_account_history_in_rocksdb(true),
+        );
 
         let provider = db.factory.database_provider_rw().unwrap();
         let result = segment.prune(&provider, input).unwrap();
@@ -828,7 +832,9 @@ mod tests {
         let segment = AccountHistory::new(prune_mode);
 
         let provider = db.factory.database_provider_rw().unwrap();
-        provider.set_storage_settings_cache(StorageSettings::v1());
+        provider.set_storage_settings_cache(
+            StorageSettings::default().with_account_changesets_in_static_files(false),
+        );
         let result = segment.prune(&provider, input).unwrap();
 
         // Should report that there's more data
@@ -886,7 +892,9 @@ mod tests {
         };
 
         let provider2 = db.factory.database_provider_rw().unwrap();
-        provider2.set_storage_settings_cache(StorageSettings::v1());
+        provider2.set_storage_settings_cache(
+            StorageSettings::default().with_account_changesets_in_static_files(false),
+        );
         let result2 = segment.prune(&provider2, input2).unwrap();
 
         assert!(result2.progress.is_finished(), "Second run should complete");

crates/prune/prune/src/segments/user/storage_history.rs

@@ -76,7 +76,7 @@ where
 
         // Check where storage history indices are stored
         #[cfg(all(unix, feature = "rocksdb"))]
-        if provider.cached_storage_settings().storage_v2 {
+        if provider.cached_storage_settings().storages_history_in_rocksdb {
             return self.prune_rocksdb(provider, input, range, range_end);
         }
 
@@ -135,7 +135,7 @@ impl StorageHistory {
             let (block_address, entry) = result?;
             let block_number = block_address.block_number();
             let address = block_address.address();
-            highest_deleted_storages.insert((address, entry.key.as_b256()), block_number);
+            highest_deleted_storages.insert((address, entry.key), block_number);
             last_changeset_pruned_block = Some(block_number);
             pruned_changesets += 1;
             limiter.increment_deleted_entries_count();
@@ -273,7 +273,7 @@ impl StorageHistory {
             let (block_address, entry) = result?;
             let block_number = block_address.block_number();
             let address = block_address.address();
-            highest_deleted_storages.insert((address, entry.key.as_b256()), block_number);
+            highest_deleted_storages.insert((address, entry.key), block_number);
             last_changeset_pruned_block = Some(block_number);
             changesets_processed += 1;
             limiter.increment_deleted_entries_count();
@@ -413,7 +413,9 @@ mod tests {
             let segment = StorageHistory::new(prune_mode);
 
             let provider = db.factory.database_provider_rw().unwrap();
-            provider.set_storage_settings_cache(StorageSettings::v1());
+            provider.set_storage_settings_cache(
+                StorageSettings::default().with_storage_changesets_in_static_files(false),
+            );
             let result = segment.prune(&provider, input).unwrap();
             limiter.increment_deleted_entries_count_by(result.pruned);
 
@@ -518,11 +520,7 @@ mod tests {
         test_prune(1200, 3, (PruneProgress::Finished, 202));
     }
 
-    /// Tests the `prune_static_files` code path. On unix with rocksdb feature, v2 storage
-    /// routes to `prune_rocksdb` instead, so this test only runs without rocksdb (the
-    /// `prune_rocksdb_path` test covers that configuration).
     #[test]
-    #[cfg(not(all(unix, feature = "rocksdb")))]
     fn prune_static_file() {
         let db = TestStageDB::default();
         let mut rng = generators::rng();
@@ -579,7 +577,9 @@ mod tests {
             let segment = StorageHistory::new(prune_mode);
 
             let provider = db.factory.database_provider_rw().unwrap();
-            provider.set_storage_settings_cache(StorageSettings::v2());
+            provider.set_storage_settings_cache(
+                StorageSettings::default().with_storage_changesets_in_static_files(true),
+            );
             let result = segment.prune(&provider, input).unwrap();
             limiter.increment_deleted_entries_count_by(result.pruned);
 
@@ -739,7 +739,9 @@ mod tests {
         let segment = StorageHistory::new(prune_mode);
 
         let provider = db.factory.database_provider_rw().unwrap();
-        provider.set_storage_settings_cache(StorageSettings::v1());
+        provider.set_storage_settings_cache(
+            StorageSettings::default().with_storage_changesets_in_static_files(false),
+        );
         let result = segment.prune(&provider, input).unwrap();
 
         // Should report that there's more data
@@ -791,7 +793,9 @@ mod tests {
         };
 
         let provider2 = db.factory.database_provider_rw().unwrap();
-        provider2.set_storage_settings_cache(StorageSettings::v1());
+        provider2.set_storage_settings_cache(
+            StorageSettings::default().with_storage_changesets_in_static_files(false),
+        );
         let result2 = segment.prune(&provider2, input2).unwrap();
 
         assert!(result2.progress.is_finished(), "Second run should complete");
@@ -891,7 +895,11 @@ mod tests {
         let segment = StorageHistory::new(prune_mode);
 
         let provider = db.factory.database_provider_rw().unwrap();
-        provider.set_storage_settings_cache(StorageSettings::v2());
+        provider.set_storage_settings_cache(
+            StorageSettings::default()
+                .with_storage_changesets_in_static_files(true)
+                .with_storages_history_in_rocksdb(true),
+        );
         let result = segment.prune(&provider, input).unwrap();
         provider.commit().expect("commit");
 

crates/prune/prune/src/segments/user/transaction_lookup.rs

@@ -96,7 +96,7 @@ where
 
         // Check where transaction hash numbers are stored
         #[cfg(all(unix, feature = "rocksdb"))]
-        if provider.cached_storage_settings().storage_v2 {
+        if provider.cached_storage_settings().transaction_hash_numbers_in_rocksdb {
             return self.prune_rocksdb(provider, input, start, end);
         }
 
@@ -491,7 +491,9 @@ mod tests {
         let segment = TransactionLookup::new(prune_mode);
 
         // Enable RocksDB storage for transaction hash numbers
-        db.factory.set_storage_settings_cache(StorageSettings::v2());
+        db.factory.set_storage_settings_cache(
+            StorageSettings::v1().with_transaction_hash_numbers_in_rocksdb(true),
+        );
 
         let provider = db.factory.database_provider_rw().unwrap();
         let result = segment.prune(&provider, input).unwrap();
@@ -576,7 +578,9 @@ mod tests {
         }
 
         // Enable RocksDB storage for transaction hash numbers
-        db.factory.set_storage_settings_cache(StorageSettings::v2());
+        db.factory.set_storage_settings_cache(
+            StorageSettings::v1().with_transaction_hash_numbers_in_rocksdb(true),
+        );
 
         let to_block: BlockNumber = 6;
         let prune_mode = PruneMode::Before(to_block);

crates/revm/src/test_utils.rs

@@ -160,14 +160,6 @@ impl StateProvider for StateProviderTest {
     ) -> ProviderResult<Option<alloy_primitives::StorageValue>> {
         Ok(self.accounts.get(&account).and_then(|(storage, _)| storage.get(&storage_key).copied()))
     }
-
-    fn storage_by_hashed_key(
-        &self,
-        _address: Address,
-        _hashed_storage_key: StorageKey,
-    ) -> ProviderResult<Option<alloy_primitives::StorageValue>> {
-        Ok(None)
-    }
 }
 
 impl BytecodeReader for StateProviderTest {

crates/revm/src/witness.rs

@@ -70,8 +70,8 @@ impl ExecutionWitnessRecord {
                 }
             }
         }
-        // BTreeMap keys are ordered, so the first key is the smallest
-        self.lowest_block_number = statedb.block_hashes.keys().next().copied()
+        // Get the lowest block number from the cache
+        self.lowest_block_number = statedb.block_hashes.lowest().map(|(n, _)| n)
     }
 
     /// Creates the record from the state after execution.

crates/rpc/rpc-eth-api/src/core.rs

@@ -20,7 +20,6 @@ use reth_primitives_traits::TxTy;
 use reth_rpc_convert::RpcTxReq;
 use reth_rpc_eth_types::FillTransaction;
 use reth_rpc_server_types::{result::internal_rpc_err, ToRpcResult};
-use std::collections::HashMap;
 use tracing::trace;
 
 /// Helper trait, unifies functionality that must be supported to implement all RPC methods for
@@ -202,14 +201,6 @@ pub trait EthApi<
         block_number: Option<BlockId>,
     ) -> RpcResult<B256>;
 
-    /// Returns values from multiple storage positions across multiple addresses.
-    #[method(name = "getStorageValues")]
-    async fn storage_values(
-        &self,
-        requests: HashMap<Address, Vec<JsonStorageKey>>,
-        block_number: Option<BlockId>,
-    ) -> RpcResult<HashMap<Address, Vec<B256>>>;
-
     /// Returns the number of transactions sent from an address at given block number.
     #[method(name = "getTransactionCount")]
     async fn transaction_count(
@@ -660,16 +651,6 @@ where
         Ok(EthState::storage_at(self, address, index, block_number).await?)
     }
 
-    /// Handler for: `eth_getStorageValues`
-    async fn storage_values(
-        &self,
-        requests: HashMap<Address, Vec<JsonStorageKey>>,
-        block_number: Option<BlockId>,
-    ) -> RpcResult<HashMap<Address, Vec<B256>>> {
-        trace!(target: "rpc::eth", ?block_number, "Serving eth_getStorageValues");
-        Ok(EthState::storage_values(self, requests, block_number).await?)
-    }
-
     /// Handler for: `eth_getTransactionCount`
     async fn transaction_count(
         &self,