fix: allow smaller header size

docs: fix incorrect API example in node-components.mdx (#20297 )
refactor(era): make era count in era file name optional (#20292 )
2026-04-30 03:01:58 -04:00 · 2025-12-16 17:08:53 +01:00 · 2025-12-16 15:09:29 +00:00 · 2025-12-16 15:08:43 +00:00 · 2025-12-16 15:06:04 +00:00 · 2025-12-16 15:04:34 +00:00
161 changed files with 3915 additions and 1278 deletions
--- a/.config/zepter.yaml
+++ b/.config/zepter.yaml
@@ -12,7 +12,7 @@ workflows:
        # Check that `A` activates the features of `B`.
        "propagate-feature",
        # These are the features to check:
-        "--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,js-tracer,portable",
+        "--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,js-tracer,portable,keccak-cache-global",
        # Do not try to add a new section to `[features]` of `A` only because `B` exposes that feature. There are edge-cases where this is still needed, but we can add them manually.
        "--left-side-feature-missing=ignore",
        # Ignore the case that `A` it outside of the workspace. Otherwise it will report errors in external dependencies that we have no influence on.
--- a/.github/workflows/hive.yml
+++ b/.github/workflows/hive.yml
@@ -67,7 +67,7 @@ jobs:
          chmod +x hive

      - name: Upload hive assets
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
        with:
          name: hive_assets
          path: ./hive_assets
@@ -187,13 +187,13 @@ jobs:
          fetch-depth: 0

      - name: Download hive assets
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
        with:
          name: hive_assets
          path: /tmp

      - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
        with:
          name: artifacts
          path: /tmp
--- a/.github/workflows/kurtosis-op.yml
+++ b/.github/workflows/kurtosis-op.yml
@@ -41,7 +41,7 @@ jobs:
          fetch-depth: 0

      - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
        with:
          name: artifacts
          path: /tmp
--- a/.github/workflows/kurtosis.yml
+++ b/.github/workflows/kurtosis.yml
@@ -39,7 +39,7 @@ jobs:
          fetch-depth: 0

      - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
        with:
          name: artifacts
          path: /tmp
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -245,12 +245,8 @@ jobs:

  # Checks that selected crates can compile with power set of features
  features:
-    name: features (${{ matrix.partition }}/${{ matrix.total_partitions }})
+    name: features
    runs-on: depot-ubuntu-latest
-    strategy:
-      matrix:
-        partition: [1, 2]
-        total_partitions: [2]
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v6
@@ -268,7 +264,7 @@ jobs:
            --package reth-primitives-traits \
            --package reth-primitives \
            --feature-powerset \
-            --partition ${{ matrix.partition }}/${{ matrix.total_partitions }}
+            --depth 2
        env:
          RUSTFLAGS: -D warnings

--- a/.github/workflows/prepare-reth.yml
+++ b/.github/workflows/prepare-reth.yml
@@ -50,7 +50,7 @@ jobs:

      - name: Upload reth image
        id: upload
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
        with:
          name: artifacts
          path: ./artifacts
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -144,14 +144,14 @@ jobs:

      - name: Upload artifact
        if: ${{ github.event.inputs.dry_run != 'true' }}
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
        with:
          name: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz
          path: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz

      - name: Upload signature
        if: ${{ github.event.inputs.dry_run != 'true' }}
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
        with:
          name: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz.asc
          path: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz.asc
@@ -173,7 +173,7 @@ jobs:
        with:
          fetch-depth: 0
      - name: Download artifacts
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
      - name: Generate full changelog
        id: changelog
        run: |
--- a/.github/workflows/reproducible-build.yml
+++ b/.github/workflows/reproducible-build.yml
@@ -42,7 +42,7 @@ jobs:
          echo "Binaries SHA256 on ${{ matrix.machine }}: $(cat checksum.sha256)"

      - name: Upload the hash
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
        with:
          name: checksum-${{ matrix.machine }}
          path: |
@@ -55,12 +55,12 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Download artifacts from machine-1
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v7
        with:
          name: checksum-machine-1
          path: machine-1/
      - name: Download artifacts from machine-2
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v7
        with:
          name: checksum-machine-2
          path: machine-2/
--- a/.github/workflows/update-superchain.yml
+++ b/.github/workflows/update-superchain.yml
@@ -27,7 +27,7 @@ jobs:
          ./fetch_superchain_config.sh

      - name: Create Pull Request
-        uses: peter-evans/create-pull-request@v7
+        uses: peter-evans/create-pull-request@v8
        with:
          commit-message: "chore: update superchain config"
          title: "chore: update superchain config"
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -329,9 +329,9 @@ dependencies = [

 [[package]]
 name = "alloy-json-abi"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5513d5e6bd1cba6bdcf5373470f559f320c05c8c59493b6e98912fbe6733943f"
+checksum = "6bfca3dbbcb7498f0f60e67aff2ad6aff57032e22eb2fd03189854be11a22c03"
 dependencies = [
 "alloy-primitives",
 "alloy-sol-type-parser",
@@ -426,9 +426,9 @@ dependencies = [

 [[package]]
 name = "alloy-primitives"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "355bf68a433e0fd7f7d33d5a9fc2583fde70bf5c530f63b80845f8da5505cf28"
+checksum = "5c850e6ccbd34b8a463a1e934ffc8fc00e1efc5e5489f2ad82d7797949f3bd4e"
 dependencies = [
 "alloy-rlp",
 "arbitrary",
@@ -447,6 +447,7 @@ dependencies = [
 "proptest",
 "proptest-derive 0.6.0",
 "rand 0.9.2",
+ "rapidhash",
 "ruint",
 "rustc-hash",
 "serde",
@@ -781,9 +782,9 @@ dependencies = [

 [[package]]
 name = "alloy-sol-macro"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3ce480400051b5217f19d6e9a82d9010cdde20f1ae9c00d53591e4a1afbb312"
+checksum = "b2218e3aeb3ee665d117fdf188db0d5acfdc3f7b7502c827421cb78f26a2aec0"
 dependencies = [
 "alloy-sol-macro-expander",
 "alloy-sol-macro-input",
@@ -795,9 +796,9 @@ dependencies = [

 [[package]]
 name = "alloy-sol-macro-expander"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d792e205ed3b72f795a8044c52877d2e6b6e9b1d13f431478121d8d4eaa9028"
+checksum = "b231cb8cc48e66dd1c6e11a1402f3ac86c3667cbc13a6969a0ac030ba7bb8c88"
 dependencies = [
 "alloy-sol-macro-input",
 "const-hex",
@@ -813,9 +814,9 @@ dependencies = [

 [[package]]
 name = "alloy-sol-macro-input"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bd1247a8f90b465ef3f1207627547ec16940c35597875cdc09c49d58b19693c"
+checksum = "49a522d79929c1bf0152b07567a38f7eaed3ab149e53e7528afa78ff11994668"
 dependencies = [
 "const-hex",
 "dunce",
@@ -829,9 +830,9 @@ dependencies = [

 [[package]]
 name = "alloy-sol-type-parser"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "954d1b2533b9b2c7959652df3076954ecb1122a28cc740aa84e7b0a49f6ac0a9"
+checksum = "0475c459859c8d9428af6ff3736614655a57efda8cc435a3b8b4796fa5ac1dd0"
 dependencies = [
 "serde",
 "winnow",
@@ -839,9 +840,9 @@ dependencies = [

 [[package]]
 name = "alloy-sol-types"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70319350969a3af119da6fb3e9bddb1bce66c9ea933600cb297c8b1850ad2a3c"
+checksum = "35287d9d821d5f26011bcd8d9101340898f761c9933cf50fca689bb7ed62fdeb"
 dependencies = [
 "alloy-json-abi",
 "alloy-primitives",
@@ -7209,6 +7210,16 @@ dependencies = [
 "rand_core 0.9.3",
 ]

+[[package]]
+name = "rapidhash"
+version = "4.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d8e65c75143ce5d47c55b510297eeb1182f3c739b6043c537670e9fc18612dae"
+dependencies = [
+ "rand 0.9.2",
+ "rustversion",
+]
+
 [[package]]
 name = "ratatui"
 version = "0.29.0"
@@ -8639,6 +8650,7 @@ dependencies = [
 "derive_more",
 "futures-util",
 "metrics",
+ "rayon",
 "reth-ethereum-forks",
 "reth-ethereum-primitives",
 "reth-execution-errors",
@@ -8944,6 +8956,7 @@ dependencies = [
 "pin-project",
 "rand 0.8.5",
 "rand 0.9.2",
+ "rayon",
 "reth-chainspec",
 "reth-consensus",
 "reth-discv4",
@@ -10894,6 +10907,7 @@ dependencies = [
 "pretty_assertions",
 "proptest",
 "proptest-arbitrary-interop",
+ "rand 0.9.2",
 "reth-ethereum-primitives",
 "reth-execution-errors",
 "reth-metrics",
@@ -12337,9 +12351,9 @@ dependencies = [

 [[package]]
 name = "syn-solidity"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff790eb176cc81bb8936aed0f7b9f14fc4670069a2d371b3e3b0ecce908b2cb3"
+checksum = "60ceeb7c95a4536de0c0e1649bd98d1a72a4bb9590b1f3e45a8a0bfdb7c188c0"
 dependencies = [
 "paste",
 "proc-macro2",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -489,10 +489,10 @@ alloy-dyn-abi = "1.4.1"
 alloy-eip2124 = { version = "0.2.0", default-features = false }
 alloy-eip7928 = { version = "0.1.0" }
 alloy-evm = { version = "0.25.1", default-features = false }
-alloy-primitives = { version = "1.4.1", default-features = false, features = ["map-foldhash"] }
+alloy-primitives = { version = "1.5.0", default-features = false, features = ["map-foldhash"] }
 alloy-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
-alloy-sol-macro = "1.4.1"
-alloy-sol-types = { version = "1.4.1", default-features = false }
+alloy-sol-macro = "1.5.0"
+alloy-sol-types = { version = "1.5.0", default-features = false }
 alloy-trie = { version = "0.9.1", default-features = false }

 alloy-hardforks = "0.4.5"
--- a/2
+++ b/2
@@ -521,5 +521,3 @@ pr:
 	make update-book-cli && \
 	cargo docs --document-private-items && \
 	make test
-
-check-features:
--- a/bin/reth/Cargo.toml
+++ b/bin/reth/Cargo.toml
@@ -81,7 +81,7 @@ backon.workspace = true
 tempfile.workspace = true

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

 otlp = [
    "reth-ethereum-cli/otlp",
@@ -102,7 +102,9 @@ asm-keccak = [
    "reth-ethereum-cli/asm-keccak",
    "reth-node-ethereum/asm-keccak",
 ]
-
+keccak-cache-global = [
+    "reth-node-ethereum/keccak-cache-global",
+]
 jemalloc = [
    "reth-cli-util/jemalloc",
    "reth-node-core/jemalloc",
--- a/crates/cli/commands/src/common.rs
+++ b/crates/cli/commands/src/common.rs
@@ -23,7 +23,7 @@ use reth_node_core::{
    dirs::{ChainPath, DataDirPath},
 };
 use reth_provider::{
-    providers::{BlockchainProvider, NodeTypesForProvider, StaticFileProvider},
+    providers::{BlockchainProvider, NodeTypesForProvider, RocksDBProvider, StaticFileProvider},
    ProviderFactory, StaticFileProviderFactory,
 };
 use reth_stages::{sets::DefaultStages, Pipeline, PipelineTarget};
@@ -75,10 +75,12 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
        let data_dir = self.datadir.clone().resolve_datadir(self.chain.chain());
        let db_path = data_dir.db();
        let sf_path = data_dir.static_files();
+        let rocksdb_path = data_dir.rocksdb();

        if access.is_read_write() {
            reth_fs_util::create_dir_all(&db_path)?;
            reth_fs_util::create_dir_all(&sf_path)?;
+            reth_fs_util::create_dir_all(&rocksdb_path)?;
        }

        let config_path = self.config.clone().unwrap_or_else(|| data_dir.config());
@@ -108,8 +110,14 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
                StaticFileProvider::read_only(sf_path, false)?,
            ),
        };
+        // TransactionDB only support read-write mode
+        let rocksdb_provider = RocksDBProvider::builder(data_dir.rocksdb())
+            .with_default_tables()
+            .with_database_log_level(self.db.log_level)
+            .build()?;

-        let provider_factory = self.create_provider_factory(&config, db, sfp, access)?;
+        let provider_factory =
+            self.create_provider_factory(&config, db, sfp, rocksdb_provider, access)?;
        if access.is_read_write() {
            debug!(target: "reth::cli", chain=%self.chain.chain(), genesis=?self.chain.genesis_hash(), "Initializing genesis");
            init_genesis_with_settings(&provider_factory, self.static_files.to_settings())?;
@@ -128,6 +136,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
        config: &Config,
        db: Arc<DatabaseEnv>,
        static_file_provider: StaticFileProvider<N::Primitives>,
+        rocksdb_provider: RocksDBProvider,
        access: AccessRights,
    ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>>
    where
@@ -138,6 +147,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
            db,
            self.chain.clone(),
            static_file_provider,
+            rocksdb_provider,
        )?
        .with_prune_modes(prune_modes.clone());

--- a/crates/cli/commands/src/db/mod.rs
+++ b/crates/cli/commands/src/db/mod.rs
@@ -2,6 +2,7 @@ use crate::common::{AccessRights, CliNodeTypes, Environment, EnvironmentArgs};
 use clap::{Parser, Subcommand};
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
+use reth_cli_runner::CliContext;
 use reth_db::version::{get_db_version, DatabaseVersionError, DB_VERSION};
 use reth_db_common::DbTool;
 use std::{
@@ -79,7 +80,10 @@ macro_rules! db_exec {

 impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C> {
    /// Execute `db` command
-    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(self) -> eyre::Result<()> {
+    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(
+        self,
+        ctx: CliContext,
+    ) -> eyre::Result<()> {
        let data_dir = self.env.datadir.clone().resolve_datadir(self.env.chain.chain());
        let db_path = data_dir.db();
        let static_files_path = data_dir.static_files();
@@ -158,7 +162,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
                let access_rights =
                    if command.dry_run { AccessRights::RO } else { AccessRights::RW };
                db_exec!(self.env, tool, N, access_rights, {
-                    command.execute(&tool)?;
+                    command.execute(&tool, ctx.task_executor.clone())?;
                });
            }
            Subcommands::StaticFileHeader(command) => {
--- a/crates/cli/commands/src/db/repair_trie.rs
+++ b/crates/cli/commands/src/db/repair_trie.rs
@@ -18,6 +18,7 @@ use reth_node_metrics::{
 };
 use reth_provider::{providers::ProviderNodeTypes, ChainSpecProvider, StageCheckpointReader};
 use reth_stages::StageId;
+use reth_tasks::TaskExecutor;
 use reth_trie::{
    verify::{Output, Verifier},
    Nibbles,
@@ -48,52 +49,37 @@ pub struct Command {

 impl Command {
    /// Execute `db repair-trie` command
-    pub fn execute<N: ProviderNodeTypes>(self, tool: &DbTool<N>) -> eyre::Result<()> {
+    pub fn execute<N: ProviderNodeTypes>(
+        self,
+        tool: &DbTool<N>,
+        task_executor: TaskExecutor,
+    ) -> eyre::Result<()> {
        // Set up metrics server if requested
        let _metrics_handle = if let Some(listen_addr) = self.metrics {
-            // Spawn an OS thread with a single-threaded tokio runtime for the metrics server
            let chain_name = tool.provider_factory.chain_spec().chain().to_string();
+            let executor = task_executor.clone();

-            let handle = std::thread::Builder::new().name("metrics-server".to_string()).spawn(
-                move || {
-                    // Create a single-threaded tokio runtime
-                    let runtime = tokio::runtime::Builder::new_current_thread()
-                        .enable_all()
-                        .build()
-                        .expect("Failed to create tokio runtime for metrics server");
+            let handle = task_executor.spawn_critical("metrics server", async move {
+                let config = MetricServerConfig::new(
+                    listen_addr,
+                    VersionInfo {
+                        version: version_metadata().cargo_pkg_version.as_ref(),
+                        build_timestamp: version_metadata().vergen_build_timestamp.as_ref(),
+                        cargo_features: version_metadata().vergen_cargo_features.as_ref(),
+                        git_sha: version_metadata().vergen_git_sha.as_ref(),
+                        target_triple: version_metadata().vergen_cargo_target_triple.as_ref(),
+                        build_profile: version_metadata().build_profile_name.as_ref(),
+                    },
+                    ChainSpecInfo { name: chain_name },
+                    executor,
+                    Hooks::builder().build(),
+                );

-                    let handle = runtime.handle().clone();
-                    runtime.block_on(async move {
-                        let task_manager = reth_tasks::TaskManager::new(handle.clone());
-                        let task_executor = task_manager.executor();
-
-                        let config = MetricServerConfig::new(
-                            listen_addr,
-                            VersionInfo {
-                                version: version_metadata().cargo_pkg_version.as_ref(),
-                                build_timestamp: version_metadata().vergen_build_timestamp.as_ref(),
-                                cargo_features: version_metadata().vergen_cargo_features.as_ref(),
-                                git_sha: version_metadata().vergen_git_sha.as_ref(),
-                                target_triple: version_metadata()
-                                    .vergen_cargo_target_triple
-                                    .as_ref(),
-                                build_profile: version_metadata().build_profile_name.as_ref(),
-                            },
-                            ChainSpecInfo { name: chain_name },
-                            task_executor,
-                            Hooks::builder().build(),
-                        );
-
-                        // Spawn the metrics server
-                        if let Err(e) = MetricServer::new(config).serve().await {
-                            tracing::error!("Metrics server error: {}", e);
-                        }
-
-                        // Block forever to keep the runtime alive
-                        std::future::pending::<()>().await
-                    });
-                },
-            )?;
+                // Spawn the metrics server
+                if let Err(e) = MetricServer::new(config).serve().await {
+                    tracing::error!("Metrics server error: {}", e);
+                }
+            });

            Some(handle)
        } else {
--- a/crates/cli/commands/src/p2p/mod.rs
+++ b/crates/cli/commands/src/p2p/mod.rs
@@ -189,7 +189,7 @@ impl<C: ChainSpecParser> DownloadArgs<C> {

        let net = NetworkConfigBuilder::<N::NetworkPrimitives>::new(p2p_secret_key)
            .peer_config(config.peers_config_with_basic_nodes_from_file(None))
-            .external_ip_resolver(self.network.nat)
+            .external_ip_resolver(self.network.nat.clone())
            .network_id(self.network.network_id)
            .boot_nodes(boot_nodes.clone())
            .apply(|builder| {
--- a/crates/cli/commands/src/stage/dump/execution.rs
+++ b/crates/cli/commands/src/stage/dump/execution.rs
@@ -9,7 +9,7 @@ use reth_evm::ConfigureEvm;
 use reth_node_builder::NodeTypesWithDB;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::ExecutionStage, Stage, StageCheckpoint, UnwindInput};
@@ -42,6 +42,7 @@ where
                Arc::new(output_db),
                db_tool.chain(),
                StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
            )?,
            to,
            from,
--- a/crates/cli/commands/src/stage/dump/hashing_account.rs
+++ b/crates/cli/commands/src/stage/dump/hashing_account.rs
@@ -6,7 +6,7 @@ use reth_db_api::{database::Database, table::TableImporter, tables};
 use reth_db_common::DbTool;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::AccountHashingStage, Stage, StageCheckpoint, UnwindInput};
@@ -39,6 +39,7 @@ pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Arc<Dat
                Arc::new(output_db),
                db_tool.chain(),
                StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
            )?,
            to,
            from,
--- a/crates/cli/commands/src/stage/dump/hashing_storage.rs
+++ b/crates/cli/commands/src/stage/dump/hashing_storage.rs
@@ -5,7 +5,7 @@ use reth_db_api::{database::Database, table::TableImporter, tables};
 use reth_db_common::DbTool;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::StorageHashingStage, Stage, StageCheckpoint, UnwindInput};
@@ -29,6 +29,7 @@ pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = Arc<Dat
                Arc::new(output_db),
                db_tool.chain(),
                StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
            )?,
            to,
            from,
--- a/crates/cli/commands/src/stage/dump/merkle.rs
+++ b/crates/cli/commands/src/stage/dump/merkle.rs
@@ -12,7 +12,7 @@ use reth_evm::ConfigureEvm;
 use reth_exex::ExExManagerHandle;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{
@@ -62,6 +62,7 @@ where
                Arc::new(output_db),
                db_tool.chain(),
                StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
            )?,
            to,
            from,
--- a/crates/cli/runner/src/lib.rs
+++ b/crates/cli/runner/src/lib.rs
@@ -97,6 +97,57 @@ impl CliRunner {
        command_res
    }

+    /// Executes a command in a blocking context with access to `CliContext`.
+    ///
+    /// See [`Runtime::spawn_blocking`](tokio::runtime::Runtime::spawn_blocking).
+    pub fn run_blocking_command_until_exit<F, E>(
+        self,
+        command: impl FnOnce(CliContext) -> F + Send + 'static,
+    ) -> Result<(), E>
+    where
+        F: Future<Output = Result<(), E>> + Send + 'static,
+        E: Send + Sync + From<std::io::Error> + From<reth_tasks::PanickedTaskError> + 'static,
+    {
+        let AsyncCliRunner { context, mut task_manager, tokio_runtime } =
+            AsyncCliRunner::new(self.tokio_runtime);
+
+        // Spawn the command on the blocking thread pool
+        let handle = tokio_runtime.handle().clone();
+        let command_handle =
+            tokio_runtime.handle().spawn_blocking(move || handle.block_on(command(context)));
+
+        // Wait for the command to complete or ctrl-c
+        let command_res = tokio_runtime.block_on(run_to_completion_or_panic(
+            &mut task_manager,
+            run_until_ctrl_c(
+                async move { command_handle.await.expect("Failed to join blocking task") },
+            ),
+        ));
+
+        if command_res.is_err() {
+            error!(target: "reth::cli", "shutting down due to error");
+        } else {
+            debug!(target: "reth::cli", "shutting down gracefully");
+            task_manager.graceful_shutdown_with_timeout(Duration::from_secs(5));
+        }
+
+        // Shutdown the runtime on a separate thread
+        let (tx, rx) = mpsc::channel();
+        std::thread::Builder::new()
+            .name("tokio-runtime-shutdown".to_string())
+            .spawn(move || {
+                drop(tokio_runtime);
+                let _ = tx.send(());
+            })
+            .unwrap();
+
+        let _ = rx.recv_timeout(Duration::from_secs(5)).inspect_err(|err| {
+            debug!(target: "reth::cli", %err, "tokio runtime shutdown timed out");
+        });
+
+        command_res
+    }
+
    /// Executes a regular future until completion or until external signal received.
    pub fn run_until_ctrl_c<F, E>(self, fut: F) -> Result<(), E>
    where
--- a/crates/consensus/common/src/validation.rs
+++ b/crates/consensus/common/src/validation.rs
@@ -279,20 +279,28 @@ pub fn validate_against_parent_hash_number<H: BlockHeader>(
    header: &H,
    parent: &SealedHeader<H>,
 ) -> Result<(), ConsensusError> {
-    // Parent number is consistent.
-    if parent.number() + 1 != header.number() {
-        return Err(ConsensusError::ParentBlockNumberMismatch {
-            parent_block_number: parent.number(),
-            block_number: header.number(),
-        })
-    }
-
    if parent.hash() != header.parent_hash() {
        return Err(ConsensusError::ParentHashMismatch(
            GotExpected { got: header.parent_hash(), expected: parent.hash() }.into(),
        ))
    }

+    let Some(parent_number) = parent.number().checked_add(1) else {
+        // parent block already reached the maximum
+        return Err(ConsensusError::ParentBlockNumberMismatch {
+            parent_block_number: parent.number(),
+            block_number: u64::MAX,
+        })
+    };
+
+    // Parent number is consistent.
+    if parent_number != header.number() {
+        return Err(ConsensusError::ParentBlockNumberMismatch {
+            parent_block_number: parent.number(),
+            block_number: header.number(),
+        })
+    }
+
    Ok(())
 }

--- a/crates/e2e-test-utils/src/setup_import.rs
+++ b/crates/e2e-test-utils/src/setup_import.rs
@@ -110,6 +110,7 @@ pub async fn setup_engine_with_chain_import(
        // Create database path and static files path
        let db_path = datadir.join("db");
        let static_files_path = datadir.join("static_files");
+        let rocksdb_dir_path = datadir.join("rocksdb");

        // Initialize the database using init_db (same as CLI import command)
        // Use the same database arguments as the node will use
@@ -125,6 +126,7 @@ pub async fn setup_engine_with_chain_import(
            db.clone(),
            chain_spec.clone(),
            reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())?,
+            reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
        )?;

        // Initialize genesis if needed
@@ -311,6 +313,7 @@ mod tests {
        std::fs::create_dir_all(&datadir).unwrap();
        let db_path = datadir.join("db");
        let static_files_path = datadir.join("static_files");
+        let rocksdb_dir_path = datadir.join("rocksdb");

        // Import the chain
        {
@@ -324,6 +327,9 @@ mod tests {
                chain_spec.clone(),
                reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())
                    .unwrap(),
+                reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path.clone())
+                    .build()
+                    .unwrap(),
            )
            .expect("failed to create provider factory");

@@ -385,6 +391,9 @@ mod tests {
                chain_spec.clone(),
                reth_provider::providers::StaticFileProvider::read_only(static_files_path, false)
                    .unwrap(),
+                reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
+                    .build()
+                    .unwrap(),
            )
            .expect("failed to create provider factory");

@@ -472,11 +481,15 @@ mod tests {
        // Create static files path
        let static_files_path = datadir.join("static_files");

+        // Create rocksdb path
+        let rocksdb_dir_path = datadir.join("rocksdb");
+
        // Create a provider factory
        let provider_factory: ProviderFactory<MockNodeTypesWithDB> = ProviderFactory::new(
            db.clone(),
            chain_spec.clone(),
            reth_provider::providers::StaticFileProvider::read_write(static_files_path).unwrap(),
+            reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
        )
        .expect("failed to create provider factory");

--- a/crates/engine/primitives/src/config.rs
+++ b/crates/engine/primitives/src/config.rs
@@ -101,7 +101,7 @@ pub struct TreeConfig {
    state_provider_metrics: bool,
    /// Cross-block cache size in bytes.
    cross_block_cache_size: u64,
-    /// Whether the host has enough parallelism to run state root in parallel.
+    /// Whether the host has enough parallelism to run state root task.
    has_enough_parallelism: bool,
    /// Whether multiproof task should chunk proof targets.
    multiproof_chunking_enabled: bool,
@@ -403,17 +403,12 @@ impl TreeConfig {
        self
    }

-    /// Setter for whether or not the host has enough parallelism to run state root in parallel.
+    /// Setter for has enough parallelism.
    pub const fn with_has_enough_parallelism(mut self, has_enough_parallelism: bool) -> Self {
        self.has_enough_parallelism = has_enough_parallelism;
        self
    }

-    /// Whether or not the host has enough parallelism to run state root in parallel.
-    pub const fn has_enough_parallelism(&self) -> bool {
-        self.has_enough_parallelism
-    }
-
    /// Setter for state provider metrics.
    pub const fn with_state_provider_metrics(mut self, state_provider_metrics: bool) -> Self {
        self.state_provider_metrics = state_provider_metrics;
--- a/crates/engine/primitives/src/lib.rs
+++ b/crates/engine/primitives/src/lib.rs
@@ -22,7 +22,8 @@ use reth_trie_common::HashedPostState;
 use serde::{de::DeserializeOwned, Serialize};

 // Re-export [`ExecutionPayload`] moved to `reth_payload_primitives`
-pub use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator};
+#[cfg(feature = "std")]
+pub use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator, ExecutableTxTuple};
 pub use reth_payload_primitives::ExecutionPayload;

 mod error;
--- a/crates/engine/tree/Cargo.toml
+++ b/crates/engine/tree/Cargo.toml
@@ -16,7 +16,7 @@ reth-chain-state.workspace = true
 reth-chainspec = { workspace = true, optional = true }
 reth-consensus.workspace = true
 reth-db.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-errors.workspace = true
 reth-execution-types.workspace = true
 reth-evm = { workspace = true, features = ["metrics"] }
--- a/crates/engine/tree/benches/state_root_task.rs
+++ b/crates/engine/tree/benches/state_root_task.rs
@@ -230,17 +230,18 @@ fn bench_state_root(c: &mut Criterion) {
                            let mut handle = payload_processor.spawn(
                                Default::default(),
                                (
-                                    core::iter::empty::<
+                                    Vec::<
                                        Result<
                                            Recovered<TransactionSigned>,
                                            core::convert::Infallible,
                                        >,
-                                    >(),
+                                    >::new(),
                                    std::convert::identity,
                                ),
                                StateProviderBuilder::new(provider.clone(), genesis_hash, None),
                                OverlayStateProviderFactory::new(provider),
                                &TreeConfig::default(),
+                                None,
                            );

                            let mut state_hook = handle.state_hook();
--- a/crates/engine/tree/src/tree/payload_processor/bal.rs
+++ b/crates/engine/tree/src/tree/payload_processor/bal.rs
@@ -0,0 +1,318 @@
+//! BAL (Block Access List, EIP-7928) related functionality.
+
+use alloy_consensus::constants::KECCAK_EMPTY;
+use alloy_eip7928::BlockAccessList;
+use alloy_primitives::{keccak256, U256};
+use reth_primitives_traits::Account;
+use reth_provider::{AccountReader, ProviderError};
+use reth_trie::{HashedPostState, HashedStorage};
+
+/// Converts a Block Access List into a [`HashedPostState`] by extracting the final state
+/// of modified accounts and storage slots.
+pub fn bal_to_hashed_post_state<P>(
+    bal: &BlockAccessList,
+    provider: &P,
+) -> Result<HashedPostState, ProviderError>
+where
+    P: AccountReader,
+{
+    let mut hashed_state = HashedPostState::with_capacity(bal.len());
+
+    for account_changes in bal {
+        let address = account_changes.address;
+        let hashed_address = keccak256(address);
+
+        // Get the latest balance (last balance change if any)
+        let balance = account_changes.balance_changes.last().map(|change| change.post_balance);
+
+        // Get the latest nonce (last nonce change if any)
+        let nonce = account_changes.nonce_changes.last().map(|change| change.new_nonce);
+
+        // Get the latest code (last code change if any)
+        let code_hash = if let Some(code_change) = account_changes.code_changes.last() {
+            if code_change.new_code.is_empty() {
+                Some(Some(KECCAK_EMPTY))
+            } else {
+                Some(Some(keccak256(&code_change.new_code)))
+            }
+        } else {
+            None
+        };
+
+        // Only fetch account from provider if we're missing any field
+        let existing_account = if balance.is_none() || nonce.is_none() || code_hash.is_none() {
+            provider.basic_account(&address)?
+        } else {
+            None
+        };
+
+        // Build the final account state
+        let account = Account {
+            balance: balance.unwrap_or_else(|| {
+                existing_account.as_ref().map(|acc| acc.balance).unwrap_or(U256::ZERO)
+            }),
+            nonce: nonce
+                .unwrap_or_else(|| existing_account.as_ref().map(|acc| acc.nonce).unwrap_or(0)),
+            bytecode_hash: code_hash.unwrap_or_else(|| {
+                existing_account.as_ref().and_then(|acc| acc.bytecode_hash).or(Some(KECCAK_EMPTY))
+            }),
+        };
+
+        hashed_state.accounts.insert(hashed_address, Some(account));
+
+        // Process storage changes
+        if !account_changes.storage_changes.is_empty() {
+            let mut storage_map = HashedStorage::new(false);
+
+            for slot_changes in &account_changes.storage_changes {
+                let hashed_slot = keccak256(slot_changes.slot);
+
+                // Get the last change for this slot
+                if let Some(last_change) = slot_changes.changes.last() {
+                    storage_map
+                        .storage
+                        .insert(hashed_slot, U256::from_be_bytes(last_change.new_value.0));
+                }
+            }
+
+            if !storage_map.storage.is_empty() {
+                hashed_state.storages.insert(hashed_address, storage_map);
+            }
+        }
+    }
+
+    Ok(hashed_state)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_eip7928::{
+        AccountChanges, BalanceChange, CodeChange, NonceChange, SlotChanges, StorageChange,
+    };
+    use alloy_primitives::{Address, Bytes, StorageKey, B256};
+    use reth_revm::test_utils::StateProviderTest;
+
+    #[test]
+    fn test_bal_to_hashed_post_state_basic() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(100))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        assert_eq!(result.accounts.len(), 1);
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        assert!(account_opt.is_some());
+
+        let account = account_opt.as_ref().unwrap();
+        assert_eq!(account.balance, U256::from(100));
+        assert_eq!(account.nonce, 1);
+        assert_eq!(account.bytecode_hash, Some(KECCAK_EMPTY));
+    }
+
+    #[test]
+    fn test_bal_with_storage_changes() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let slot = StorageKey::random();
+        let value = B256::random();
+
+        let slot_changes = SlotChanges { slot, changes: vec![StorageChange::new(0, value)] };
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![slot_changes],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(500))],
+            nonce_changes: vec![NonceChange::new(0, 2)],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        assert!(result.storages.contains_key(&hashed_address));
+
+        let storage = result.storages.get(&hashed_address).unwrap();
+        let hashed_slot = keccak256(slot);
+
+        let stored_value = storage.storage.get(&hashed_slot).unwrap();
+        assert_eq!(*stored_value, U256::from_be_bytes(value.0));
+    }
+
+    #[test]
+    fn test_bal_with_code_change() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let code = Bytes::from(vec![0x60, 0x80, 0x60, 0x40]); // Some bytecode
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![CodeChange::new(0, code.clone())],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        let expected_code_hash = keccak256(&code);
+        assert_eq!(account.bytecode_hash, Some(expected_code_hash));
+    }
+
+    #[test]
+    fn test_bal_with_empty_code() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let empty_code = Bytes::default();
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![CodeChange::new(0, empty_code)],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        assert_eq!(account.bytecode_hash, Some(KECCAK_EMPTY));
+    }
+
+    #[test]
+    fn test_bal_multiple_changes_takes_last() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+
+        // Multiple balance changes - should take the last one
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![
+                BalanceChange::new(0, U256::from(100)),
+                BalanceChange::new(1, U256::from(200)),
+                BalanceChange::new(2, U256::from(300)),
+            ],
+            nonce_changes: vec![
+                NonceChange::new(0, 1),
+                NonceChange::new(1, 2),
+                NonceChange::new(2, 3),
+            ],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        // Should have the last values
+        assert_eq!(account.balance, U256::from(300));
+        assert_eq!(account.nonce, 3);
+    }
+
+    #[test]
+    fn test_bal_uses_provider_for_missing_fields() {
+        let mut provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let code_hash = B256::random();
+        let existing_account =
+            Account { balance: U256::from(999), nonce: 42, bytecode_hash: Some(code_hash) };
+        provider.insert_account(address, existing_account, None, Default::default());
+
+        // Only change balance, nonce and code should come from provider
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(1500))],
+            nonce_changes: vec![],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        // Balance should be updated
+        assert_eq!(account.balance, U256::from(1500));
+        // Nonce and bytecode_hash should come from provider
+        assert_eq!(account.nonce, 42);
+        assert_eq!(account.bytecode_hash, Some(code_hash));
+    }
+
+    #[test]
+    fn test_bal_multiple_storage_changes_per_slot() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let slot = StorageKey::random();
+
+        // Multiple changes to the same slot - should take the last one
+        let slot_changes = SlotChanges {
+            slot,
+            changes: vec![
+                StorageChange::new(0, B256::from(U256::from(100).to_be_bytes::<32>())),
+                StorageChange::new(1, B256::from(U256::from(200).to_be_bytes::<32>())),
+                StorageChange::new(2, B256::from(U256::from(300).to_be_bytes::<32>())),
+            ],
+        };
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![slot_changes],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(100))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let storage = result.storages.get(&hashed_address).unwrap();
+        let hashed_slot = keccak256(slot);
+
+        let stored_value = storage.storage.get(&hashed_slot).unwrap();
+
+        // Should have the last value
+        assert_eq!(*stored_value, U256::from(300));
+    }
+}
--- a/crates/engine/tree/src/tree/payload_processor/mod.rs
+++ b/crates/engine/tree/src/tree/payload_processor/mod.rs
@@ -13,6 +13,7 @@ use crate::tree::{
    sparse_trie::SparseTrieTask,
    StateProviderBuilder, TreeConfig,
 };
+use alloy_eip7928::BlockAccessList;
 use alloy_eips::eip1898::BlockWithParent;
 use alloy_evm::{block::StateChangeSource, ToTxEnv};
 use alloy_primitives::B256;
@@ -21,7 +22,7 @@ use executor::WorkloadExecutor;
 use multiproof::{SparseTrieUpdate, *};
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
-use rayon::iter::{ParallelBridge, ParallelIterator};
+use rayon::prelude::*;
 use reth_engine_primitives::ExecutableTxIterator;
 use reth_evm::{
    execute::{ExecutableTxFor, WithTxEnv},
@@ -49,8 +50,9 @@ use std::{
    },
    time::Instant,
 };
-use tracing::{debug, debug_span, instrument, warn, Span};
+use tracing::{debug, debug_span, error, instrument, warn, Span};

+pub mod bal;
 mod configured_sparse_trie;
 pub mod executor;
 pub mod multiproof;
@@ -212,6 +214,7 @@ where
        provider_builder: StateProviderBuilder<N, P>,
        multiproof_provider_factory: F,
        config: &TreeConfig,
+        bal: Option<Arc<BlockAccessList>>,
    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
@@ -252,19 +255,45 @@ where
        // wire the multiproof task to the prewarm task
        let to_multi_proof = Some(multi_proof_task.state_root_message_sender());

-        let prewarm_handle = self.spawn_caching_with(
-            env,
-            prewarm_rx,
-            transaction_count_hint,
-            provider_builder,
-            to_multi_proof.clone(),
-        );
+        // Handle BAL-based optimization if available
+        let prewarm_handle = if let Some(bal) = bal {
+            // When BAL is present, skip spawning prewarm tasks entirely and send BAL to multiproof
+            debug!(target: "engine::tree::payload_processor", "BAL present, skipping prewarm tasks");
+
+            // Send BAL message immediately to MultiProofTask
+            if let Some(ref sender) = to_multi_proof &&
+                let Err(err) = sender.send(MultiProofMessage::BlockAccessList(bal))
+            {
+                // In this case state root validation will simply fail
+                error!(target: "engine::tree::payload_processor", ?err, "Failed to send BAL to MultiProofTask");
+            }
+
+            // Spawn minimal cache-only task without prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                transaction_count_hint,
+                provider_builder.clone(),
+                None, // Don't send proof targets when BAL is present
+            )
+        } else {
+            // Normal path: spawn with full prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                transaction_count_hint,
+                provider_builder.clone(),
+                to_multi_proof.clone(),
+            )
+        };

        // spawn multi-proof task
        let parent_span = span.clone();
        self.executor.spawn_blocking(move || {
            let _enter = parent_span.entered();
-            multi_proof_task.run();
+            // Build a state provider for the multiproof task
+            let provider = provider_builder.build().expect("failed to build provider");
+            multi_proof_task.run(provider);
        });

        // wire the sparse trie to the state root response receiver
@@ -318,36 +347,32 @@ where
        usize,
    ) {
        let (transactions, convert) = transactions.into();
-        let transactions = transactions.into_iter();
-        // Get the transaction count for prewarming task
-        // Use upper bound if available (more accurate), otherwise use lower bound
-        let (lower, upper) = transactions.size_hint();
-        let transaction_count_hint = upper.unwrap_or(lower);
+        let transactions = transactions.into_par_iter();
+        let transaction_count_hint = transactions.len();

-        // Spawn a task that iterates through all transactions in parallel and sends them to the
-        // main task.
-        let (tx, rx) = mpsc::channel();
+        let (ooo_tx, ooo_rx) = mpsc::channel();
+        let (prewarm_tx, prewarm_rx) = mpsc::channel();
+        let (execute_tx, execute_rx) = mpsc::channel();
+
+        // Spawn a task that `convert`s all transactions in parallel and sends them out-of-order.
        self.executor.spawn_blocking(move || {
-            transactions.enumerate().par_bridge().for_each_with(tx, |sender, (idx, tx)| {
+            transactions.enumerate().for_each_with(ooo_tx, |ooo_tx, (idx, tx)| {
                let tx = convert(tx);
                let tx = tx.map(|tx| WithTxEnv { tx_env: tx.to_tx_env(), tx: Arc::new(tx) });
-                let _ = sender.send((idx, tx));
+                // Only send Ok(_) variants to prewarming task.
+                if let Ok(tx) = &tx {
+                    let _ = prewarm_tx.send(tx.clone());
+                }
+                let _ = ooo_tx.send((idx, tx));
            });
        });

        // Spawn a task that processes out-of-order transactions from the task above and sends them
-        // to prewarming and execution tasks.
-        let (prewarm_tx, prewarm_rx) = mpsc::channel();
-        let (execute_tx, execute_rx) = mpsc::channel();
+        // 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)) = rx.recv() {
-                // only send Ok(_) variants to prewarming task
-                if let Ok(tx) = &tx {
-                    let _ = prewarm_tx.send(tx.clone());
-                }
-
+            while let Ok((idx, tx)) = ooo_rx.recv() {
                if next_for_execution == idx {
                    let _ = execute_tx.send(tx);
                    next_for_execution += 1;
@@ -599,7 +624,7 @@ impl<Tx, Err> PayloadHandle<Tx, Err> {

        move |source: StateChangeSource, state: &EvmState| {
            if let Some(sender) = &to_multi_proof {
-                let _ = sender.send(MultiProofMessage::StateUpdate(source, state.clone()));
+                let _ = sender.send(MultiProofMessage::StateUpdate(source.into(), state.clone()));
            }
        }
    }
@@ -1057,19 +1082,17 @@ mod tests {

        let provider_factory = BlockchainProvider::new(factory).unwrap();

-        let mut handle =
-            payload_processor.spawn(
-                Default::default(),
-                (
-                    core::iter::empty::<
-                        Result<Recovered<TransactionSigned>, core::convert::Infallible>,
-                    >(),
-                    std::convert::identity,
-                ),
-                StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
-                OverlayStateProviderFactory::new(provider_factory),
-                &TreeConfig::default(),
-            );
+        let mut handle = payload_processor.spawn(
+            Default::default(),
+            (
+                Vec::<Result<Recovered<TransactionSigned>, core::convert::Infallible>>::new(),
+                std::convert::identity,
+            ),
+            StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
+            OverlayStateProviderFactory::new(provider_factory),
+            &TreeConfig::default(),
+            None, // No BAL for test
+        );

        let mut state_hook = handle.state_hook();

--- a/crates/engine/tree/src/tree/payload_processor/multiproof.rs
+++ b/crates/engine/tree/src/tree/payload_processor/multiproof.rs
@@ -1,5 +1,7 @@
 //! Multiproof task related functionality.

+use crate::tree::payload_processor::bal::bal_to_hashed_post_state;
+use alloy_eip7928::BlockAccessList;
 use alloy_evm::block::StateChangeSource;
 use alloy_primitives::{
    keccak256,
@@ -11,6 +13,7 @@ use dashmap::DashMap;
 use derive_more::derive::Deref;
 use metrics::{Gauge, Histogram};
 use reth_metrics::Metrics;
+use reth_provider::AccountReader;
 use reth_revm::state::EvmState;
 use reth_trie::{
    added_removed_keys::MultiAddedRemovedKeys, DecodedMultiProof, HashedPostState, HashedStorage,
@@ -26,6 +29,30 @@ use reth_trie_parallel::{
 use std::{collections::BTreeMap, mem, ops::DerefMut, sync::Arc, time::Instant};
 use tracing::{debug, error, instrument, trace};

+/// Source of state changes, either from EVM execution or from a Block Access List.
+#[derive(Clone, Copy)]
+pub enum Source {
+    /// State changes from EVM execution.
+    Evm(StateChangeSource),
+    /// State changes from Block Access List (EIP-7928).
+    BlockAccessList,
+}
+
+impl std::fmt::Debug for Source {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Evm(source) => source.fmt(f),
+            Self::BlockAccessList => f.write_str("BlockAccessList"),
+        }
+    }
+}
+
+impl From<StateChangeSource> for Source {
+    fn from(source: StateChangeSource) -> Self {
+        Self::Evm(source)
+    }
+}
+
 /// Maximum number of targets to batch together for prefetch batching.
 /// Prefetches are just proof requests (no state merging), so we allow a higher cap than state
 /// updates
@@ -82,7 +109,7 @@ pub(super) enum MultiProofMessage {
    /// Prefetch proof targets
    PrefetchProofs(MultiProofTargets),
    /// New state update from transaction execution with its source
-    StateUpdate(StateChangeSource, EvmState),
+    StateUpdate(Source, EvmState),
    /// State update that can be applied to the sparse trie without any new proofs.
    ///
    /// It can be the case when all accounts and storage slots from the state update were already
@@ -93,6 +120,11 @@ pub(super) enum MultiProofMessage {
        /// The state update that was used to calculate the proof
        state: HashedPostState,
    },
+    /// Block Access List (EIP-7928; BAL) containing complete state changes for the block.
+    ///
+    /// When received, the task generates a single state update from the BAL and processes it.
+    /// No further messages are expected after receiving this variant.
+    BlockAccessList(Arc<BlockAccessList>),
    /// Signals state update stream end.
    ///
    /// This is triggered by block execution, indicating that no additional state updates are
@@ -280,7 +312,7 @@ impl StorageMultiproofInput {
 /// Input parameters for dispatching a multiproof calculation.
 #[derive(Debug)]
 struct MultiproofInput {
-    source: Option<StateChangeSource>,
+    source: Option<Source>,
    hashed_state_update: HashedPostState,
    proof_targets: MultiProofTargets,
    proof_sequence_number: u64,
@@ -883,9 +915,19 @@ impl MultiProofTask {
        skip(self, update),
        fields(accounts = update.len(), chunks = 0)
    )]
-    fn on_state_update(&mut self, source: StateChangeSource, update: EvmState) -> u64 {
+    fn on_state_update(&mut self, source: Source, update: EvmState) -> u64 {
        let hashed_state_update = evm_state_to_hashed_post_state(update);
+        self.on_hashed_state_update(source, hashed_state_update)
+    }

+    /// Processes a hashed state update and dispatches multiproofs as needed.
+    ///
+    /// Returns the number of state updates dispatched (both `EmptyProof` and regular multiproofs).
+    fn on_hashed_state_update(
+        &mut self,
+        source: Source,
+        hashed_state_update: HashedPostState,
+    ) -> u64 {
        // Update removed keys based on the state update.
        self.multi_added_removed_keys.update_with_state(&hashed_state_update);

@@ -982,12 +1024,16 @@ impl MultiProofTask {
    /// This preserves ordering without requeuing onto the channel.
    ///
    /// Returns `true` if done, `false` to continue.
-    fn process_multiproof_message(
+    fn process_multiproof_message<P>(
        &mut self,
        msg: MultiProofMessage,
        ctx: &mut MultiproofBatchCtx,
        batch_metrics: &mut MultiproofBatchMetrics,
-    ) -> bool {
+        provider: &P,
+    ) -> bool
+    where
+        P: AccountReader,
+    {
        match msg {
            // Prefetch proofs: batch consecutive prefetch requests up to target/message limits
            MultiProofMessage::PrefetchProofs(targets) => {
@@ -1146,6 +1192,56 @@ impl MultiProofTask {

                false
            }
+            // Process Block Access List (BAL) - complete state changes provided upfront
+            MultiProofMessage::BlockAccessList(bal) => {
+                trace!(target: "engine::tree::payload_processor::multiproof", "processing MultiProofMessage::BAL");
+
+                if ctx.first_update_time.is_none() {
+                    self.metrics
+                        .first_update_wait_time_histogram
+                        .record(ctx.start.elapsed().as_secs_f64());
+                    ctx.first_update_time = Some(Instant::now());
+                    debug!(target: "engine::tree::payload_processor::multiproof", "Started state root calculation from BAL");
+                }
+
+                // Convert BAL to HashedPostState and process it
+                match bal_to_hashed_post_state(&bal, &provider) {
+                    Ok(hashed_state) => {
+                        debug!(
+                            target: "engine::tree::payload_processor::multiproof",
+                            accounts = hashed_state.accounts.len(),
+                            storages = hashed_state.storages.len(),
+                            "Processing BAL state update"
+                        );
+
+                        // Use BlockAccessList as source for BAL-derived state updates
+                        batch_metrics.state_update_proofs_requested +=
+                            self.on_hashed_state_update(Source::BlockAccessList, hashed_state);
+                    }
+                    Err(err) => {
+                        error!(target: "engine::tree::payload_processor::multiproof", ?err, "Failed to convert BAL to hashed state");
+                        return true;
+                    }
+                }
+
+                // Mark updates as finished since BAL provides complete state
+                ctx.updates_finished_time = Some(Instant::now());
+
+                // Check if we're done (might need to wait for proofs to complete)
+                if self.is_done(
+                    batch_metrics.proofs_processed,
+                    batch_metrics.state_update_proofs_requested,
+                    batch_metrics.prefetch_proofs_requested,
+                    ctx.updates_finished(),
+                ) {
+                    debug!(
+                        target: "engine::tree::payload_processor::multiproof",
+                        "BAL processed and all proofs complete, ending calculation"
+                    );
+                    return true;
+                }
+                false
+            }
            // Signal that no more state updates will arrive
            MultiProofMessage::FinishedStateUpdates => {
                trace!(target: "engine::tree::payload_processor::multiproof", "processing MultiProofMessage::FinishedStateUpdates");
@@ -1238,7 +1334,10 @@ impl MultiProofTask {
        target = "engine::tree::payload_processor::multiproof",
        skip_all
    )]
-    pub(crate) fn run(mut self) {
+    pub(crate) fn run<P>(mut self, provider: P)
+    where
+        P: AccountReader,
+    {
        let mut ctx = MultiproofBatchCtx::new(Instant::now());
        let mut batch_metrics = MultiproofBatchMetrics::default();

@@ -1248,7 +1347,7 @@ impl MultiProofTask {
            trace!(target: "engine::tree::payload_processor::multiproof", "entering main channel receiving loop");

            if let Some(msg) = ctx.pending_msg.take() {
-                if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
+                if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics, &provider) {
                    break 'main;
                }
                continue;
@@ -1323,7 +1422,7 @@ impl MultiProofTask {
                        }
                    };

-                    if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
+                    if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics, &provider) {
                        break 'main;
                    }
                }
@@ -1374,7 +1473,7 @@ struct MultiproofBatchCtx {
    /// Reusable buffer for accumulating prefetch targets during batching.
    accumulated_prefetch_targets: Vec<MultiProofTargets>,
    /// Reusable buffer for accumulating state updates during batching.
-    accumulated_state_updates: Vec<(StateChangeSource, EvmState)>,
+    accumulated_state_updates: Vec<(Source, EvmState)>,
 }

 impl MultiproofBatchCtx {
@@ -1492,34 +1591,44 @@ where
 /// are safe to merge because they originate from the same logical execution and can be
 /// coalesced to amortize proof work.
 fn can_batch_state_update(
-    batch_source: StateChangeSource,
+    batch_source: Source,
    batch_update: &EvmState,
-    next_source: StateChangeSource,
+    next_source: Source,
    next_update: &EvmState,
 ) -> bool {
-    if !same_state_change_source(batch_source, next_source) {
+    if !same_source(batch_source, next_source) {
        return false;
    }

    match (batch_source, next_source) {
-        (StateChangeSource::PreBlock(_), StateChangeSource::PreBlock(_)) |
-        (StateChangeSource::PostBlock(_), StateChangeSource::PostBlock(_)) => {
-            batch_update == next_update
-        }
+        (
+            Source::Evm(StateChangeSource::PreBlock(_)),
+            Source::Evm(StateChangeSource::PreBlock(_)),
+        ) |
+        (
+            Source::Evm(StateChangeSource::PostBlock(_)),
+            Source::Evm(StateChangeSource::PostBlock(_)),
+        ) => batch_update == next_update,
        _ => true,
    }
 }

-/// Checks whether two state change sources refer to the same origin.
-fn same_state_change_source(lhs: StateChangeSource, rhs: StateChangeSource) -> bool {
+/// Checks whether two sources refer to the same origin.
+fn same_source(lhs: Source, rhs: Source) -> bool {
    match (lhs, rhs) {
-        (StateChangeSource::Transaction(a), StateChangeSource::Transaction(b)) => a == b,
-        (StateChangeSource::PreBlock(a), StateChangeSource::PreBlock(b)) => {
-            mem::discriminant(&a) == mem::discriminant(&b)
-        }
-        (StateChangeSource::PostBlock(a), StateChangeSource::PostBlock(b)) => {
-            mem::discriminant(&a) == mem::discriminant(&b)
-        }
+        (
+            Source::Evm(StateChangeSource::Transaction(a)),
+            Source::Evm(StateChangeSource::Transaction(b)),
+        ) => a == b,
+        (
+            Source::Evm(StateChangeSource::PreBlock(a)),
+            Source::Evm(StateChangeSource::PreBlock(b)),
+        ) => mem::discriminant(&a) == mem::discriminant(&b),
+        (
+            Source::Evm(StateChangeSource::PostBlock(a)),
+            Source::Evm(StateChangeSource::PostBlock(b)),
+        ) => mem::discriminant(&a) == mem::discriminant(&b),
+        (Source::BlockAccessList, Source::BlockAccessList) => true,
        _ => false,
    }
 }
@@ -1539,7 +1648,8 @@ fn estimate_evm_state_targets(state: &EvmState) -> usize {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use alloy_primitives::map::B256Set;
+    use alloy_eip7928::{AccountChanges, BalanceChange};
+    use alloy_primitives::{map::B256Set, Address};
    use reth_provider::{
        providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
        BlockReader, DatabaseProviderFactory, PruneCheckpointReader, StageCheckpointReader,
@@ -1548,7 +1658,7 @@ mod tests {
    use reth_trie::MultiProof;
    use reth_trie_parallel::proof_task::{ProofTaskCtx, ProofWorkerHandle};
    use revm_primitives::{B256, U256};
-    use std::sync::OnceLock;
+    use std::sync::{Arc, OnceLock};
    use tokio::runtime::{Handle, Runtime};

    /// Get a handle to the test runtime, creating it if necessary
@@ -2109,8 +2219,8 @@ mod tests {
        let source = StateChangeSource::Transaction(0);

        let tx = task.state_root_message_sender();
-        tx.send(MultiProofMessage::StateUpdate(source, update1.clone())).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, update2.clone())).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), update1.clone())).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), update2.clone())).unwrap();

        let proofs_requested =
            if let Ok(MultiProofMessage::StateUpdate(_src, update)) = task.rx.recv() {
@@ -2129,7 +2239,7 @@ mod tests {
                assert!(merged_update.contains_key(&addr1));
                assert!(merged_update.contains_key(&addr2));

-                task.on_state_update(source, merged_update)
+                task.on_state_update(source.into(), merged_update)
            } else {
                panic!("Expected StateUpdate message");
            };
@@ -2173,20 +2283,20 @@ mod tests {

        // Queue: A1 (immediate dispatch), B1 (batched), A2 (should become pending)
        let tx = task.state_root_message_sender();
-        tx.send(MultiProofMessage::StateUpdate(source_a, create_state_update(addr_a1, 100)))
+        tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a1, 100)))
            .unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source_b, create_state_update(addr_b1, 200)))
+        tx.send(MultiProofMessage::StateUpdate(source_b.into(), create_state_update(addr_b1, 200)))
            .unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source_a, create_state_update(addr_a2, 300)))
+        tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a2, 300)))
            .unwrap();

        let mut pending_msg: Option<MultiProofMessage> = None;

        if let Ok(MultiProofMessage::StateUpdate(first_source, _)) = task.rx.recv() {
-            assert!(same_state_change_source(first_source, source_a));
+            assert!(same_source(first_source, source_a.into()));

            // Simulate batching loop for remaining messages
-            let mut accumulated_updates: Vec<(StateChangeSource, EvmState)> = Vec::new();
+            let mut accumulated_updates: Vec<(Source, EvmState)> = Vec::new();
            let mut accumulated_targets = 0usize;

            loop {
@@ -2234,7 +2344,7 @@ mod tests {

            assert_eq!(accumulated_updates.len(), 1, "Should only batch matching sources");
            let batch_source = accumulated_updates[0].0;
-            assert!(same_state_change_source(batch_source, source_b));
+            assert!(same_source(batch_source, source_b.into()));

            let batch_source = accumulated_updates[0].0;
            let mut merged_update = accumulated_updates.remove(0).1;
@@ -2242,10 +2352,7 @@ mod tests {
                merged_update.extend(next_update);
            }

-            assert!(
-                same_state_change_source(batch_source, source_b),
-                "Batch should use matching source"
-            );
+            assert!(same_source(batch_source, source_b.into()), "Batch should use matching source");
            assert!(merged_update.contains_key(&addr_b1));
            assert!(!merged_update.contains_key(&addr_a1));
            assert!(!merged_update.contains_key(&addr_a2));
@@ -2255,7 +2362,7 @@ mod tests {

        match pending_msg {
            Some(MultiProofMessage::StateUpdate(pending_source, pending_update)) => {
-                assert!(same_state_change_source(pending_source, source_a));
+                assert!(same_source(pending_source, source_a.into()));
                assert!(pending_update.contains_key(&addr_a2));
            }
            other => panic!("Expected pending StateUpdate with source_a, got {:?}", other),
@@ -2298,17 +2405,20 @@ mod tests {

        // Queue: first update dispatched immediately, next two should not merge
        let tx = task.state_root_message_sender();
-        tx.send(MultiProofMessage::StateUpdate(source, create_state_update(addr1, 100))).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, create_state_update(addr2, 200))).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, create_state_update(addr3, 300))).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr1, 100)))
+            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr2, 200)))
+            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr3, 300)))
+            .unwrap();

        let mut pending_msg: Option<MultiProofMessage> = None;

        if let Ok(MultiProofMessage::StateUpdate(first_source, first_update)) = task.rx.recv() {
-            assert!(same_state_change_source(first_source, source));
+            assert!(same_source(first_source, source.into()));
            assert!(first_update.contains_key(&addr1));

-            let mut accumulated_updates: Vec<(StateChangeSource, EvmState)> = Vec::new();
+            let mut accumulated_updates: Vec<(Source, EvmState)> = Vec::new();
            let mut accumulated_targets = 0usize;

            loop {
@@ -2360,7 +2470,7 @@ mod tests {
                "Second pre-block update should not merge with a different payload"
            );
            let (batched_source, batched_update) = accumulated_updates.remove(0);
-            assert!(same_state_change_source(batched_source, source));
+            assert!(same_source(batched_source, source.into()));
            assert!(batched_update.contains_key(&addr2));
            assert!(!batched_update.contains_key(&addr3));

@@ -2440,8 +2550,8 @@ mod tests {
        let tx = task.state_root_message_sender();
        tx.send(MultiProofMessage::PrefetchProofs(targets1)).unwrap();
        tx.send(MultiProofMessage::PrefetchProofs(targets2)).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, state_update1)).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, state_update2)).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), state_update1)).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), state_update2)).unwrap();
        tx.send(MultiProofMessage::PrefetchProofs(targets3.clone())).unwrap();

        // Step 1: Receive and batch PrefetchProofs (should get targets1 + targets2)
@@ -2508,6 +2618,7 @@ mod tests {
        use revm_state::Account;

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

        // Queue: Prefetch1, StateUpdate, Prefetch2
@@ -2539,7 +2650,7 @@ mod tests {

        let tx = task.state_root_message_sender();
        tx.send(MultiProofMessage::PrefetchProofs(prefetch1)).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, state_update)).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), state_update)).unwrap();
        tx.send(MultiProofMessage::PrefetchProofs(prefetch2.clone())).unwrap();

        let mut ctx = MultiproofBatchCtx::new(Instant::now());
@@ -2548,12 +2659,22 @@ mod tests {
        // First message: Prefetch1 batches; StateUpdate becomes pending.
        let first = task.rx.recv().unwrap();
        assert!(matches!(first, MultiProofMessage::PrefetchProofs(_)));
-        assert!(!task.process_multiproof_message(first, &mut ctx, &mut batch_metrics));
+        assert!(!task.process_multiproof_message(
+            first,
+            &mut ctx,
+            &mut batch_metrics,
+            &test_provider
+        ));
        let pending = ctx.pending_msg.take().expect("pending message captured");
        assert!(matches!(pending, MultiProofMessage::StateUpdate(_, _)));

        // Pending message should be handled before the next select loop.
-        assert!(!task.process_multiproof_message(pending, &mut ctx, &mut batch_metrics));
+        assert!(!task.process_multiproof_message(
+            pending,
+            &mut ctx,
+            &mut batch_metrics,
+            &test_provider
+        ));

        // Prefetch2 should now be in pending_msg (captured by StateUpdate's batching loop).
        match ctx.pending_msg.take() {
@@ -2625,12 +2746,21 @@ mod tests {
        // Queue: [Prefetch1, State1, State2, State3, Prefetch2]
        let tx = task.state_root_message_sender();
        tx.send(MultiProofMessage::PrefetchProofs(prefetch1.clone())).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, create_state_update(state_addr1, 100)))
-            .unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, create_state_update(state_addr2, 200)))
-            .unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source, create_state_update(state_addr3, 300)))
-            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(
+            source.into(),
+            create_state_update(state_addr1, 100),
+        ))
+        .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(
+            source.into(),
+            create_state_update(state_addr2, 200),
+        ))
+        .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(
+            source.into(),
+            create_state_update(state_addr3, 300),
+        ))
+        .unwrap();
        tx.send(MultiProofMessage::PrefetchProofs(prefetch2.clone())).unwrap();

        // Simulate the state-machine loop behavior
@@ -2703,4 +2833,44 @@ mod tests {
            _ => panic!("Prefetch2 was lost!"),
        }
    }
+
+    /// Verifies that BAL messages are processed correctly and generate state updates.
+    #[test]
+    fn test_bal_message_processing() {
+        let test_provider_factory = create_test_provider_factory();
+        let test_provider = test_provider_factory.latest().unwrap();
+        let mut task = create_test_state_root_task(test_provider_factory);
+
+        // Create a simple BAL with one account change
+        let account_address = Address::random();
+        let account_changes = AccountChanges {
+            address: account_address,
+            balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
+            nonce_changes: vec![],
+            code_changes: vec![],
+            storage_changes: vec![],
+            storage_reads: vec![],
+        };
+
+        let bal = Arc::new(vec![account_changes]);
+
+        let mut ctx = MultiproofBatchCtx::new(Instant::now());
+        let mut batch_metrics = MultiproofBatchMetrics::default();
+
+        let should_finish = task.process_multiproof_message(
+            MultiProofMessage::BlockAccessList(bal),
+            &mut ctx,
+            &mut batch_metrics,
+            &test_provider,
+        );
+
+        // BAL should mark updates as finished
+        assert!(ctx.updates_finished_time.is_some());
+
+        // Should have dispatched state update proofs
+        assert!(batch_metrics.state_update_proofs_requested > 0);
+
+        // Should need to wait for the results of those proofs to arrive
+        assert!(!should_finish, "Should continue waiting for proofs");
+    }
 }
--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/crates/engine/tree/src/tree/payload_validator.rs
@@ -15,6 +15,7 @@ use alloy_eip7928::BlockAccessList;
 use alloy_eips::{eip1898::BlockWithParent, NumHash};
 use alloy_evm::Evm;
 use alloy_primitives::B256;
+use rayon::prelude::*;
 use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock};
 use reth_consensus::{ConsensusError, FullConsensus};
 use reth_engine_primitives::{
@@ -221,7 +222,7 @@ where
                    .map_err(NewPayloadError::other)?
                    .into();

-                let iter = Either::Left(iter.into_iter().map(Either::Left));
+                let iter = Either::Left(iter.into_par_iter().map(Either::Left));
                let convert = move |tx| {
                    let Either::Left(tx) = tx else { unreachable!() };
                    convert(tx).map(Either::Left).map_err(Either::Left)
@@ -231,8 +232,9 @@ where
                Ok((iter, Box::new(convert) as Box<dyn Fn(_) -> _ + Send + Sync + 'static>))
            }
            BlockOrPayload::Block(block) => {
-                let iter =
-                    Either::Right(block.body().clone_transactions().into_iter().map(Either::Right));
+                let iter = Either::Right(
+                    block.body().clone_transactions().into_par_iter().map(Either::Right),
+                );
                let convert = move |tx: Either<_, N::SignedTx>| {
                    let Either::Right(tx) = tx else { unreachable!() };
                    tx.try_into_recovered().map(Either::Right).map_err(Either::Right)
@@ -400,6 +402,14 @@ where
        // use prewarming background task
        let txs = self.tx_iterator_for(&input)?;

+        // Extract the BAL, if valid and available
+        let block_access_list = ensure_ok!(input
+            .block_access_list()
+            .transpose()
+            // Eventually gets converted to a `InsertBlockErrorKind::Other`
+            .map_err(Box::<dyn std::error::Error + Send + Sync>::from))
+        .map(Arc::new);
+
        // Spawn the appropriate processor based on strategy
        let mut handle = ensure_ok!(self.spawn_payload_processor(
            env.clone(),
@@ -408,6 +418,7 @@ where
            parent_hash,
            ctx.state(),
            strategy,
+            block_access_list,
        ));

        // Use cached state provider before executing, used in execution after prewarming threads
@@ -777,6 +788,7 @@ where
        parent_hash: B256,
        state: &EngineApiTreeState<N>,
        strategy: StateRootStrategy,
+        block_access_list: Option<Arc<BlockAccessList>>,
    ) -> Result<
        PayloadHandle<
            impl ExecutableTxFor<Evm> + use<N, P, Evm, V, T>,
@@ -806,12 +818,14 @@ where
                    .record(trie_input_start.elapsed().as_secs_f64());

                let spawn_start = Instant::now();
+
                let handle = self.payload_processor.spawn(
                    env,
                    txs,
                    provider_builder,
                    multiproof_provider_factory,
                    &self.config,
+                    block_access_list,
                );

                // record prewarming initialization duration
@@ -874,7 +888,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.
    const fn plan_state_root_computation(&self) -> StateRootStrategy {
-        if self.config.state_root_fallback() || !self.config.has_enough_parallelism() {
+        if self.config.state_root_fallback() {
            StateRootStrategy::Synchronous
        } else if self.config.use_state_root_task() {
            StateRootStrategy::StateRootTask
--- a/crates/engine/util/Cargo.toml
+++ b/crates/engine/util/Cargo.toml
@@ -16,7 +16,7 @@ reth-primitives-traits.workspace = true
 reth-errors.workspace = true
 reth-chainspec.workspace = true
 reth-fs-util.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-engine-tree.workspace = true
 reth-evm.workspace = true
 reth-revm.workspace = true
--- a/crates/era-utils/src/export.rs
+++ b/crates/era-utils/src/export.rs
@@ -150,6 +150,12 @@ where
        let era1_id = Era1Id::new(&config.network, start_block, block_count as u32)
            .with_hash(historical_root);

+        let era1_id = if config.max_blocks_per_file == MAX_BLOCKS_PER_ERA1 as u64 {
+            era1_id
+        } else {
+            era1_id.with_era_count()
+        };
+
        debug!("Final file name {}", era1_id.to_file_name());
        let file_path = config.dir.join(era1_id.to_file_name());
        let file = std::fs::File::create(&file_path)?;
--- a/crates/era-utils/tests/it/genesis.rs
+++ b/crates/era-utils/tests/it/genesis.rs
@@ -24,7 +24,7 @@ fn test_export_with_genesis_only() {
    assert!(file_path.exists(), "Exported file should exist on disk");
    let file_name = file_path.file_name().unwrap().to_str().unwrap();
    assert!(
-        file_name.starts_with("mainnet-00000-00001-"),
+        file_name.starts_with("mainnet-00000-"),
        "File should have correct prefix with era format"
    );
    assert!(file_name.ends_with(".era1"), "File should have correct extension");
--- a/crates/era/src/common/file_ops.rs
+++ b/crates/era/src/common/file_ops.rs
@@ -30,8 +30,11 @@ pub trait EraFileFormat: Sized {

 /// Era file identifiers
 pub trait EraFileId: Clone {
-    /// Convert to standardized file name
-    fn to_file_name(&self) -> String;
+    /// File type for this identifier
+    const FILE_TYPE: EraFileType;
+
+    /// Number of items, slots for `era`, blocks for `era1`, per era
+    const ITEMS_PER_ERA: u64;

    /// Get the network name
    fn network_name(&self) -> &str;
@@ -41,6 +44,43 @@ pub trait EraFileId: Clone {

    /// Get the count of items
    fn count(&self) -> u32;
+
+    /// Get the optional hash identifier
+    fn hash(&self) -> Option<[u8; 4]>;
+
+    /// Whether to include era count in filename
+    fn include_era_count(&self) -> bool;
+
+    /// Calculate era number
+    fn era_number(&self) -> u64 {
+        self.start_number() / Self::ITEMS_PER_ERA
+    }
+
+    /// Calculate the number of eras spanned per file.
+    ///
+    /// If the user can decide how many slots/blocks per era file there are, we need to calculate
+    /// it. Most of the time it should be 1, but it can never be more than 2 eras per file
+    /// as there is a maximum of 8192 slots/blocks per era file.
+    fn era_count(&self) -> u64 {
+        if self.count() == 0 {
+            return 0;
+        }
+        let first_era = self.era_number();
+        let last_number = self.start_number() + self.count() as u64 - 1;
+        let last_era = last_number / Self::ITEMS_PER_ERA;
+        last_era - first_era + 1
+    }
+
+    /// Convert to standardized file name.
+    fn to_file_name(&self) -> String {
+        Self::FILE_TYPE.format_filename(
+            self.network_name(),
+            self.era_number(),
+            self.hash(),
+            self.include_era_count(),
+            self.era_count(),
+        )
+    }
 }

 /// [`StreamReader`] for reading era-format files
@@ -154,6 +194,37 @@ impl EraFileType {
        }
    }

+    /// Generate era file name.
+    ///
+    /// Standard format: `<config-name>-<era-number>-<short-historical-root>.<ext>`
+    /// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
+    ///
+    /// With era count (for custom exports):
+    /// `<config-name>-<era-number>-<era-count>-<short-historical-root>.<ext>`
+    pub fn format_filename(
+        &self,
+        network_name: &str,
+        era_number: u64,
+        hash: Option<[u8; 4]>,
+        include_era_count: bool,
+        era_count: u64,
+    ) -> String {
+        let hash = format_hash(hash);
+
+        if include_era_count {
+            format!(
+                "{}-{:05}-{:05}-{}{}",
+                network_name,
+                era_number,
+                era_count,
+                hash,
+                self.extension()
+            )
+        } else {
+            format!("{}-{:05}-{}{}", network_name, era_number, hash, self.extension())
+        }
+    }
+
    /// Detect file type from URL
    /// By default, it assumes `Era` type
    pub fn from_url(url: &str) -> Self {
@@ -164,3 +235,11 @@ impl EraFileType {
        }
    }
 }
+
+/// Format hash as hex string, or placeholder if none
+pub fn format_hash(hash: Option<[u8; 4]>) -> String {
+    match hash {
+        Some(h) => format!("{:02x}{:02x}{:02x}{:02x}", h[0], h[1], h[2], h[3]),
+        None => "00000000".to_string(),
+    }
+}
--- a/crates/era/src/era/types/group.rs
+++ b/crates/era/src/era/types/group.rs
@@ -3,7 +3,7 @@
 //! See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md>

 use crate::{
-    common::file_ops::EraFileId,
+    common::file_ops::{EraFileId, EraFileType},
    e2s::types::{Entry, IndexEntry, SLOT_INDEX},
    era::types::consensus::{CompressedBeaconState, CompressedSignedBeaconBlock},
 };
@@ -163,12 +163,22 @@ pub struct EraId {
    /// Optional hash identifier for this file
    /// First 4 bytes of the last historical root in the last state in the era file
    pub hash: Option<[u8; 4]>,
+
+    /// Whether to include era count in filename
+    /// It is used for custom exports when we don't use the max number of items per file
+    include_era_count: bool,
 }

 impl EraId {
    /// Create a new [`EraId`]
    pub fn new(network_name: impl Into<String>, start_slot: u64, slot_count: u32) -> Self {
-        Self { network_name: network_name.into(), start_slot, slot_count, hash: None }
+        Self {
+            network_name: network_name.into(),
+            start_slot,
+            slot_count,
+            hash: None,
+            include_era_count: false,
+        }
    }

    /// Add a hash identifier to  [`EraId`]
@@ -177,32 +187,18 @@ impl EraId {
        self
    }

-    /// Calculate which era number the file starts at
-    pub const fn era_number(&self) -> u64 {
-        self.start_slot / SLOTS_PER_HISTORICAL_ROOT
-    }
-
-    // Helper function to calculate the number of eras per era1 file,
-    // If the user can decide how many blocks per era1 file there are, we need to calculate it.
-    // Most of the time it should be 1, but it can never be more than 2 eras per file
-    // as there is a maximum of 8192 blocks per era1 file.
-    const fn calculate_era_count(&self) -> u64 {
-        if self.slot_count == 0 {
-            return 0;
-        }
-
-        let first_era = self.era_number();
-
-        // Calculate the actual last slot number in the range
-        let last_slot = self.start_slot + self.slot_count as u64 - 1;
-        // Find which era the last block belongs to
-        let last_era = last_slot / SLOTS_PER_HISTORICAL_ROOT;
-        // Count how many eras we span
-        last_era - first_era + 1
+    /// Include era count in filename, for custom slot-per-file exports
+    pub const fn with_era_count(mut self) -> Self {
+        self.include_era_count = true;
+        self
    }
 }

 impl EraFileId for EraId {
+    const FILE_TYPE: EraFileType = EraFileType::Era;
+
+    const ITEMS_PER_ERA: u64 = SLOTS_PER_HISTORICAL_ROOT;
+
    fn network_name(&self) -> &str {
        &self.network_name
    }
@@ -214,24 +210,13 @@ impl EraFileId for EraId {
    fn count(&self) -> u32 {
        self.slot_count
    }
-    /// Convert to file name following the era file naming:
-    /// `<config-name>-<era-number>-<era-count>-<short-historical-root>.era`
-    /// <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
-    /// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md>
-    fn to_file_name(&self) -> String {
-        let era_number = self.era_number();
-        let era_count = self.calculate_era_count();

-        if let Some(hash) = self.hash {
-            format!(
-                "{}-{:05}-{:05}-{:02x}{:02x}{:02x}{:02x}.era",
-                self.network_name, era_number, era_count, hash[0], hash[1], hash[2], hash[3]
-            )
-        } else {
-            // era spec format with placeholder hash when no hash available
-            // Format: `<config-name>-<era-number>-<era-count>-00000000.era`
-            format!("{}-{:05}-{:05}-00000000.era", self.network_name, era_number, era_count)
-        }
+    fn hash(&self) -> Option<[u8; 4]> {
+        self.hash
+    }
+
+    fn include_era_count(&self) -> bool {
+        self.include_era_count
    }
 }

@@ -399,4 +384,40 @@ mod tests {
        let parsed_offset = index.offsets[0];
        assert_eq!(parsed_offset, -1024);
    }
+
+    #[test_case::test_case(
+        EraId::new("mainnet", 0, 8192).with_hash([0x4b, 0x36, 0x3d, 0xb9]),
+        "mainnet-00000-4b363db9.era";
+        "Mainnet era 0"
+    )]
+    #[test_case::test_case(
+        EraId::new("mainnet", 8192, 8192).with_hash([0x40, 0xcf, 0x2f, 0x3c]),
+        "mainnet-00001-40cf2f3c.era";
+        "Mainnet era 1"
+    )]
+    #[test_case::test_case(
+        EraId::new("mainnet", 0, 8192),
+        "mainnet-00000-00000000.era";
+        "Without hash"
+    )]
+    fn test_era_id_file_naming(id: EraId, expected_file_name: &str) {
+        let actual_file_name = id.to_file_name();
+        assert_eq!(actual_file_name, expected_file_name);
+    }
+
+    // File naming with era-count, for custom exports
+    #[test_case::test_case(
+        EraId::new("mainnet", 0, 8192).with_hash([0x4b, 0x36, 0x3d, 0xb9]).with_era_count(),
+        "mainnet-00000-00001-4b363db9.era";
+        "Mainnet era 0 with count"
+    )]
+    #[test_case::test_case(
+        EraId::new("mainnet", 8000, 500).with_hash([0xab, 0xcd, 0xef, 0x12]).with_era_count(),
+        "mainnet-00000-00002-abcdef12.era";
+        "Spanning two eras with count"
+    )]
+    fn test_era_id_file_naming_with_era_count(id: EraId, expected_file_name: &str) {
+        let actual_file_name = id.to_file_name();
+        assert_eq!(actual_file_name, expected_file_name);
+    }
 }
--- a/crates/era/src/era1/types/group.rs
+++ b/crates/era/src/era1/types/group.rs
@@ -3,7 +3,7 @@
 //! See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md>

 use crate::{
-    common::file_ops::EraFileId,
+    common::file_ops::{EraFileId, EraFileType},
    e2s::types::{Entry, IndexEntry},
    era1::types::execution::{Accumulator, BlockTuple, MAX_BLOCKS_PER_ERA1},
 };
@@ -105,6 +105,10 @@ pub struct Era1Id {
    /// Optional hash identifier for this file
    /// First 4 bytes of the last historical root in the last state in the era file
    pub hash: Option<[u8; 4]>,
+
+    /// Whether to include era count in filename
+    /// It is used for custom exports when we don't use the max number of items per file
+    pub include_era_count: bool,
 }

 impl Era1Id {
@@ -114,7 +118,13 @@ impl Era1Id {
        start_block: BlockNumber,
        block_count: u32,
    ) -> Self {
-        Self { network_name: network_name.into(), start_block, block_count, hash: None }
+        Self {
+            network_name: network_name.into(),
+            start_block,
+            block_count,
+            hash: None,
+            include_era_count: false,
+        }
    }

    /// Add a hash identifier to  [`Era1Id`]
@@ -123,21 +133,17 @@ impl Era1Id {
        self
    }

-    // Helper function to calculate the number of eras per era1 file,
-    // If the user can decide how many blocks per era1 file there are, we need to calculate it.
-    // Most of the time it should be 1, but it can never be more than 2 eras per file
-    // as there is a maximum of 8192 blocks per era1 file.
-    const fn calculate_era_count(&self, first_era: u64) -> u64 {
-        // Calculate the actual last block number in the range
-        let last_block = self.start_block + self.block_count as u64 - 1;
-        // Find which era the last block belongs to
-        let last_era = last_block / MAX_BLOCKS_PER_ERA1 as u64;
-        // Count how many eras we span
-        last_era - first_era + 1
+    /// Include era count in filename, for custom block-per-file exports
+    pub const fn with_era_count(mut self) -> Self {
+        self.include_era_count = true;
+        self
    }
 }

 impl EraFileId for Era1Id {
+    const FILE_TYPE: EraFileType = EraFileType::Era1;
+
+    const ITEMS_PER_ERA: u64 = MAX_BLOCKS_PER_ERA1 as u64;
    fn network_name(&self) -> &str {
        &self.network_name
    }
@@ -149,24 +155,13 @@ impl EraFileId for Era1Id {
    fn count(&self) -> u32 {
        self.block_count
    }
-    /// Convert to file name following the era file naming:
-    /// `<config-name>-<era-number>-<era-count>-<short-historical-root>.era(1)`
-    /// <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
-    /// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md>
-    fn to_file_name(&self) -> String {
-        // Find which era the first block belongs to
-        let era_number = self.start_block / MAX_BLOCKS_PER_ERA1 as u64;
-        let era_count = self.calculate_era_count(era_number);
-        if let Some(hash) = self.hash {
-            format!(
-                "{}-{:05}-{:05}-{:02x}{:02x}{:02x}{:02x}.era1",
-                self.network_name, era_number, era_count, hash[0], hash[1], hash[2], hash[3]
-            )
-        } else {
-            // era spec format with placeholder hash when no hash available
-            // Format: `<config-name>-<era-number>-<era-count>-00000000.era1`
-            format!("{}-{:05}-{:05}-00000000.era1", self.network_name, era_number, era_count)
-        }
+
+    fn hash(&self) -> Option<[u8; 4]> {
+        self.hash
+    }
+
+    fn include_era_count(&self) -> bool {
+        self.include_era_count
    }
 }

@@ -314,35 +309,51 @@ mod tests {

    #[test_case::test_case(
        Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]),
-        "mainnet-00000-00001-5ec1ffb8.era1";
+        "mainnet-00000-5ec1ffb8.era1";
        "Mainnet era 0"
    )]
    #[test_case::test_case(
        Era1Id::new("mainnet", 8192, 8192).with_hash([0x5e, 0xcb, 0x9b, 0xf9]),
-        "mainnet-00001-00001-5ecb9bf9.era1";
+        "mainnet-00001-5ecb9bf9.era1";
        "Mainnet era 1"
    )]
    #[test_case::test_case(
        Era1Id::new("sepolia", 0, 8192).with_hash([0x90, 0x91, 0x84, 0x72]),
-        "sepolia-00000-00001-90918472.era1";
+        "sepolia-00000-90918472.era1";
        "Sepolia era 0"
    )]
    #[test_case::test_case(
        Era1Id::new("sepolia", 155648, 8192).with_hash([0xfa, 0x77, 0x00, 0x19]),
-        "sepolia-00019-00001-fa770019.era1";
+        "sepolia-00019-fa770019.era1";
        "Sepolia era 19"
    )]
    #[test_case::test_case(
        Era1Id::new("mainnet", 1000, 100),
-        "mainnet-00000-00001-00000000.era1";
+        "mainnet-00000-00000000.era1";
        "ID without hash"
    )]
    #[test_case::test_case(
        Era1Id::new("sepolia", 101130240, 8192).with_hash([0xab, 0xcd, 0xef, 0x12]),
-        "sepolia-12345-00001-abcdef12.era1";
+        "sepolia-12345-abcdef12.era1";
        "Large block number era 12345"
    )]
-    fn test_era1id_file_naming(id: Era1Id, expected_file_name: &str) {
+    fn test_era1_id_file_naming(id: Era1Id, expected_file_name: &str) {
+        let actual_file_name = id.to_file_name();
+        assert_eq!(actual_file_name, expected_file_name);
+    }
+
+    // File naming with era-count, for custom exports
+    #[test_case::test_case(
+        Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]).with_era_count(),
+        "mainnet-00000-00001-5ec1ffb8.era1";
+        "Mainnet era 0 with count"
+    )]
+    #[test_case::test_case(
+        Era1Id::new("mainnet", 8000, 500).with_hash([0xab, 0xcd, 0xef, 0x12]).with_era_count(),
+        "mainnet-00000-00002-abcdef12.era1";
+        "Spanning two eras with count"
+    )]
+    fn test_era1_id_file_naming_with_era_count(id: Era1Id, expected_file_name: &str) {
        let actual_file_name = id.to_file_name();
        assert_eq!(actual_file_name, expected_file_name);
    }
--- a/crates/ethereum/cli/src/app.rs
+++ b/crates/ethereum/cli/src/app.rs
@@ -154,7 +154,9 @@ where
        Commands::ImportEra(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
        Commands::ExportEra(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
        Commands::DumpGenesis(command) => runner.run_blocking_until_ctrl_c(command.execute()),
-        Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
+        Commands::Db(command) => {
+            runner.run_blocking_command_until_exit(|ctx| command.execute::<N>(ctx))
+        }
        Commands::Download(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
        Commands::Stage(command) => {
            runner.run_command_until_exit(|ctx| command.execute::<N, _>(ctx, components))
--- a/crates/ethereum/evm/src/lib.rs
+++ b/crates/ethereum/evm/src/lib.rs
@@ -19,32 +19,37 @@ extern crate alloc;

 use alloc::{borrow::Cow, sync::Arc};
 use alloy_consensus::Header;
-use alloy_eips::Decodable2718;
-pub use alloy_evm::EthEvm;
 use alloy_evm::{
    eth::{EthBlockExecutionCtx, EthBlockExecutorFactory},
    EthEvmFactory, FromRecoveredTx, FromTxWithEncoded,
 };
-use alloy_primitives::{Bytes, U256};
-use alloy_rpc_types_engine::ExecutionData;
 use core::{convert::Infallible, fmt::Debug};
-use reth_chainspec::{ChainSpec, EthChainSpec, EthereumHardforks, MAINNET};
+use reth_chainspec::{ChainSpec, EthChainSpec, MAINNET};
 use reth_ethereum_primitives::{Block, EthPrimitives, TransactionSigned};
 use reth_evm::{
-    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEngineEvm, ConfigureEvm,
-    EvmEnv, EvmEnvFor, EvmFactory, ExecutableTxIterator, ExecutionCtxFor, NextBlockEnvAttributes,
-    TransactionEnv,
+    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEvm, EvmEnv, EvmFactory,
+    NextBlockEnvAttributes, TransactionEnv,
 };
-use reth_primitives_traits::{
-    constants::MAX_TX_GAS_LIMIT_OSAKA, SealedBlock, SealedHeader, SignedTransaction, TxTy,
-};
-use reth_storage_errors::any::AnyError;
-use revm::{
-    context::{BlockEnv, CfgEnv},
-    context_interface::block::BlobExcessGasAndPrice,
-    primitives::hardfork::SpecId,
+use reth_primitives_traits::{SealedBlock, SealedHeader};
+use revm::{context::BlockEnv, primitives::hardfork::SpecId};
+
+#[cfg(feature = "std")]
+use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator};
+#[allow(unused_imports)]
+use {
+    alloy_eips::Decodable2718,
+    alloy_primitives::{Bytes, U256},
+    alloy_rpc_types_engine::ExecutionData,
+    reth_chainspec::EthereumHardforks,
+    reth_evm::{EvmEnvFor, ExecutionCtxFor},
+    reth_primitives_traits::{constants::MAX_TX_GAS_LIMIT_OSAKA, SignedTransaction, TxTy},
+    reth_storage_errors::any::AnyError,
+    revm::context::CfgEnv,
+    revm::context_interface::block::BlobExcessGasAndPrice,
 };

+pub use alloy_evm::EthEvm;
+
 mod config;
 use alloy_evm::eth::spec::EthExecutorSpec;
 pub use config::{revm_spec, revm_spec_by_timestamp_and_block_number};
@@ -206,6 +211,7 @@ where
    }
 }

+#[cfg(feature = "std")]
 impl<ChainSpec, EvmF> ConfigureEngineEvm<ExecutionData> for EthEvmConfig<ChainSpec, EvmF>
 where
    ChainSpec: EthExecutorSpec + EthChainSpec<Header = Header> + Hardforks + 'static,
@@ -286,7 +292,7 @@ where
        &self,
        payload: &ExecutionData,
    ) -> Result<impl ExecutableTxIterator<Self>, Self::Error> {
-        let txs = payload.payload.transactions().clone().into_iter();
+        let txs = payload.payload.transactions().clone();
        let convert = |tx: Bytes| {
            let tx =
                TxTy::<Self::Primitives>::decode_2718_exact(tx.as_ref()).map_err(AnyError::new)?;
--- a/crates/ethereum/node/Cargo.toml
+++ b/crates/ethereum/node/Cargo.toml
@@ -24,7 +24,7 @@ reth-provider.workspace = true
 reth-transaction-pool.workspace = true
 reth-network.workspace = true
 reth-evm.workspace = true
-reth-evm-ethereum.workspace = true
+reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-rpc.workspace = true
 reth-rpc-api.workspace = true
 reth-rpc-eth-api.workspace = true
@@ -35,7 +35,7 @@ reth-chainspec.workspace = true
 reth-revm = { workspace = true, features = ["std"] }
 reth-rpc-eth-types.workspace = true
 reth-engine-local.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-payload-primitives.workspace = true

 # ethereum
@@ -88,6 +88,9 @@ asm-keccak = [
    "reth-node-core/asm-keccak",
    "revm/asm-keccak",
 ]
+keccak-cache-global = [
+    "alloy-primitives/keccak-cache-global",
+]
 js-tracer = [
    "reth-node-builder/js-tracer",
    "reth-rpc/js-tracer",
--- a/crates/ethereum/node/src/node.rs
+++ b/crates/ethereum/node/src/node.rs
@@ -118,13 +118,14 @@ impl EthereumNode {
    /// use reth_chainspec::ChainSpecBuilder;
    /// use reth_db::open_db_read_only;
    /// use reth_node_ethereum::EthereumNode;
-    /// use reth_provider::providers::StaticFileProvider;
+    /// use reth_provider::providers::{RocksDBProvider, StaticFileProvider};
    /// use std::sync::Arc;
    ///
    /// let factory = EthereumNode::provider_factory_builder()
    ///     .db(Arc::new(open_db_read_only("db", Default::default()).unwrap()))
    ///     .chainspec(ChainSpecBuilder::mainnet().build().into())
    ///     .static_file(StaticFileProvider::read_only("db/static_files", false).unwrap())
+    ///     .rocksdb_provider(RocksDBProvider::builder("db/rocksdb").build().unwrap())
    ///     .build_provider_factory();
    /// ```
    pub fn provider_factory_builder() -> ProviderFactoryBuilder<Self> {
--- a/crates/ethereum/node/tests/it/testing.rs
+++ b/crates/ethereum/node/tests/it/testing.rs
@@ -28,6 +28,7 @@ async fn testing_rpc_build_block_works() -> eyre::Result<()> {
        datadir: MaybePlatformPath::<DataDirPath>::from_str(tempdir.path().to_str().unwrap())
            .expect("valid datadir"),
        static_files_path: Some(tempdir.path().join("static")),
+        rocksdb_path: Some(tempdir.path().join("rocksdb")),
    };
    let config = NodeConfig::test().with_datadir_args(datadir_args).with_rpc(rpc_args);
    let db = create_test_rw_db();
--- a/crates/ethereum/payload/Cargo.toml
+++ b/crates/ethereum/payload/Cargo.toml
@@ -24,7 +24,7 @@ reth-payload-builder-primitives.workspace = true
 reth-payload-primitives.workspace = true
 reth-basic-payload-builder.workspace = true
 reth-evm.workspace = true
-reth-evm-ethereum.workspace = true
+reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-errors.workspace = true
 reth-chainspec.workspace = true
 reth-payload-validator.workspace = true
--- a/crates/ethereum/reth/Cargo.toml
+++ b/crates/ethereum/reth/Cargo.toml
@@ -79,7 +79,9 @@ arbitrary = [
    "alloy-rpc-types-engine?/arbitrary",
    "reth-codecs?/arbitrary",
 ]
-
+keccak-cache-global = [
+    "reth-node-ethereum?/keccak-cache-global",
+]
 test-utils = [
    "reth-chainspec/test-utils",
    "reth-consensus?/test-utils",
--- a/crates/evm/evm/Cargo.toml
+++ b/crates/evm/evm/Cargo.toml
@@ -32,6 +32,7 @@ auto_impl.workspace = true
 derive_more.workspace = true
 futures-util.workspace = true
 metrics = { workspace = true, optional = true }
+rayon = { workspace = true, optional = true }

 [dev-dependencies]
 reth-ethereum-primitives.workspace = true
@@ -40,6 +41,7 @@ reth-ethereum-forks.workspace = true
 [features]
 default = ["std"]
 std = [
+    "dep:rayon",
    "reth-primitives-traits/std",
    "alloy-eips/std",
    "alloy-primitives/std",
--- a/crates/evm/evm/src/engine.rs
+++ b/crates/evm/evm/src/engine.rs
@@ -1,4 +1,5 @@
 use crate::{execute::ExecutableTxFor, ConfigureEvm, EvmEnvFor, ExecutionCtxFor};
+use rayon::prelude::*;

 /// [`ConfigureEvm`] extension providing methods for executing payloads.
 pub trait ConfigureEngineEvm<ExecutionData>: ConfigureEvm {
@@ -21,7 +22,7 @@ pub trait ConfigureEngineEvm<ExecutionData>: ConfigureEvm {
 /// A helper trait representing a pair of a "raw" transactions iterator and a closure that can be
 /// used to convert them to an executable transaction. This tuple is used in the engine to
 /// parallelize heavy work like decoding or recovery.
-pub trait ExecutableTxTuple: Into<(Self::Iter, Self::Convert)> + Send + 'static {
+pub trait ExecutableTxTuple: Into<(Self::IntoIter, Self::Convert)> + Send + 'static {
    /// Raw transaction that can be converted to an [`ExecutableTxTuple::Tx`]
    ///
    /// This can be any type that can be converted to an [`ExecutableTxTuple::Tx`]. For example,
@@ -32,8 +33,10 @@ pub trait ExecutableTxTuple: Into<(Self::Iter, Self::Convert)> + Send + 'static
    /// Errors that may occur while recovering or decoding transactions.
    type Error: core::error::Error + Send + Sync + 'static;

-    /// Iterator over [`ExecutableTxTuple::Tx`]
-    type Iter: Iterator<Item = Self::RawTx> + Send + 'static;
+    /// Iterator over [`ExecutableTxTuple::Tx`].
+    type IntoIter: IntoParallelIterator<Item = Self::RawTx, Iter: IndexedParallelIterator>
+        + Send
+        + 'static;
    /// Closure that can be used to convert a [`ExecutableTxTuple::RawTx`] to a
    /// [`ExecutableTxTuple::Tx`]. This might involve heavy work like decoding or recovery
    /// and will be parallelized in the engine.
@@ -45,14 +48,14 @@ where
    RawTx: Send + Sync + 'static,
    Tx: Clone + Send + Sync + 'static,
    Err: core::error::Error + Send + Sync + 'static,
-    I: Iterator<Item = RawTx> + Send + 'static,
+    I: IntoParallelIterator<Item = RawTx, Iter: IndexedParallelIterator> + Send + 'static,
    F: Fn(RawTx) -> Result<Tx, Err> + Send + Sync + 'static,
 {
    type RawTx = RawTx;
    type Tx = Tx;
    type Error = Err;

-    type Iter = I;
+    type IntoIter = I;
    type Convert = F;
 }

--- a/crates/evm/evm/src/lib.rs
+++ b/crates/evm/evm/src/lib.rs
@@ -44,8 +44,10 @@ pub mod execute;
 mod aliases;
 pub use aliases::*;

+#[cfg(feature = "std")]
 mod engine;
-pub use engine::{ConfigureEngineEvm, ExecutableTxIterator};
+#[cfg(feature = "std")]
+pub use engine::{ConfigureEngineEvm, ExecutableTxIterator, ExecutableTxTuple};

 #[cfg(feature = "metrics")]
 pub mod metrics;
--- a/crates/exex/test-utils/src/lib.rs
+++ b/crates/exex/test-utils/src/lib.rs
@@ -20,7 +20,9 @@ use futures_util::FutureExt;
 use reth_chainspec::{ChainSpec, MAINNET};
 use reth_consensus::test_utils::TestConsensus;
 use reth_db::{
-    test_utils::{create_test_rw_db, create_test_static_files_dir, TempDatabase},
+    test_utils::{
+        create_test_rocksdb_dir, create_test_rw_db, create_test_static_files_dir, TempDatabase,
+    },
    DatabaseEnv,
 };
 use reth_db_common::init::init_genesis;
@@ -50,7 +52,7 @@ use reth_node_ethereum::{
 use reth_payload_builder::noop::NoopPayloadBuilderService;
 use reth_primitives_traits::{Block as _, RecoveredBlock};
 use reth_provider::{
-    providers::{BlockchainProvider, StaticFileProvider},
+    providers::{BlockchainProvider, RocksDBProvider, StaticFileProvider},
    BlockReader, EthStorage, ProviderFactory,
 };
 use reth_tasks::TaskManager;
@@ -239,11 +241,13 @@ pub async fn test_exex_context_with_chain_spec(
    let consensus = Arc::new(TestConsensus::default());

    let (static_dir, _) = create_test_static_files_dir();
+    let (rocksdb_dir, _) = create_test_rocksdb_dir();
    let db = create_test_rw_db();
    let provider_factory = ProviderFactory::<NodeTypesWithDBAdapter<TestNode, _>>::new(
        db,
        chain_spec.clone(),
        StaticFileProvider::read_write(static_dir.keep()).expect("static file provider"),
+        RocksDBProvider::builder(rocksdb_dir.keep()).build().unwrap(),
    )?;

    let genesis_hash = init_genesis(&provider_factory)?;
--- a/crates/net/discv4/src/config.rs
+++ b/crates/net/discv4/src/config.rs
@@ -24,7 +24,7 @@ pub struct Discv4Config {
    /// The number of allowed consecutive failures for `FindNode` requests. Default: 5.
    pub max_find_node_failures: u8,
    /// The interval to use when checking for expired nodes that need to be re-pinged. Default:
-    /// 10min.
+    /// 10 seconds.
    pub ping_interval: Duration,
    /// The duration of we consider a ping timed out.
    pub ping_expiration: Duration,
@@ -93,7 +93,7 @@ impl Discv4Config {
    /// Returns the corresponding [`ResolveNatInterval`], if a [`NatResolver`] and an interval was
    /// configured
    pub fn resolve_external_ip_interval(&self) -> Option<ResolveNatInterval> {
-        let resolver = self.external_ip_resolver?;
+        let resolver = self.external_ip_resolver.clone()?;
        let interval = self.resolve_external_ip_interval?;
        Some(ResolveNatInterval::interval_at(resolver, tokio::time::Instant::now(), interval))
    }
@@ -275,10 +275,7 @@ impl Discv4ConfigBuilder {
    }

    /// Configures if and how the external IP of the node should be resolved.
-    pub const fn external_ip_resolver(
-        &mut self,
-        external_ip_resolver: Option<NatResolver>,
-    ) -> &mut Self {
+    pub fn external_ip_resolver(&mut self, external_ip_resolver: Option<NatResolver>) -> &mut Self {
        self.config.external_ip_resolver = external_ip_resolver;
        self
    }
--- a/crates/net/discv4/src/lib.rs
+++ b/crates/net/discv4/src/lib.rs
@@ -625,10 +625,13 @@ impl Discv4Service {
        self.lookup_interval = tokio::time::interval(duration);
    }

-    /// Sets the external Ip to the configured external IP if [`NatResolver::ExternalIp`].
+    /// Sets the external Ip to the configured external IP if [`NatResolver::ExternalIp`] or
+    /// [`NatResolver::ExternalAddr`]. In the case of [`NatResolver::ExternalAddr`], it will return
+    /// the first IP address found for the domain associated with the discv4 UDP port.
    fn resolve_external_ip(&mut self) {
        if let Some(r) = &self.resolve_external_ip_interval &&
-            let Some(external_ip) = r.resolver().as_external_ip()
+            let Some(external_ip) =
+                r.resolver().clone().as_external_ip(self.local_node_record.udp_port)
        {
            self.set_external_ip_addr(external_ip);
        }
--- a/crates/net/downloaders/src/headers/reverse_headers.rs
+++ b/crates/net/downloaders/src/headers/reverse_headers.rs
@@ -1218,7 +1218,9 @@ impl ReverseHeadersDownloaderBuilder {
            next_request_block_number: 0,
            next_chain_tip_block_number: 0,
            lowest_validated_header: None,
-            request_limit,
+            // TODO(mattsse): tmp hotfix to prevent issues with syncing from besu which has an upper
+            // limit of 512
+            request_limit: request_limit.min(512),
            min_concurrent_requests,
            max_concurrent_requests,
            stream_batch_size,
--- a/crates/net/eth-wire-types/src/broadcast.rs
+++ b/crates/net/eth-wire-types/src/broadcast.rs
@@ -169,7 +169,7 @@ impl NewPooledTransactionHashes {
                matches!(version, EthVersion::Eth67 | EthVersion::Eth66)
            }
            Self::Eth68(_) => {
-                matches!(version, EthVersion::Eth68 | EthVersion::Eth69)
+                matches!(version, EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70)
            }
        }
    }
--- a/crates/net/eth-wire-types/src/capability.rs
+++ b/crates/net/eth-wire-types/src/capability.rs
@@ -100,6 +100,16 @@ impl Capability {
        Self::eth(EthVersion::Eth68)
    }

+    /// Returns the [`EthVersion::Eth69`] capability.
+    pub const fn eth_69() -> Self {
+        Self::eth(EthVersion::Eth69)
+    }
+
+    /// Returns the [`EthVersion::Eth70`] capability.
+    pub const fn eth_70() -> Self {
+        Self::eth(EthVersion::Eth70)
+    }
+
    /// Whether this is eth v66 protocol.
    #[inline]
    pub fn is_eth_v66(&self) -> bool {
@@ -118,10 +128,26 @@ impl Capability {
        self.name == "eth" && self.version == 68
    }

+    /// Whether this is eth v69.
+    #[inline]
+    pub fn is_eth_v69(&self) -> bool {
+        self.name == "eth" && self.version == 69
+    }
+
+    /// Whether this is eth v70.
+    #[inline]
+    pub fn is_eth_v70(&self) -> bool {
+        self.name == "eth" && self.version == 70
+    }
+
    /// Whether this is any eth version.
    #[inline]
    pub fn is_eth(&self) -> bool {
-        self.is_eth_v66() || self.is_eth_v67() || self.is_eth_v68()
+        self.is_eth_v66() ||
+            self.is_eth_v67() ||
+            self.is_eth_v68() ||
+            self.is_eth_v69() ||
+            self.is_eth_v70()
    }
 }

@@ -141,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..=69)?; // Valid eth protocol versions are 66-69
+        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))
    }
@@ -155,6 +181,8 @@ pub struct Capabilities {
    eth_66: bool,
    eth_67: bool,
    eth_68: bool,
+    eth_69: bool,
+    eth_70: bool,
 }

 impl Capabilities {
@@ -164,6 +192,8 @@ impl Capabilities {
            eth_66: value.iter().any(Capability::is_eth_v66),
            eth_67: value.iter().any(Capability::is_eth_v67),
            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),
            inner: value,
        }
    }
@@ -182,7 +212,7 @@ impl Capabilities {
    /// Whether the peer supports `eth` sub-protocol.
    #[inline]
    pub const fn supports_eth(&self) -> bool {
-        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.
@@ -202,6 +232,18 @@ impl Capabilities {
    pub const fn supports_eth_v68(&self) -> bool {
        self.eth_68
    }
+
+    /// Whether this peer supports eth v69 protocol.
+    #[inline]
+    pub const fn supports_eth_v69(&self) -> bool {
+        self.eth_69
+    }
+
+    /// Whether this peer supports eth v70 protocol.
+    #[inline]
+    pub const fn supports_eth_v70(&self) -> bool {
+        self.eth_70
+    }
 }

 impl From<Vec<Capability>> for Capabilities {
@@ -224,6 +266,8 @@ impl Decodable for Capabilities {
            eth_66: inner.iter().any(Capability::is_eth_v66),
            eth_67: inner.iter().any(Capability::is_eth_v67),
            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),
            inner,
        })
    }
--- a/crates/net/eth-wire-types/src/message.rs
+++ b/crates/net/eth-wire-types/src/message.rs
@@ -1,4 +1,4 @@
-//! Implements Ethereum wire protocol for versions 66, 67, and 68.
+//! Implements Ethereum wire protocol for versions 66 through 70.
 //! Defines structs/enums for messages, request-response pairs, and broadcasts.
 //! Handles compatibility with [`EthVersion`].
 //!
@@ -8,13 +8,13 @@

 use super::{
    broadcast::NewBlockHashes, BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders,
-    GetNodeData, GetPooledTransactions, GetReceipts, NewPooledTransactionHashes66,
+    GetNodeData, GetPooledTransactions, GetReceipts, GetReceipts70, NewPooledTransactionHashes66,
    NewPooledTransactionHashes68, NodeData, PooledTransactions, Receipts, Status, StatusEth69,
    Transactions,
 };
 use crate::{
    status::StatusMessage, BlockRangeUpdate, EthNetworkPrimitives, EthVersion, NetworkPrimitives,
-    RawCapabilityMessage, Receipts69, SharedTransactions,
+    RawCapabilityMessage, Receipts69, Receipts70, SharedTransactions,
 };
 use alloc::{boxed::Box, string::String, sync::Arc};
 use alloy_primitives::{
@@ -111,13 +111,29 @@ impl<N: NetworkPrimitives> ProtocolMessage<N> {
                }
                EthMessage::NodeData(RequestPair::decode(buf)?)
            }
-            EthMessageID::GetReceipts => EthMessage::GetReceipts(RequestPair::decode(buf)?),
-            EthMessageID::Receipts => {
-                if version < EthVersion::Eth69 {
-                    EthMessage::Receipts(RequestPair::decode(buf)?)
+            EthMessageID::GetReceipts => {
+                if version >= EthVersion::Eth70 {
+                    EthMessage::GetReceipts70(RequestPair::decode(buf)?)
                } else {
-                    // with eth69, receipts no longer include the bloom
-                    EthMessage::Receipts69(RequestPair::decode(buf)?)
+                    EthMessage::GetReceipts(RequestPair::decode(buf)?)
+                }
+            }
+            EthMessageID::Receipts => {
+                match version {
+                    v if v >= EthVersion::Eth70 => {
+                        // eth/70 continues to omit bloom filters and adds the
+                        // `lastBlockIncomplete` flag, encoded as
+                        // `[request-id, lastBlockIncomplete, [[receipt₁, receipt₂], ...]]`.
+                        EthMessage::Receipts70(RequestPair::decode(buf)?)
+                    }
+                    EthVersion::Eth69 => {
+                        // with eth69, receipts no longer include the bloom
+                        EthMessage::Receipts69(RequestPair::decode(buf)?)
+                    }
+                    _ => {
+                        // before eth69 we need to decode the bloom  as well
+                        EthMessage::Receipts(RequestPair::decode(buf)?)
+                    }
                }
            }
            EthMessageID::BlockRangeUpdate => {
@@ -205,6 +221,9 @@ impl<N: NetworkPrimitives> From<EthBroadcastMessage<N>> for ProtocolBroadcastMes
 ///
 /// The `eth/69` announces the historical block range served by the node. Removes total difficulty
 /// information. And removes the Bloom field from receipts transferred over the protocol.
+///
+/// The `eth/70` (EIP-7975) keeps the eth/69 status format and introduces partial receipts.
+/// requests/responses.
 #[derive(Clone, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
@@ -259,6 +278,12 @@ pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
    NodeData(RequestPair<NodeData>),
    /// Represents a `GetReceipts` request-response pair.
    GetReceipts(RequestPair<GetReceipts>),
+    /// Represents a `GetReceipts` request for eth/70.
+    ///
+    /// Note: Unlike earlier protocol versions, the eth/70 encoding for
+    /// `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 Receipts request-response pair.
    #[cfg_attr(
        feature = "serde",
@@ -271,6 +296,16 @@ pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
        serde(bound = "N::Receipt: serde::Serialize + serde::de::DeserializeOwned")
    )]
    Receipts69(RequestPair<Receipts69<N::Receipt>>),
+    /// Represents a Receipts request-response pair for eth/70.
+    #[cfg_attr(
+        feature = "serde",
+        serde(bound = "N::Receipt: serde::Serialize + serde::de::DeserializeOwned")
+    )]
+    ///
+    /// Note: The eth/70 encoding for `Receipts` in EIP-7975 inlines the
+    /// request id. The type still wraps a [`RequestPair`], but with a custom
+    /// inline encoding.
+    Receipts70(RequestPair<Receipts70<N::Receipt>>),
    /// Represents a `BlockRangeUpdate` message broadcast to the network.
    #[cfg_attr(
        feature = "serde",
@@ -300,8 +335,8 @@ impl<N: NetworkPrimitives> EthMessage<N> {
            Self::PooledTransactions(_) => EthMessageID::PooledTransactions,
            Self::GetNodeData(_) => EthMessageID::GetNodeData,
            Self::NodeData(_) => EthMessageID::NodeData,
-            Self::GetReceipts(_) => EthMessageID::GetReceipts,
-            Self::Receipts(_) | Self::Receipts69(_) => EthMessageID::Receipts,
+            Self::GetReceipts(_) | Self::GetReceipts70(_) => EthMessageID::GetReceipts,
+            Self::Receipts(_) | Self::Receipts69(_) | Self::Receipts70(_) => EthMessageID::Receipts,
            Self::BlockRangeUpdate(_) => EthMessageID::BlockRangeUpdate,
            Self::Other(msg) => EthMessageID::Other(msg.id as u8),
        }
@@ -314,6 +349,7 @@ impl<N: NetworkPrimitives> EthMessage<N> {
            Self::GetBlockBodies(_) |
                Self::GetBlockHeaders(_) |
                Self::GetReceipts(_) |
+                Self::GetReceipts70(_) |
                Self::GetPooledTransactions(_) |
                Self::GetNodeData(_)
        )
@@ -326,11 +362,40 @@ impl<N: NetworkPrimitives> EthMessage<N> {
            Self::PooledTransactions(_) |
                Self::Receipts(_) |
                Self::Receipts69(_) |
+                Self::Receipts70(_) |
                Self::BlockHeaders(_) |
                Self::BlockBodies(_) |
                Self::NodeData(_)
        )
    }
+
+    /// Converts the message types where applicable.
+    ///
+    /// This handles up/downcasting where appropriate, for example for different receipt request
+    /// types.
+    pub fn map_versioned(self, version: EthVersion) -> Self {
+        // For eth/70 peers we send `GetReceipts` using the new eth/70
+        // encoding with `firstBlockReceiptIndex = 0`, while keeping the
+        // user-facing `PeerRequest` API unchanged.
+        if version >= EthVersion::Eth70 {
+            return match self {
+                Self::GetReceipts(pair) => {
+                    let RequestPair { request_id, message } = pair;
+                    let req = RequestPair {
+                        request_id,
+                        message: GetReceipts70 {
+                            first_block_receipt_index: 0,
+                            block_hashes: message.0,
+                        },
+                    };
+                    Self::GetReceipts70(req)
+                }
+                other => other,
+            }
+        }
+
+        self
+    }
 }

 impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
@@ -351,8 +416,10 @@ impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
            Self::GetNodeData(request) => request.encode(out),
            Self::NodeData(data) => data.encode(out),
            Self::GetReceipts(request) => request.encode(out),
+            Self::GetReceipts70(request) => request.encode(out),
            Self::Receipts(receipts) => receipts.encode(out),
            Self::Receipts69(receipt69) => receipt69.encode(out),
+            Self::Receipts70(receipt70) => receipt70.encode(out),
            Self::BlockRangeUpdate(block_range_update) => block_range_update.encode(out),
            Self::Other(unknown) => out.put_slice(&unknown.payload),
        }
@@ -374,8 +441,10 @@ impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
            Self::GetNodeData(request) => request.length(),
            Self::NodeData(data) => data.length(),
            Self::GetReceipts(request) => request.length(),
+            Self::GetReceipts70(request) => request.length(),
            Self::Receipts(receipts) => receipts.length(),
            Self::Receipts69(receipt69) => receipt69.length(),
+            Self::Receipts70(receipt70) => receipt70.length(),
            Self::BlockRangeUpdate(block_range_update) => block_range_update.length(),
            Self::Other(unknown) => unknown.length(),
        }
--- a/crates/net/eth-wire-types/src/receipts.rs
+++ b/crates/net/eth-wire-types/src/receipts.rs
@@ -17,6 +17,42 @@ pub struct GetReceipts(
    pub Vec<B256>,
 );

+/// Eth/70 `GetReceipts` request payload that supports partial receipt queries.
+///
+/// When used with eth/70, the request id is carried by the surrounding
+/// [`crate::message::RequestPair`], and the on-wire shape is the flattened list
+/// `firstBlockReceiptIndex, [blockhash₁, ...]`.
+///
+/// See also [eip-7975](https://eips.ethereum.org/EIPS/eip-7975)
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
+pub struct GetReceipts70 {
+    /// Index into the receipts of the first requested block hash.
+    pub first_block_receipt_index: u64,
+    /// The block hashes to request receipts for.
+    pub block_hashes: Vec<B256>,
+}
+
+impl alloy_rlp::Encodable for GetReceipts70 {
+    fn encode(&self, out: &mut dyn alloy_rlp::BufMut) {
+        self.first_block_receipt_index.encode(out);
+        self.block_hashes.encode(out);
+    }
+
+    fn length(&self) -> usize {
+        self.first_block_receipt_index.length() + self.block_hashes.length()
+    }
+}
+
+impl alloy_rlp::Decodable for GetReceipts70 {
+    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
+        let first_block_receipt_index = u64::decode(buf)?;
+        let block_hashes = Vec::<B256>::decode(buf)?;
+        Ok(Self { first_block_receipt_index, block_hashes })
+    }
+}
+
 /// The response to [`GetReceipts`], containing receipt lists that correspond to each block
 /// requested.
 #[derive(Clone, Debug, PartialEq, Eq, Default)]
@@ -58,7 +94,13 @@ pub struct Receipts69<T = Receipt>(pub Vec<Vec<T>>);
 impl<T: TxReceipt> Receipts69<T> {
    /// Encodes all receipts with the bloom filter.
    ///
-    /// Note: This is an expensive operation that recalculates the bloom for each receipt.
+    /// Eth/69 omits bloom filters on the wire, while some internal callers
+    /// (and legacy APIs) still operate on [`Receipts`] with
+    /// [`ReceiptWithBloom`]. This helper reconstructs the bloom locally from
+    /// each receipt's logs so the older API can be used on top of eth/69 data.
+    ///
+    /// Note: This is an expensive operation that recalculates the bloom for
+    /// every receipt.
    pub fn into_with_bloom(self) -> Receipts<T> {
        Receipts(
            self.0
@@ -75,6 +117,68 @@ impl<T: TxReceipt> From<Receipts69<T>> for Receipts<T> {
    }
 }

+/// Eth/70 `Receipts` response payload.
+///
+/// This is used in conjunction with [`crate::message::RequestPair`] to encode the full wire
+/// message `[request-id, lastBlockIncomplete, [[receipt₁, receipt₂], ...]]`.
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
+pub struct Receipts70<T = Receipt> {
+    /// Whether the receipts list for the last block is incomplete.
+    pub last_block_incomplete: bool,
+    /// Receipts grouped by block.
+    pub receipts: Vec<Vec<T>>,
+}
+
+impl<T> alloy_rlp::Encodable for Receipts70<T>
+where
+    T: alloy_rlp::Encodable,
+{
+    fn encode(&self, out: &mut dyn alloy_rlp::BufMut) {
+        self.last_block_incomplete.encode(out);
+        self.receipts.encode(out);
+    }
+
+    fn length(&self) -> usize {
+        self.last_block_incomplete.length() + self.receipts.length()
+    }
+}
+
+impl<T> alloy_rlp::Decodable for Receipts70<T>
+where
+    T: alloy_rlp::Decodable,
+{
+    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
+        let last_block_incomplete = bool::decode(buf)?;
+        let receipts = Vec::<Vec<T>>::decode(buf)?;
+        Ok(Self { last_block_incomplete, receipts })
+    }
+}
+
+impl<T: TxReceipt> Receipts70<T> {
+    /// Encodes all receipts with the bloom filter.
+    ///
+    /// Just like eth/69, eth/70 does not transmit bloom filters over the wire.
+    /// When higher layers still expect the older bloom-bearing [`Receipts`]
+    /// type, this helper converts the eth/70 payload into that shape by
+    /// recomputing the bloom locally from the contained receipts.
+    ///
+    /// Note: This is an expensive operation that recalculates the bloom for
+    /// every receipt.
+    pub fn into_with_bloom(self) -> Receipts<T> {
+        // Reuse the eth/69 helper, since both variants carry the same
+        // receipt list shape (only eth/70 adds request metadata).
+        Receipts69(self.receipts).into_with_bloom()
+    }
+}
+
+impl<T: TxReceipt> From<Receipts70<T>> for Receipts<T> {
+    fn from(receipts: Receipts70<T>) -> Self {
+        receipts.into_with_bloom()
+    }
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -225,4 +329,70 @@ mod tests {
        let encoded = alloy_rlp::encode(&request);
        assert_eq!(encoded, data);
    }
+
+    #[test]
+    fn encode_get_receipts70_inline_shape() {
+        let req = RequestPair {
+            request_id: 1111,
+            message: GetReceipts70 {
+                first_block_receipt_index: 0,
+                block_hashes: vec![
+                    hex!("00000000000000000000000000000000000000000000000000000000deadc0de").into(),
+                    hex!("00000000000000000000000000000000000000000000000000000000feedbeef").into(),
+                ],
+            },
+        };
+
+        let mut out = vec![];
+        req.encode(&mut out);
+
+        let mut buf = out.as_slice();
+        let header = alloy_rlp::Header::decode(&mut buf).unwrap();
+        let payload_start = buf.len();
+        let request_id = u64::decode(&mut buf).unwrap();
+        let first_block_receipt_index = u64::decode(&mut buf).unwrap();
+        let block_hashes = Vec::<B256>::decode(&mut buf).unwrap();
+
+        assert!(buf.is_empty(), "buffer not fully consumed");
+        assert_eq!(request_id, 1111);
+        assert_eq!(first_block_receipt_index, 0);
+        assert_eq!(block_hashes.len(), 2);
+        // ensure payload length matches header
+        assert_eq!(payload_start - buf.len(), header.payload_length);
+
+        let mut buf = out.as_slice();
+        let decoded = RequestPair::<GetReceipts70>::decode(&mut buf).unwrap();
+        assert!(buf.is_empty(), "buffer not fully consumed on decode");
+        assert_eq!(decoded, req);
+    }
+
+    #[test]
+    fn encode_receipts70_inline_shape() {
+        let payload: Receipts70<Receipt> =
+            Receipts70 { last_block_incomplete: true, receipts: vec![vec![Receipt::default()]] };
+
+        let resp = RequestPair { request_id: 7, message: payload };
+
+        let mut out = vec![];
+        resp.encode(&mut out);
+
+        let mut buf = out.as_slice();
+        let header = alloy_rlp::Header::decode(&mut buf).unwrap();
+        let payload_start = buf.len();
+        let request_id = u64::decode(&mut buf).unwrap();
+        let last_block_incomplete = bool::decode(&mut buf).unwrap();
+        let receipts = Vec::<Vec<Receipt>>::decode(&mut buf).unwrap();
+
+        assert!(buf.is_empty(), "buffer not fully consumed");
+        assert_eq!(payload_start - buf.len(), header.payload_length);
+        assert_eq!(request_id, 7);
+        assert!(last_block_incomplete);
+        assert_eq!(receipts.len(), 1);
+        assert_eq!(receipts[0].len(), 1);
+
+        let mut buf = out.as_slice();
+        let decoded = RequestPair::<Receipts70>::decode(&mut buf).unwrap();
+        assert!(buf.is_empty(), "buffer not fully consumed on decode");
+        assert_eq!(decoded, resp);
+    }
 }
--- a/crates/net/eth-wire-types/src/status.rs
+++ b/crates/net/eth-wire-types/src/status.rs
@@ -13,7 +13,7 @@ use reth_codecs_derive::add_arbitrary_tests;
 /// unsupported fields are stripped out.
 #[derive(Clone, Debug, PartialEq, Eq, Copy)]
 pub struct UnifiedStatus {
-    /// The eth protocol version (e.g. eth/66 to eth/69).
+    /// The eth protocol version (e.g. eth/66 to eth/70).
    pub version: EthVersion,
    /// The chain ID identifying the peer’s network.
    pub chain: Chain,
@@ -157,7 +157,7 @@ impl StatusBuilder {
        self.status
    }

-    /// Sets the eth protocol version (e.g., eth/66, eth/69).
+    /// Sets the eth protocol version (e.g., eth/66, eth/70).
    pub const fn version(mut self, version: EthVersion) -> Self {
        self.status.version = version;
        self
@@ -378,8 +378,8 @@ impl Debug for StatusEth69 {
    }
 }

-/// `StatusMessage` can store either the Legacy version (with TD) or the
-/// eth/69 version (omits TD).
+/// `StatusMessage` can store either the Legacy version (with TD), or the eth/69+/eth/70 version
+/// (omits TD, includes block range).
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum StatusMessage {
@@ -546,6 +546,24 @@ mod tests {
        assert_eq!(unified_status, roundtripped_unified_status);
    }

+    #[test]
+    fn roundtrip_eth70() {
+        let unified_status = UnifiedStatus::builder()
+            .version(EthVersion::Eth70)
+            .chain(Chain::mainnet())
+            .genesis(MAINNET_GENESIS_HASH)
+            .forkid(ForkId { hash: ForkHash([0xb7, 0x15, 0x07, 0x7d]), next: 0 })
+            .blockhash(b256!("0xfeb27336ca7923f8fab3bd617fcb6e75841538f71c1bcfc267d7838489d9e13d"))
+            .total_difficulty(None)
+            .earliest_block(Some(1))
+            .latest_block(Some(2))
+            .build();
+
+        let status_message = unified_status.into_message();
+        let roundtripped_unified_status = UnifiedStatus::from_message(status_message);
+        assert_eq!(unified_status, roundtripped_unified_status);
+    }
+
    #[test]
    fn encode_eth69_status_message() {
        let expected = hex!("f8544501a0d4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3c684b715077d8083ed14f2840112a880a0feb27336ca7923f8fab3bd617fcb6e75841538f71c1bcfc267d7838489d9e13d");
--- a/crates/net/eth-wire-types/src/version.rs
+++ b/crates/net/eth-wire-types/src/version.rs
@@ -27,6 +27,8 @@ pub enum EthVersion {
    Eth68 = 68,
    /// The `eth` protocol version 69.
    Eth69 = 69,
+    /// The `eth` protocol version 70.
+    Eth70 = 70,
 }

 impl EthVersion {
@@ -55,6 +57,11 @@ impl EthVersion {
    pub const fn is_eth69(&self) -> bool {
        matches!(self, Self::Eth69)
    }
+
+    /// Returns true if the version is eth/70
+    pub const fn is_eth70(&self) -> bool {
+        matches!(self, Self::Eth70)
+    }
 }

 /// RLP encodes `EthVersion` as a single byte (66-69).
@@ -96,6 +103,7 @@ impl TryFrom<&str> for EthVersion {
            "67" => Ok(Self::Eth67),
            "68" => Ok(Self::Eth68),
            "69" => Ok(Self::Eth69),
+            "70" => Ok(Self::Eth70),
            _ => Err(ParseVersionError(s.to_string())),
        }
    }
@@ -120,6 +128,7 @@ impl TryFrom<u8> for EthVersion {
            67 => Ok(Self::Eth67),
            68 => Ok(Self::Eth68),
            69 => Ok(Self::Eth69),
+            70 => Ok(Self::Eth70),
            _ => Err(ParseVersionError(u.to_string())),
        }
    }
@@ -149,6 +158,7 @@ impl From<EthVersion> for &'static str {
            EthVersion::Eth67 => "67",
            EthVersion::Eth68 => "68",
            EthVersion::Eth69 => "69",
+            EthVersion::Eth70 => "70",
        }
    }
 }
@@ -195,7 +205,7 @@ impl Decodable for ProtocolVersion {

 #[cfg(test)]
 mod tests {
-    use super::{EthVersion, ParseVersionError};
+    use super::EthVersion;
    use alloy_rlp::{Decodable, Encodable, Error as RlpError};
    use bytes::BytesMut;

@@ -205,7 +215,7 @@ mod tests {
        assert_eq!(EthVersion::Eth67, EthVersion::try_from("67").unwrap());
        assert_eq!(EthVersion::Eth68, EthVersion::try_from("68").unwrap());
        assert_eq!(EthVersion::Eth69, EthVersion::try_from("69").unwrap());
-        assert_eq!(Err(ParseVersionError("70".to_string())), EthVersion::try_from("70"));
+        assert_eq!(EthVersion::Eth70, EthVersion::try_from("70").unwrap());
    }

    #[test]
@@ -214,12 +224,18 @@ mod tests {
        assert_eq!(EthVersion::Eth67, "67".parse().unwrap());
        assert_eq!(EthVersion::Eth68, "68".parse().unwrap());
        assert_eq!(EthVersion::Eth69, "69".parse().unwrap());
-        assert_eq!(Err(ParseVersionError("70".to_string())), "70".parse::<EthVersion>());
+        assert_eq!(EthVersion::Eth70, "70".parse().unwrap());
    }

    #[test]
    fn test_eth_version_rlp_encode() {
-        let versions = [EthVersion::Eth66, EthVersion::Eth67, EthVersion::Eth68, EthVersion::Eth69];
+        let versions = [
+            EthVersion::Eth66,
+            EthVersion::Eth67,
+            EthVersion::Eth68,
+            EthVersion::Eth69,
+            EthVersion::Eth70,
+        ];

        for version in versions {
            let mut encoded = BytesMut::new();
@@ -236,7 +252,7 @@ mod tests {
            (67_u8, Ok(EthVersion::Eth67)),
            (68_u8, Ok(EthVersion::Eth68)),
            (69_u8, Ok(EthVersion::Eth69)),
-            (70_u8, Err(RlpError::Custom("invalid eth version"))),
+            (70_u8, Ok(EthVersion::Eth70)),
            (65_u8, Err(RlpError::Custom("invalid eth version"))),
        ];

--- a/crates/net/eth-wire/src/capability.rs
+++ b/crates/net/eth-wire/src/capability.rs
@@ -418,6 +418,8 @@ mod tests {
            Capability::new_static("eth", 66),
            Capability::new_static("eth", 67),
            Capability::new_static("eth", 68),
+            Capability::new_static("eth", 69),
+            Capability::new_static("eth", 70),
        ]
        .into();

@@ -425,6 +427,8 @@ mod tests {
        assert!(capabilities.supports_eth_v66());
        assert!(capabilities.supports_eth_v67());
        assert!(capabilities.supports_eth_v68());
+        assert!(capabilities.supports_eth_v69());
+        assert!(capabilities.supports_eth_v70());
    }

    #[test]
--- a/crates/net/eth-wire/src/hello.rs
+++ b/crates/net/eth-wire/src/hello.rs
@@ -260,10 +260,11 @@ mod tests {

        assert_eq!(hello_encoded.len(), hello.length());
    }
+    //TODO: add test for eth70 here once we have fully support it

    #[test]
-    fn test_default_protocols_include_eth69() {
-        // ensure that the default protocol list includes Eth69 as the latest version
+    fn test_default_protocols_still_include_eth69() {
+        // ensure that older eth/69 remains advertised for compatibility
        let secret_key = SecretKey::new(&mut rand_08::thread_rng());
        let id = pk2id(&secret_key.public_key(SECP256K1));
        let hello = HelloMessageWithProtocols::builder(id).build();
--- a/crates/net/nat/src/lib.rs
+++ b/crates/net/nat/src/lib.rs
@@ -19,7 +19,7 @@ pub use net_if::{NetInterfaceError, DEFAULT_NET_IF_NAME};
 use std::{
    fmt,
    future::{poll_fn, Future},
-    net::{AddrParseError, IpAddr},
+    net::{AddrParseError, IpAddr, ToSocketAddrs},
    pin::Pin,
    str::FromStr,
    task::{Context, Poll},
@@ -38,7 +38,7 @@ const EXTERNAL_IP_APIS: &[&str] =
    &["https://ipinfo.io/ip", "https://icanhazip.com", "https://ifconfig.me"];

 /// All builtin resolvers.
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Default, Hash)]
+#[derive(Debug, Clone, Eq, PartialEq, Default, Hash)]
 #[cfg_attr(feature = "serde", derive(SerializeDisplay, DeserializeFromStr))]
 pub enum NatResolver {
    /// Resolve with any available resolver.
@@ -50,6 +50,14 @@ pub enum NatResolver {
    PublicIp,
    /// Use the given [`IpAddr`]
    ExternalIp(IpAddr),
+    /// Use the given domain name as the external address to expose to peers.
+    /// This is behaving essentially the same as [`NatResolver::ExternalIp`], but supports domain
+    /// names. Domain names are resolved to IP addresses using the OS's resolver. The first IP
+    /// address found is used.
+    /// This may be useful in docker bridge networks where containers are usually queried by DNS
+    /// instead of direct IP addresses.
+    /// Note: the domain shouldn't include a port number. Only the IP address is resolved.
+    ExternalAddr(String),
    /// Resolve external IP via the network interface.
    NetIf,
    /// Resolve nothing
@@ -62,10 +70,17 @@ impl NatResolver {
        external_addr_with(self).await
    }

-    /// Returns the external ip, if it is [`NatResolver::ExternalIp`]
-    pub const fn as_external_ip(self) -> Option<IpAddr> {
+    /// Returns the fixed ip, if it is [`NatResolver::ExternalIp`] or [`NatResolver::ExternalAddr`].
+    ///
+    /// In the case of [`NatResolver::ExternalAddr`], it will return the first IP address found for
+    /// the domain.
+    pub fn as_external_ip(self, port: u16) -> Option<IpAddr> {
        match self {
            Self::ExternalIp(ip) => Some(ip),
+            Self::ExternalAddr(domain) => format!("{domain}:{port}")
+                .to_socket_addrs()
+                .ok()
+                .and_then(|mut addrs| addrs.next().map(|addr| addr.ip())),
            _ => None,
        }
    }
@@ -78,6 +93,7 @@ impl fmt::Display for NatResolver {
            Self::Upnp => f.write_str("upnp"),
            Self::PublicIp => f.write_str("publicip"),
            Self::ExternalIp(ip) => write!(f, "extip:{ip}"),
+            Self::ExternalAddr(domain) => write!(f, "extaddr:{domain}"),
            Self::NetIf => f.write_str("netif"),
            Self::None => f.write_str("none"),
        }
@@ -106,12 +122,15 @@ impl FromStr for NatResolver {
            "publicip" | "public-ip" => Self::PublicIp,
            "netif" => Self::NetIf,
            s => {
-                let Some(ip) = s.strip_prefix("extip:") else {
+                if let Some(ip) = s.strip_prefix("extip:") {
+                    Self::ExternalIp(ip.parse()?)
+                } else if let Some(domain) = s.strip_prefix("extaddr:") {
+                    Self::ExternalAddr(domain.to_string())
+                } else {
                    return Err(ParseNatResolverError::UnknownVariant(format!(
                        "Unknown Nat Resolver: {s}"
-                    )))
-                };
-                Self::ExternalIp(ip.parse()?)
+                    )));
+                }
            }
        };
        Ok(r)
@@ -180,7 +199,7 @@ impl ResolveNatInterval {
    ///    `None` if the attempt was unsuccessful.
    pub fn poll_tick(&mut self, cx: &mut Context<'_>) -> Poll<Option<IpAddr>> {
        if self.interval.poll_tick(cx).is_ready() {
-            self.future = Some(Box::pin(self.resolver.external_addr()));
+            self.future = Some(Box::pin(self.resolver.clone().external_addr()));
        }

        if let Some(mut fut) = self.future.take() {
@@ -212,6 +231,9 @@ pub async fn external_addr_with(resolver: NatResolver) -> Option<IpAddr> {
                );
            })
            .ok(),
+        NatResolver::ExternalAddr(domain) => {
+            domain.to_socket_addrs().ok().and_then(|mut addrs| addrs.next().map(|addr| addr.ip()))
+        }
        NatResolver::None => None,
    }
 }
@@ -245,7 +267,7 @@ async fn resolve_external_ip_url(url: &str) -> Option<IpAddr> {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use std::net::Ipv4Addr;
+    use std::net::{Ipv4Addr, Ipv6Addr};

    #[tokio::test]
    #[ignore]
@@ -267,6 +289,18 @@ mod tests {
        dbg!(ip);
    }

+    #[test]
+    fn as_external_ip_test() {
+        let resolver = NatResolver::ExternalAddr("localhost".to_string());
+        let ip = resolver.as_external_ip(30303).expect("localhost should be resolvable");
+
+        if ip.is_ipv4() {
+            assert_eq!(ip, IpAddr::V4(Ipv4Addr::LOCALHOST));
+        } else {
+            assert_eq!(ip, IpAddr::V6(Ipv6Addr::LOCALHOST));
+        }
+    }
+
    #[test]
    fn test_from_str() {
        assert_eq!(NatResolver::Any, "any".parse().unwrap());
@@ -275,6 +309,6 @@ mod tests {
        let ip = NatResolver::ExternalIp(IpAddr::V4(Ipv4Addr::UNSPECIFIED));
        let s = "extip:0.0.0.0";
        assert_eq!(ip, s.parse().unwrap());
-        assert_eq!(ip.to_string().as_str(), s);
+        assert_eq!(ip.to_string(), s);
    }
 }
--- a/crates/net/network-api/src/events.rs
+++ b/crates/net/network-api/src/events.rs
@@ -3,8 +3,8 @@
 use reth_eth_wire_types::{
    message::RequestPair, BlockBodies, BlockHeaders, Capabilities, DisconnectReason, EthMessage,
    EthNetworkPrimitives, EthVersion, GetBlockBodies, GetBlockHeaders, GetNodeData,
-    GetPooledTransactions, GetReceipts, NetworkPrimitives, NodeData, PooledTransactions, Receipts,
-    Receipts69, UnifiedStatus,
+    GetPooledTransactions, GetReceipts, GetReceipts70, NetworkPrimitives, NodeData,
+    PooledTransactions, Receipts, Receipts69, Receipts70, UnifiedStatus,
 };
 use reth_ethereum_forks::ForkId;
 use reth_network_p2p::error::{RequestError, RequestResult};
@@ -238,6 +238,15 @@ pub enum PeerRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
        /// The channel to send the response for receipts.
        response: oneshot::Sender<RequestResult<Receipts69<N::Receipt>>>,
    },
+    /// Requests receipts from the peer using eth/70 (supports `firstBlockReceiptIndex`).
+    ///
+    /// The response should be sent through the channel.
+    GetReceipts70 {
+        /// The request for receipts.
+        request: GetReceipts70,
+        /// The channel to send the response for receipts.
+        response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
+    },
 }

 // === impl PeerRequest ===
@@ -257,6 +266,7 @@ impl<N: NetworkPrimitives> PeerRequest<N> {
            Self::GetNodeData { response, .. } => response.send(Err(err)).ok(),
            Self::GetReceipts { response, .. } => response.send(Err(err)).ok(),
            Self::GetReceipts69 { response, .. } => response.send(Err(err)).ok(),
+            Self::GetReceipts70 { response, .. } => response.send(Err(err)).ok(),
        };
    }

@@ -281,6 +291,9 @@ impl<N: NetworkPrimitives> PeerRequest<N> {
            Self::GetReceipts { request, .. } | Self::GetReceipts69 { request, .. } => {
                EthMessage::GetReceipts(RequestPair { request_id, message: request.clone() })
            }
+            Self::GetReceipts70 { request, .. } => {
+                EthMessage::GetReceipts70(RequestPair { request_id, message: request.clone() })
+            }
        }
    }

--- a/crates/net/network/Cargo.toml
+++ b/crates/net/network/Cargo.toml
@@ -66,6 +66,7 @@ tracing.workspace = true
 rustc-hash.workspace = true
 thiserror.workspace = true
 parking_lot.workspace = true
+rayon.workspace = true
 rand.workspace = true
 rand_08.workspace = true
 secp256k1 = { workspace = true, features = ["global-context", "std", "recovery"] }
--- a/crates/net/network/src/config.rs
+++ b/crates/net/network/src/config.rs
@@ -433,7 +433,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
    pub fn external_ip_resolver(mut self, resolver: NatResolver) -> Self {
        self.discovery_v4_builder
            .get_or_insert_with(Discv4Config::builder)
-            .external_ip_resolver(Some(resolver));
+            .external_ip_resolver(Some(resolver.clone()));
        self.nat = Some(resolver);
        self
    }
@@ -484,7 +484,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
    }

    // Disable nat
-    pub const fn disable_nat(mut self) -> Self {
+    pub fn disable_nat(mut self) -> Self {
        self.nat = None;
        self
    }
@@ -579,7 +579,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
    }

    /// Sets the NAT resolver for external IP.
-    pub const fn add_nat(mut self, nat: Option<NatResolver>) -> Self {
+    pub fn add_nat(mut self, nat: Option<NatResolver>) -> Self {
        self.nat = nat;
        self
    }
--- a/crates/net/network/src/eth_requests.rs
+++ b/crates/net/network/src/eth_requests.rs
@@ -10,7 +10,8 @@ use alloy_rlp::Encodable;
 use futures::StreamExt;
 use reth_eth_wire::{
    BlockBodies, BlockHeaders, EthNetworkPrimitives, GetBlockBodies, GetBlockHeaders, GetNodeData,
-    GetReceipts, HeadersDirection, NetworkPrimitives, NodeData, Receipts, Receipts69,
+    GetReceipts, GetReceipts70, HeadersDirection, NetworkPrimitives, NodeData, Receipts,
+    Receipts69, Receipts70,
 };
 use reth_network_api::test_utils::PeersHandle;
 use reth_network_p2p::error::RequestResult;
@@ -217,6 +218,69 @@ where
        let _ = response.send(Ok(Receipts69(receipts)));
    }

+    /// Handles partial responses for [`GetReceipts70`] queries.
+    ///
+    /// This will adhere to the soft limit but allow filling the last vec partially.
+    fn on_receipts70_request(
+        &self,
+        _peer_id: PeerId,
+        request: GetReceipts70,
+        response: oneshot::Sender<RequestResult<Receipts70<C::Receipt>>>,
+    ) {
+        self.metrics.eth_receipts_requests_received_total.increment(1);
+
+        let GetReceipts70 { first_block_receipt_index, block_hashes } = request;
+
+        let mut receipts = Vec::new();
+        let mut total_bytes = 0usize;
+        let mut last_block_incomplete = false;
+
+        for (idx, hash) in block_hashes.into_iter().enumerate() {
+            if idx >= MAX_RECEIPTS_SERVE {
+                break
+            }
+
+            let Some(mut block_receipts) =
+                self.client.receipts_by_block(BlockHashOrNumber::Hash(hash)).unwrap_or_default()
+            else {
+                break
+            };
+
+            if idx == 0 && first_block_receipt_index > 0 {
+                let skip = first_block_receipt_index as usize;
+                if skip >= block_receipts.len() {
+                    block_receipts.clear();
+                } else {
+                    block_receipts.drain(0..skip);
+                }
+            }
+
+            let block_size = block_receipts.length();
+
+            if total_bytes + block_size <= SOFT_RESPONSE_LIMIT {
+                total_bytes += block_size;
+                receipts.push(block_receipts);
+                continue;
+            }
+
+            let mut partial_block = Vec::new();
+            for receipt in block_receipts {
+                let receipt_size = receipt.length();
+                if total_bytes + receipt_size > SOFT_RESPONSE_LIMIT {
+                    break;
+                }
+                total_bytes += receipt_size;
+                partial_block.push(receipt);
+            }
+
+            receipts.push(partial_block);
+            last_block_incomplete = true;
+            break;
+        }
+
+        let _ = response.send(Ok(Receipts70 { last_block_incomplete, receipts }));
+    }
+
    #[inline]
    fn get_receipts_response<T, F>(&self, request: GetReceipts, transform_fn: F) -> Vec<Vec<T>>
    where
@@ -285,6 +349,9 @@ where
                    IncomingEthRequest::GetReceipts69 { peer_id, request, response } => {
                        this.on_receipts69_request(peer_id, request, response)
                    }
+                    IncomingEthRequest::GetReceipts70 { peer_id, request, response } => {
+                        this.on_receipts70_request(peer_id, request, response)
+                    }
                }
            },
        );
@@ -359,4 +426,15 @@ pub enum IncomingEthRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
        /// The channel sender for the response containing Receipts69.
        response: oneshot::Sender<RequestResult<Receipts69<N::Receipt>>>,
    },
+    /// Request Receipts from the peer using eth/70.
+    ///
+    /// The response should be sent through the channel.
+    GetReceipts70 {
+        /// The ID of the peer to request receipts from.
+        peer_id: PeerId,
+        /// The specific receipts requested including the `firstBlockReceiptIndex`.
+        request: GetReceipts70,
+        /// The channel sender for the response containing Receipts70.
+        response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
+    },
 }
--- a/crates/net/network/src/manager.rs
+++ b/crates/net/network/src/manager.rs
@@ -532,6 +532,13 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                    response,
                })
            }
+            PeerRequest::GetReceipts70 { request, response } => {
+                self.delegate_eth_request(IncomingEthRequest::GetReceipts70 {
+                    peer_id,
+                    request,
+                    response,
+                })
+            }
            PeerRequest::GetPooledTransactions { request, response } => {
                self.notify_tx_manager(NetworkTransactionEvent::GetPooledTransactions {
                    peer_id,
--- a/crates/net/network/src/message.rs
+++ b/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::Receipts69;
+use crate::types::{Receipts69, Receipts70};
 use alloy_consensus::{BlockHeader, ReceiptWithBloom};
 use alloy_primitives::{Bytes, B256};
 use futures::FutureExt;
@@ -116,6 +116,11 @@ pub enum PeerResponse<N: NetworkPrimitives = EthNetworkPrimitives> {
        /// The receiver channel for the response to a receipts request.
        response: oneshot::Receiver<RequestResult<Receipts69<N::Receipt>>>,
    },
+    /// Represents a response to a request for receipts using eth/70.
+    Receipts70 {
+        /// The receiver channel for the response to a receipts request.
+        response: oneshot::Receiver<RequestResult<Receipts70<N::Receipt>>>,
+    },
 }

 // === impl PeerResponse ===
@@ -151,6 +156,10 @@ impl<N: NetworkPrimitives> PeerResponse<N> {
            Self::Receipts69 { response } => {
                poll_request!(response, Receipts69, cx)
            }
+            Self::Receipts70 { response } => match ready!(response.poll_unpin(cx)) {
+                Ok(res) => PeerResponseResult::Receipts70(res),
+                Err(err) => PeerResponseResult::Receipts70(Err(err.into())),
+            },
        };
        Poll::Ready(res)
    }
@@ -171,6 +180,8 @@ pub enum PeerResponseResult<N: NetworkPrimitives = EthNetworkPrimitives> {
    Receipts(RequestResult<Vec<Vec<ReceiptWithBloom<N::Receipt>>>>),
    /// Represents a result containing receipts or an error for eth/69.
    Receipts69(RequestResult<Vec<Vec<N::Receipt>>>),
+    /// Represents a result containing receipts or an error for eth/70.
+    Receipts70(RequestResult<Receipts70<N::Receipt>>),
 }

 // === impl PeerResponseResult ===
@@ -208,6 +219,13 @@ impl<N: NetworkPrimitives> PeerResponseResult<N> {
            Self::Receipts69(resp) => {
                to_message!(resp, Receipts69, id)
            }
+            Self::Receipts70(resp) => match resp {
+                Ok(res) => {
+                    let request = RequestPair { request_id: id, message: res };
+                    Ok(EthMessage::Receipts70(request))
+                }
+                Err(err) => Err(err),
+            },
        }
    }

@@ -220,6 +238,7 @@ impl<N: NetworkPrimitives> PeerResponseResult<N> {
            Self::NodeData(res) => res.as_ref().err(),
            Self::Receipts(res) => res.as_ref().err(),
            Self::Receipts69(res) => res.as_ref().err(),
+            Self::Receipts70(res) => res.as_ref().err(),
        }
    }

--- a/crates/net/network/src/network.rs
+++ b/crates/net/network/src/network.rs
@@ -237,7 +237,9 @@ impl<N: NetworkPrimitives> PeersInfo for NetworkHandle<N> {
            discv4.node_record()
        } else if let Some(discv5) = self.inner.discv5.as_ref() {
            // for disv5 we must check if we have an external ip configured
-            if let Some(external) = self.inner.nat.and_then(|nat| nat.as_external_ip()) {
+            if let Some(external) =
+                self.inner.nat.clone().and_then(|nat| nat.as_external_ip(discv5.local_port()))
+            {
                NodeRecord::new((external, discv5.local_port()).into(), *self.peer_id())
            } else {
                // use the node record that discv5 tracks or use localhost
@@ -252,9 +254,11 @@ impl<N: NetworkPrimitives> PeersInfo for NetworkHandle<N> {
            // also use the tcp port
            .with_tcp_port(self.inner.listener_address.lock().port())
        } else {
-            let external_ip = self.inner.nat.and_then(|nat| nat.as_external_ip());
-
            let mut socket_addr = *self.inner.listener_address.lock();
+
+            let external_ip =
+                self.inner.nat.clone().and_then(|nat| nat.as_external_ip(socket_addr.port()));
+
            if let Some(ip) = external_ip {
                // if able to resolve external ip, use it instead and also set the local address
                socket_addr.set_ip(ip)
--- a/crates/net/network/src/session/active.rs
+++ b/crates/net/network/src/session/active.rs
@@ -25,10 +25,10 @@ use futures::{stream::Fuse, SinkExt, StreamExt};
 use metrics::Gauge;
 use reth_eth_wire::{
    errors::{EthHandshakeError, EthStreamError},
-    message::{EthBroadcastMessage, MessageError, RequestPair},
+    message::{EthBroadcastMessage, MessageError},
    Capabilities, DisconnectP2P, DisconnectReason, EthMessage, NetworkPrimitives, NewBlockPayload,
 };
-use reth_eth_wire_types::RawCapabilityMessage;
+use reth_eth_wire_types::{message::RequestPair, RawCapabilityMessage};
 use reth_metrics::common::mpsc::MeteredPollSender;
 use reth_network_api::PeerRequest;
 use reth_network_p2p::error::RequestError;
@@ -270,12 +270,18 @@ impl<N: NetworkPrimitives> ActiveSession<N> {
                    on_request!(req, Receipts, GetReceipts)
                }
            }
+            EthMessage::GetReceipts70(req) => {
+                on_request!(req, Receipts70, GetReceipts70)
+            }
            EthMessage::Receipts(resp) => {
                on_response!(resp, GetReceipts)
            }
            EthMessage::Receipts69(resp) => {
                on_response!(resp, GetReceipts69)
            }
+            EthMessage::Receipts70(resp) => {
+                on_response!(resp, GetReceipts70)
+            }
            EthMessage::BlockRangeUpdate(msg) => {
                // Validate that earliest <= latest according to the spec
                if msg.earliest > msg.latest {
@@ -311,9 +317,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 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);
+        let msg = request.create_request_message(request_id).map_versioned(self.conn.version());
+
        self.queued_outgoing.push_back(msg.into());
        let req = InflightRequest {
            request: RequestState::Waiting(request),
--- a/crates/net/network/src/transactions/mod.rs
+++ b/crates/net/network/src/transactions/mod.rs
@@ -1,6 +1,7 @@
 //! Transactions management for the p2p network.

 use alloy_consensus::transaction::TxHashRef;
+use rayon::iter::{IntoParallelIterator, ParallelIterator};

 /// Aggregation on configurable parameters for [`TransactionsManager`].
 pub mod config;
@@ -1368,51 +1369,49 @@ where
        // tracks the quality of the given transactions
        let mut has_bad_transactions = false;

-        // 2. filter out transactions that are invalid or already pending import pre-size to avoid
-        //    reallocations
-        let mut new_txs = Vec::with_capacity(transactions.len());
-        for tx in transactions {
-            match self.transactions_by_peers.entry(*tx.tx_hash()) {
-                Entry::Occupied(mut entry) => {
-                    // transaction was already inserted
-                    entry.get_mut().insert(peer_id);
-                }
-                Entry::Vacant(entry) => {
-                    if self.bad_imports.contains(tx.tx_hash()) {
-                        trace!(target: "net::tx",
-                            peer_id=format!("{peer_id:#}"),
-                            hash=%tx.tx_hash(),
-                            client_version=%peer.client_version,
-                            "received a known bad transaction from peer"
-                        );
-                        has_bad_transactions = true;
-                    } else {
-                        // this is a new transaction that should be imported into the pool
-
-                        // recover transaction
-                        let tx = match tx.try_into_recovered() {
-                            Ok(tx) => tx,
-                            Err(badtx) => {
-                                trace!(target: "net::tx",
-                                    peer_id=format!("{peer_id:#}"),
-                                    hash=%badtx.tx_hash(),
-                                    client_version=%peer.client_version,
-                                    "failed ecrecovery for transaction"
-                                );
-                                has_bad_transactions = true;
-                                continue
-                            }
-                        };
-
-                        let pool_transaction = Pool::Transaction::from_pooled(tx);
-                        new_txs.push(pool_transaction);
-
-                        entry.insert(HashSet::from([peer_id]));
-                    }
-                }
+        // Remove known and invalid transactions
+        transactions.retain(|tx| {
+            if let Entry::Occupied(mut entry) = self.transactions_by_peers.entry(*tx.tx_hash()) {
+                entry.get_mut().insert(peer_id);
+                return false
            }
+            if self.bad_imports.contains(tx.tx_hash()) {
+                trace!(target: "net::tx",
+                    peer_id=format!("{peer_id:#}"),
+                    hash=%tx.tx_hash(),
+                    client_version=%peer.client_version,
+                    "received a known bad transaction from peer"
+                );
+                has_bad_transactions = true;
+                return false;
+            }
+            true
+        });
+
+        let txs_len = transactions.len();
+
+        let new_txs = transactions
+            .into_par_iter()
+            .filter_map(|tx| match tx.try_into_recovered() {
+                Ok(tx) => Some(Pool::Transaction::from_pooled(tx)),
+                Err(badtx) => {
+                    trace!(target: "net::tx",
+                        peer_id=format!("{peer_id:#}"),
+                        hash=%badtx.tx_hash(),
+                        client_version=%peer.client_version,
+                        "failed ecrecovery for transaction"
+                    );
+                    None
+                }
+            })
+            .collect::<Vec<_>>();
+
+        has_bad_transactions |= new_txs.len() != txs_len;
+
+        // Record the transactions as seen by the peer
+        for tx in &new_txs {
+            self.transactions_by_peers.insert(*tx.hash(), HashSet::from([peer_id]));
        }
-        new_txs.shrink_to_fit();

        // 3. import new transactions as a batch to minimize lock contention on the underlying
        // pool
@@ -1925,7 +1924,9 @@ impl PooledTransactionsHashesBuilder {
    fn new(version: EthVersion) -> Self {
        match version {
            EthVersion::Eth66 | EthVersion::Eth67 => Self::Eth66(Default::default()),
-            EthVersion::Eth68 | EthVersion::Eth69 => Self::Eth68(Default::default()),
+            EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 => {
+                Self::Eth68(Default::default())
+            }
        }
    }

--- a/crates/node/builder/src/launch/common.rs
+++ b/crates/node/builder/src/launch/common.rs
@@ -65,7 +65,7 @@ use reth_node_metrics::{
    version::VersionInfo,
 };
 use reth_provider::{
-    providers::{NodeTypesForProvider, ProviderNodeTypes, StaticFileProvider},
+    providers::{NodeTypesForProvider, ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
    BlockHashReader, BlockNumReader, ProviderError, ProviderFactory, ProviderResult,
    StageCheckpointReader, StaticFileProviderBuilder, StaticFileProviderFactory,
 };
@@ -485,9 +485,20 @@ where
                .with_blocks_per_file_for_segments(static_files_config.as_blocks_per_file_map())
                .build()?;

-        let factory =
-            ProviderFactory::new(self.right().clone(), self.chain_spec(), static_file_provider)?
-                .with_prune_modes(self.prune_modes());
+        // Initialize RocksDB provider with metrics, statistics, and default tables
+        let rocksdb_provider = RocksDBProvider::builder(self.data_dir().rocksdb())
+            .with_default_tables()
+            .with_metrics()
+            .with_statistics()
+            .build()?;
+
+        let factory = ProviderFactory::new(
+            self.right().clone(),
+            self.chain_spec(),
+            static_file_provider,
+            rocksdb_provider,
+        )?
+        .with_prune_modes(self.prune_modes());

        // Check for consistency between database and static files. If it fails, it unwinds to
        // the first block that's consistent between database and static files.
--- a/crates/node/core/src/args/datadir_args.rs
+++ b/crates/node/core/src/args/datadir_args.rs
@@ -27,6 +27,10 @@ pub struct DatadirArgs {
        verbatim_doc_comment
    )]
    pub static_files_path: Option<PathBuf>,
+
+    /// The absolute path to store `RocksDB` database in.
+    #[arg(long = "datadir.rocksdb", value_name = "PATH", verbatim_doc_comment)]
+    pub rocksdb_path: Option<PathBuf>,
 }

 impl DatadirArgs {
--- a/crates/node/core/src/args/network.rs
+++ b/crates/node/core/src/args/network.rs
@@ -337,7 +337,7 @@ impl NetworkArgs {

        // Configure basic network stack
        NetworkConfigBuilder::<N>::new(secret_key)
-            .external_ip_resolver(self.nat)
+            .external_ip_resolver(self.nat.clone())
            .sessions_config(
                SessionsConfig::default().with_upscaled_event_buffer(peers_config.max_peers()),
            )
@@ -399,7 +399,7 @@ impl NetworkArgs {
    }

    /// Configures the [`NatResolver`]
-    pub const fn with_nat_resolver(mut self, nat: NatResolver) -> Self {
+    pub fn with_nat_resolver(mut self, nat: NatResolver) -> Self {
        self.nat = nat;
        self
    }
@@ -782,10 +782,11 @@ mod tests {
        let tests = vec![0, 10];

        for retries in tests {
+            let retries_str = retries.to_string();
            let args = CommandParser::<NetworkArgs>::parse_from([
                "reth",
                "--dns-retries",
-                retries.to_string().as_str(),
+                retries_str.as_str(),
            ])
            .args;

--- a/crates/node/core/src/dirs.rs
+++ b/crates/node/core/src/dirs.rs
@@ -301,6 +301,18 @@ impl<D> ChainPath<D> {
        }
    }

+    /// Returns the path to the `RocksDB` database directory for this chain.
+    ///
+    /// `<DIR>/<CHAIN_ID>/rocksdb`
+    pub fn rocksdb(&self) -> PathBuf {
+        let datadir_args = &self.2;
+        if let Some(rocksdb_path) = &datadir_args.rocksdb_path {
+            rocksdb_path.clone()
+        } else {
+            self.data_dir().join("rocksdb")
+        }
+    }
+
    /// Returns the path to the reth p2p secret key for this chain.
    ///
    /// `<DIR>/<CHAIN_ID>/discovery-secret`
--- a/crates/optimism/bin/Cargo.toml
+++ b/crates/optimism/bin/Cargo.toml
@@ -27,7 +27,7 @@ tracing.workspace = true
 workspace = true

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

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

@@ -40,7 +40,9 @@ jemalloc-prof = ["reth-cli-util/jemalloc-prof"]
 tracy-allocator = ["reth-cli-util/tracy-allocator"]

 asm-keccak = ["reth-optimism-cli/asm-keccak", "reth-optimism-node/asm-keccak"]
-
+keccak-cache-global = [
+    "reth-optimism-node/keccak-cache-global",
+]
 dev = [
    "reth-optimism-cli/dev",
    "reth-optimism-primitives/arbitrary",
--- a/crates/optimism/cli/src/app.rs
+++ b/crates/optimism/cli/src/app.rs
@@ -98,7 +98,9 @@ where
                runner.run_blocking_until_ctrl_c(command.execute::<OpNode>())
            }
            Commands::DumpGenesis(command) => runner.run_blocking_until_ctrl_c(command.execute()),
-            Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute::<OpNode>()),
+            Commands::Db(command) => {
+                runner.run_blocking_command_until_exit(|ctx| command.execute::<OpNode>(ctx))
+            }
            Commands::Stage(command) => {
                runner.run_command_until_exit(|ctx| command.execute::<OpNode, _>(ctx, components))
            }
--- a/crates/optimism/evm/src/lib.rs
+++ b/crates/optimism/evm/src/lib.rs
@@ -13,32 +13,38 @@ extern crate alloc;

 use alloc::sync::Arc;
 use alloy_consensus::{BlockHeader, Header};
-use alloy_eips::Decodable2718;
 use alloy_evm::{EvmFactory, FromRecoveredTx, FromTxWithEncoded};
 use alloy_op_evm::block::{receipt_builder::OpReceiptBuilder, OpTxEnv};
-use alloy_primitives::{Bytes, U256};
 use core::fmt::Debug;
 use op_alloy_consensus::EIP1559ParamError;
-use op_alloy_rpc_types_engine::OpExecutionData;
 use op_revm::{OpSpecId, OpTransaction};
 use reth_chainspec::EthChainSpec;
 use reth_evm::{
-    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEngineEvm, ConfigureEvm,
-    EvmEnv, EvmEnvFor, ExecutableTxIterator, ExecutionCtxFor, TransactionEnv,
+    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEvm, EvmEnv, TransactionEnv,
 };
 use reth_optimism_chainspec::OpChainSpec;
 use reth_optimism_forks::OpHardforks;
 use reth_optimism_primitives::{DepositReceipt, OpPrimitives};
-use reth_primitives_traits::{
-    NodePrimitives, SealedBlock, SealedHeader, SignedTransaction, TxTy, WithEncoded,
-};
-use reth_storage_errors::any::AnyError;
-use revm::{
-    context::{BlockEnv, CfgEnv, TxEnv},
-    context_interface::block::BlobExcessGasAndPrice,
-    primitives::hardfork::SpecId,
+use reth_primitives_traits::{NodePrimitives, SealedBlock, SealedHeader, SignedTransaction};
+use revm::context::{BlockEnv, TxEnv};
+
+#[allow(unused_imports)]
+use {
+    alloy_eips::Decodable2718,
+    alloy_primitives::{Bytes, U256},
+    op_alloy_rpc_types_engine::OpExecutionData,
+    reth_evm::{EvmEnvFor, ExecutionCtxFor},
+    reth_primitives_traits::{TxTy, WithEncoded},
+    reth_storage_errors::any::AnyError,
+    revm::{
+        context::CfgEnv, context_interface::block::BlobExcessGasAndPrice,
+        primitives::hardfork::SpecId,
+    },
 };

+#[cfg(feature = "std")]
+use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator};
+
 mod config;
 pub use config::{revm_spec, revm_spec_by_timestamp_after_bedrock, OpNextBlockEnvAttributes};
 mod execute;
@@ -200,6 +206,7 @@ where
    }
 }

+#[cfg(feature = "std")]
 impl<ChainSpec, N, R> ConfigureEngineEvm<OpExecutionData> for OpEvmConfig<ChainSpec, N, R>
 where
    ChainSpec: EthChainSpec<Header = Header> + OpHardforks,
@@ -265,7 +272,7 @@ where
        &self,
        payload: &OpExecutionData,
    ) -> Result<impl ExecutableTxIterator<Self>, Self::Error> {
-        let transactions = payload.payload.transactions().clone().into_iter();
+        let transactions = payload.payload.transactions().clone();
        let convert = |encoded: Bytes| {
            let tx = TxTy::<Self::Primitives>::decode_2718_exact(encoded.as_ref())
                .map_err(AnyError::new)?;
--- a/crates/optimism/flashblocks/src/lib.rs
+++ b/crates/optimism/flashblocks/src/lib.rs
@@ -34,7 +34,7 @@ mod cache;
 mod test_utils;

 mod ws;
-pub use ws::{WsConnect, WsFlashBlockStream};
+pub use ws::{FlashBlockDecoder, WsConnect, WsFlashBlockStream};

 /// Receiver of the most recent [`PendingFlashBlock`] built out of [`FlashBlock`]s.
 ///
--- a/crates/optimism/flashblocks/src/ws/mod.rs
+++ b/crates/optimism/flashblocks/src/ws/mod.rs
@@ -1,6 +1,6 @@
 pub use stream::{WsConnect, WsFlashBlockStream};

 mod decoding;
-pub(crate) use decoding::FlashBlockDecoder;
+pub use decoding::FlashBlockDecoder;

 mod stream;
--- a/crates/optimism/node/Cargo.toml
+++ b/crates/optimism/node/Cargo.toml
@@ -34,7 +34,7 @@ reth-rpc-api.workspace = true

 # op-reth
 reth-optimism-payload-builder.workspace = true
-reth-optimism-evm = { workspace = true, features = ["rpc"] }
+reth-optimism-evm = { workspace = true, features = ["std", "rpc"] }
 reth-optimism-rpc.workspace = true
 reth-optimism-storage.workspace = true
 reth-optimism-txpool.workspace = true
@@ -93,6 +93,10 @@ asm-keccak = [
    "reth-node-core/asm-keccak",
    "revm/asm-keccak",
 ]
+keccak-cache-global = [
+    "alloy-primitives/keccak-cache-global",
+    "reth-optimism-node/keccak-cache-global",
+]
 js-tracer = [
    "reth-node-builder/js-tracer",
    "reth-optimism-node/js-tracer",
--- a/crates/optimism/node/src/node.rs
+++ b/crates/optimism/node/src/node.rs
@@ -218,13 +218,14 @@ impl OpNode {
    /// use reth_db::open_db_read_only;
    /// use reth_optimism_chainspec::OpChainSpecBuilder;
    /// use reth_optimism_node::OpNode;
-    /// use reth_provider::providers::StaticFileProvider;
+    /// use reth_provider::providers::{RocksDBProvider, StaticFileProvider};
    /// use std::sync::Arc;
    ///
    /// let factory = OpNode::provider_factory_builder()
    ///     .db(Arc::new(open_db_read_only("db", Default::default()).unwrap()))
    ///     .chainspec(OpChainSpecBuilder::base_mainnet().build().into())
    ///     .static_file(StaticFileProvider::read_only("db/static_files", false).unwrap())
+    ///     .rocksdb_provider(RocksDBProvider::builder("db/rocksdb").build().unwrap())
    ///     .build_provider_factory();
    /// ```
    pub fn provider_factory_builder() -> ProviderFactoryBuilder<Self> {
--- a/crates/optimism/reth/Cargo.toml
+++ b/crates/optimism/reth/Cargo.toml
@@ -74,7 +74,9 @@ arbitrary = [
    "reth-eth-wire?/arbitrary",
    "reth-codecs?/arbitrary",
 ]
-
+keccak-cache-global = [
+    "reth-optimism-node?/keccak-cache-global",
+]
 test-utils = [
    "reth-chainspec/test-utils",
    "reth-consensus?/test-utils",
--- a/crates/primitives-traits/src/block/recovered.rs
+++ b/crates/primitives-traits/src/block/recovered.rs
@@ -218,7 +218,7 @@ impl<B: Block> RecoveredBlock<B> {

    /// A safer variant of [`Self::new_sealed`] that checks if the number of senders is equal to
    /// the number of transactions in the block and recovers the senders from the transactions, if
-    /// not using [`SignedTransaction::recover_signer_unchecked`](crate::transaction::signed::SignedTransaction)
+    /// not using [`SignedTransaction::recover_signer`](crate::transaction::signed::SignedTransaction)
    /// to recover the senders.
    ///
    /// Returns an error if any of the transactions fail to recover the sender.
--- a/crates/rpc/rpc-builder/src/lib.rs
+++ b/crates/rpc/rpc-builder/src/lib.rs
@@ -2342,7 +2342,7 @@ mod tests {
                $(
                    let val: RethRpcModule  = $s.parse().unwrap();
                    assert_eq!(val, $v);
-                    assert_eq!(val.to_string().as_str(), $s);
+                    assert_eq!(val.to_string(), $s);
                )*
            };
        }
--- a/crates/rpc/rpc-engine-api/src/engine_api.rs
+++ b/crates/rpc/rpc-engine-api/src/engine_api.rs
@@ -463,7 +463,7 @@ where
    /// Returns the most recent version of the payload that is available in the corresponding
    /// payload build process at the time of receiving this call.
    ///
-    /// See also <https://github.com/ethereum/execution-apis/blob/7907424db935b93c2fe6a3c0faab943adebe8557/src/engine/prague.md#engine_newpayloadv4>
+    /// See also <https://github.com/ethereum/execution-apis/blob/7907424db935b93c2fe6a3c0faab943adebe8557/src/engine/prague.md#engine_getpayloadv4>
    ///
    /// Note:
    /// > Provider software MAY stop the corresponding build process after serving this call.
@@ -933,7 +933,7 @@ where
        Ok(self.fork_choice_updated_v2_metered(fork_choice_state, payload_attributes).await?)
    }

-    /// Handler for `engine_forkchoiceUpdatedV2`
+    /// Handler for `engine_forkchoiceUpdatedV3`
    ///
    /// See also <https://github.com/ethereum/execution-apis/blob/main/src/engine/cancun.md#engine_forkchoiceupdatedv3>
    async fn fork_choice_updated_v3(
--- a/crates/rpc/rpc-eth-api/src/helpers/block.rs
+++ b/crates/rpc/rpc-eth-api/src/helpers/block.rs
@@ -74,7 +74,11 @@ pub trait EthBlocks: LoadBlock<RpcConvert: RpcConvert<Primitives = Self::Primiti
        block_id: BlockId,
    ) -> impl Future<Output = Result<Option<usize>, Self::Error>> + Send {
        async move {
+            // If no pending block from provider, build the pending block locally.
            if block_id.is_pending() {
+                if let Some(pending) = self.local_pending_block().await? {
+                    return Ok(Some(pending.block.body().transaction_count()));
+                }
                // Pending block can be fetched directly without need for caching
                return Ok(self
                    .provider()
--- a/crates/rpc/rpc-testing-util/src/trace.rs
+++ b/crates/rpc/rpc-testing-util/src/trace.rs
@@ -71,7 +71,7 @@ pub trait TraceApiExt {

    /// Returns a new stream that yields the traces the opcodes for the given blocks.
    ///
-    /// See also [`StreamExt::buffered`].
+    /// See also [`StreamExt::buffer_unordered`].
    fn trace_block_opcode_gas_unordered<I, B>(
        &self,
        params: I,
@@ -301,7 +301,7 @@ impl<T: TraceApiClient<TransactionRequest> + Sync> TraceApiExt for T {
                Err(err) => Err((err, block)),
            }
        }))
-        .buffered(n);
+        .buffer_unordered(n);
        TraceBlockOpcodeGasStream { stream: Box::pin(stream) }
    }

--- a/crates/rpc/rpc/src/trace.rs
+++ b/crates/rpc/rpc/src/trace.rs
@@ -175,8 +175,6 @@ where
                    // need to apply the state changes of this call before executing the
                    // next call
                    if calls.peek().is_some() {
-                        // need to apply the state changes of this call before executing
-                        // the next call
                        db.commit(res.state)
                    }
                }
--- a/crates/stages/api/src/stage.rs
+++ b/crates/stages/api/src/stage.rs
@@ -324,11 +324,13 @@ impl<Provider, S: Stage<Provider> + ?Sized> StageExt<Provider> for S {}
 #[cfg(test)]
 mod tests {
    use reth_chainspec::MAINNET;
-    use reth_db::test_utils::{create_test_rw_db, create_test_static_files_dir};
+    use reth_db::test_utils::{
+        create_test_rocksdb_dir, create_test_rw_db, create_test_static_files_dir,
+    };
    use reth_db_api::{models::StoredBlockBodyIndices, tables, transaction::DbTxMut};
    use reth_provider::{
-        test_utils::MockNodeTypesWithDB, ProviderFactory, StaticFileProviderBuilder,
-        StaticFileProviderFactory, StaticFileSegment,
+        providers::RocksDBProvider, test_utils::MockNodeTypesWithDB, ProviderFactory,
+        StaticFileProviderBuilder, StaticFileProviderFactory, StaticFileSegment,
    };
    use reth_stages_types::StageCheckpoint;
    use reth_testing_utils::generators::{self, random_signed_tx};
@@ -346,6 +348,7 @@ mod tests {
                .with_blocks_per_file(1)
                .build()
                .unwrap(),
+            RocksDBProvider::builder(create_test_rocksdb_dir().0.keep()).build().unwrap(),
        )
        .unwrap();

--- a/crates/stages/stages/src/test_utils/test_db.rs
+++ b/crates/stages/stages/src/test_utils/test_db.rs
@@ -1,7 +1,10 @@
 use alloy_primitives::{keccak256, Address, BlockNumber, TxHash, TxNumber, B256};
 use reth_chainspec::MAINNET;
 use reth_db::{
-    test_utils::{create_test_rw_db, create_test_rw_db_with_path, create_test_static_files_dir},
+    test_utils::{
+        create_test_rocksdb_dir, create_test_rw_db, create_test_rw_db_with_path,
+        create_test_static_files_dir,
+    },
    DatabaseEnv,
 };
 use reth_db_api::{
@@ -17,7 +20,9 @@ use reth_db_api::{
 use reth_ethereum_primitives::{Block, EthPrimitives, Receipt};
 use reth_primitives_traits::{Account, SealedBlock, SealedHeader, StorageEntry};
 use reth_provider::{
-    providers::{StaticFileProvider, StaticFileProviderRWRefMut, StaticFileWriter},
+    providers::{
+        RocksDBProvider, StaticFileProvider, StaticFileProviderRWRefMut, StaticFileWriter,
+    },
    test_utils::MockNodeTypesWithDB,
    HistoryWriter, ProviderError, ProviderFactory, StaticFileProviderFactory, StatsReader,
 };
@@ -38,12 +43,14 @@ impl Default for TestStageDB {
    /// Create a new instance of [`TestStageDB`]
    fn default() -> Self {
        let (static_dir, static_dir_path) = create_test_static_files_dir();
+        let (_, rocksdb_dir_path) = create_test_rocksdb_dir();
        Self {
            temp_static_files_dir: static_dir,
            factory: ProviderFactory::new(
                create_test_rw_db(),
                MAINNET.clone(),
                StaticFileProvider::read_write(static_dir_path).unwrap(),
+                RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
            )
            .expect("failed to create test provider factory"),
        }
@@ -53,6 +60,7 @@ impl Default for TestStageDB {
 impl TestStageDB {
    pub fn new(path: &Path) -> Self {
        let (static_dir, static_dir_path) = create_test_static_files_dir();
+        let (_, rocksdb_dir_path) = create_test_rocksdb_dir();

        Self {
            temp_static_files_dir: static_dir,
@@ -60,6 +68,7 @@ impl TestStageDB {
                create_test_rw_db_with_path(path),
                MAINNET.clone(),
                StaticFileProvider::read_write(static_dir_path).unwrap(),
+                RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
            )
            .expect("failed to create test provider factory"),
        }
--- a/crates/storage/db-common/src/init.rs
+++ b/crates/storage/db-common/src/init.rs
@@ -711,7 +711,7 @@ mod tests {
    };
    use reth_provider::{
        test_utils::{create_test_provider_factory_with_chain_spec, MockNodeTypesWithDB},
-        ProviderFactory,
+        ProviderFactory, RocksDBProviderFactory,
    };
    use std::{collections::BTreeMap, sync::Arc};

@@ -756,6 +756,7 @@ mod tests {
    fn fail_init_inconsistent_db() {
        let factory = create_test_provider_factory_with_chain_spec(SEPOLIA.clone());
        let static_file_provider = factory.static_file_provider();
+        let rocksdb_provider = factory.rocksdb_provider();
        init_genesis(&factory).unwrap();

        // Try to init db with a different genesis block
@@ -764,6 +765,7 @@ mod tests {
                factory.into_db(),
                MAINNET.clone(),
                static_file_provider,
+                rocksdb_provider,
            )
            .unwrap(),
        );
--- a/crates/storage/db/src/lib.rs
+++ b/crates/storage/db/src/lib.rs
@@ -159,6 +159,14 @@ pub mod test_utils {
        (temp_dir, path)
    }

+    /// Create `rocksdb` path for testing
+    #[track_caller]
+    pub fn create_test_rocksdb_dir() -> (TempDir, PathBuf) {
+        let temp_dir = TempDir::with_prefix("reth-test-rocksdb-").expect(ERROR_TEMPDIR);
+        let path = temp_dir.path().to_path_buf();
+        (temp_dir, path)
+    }
+
    /// Get a temporary directory path to use for the database
    pub fn tempdir_path() -> PathBuf {
        let builder = tempfile::Builder::new().prefix("reth-test-").rand_bytes(8).tempdir();
--- a/crates/storage/provider/src/either_writer.rs
+++ b/crates/storage/provider/src/either_writer.rs
@@ -4,19 +4,24 @@
 use std::{marker::PhantomData, ops::Range};

 #[cfg(all(unix, feature = "rocksdb"))]
-use crate::providers::rocksdb::RocksDBWriteMode;
+use crate::providers::rocksdb::RocksDBBatch;
 use crate::{
    providers::{StaticFileProvider, StaticFileProviderRWRefMut},
    StaticFileProviderFactory,
 };
-use alloy_primitives::{map::HashMap, Address, BlockNumber, TxNumber};
+use alloy_primitives::{map::HashMap, Address, BlockNumber, TxHash, TxNumber};
 use reth_db::{
    cursor::DbCursorRO,
    static_file::TransactionSenderMask,
    table::Value,
    transaction::{CursorMutTy, CursorTy, DbTx, DbTxMut},
 };
-use reth_db_api::{cursor::DbCursorRW, tables};
+use reth_db_api::{
+    cursor::DbCursorRW,
+    models::{storage_sharded_key::StorageShardedKey, ShardedKey},
+    tables,
+    tables::BlockNumberList,
+};
 use reth_errors::ProviderError;
 use reth_node_types::NodePrimitives;
 use reth_primitives_traits::ReceiptTy;
@@ -37,11 +42,11 @@ type EitherWriterTy<'a, P, T> = EitherWriter<
 >;

 // Helper types so constructors stay exported even when RocksDB feature is off.
-// RocksDBWriteMode encapsulates the choice between Transaction and Batch writes.
+// Historical data tables use a write-only RocksDB batch (no read-your-writes needed).
 #[cfg(all(unix, feature = "rocksdb"))]
-type RocksWriteModeArg<'a> = crate::providers::rocksdb::RocksDBWriteMode<'a>;
+type RocksBatchArg<'a> = crate::providers::rocksdb::RocksDBBatch<'a>;
 #[cfg(not(all(unix, feature = "rocksdb")))]
-type RocksWriteModeArg<'a> = ();
+type RocksBatchArg<'a> = ();

 #[cfg(all(unix, feature = "rocksdb"))]
 type RocksTxRefArg<'a> = &'a crate::providers::rocksdb::RocksTx<'a>;
@@ -55,12 +60,9 @@ pub enum EitherWriter<'a, CURSOR, N> {
    Database(CURSOR),
    /// Write to static file
    StaticFile(StaticFileProviderRWRefMut<'a, N>),
-    /// Write to `RocksDB` (transaction or batch - internal detail).
-    ///
-    /// Uses [`RocksDBWriteMode`] to encapsulate the choice between full transaction
-    /// semantics and high-throughput batch writes.
+    /// Write to `RocksDB` using a write-only batch (historical tables).
    #[cfg(all(unix, feature = "rocksdb"))]
-    RocksDB(RocksDBWriteMode<'a>),
+    RocksDB(RocksDBBatch<'a>),
 }

 impl<'a> EitherWriter<'a, (), ()> {
@@ -131,11 +133,9 @@ impl<'a> EitherWriter<'a, (), ()> {
    }

    /// Creates a new [`EitherWriter`] for storages history based on storage settings.
-    ///
-    /// Accepts either Transaction or Batch mode via [`RocksDBWriteMode`].
    pub fn new_storages_history<P>(
        provider: &P,
-        _rocksdb_mode: RocksWriteModeArg<'a>,
+        _rocksdb_batch: RocksBatchArg<'a>,
    ) -> ProviderResult<EitherWriterTy<'a, P, tables::StoragesHistory>>
    where
        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
@@ -143,37 +143,16 @@ impl<'a> EitherWriter<'a, (), ()> {
    {
        #[cfg(all(unix, feature = "rocksdb"))]
        if provider.cached_storage_settings().storages_history_in_rocksdb {
-            return Ok(EitherWriter::RocksDB(_rocksdb_mode));
+            return Ok(EitherWriter::RocksDB(_rocksdb_batch));
        }

        Ok(EitherWriter::Database(provider.tx_ref().cursor_write::<tables::StoragesHistory>()?))
    }

-    /// Creates a new [`EitherWriter`] for accounts history based on storage settings.
-    ///
-    /// Accepts either Transaction or Batch mode via [`RocksDBWriteMode`].
-    pub fn new_accounts_history<P>(
-        provider: &P,
-        _rocksdb_mode: RocksWriteModeArg<'a>,
-    ) -> ProviderResult<EitherWriterTy<'a, P, tables::AccountsHistory>>
-    where
-        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
-        P::Tx: DbTxMut,
-    {
-        #[cfg(all(unix, feature = "rocksdb"))]
-        if provider.cached_storage_settings().account_history_in_rocksdb {
-            return Ok(EitherWriter::RocksDB(_rocksdb_mode));
-        }
-
-        Ok(EitherWriter::Database(provider.tx_ref().cursor_write::<tables::AccountsHistory>()?))
-    }
-
    /// Creates a new [`EitherWriter`] for transaction hash numbers based on storage settings.
-    ///
-    /// Accepts either Transaction or Batch mode via [`RocksDBWriteMode`].
    pub fn new_transaction_hash_numbers<P>(
        provider: &P,
-        _rocksdb_mode: RocksWriteModeArg<'a>,
+        _rocksdb_batch: RocksBatchArg<'a>,
    ) -> ProviderResult<EitherWriterTy<'a, P, tables::TransactionHashNumbers>>
    where
        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
@@ -181,13 +160,30 @@ impl<'a> EitherWriter<'a, (), ()> {
    {
        #[cfg(all(unix, feature = "rocksdb"))]
        if provider.cached_storage_settings().transaction_hash_numbers_in_rocksdb {
-            return Ok(EitherWriter::RocksDB(_rocksdb_mode));
+            return Ok(EitherWriter::RocksDB(_rocksdb_batch));
        }

        Ok(EitherWriter::Database(
            provider.tx_ref().cursor_write::<tables::TransactionHashNumbers>()?,
        ))
    }
+
+    /// Creates a new [`EitherWriter`] for account history based on storage settings.
+    pub fn new_accounts_history<P>(
+        provider: &P,
+        _rocksdb_batch: RocksBatchArg<'a>,
+    ) -> ProviderResult<EitherWriterTy<'a, P, tables::AccountsHistory>>
+    where
+        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
+        P::Tx: DbTxMut,
+    {
+        #[cfg(all(unix, feature = "rocksdb"))]
+        if provider.cached_storage_settings().account_history_in_rocksdb {
+            return Ok(EitherWriter::RocksDB(_rocksdb_batch));
+        }
+
+        Ok(EitherWriter::Database(provider.tx_ref().cursor_write::<tables::AccountsHistory>()?))
+    }
 }

 impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N> {
@@ -217,24 +213,6 @@ impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N> {
            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
        }
    }
-
-    /// Commits `RocksDB` writes if this is a `RocksDB` writer.
-    ///
-    /// For [`Self::Database`] and [`Self::StaticFile`], this is a no-op as they use
-    /// different commit patterns (MDBX transaction commit, static file sync).
-    ///
-    /// # Commit Order
-    ///
-    /// Call this AFTER the outer MDBX transaction commits successfully. This ensures
-    /// that if `RocksDB` commit fails, the primary data (MDBX) is still intact and
-    /// the `RocksDB` data (which is derived) can be rebuilt.
-    #[cfg(all(unix, feature = "rocksdb"))]
-    pub fn commit(self) -> ProviderResult<()> {
-        match self {
-            Self::Database(_) | Self::StaticFile(_) => Ok(()),
-            Self::RocksDB(mode) => mode.commit(),
-        }
-    }
 }

 impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
@@ -321,6 +299,108 @@ where
    }
 }

+impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
+where
+    CURSOR: DbCursorRW<tables::TransactionHashNumbers> + DbCursorRO<tables::TransactionHashNumbers>,
+{
+    /// Puts a transaction hash number mapping.
+    pub fn put_transaction_hash_number(
+        &mut self,
+        hash: TxHash,
+        tx_num: TxNumber,
+    ) -> ProviderResult<()> {
+        match self {
+            Self::Database(cursor) => Ok(cursor.upsert(hash, &tx_num)?),
+            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(batch) => batch.put::<tables::TransactionHashNumbers>(hash, &tx_num),
+        }
+    }
+
+    /// Deletes a transaction hash number mapping.
+    pub fn delete_transaction_hash_number(&mut self, hash: TxHash) -> ProviderResult<()> {
+        match self {
+            Self::Database(cursor) => {
+                if cursor.seek_exact(hash)?.is_some() {
+                    cursor.delete_current()?;
+                }
+                Ok(())
+            }
+            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(batch) => batch.delete::<tables::TransactionHashNumbers>(hash),
+        }
+    }
+}
+
+impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
+where
+    CURSOR: DbCursorRW<tables::StoragesHistory> + DbCursorRO<tables::StoragesHistory>,
+{
+    /// Puts a storage history entry.
+    pub fn put_storage_history(
+        &mut self,
+        key: StorageShardedKey,
+        value: &BlockNumberList,
+    ) -> ProviderResult<()> {
+        match self {
+            Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
+            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(batch) => batch.put::<tables::StoragesHistory>(key, value),
+        }
+    }
+
+    /// Deletes a storage history entry.
+    pub fn delete_storage_history(&mut self, key: StorageShardedKey) -> ProviderResult<()> {
+        match self {
+            Self::Database(cursor) => {
+                if cursor.seek_exact(key)?.is_some() {
+                    cursor.delete_current()?;
+                }
+                Ok(())
+            }
+            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(batch) => batch.delete::<tables::StoragesHistory>(key),
+        }
+    }
+}
+
+impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
+where
+    CURSOR: DbCursorRW<tables::AccountsHistory> + DbCursorRO<tables::AccountsHistory>,
+{
+    /// Puts an account history entry.
+    pub fn put_account_history(
+        &mut self,
+        key: ShardedKey<Address>,
+        value: &BlockNumberList,
+    ) -> ProviderResult<()> {
+        match self {
+            Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
+            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(batch) => batch.put::<tables::AccountsHistory>(key, value),
+        }
+    }
+
+    /// Deletes an account history entry.
+    pub fn delete_account_history(&mut self, key: ShardedKey<Address>) -> ProviderResult<()> {
+        match self {
+            Self::Database(cursor) => {
+                if cursor.seek_exact(key)?.is_some() {
+                    cursor.delete_current()?;
+                }
+                Ok(())
+            }
+            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(batch) => batch.delete::<tables::AccountsHistory>(key),
+        }
+    }
+}
+
 /// Represents a source for reading data, either from database, static files, or `RocksDB`.
 #[derive(Debug, Display)]
 pub enum EitherReader<'a, CURSOR, N> {
@@ -391,6 +471,26 @@ impl<'a> EitherReader<'a, (), ()> {
            PhantomData,
        ))
    }
+
+    /// Creates a new [`EitherReader`] for account history based on storage settings.
+    pub fn new_accounts_history<P>(
+        provider: &P,
+        _rocksdb_tx: RocksTxRefArg<'a>,
+    ) -> ProviderResult<EitherReaderTy<'a, P, tables::AccountsHistory>>
+    where
+        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
+        P::Tx: DbTx,
+    {
+        #[cfg(all(unix, feature = "rocksdb"))]
+        if provider.cached_storage_settings().account_history_in_rocksdb {
+            return Ok(EitherReader::RocksDB(_rocksdb_tx));
+        }
+
+        Ok(EitherReader::Database(
+            provider.tx_ref().cursor_read::<tables::AccountsHistory>()?,
+            PhantomData,
+        ))
+    }
 }

 impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
@@ -425,6 +525,60 @@ where
    }
 }

+impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
+where
+    CURSOR: DbCursorRO<tables::TransactionHashNumbers>,
+{
+    /// Gets a transaction number by its hash.
+    pub fn get_transaction_hash_number(
+        &mut self,
+        hash: TxHash,
+    ) -> ProviderResult<Option<TxNumber>> {
+        match self {
+            Self::Database(cursor, _) => Ok(cursor.seek_exact(hash)?.map(|(_, v)| v)),
+            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(tx) => tx.get::<tables::TransactionHashNumbers>(hash),
+        }
+    }
+}
+
+impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
+where
+    CURSOR: DbCursorRO<tables::StoragesHistory>,
+{
+    /// Gets a storage history entry.
+    pub fn get_storage_history(
+        &mut self,
+        key: StorageShardedKey,
+    ) -> ProviderResult<Option<BlockNumberList>> {
+        match self {
+            Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
+            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(tx) => tx.get::<tables::StoragesHistory>(key),
+        }
+    }
+}
+
+impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
+where
+    CURSOR: DbCursorRO<tables::AccountsHistory>,
+{
+    /// Gets an account history entry.
+    pub fn get_account_history(
+        &mut self,
+        key: ShardedKey<Address>,
+    ) -> ProviderResult<Option<BlockNumberList>> {
+        match self {
+            Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
+            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
+            #[cfg(all(unix, feature = "rocksdb"))]
+            Self::RocksDB(tx) => tx.get::<tables::AccountsHistory>(key),
+        }
+    }
+}
+
 /// Destination for writing data.
 #[derive(Debug, EnumIs)]
 pub enum EitherWriterDestination {
@@ -506,3 +660,160 @@ mod tests {
        }
    }
 }
+
+#[cfg(all(test, unix, feature = "rocksdb"))]
+mod rocksdb_tests {
+    use crate::providers::rocksdb::{RocksDBBuilder, RocksDBProvider};
+    use alloy_primitives::{Address, B256};
+    use reth_db_api::{
+        models::{storage_sharded_key::StorageShardedKey, IntegerList, ShardedKey},
+        tables,
+    };
+    use tempfile::TempDir;
+
+    fn create_rocksdb_provider() -> (TempDir, RocksDBProvider) {
+        let temp_dir = TempDir::new().unwrap();
+        let provider = RocksDBBuilder::new(temp_dir.path())
+            .with_table::<tables::TransactionHashNumbers>()
+            .with_table::<tables::StoragesHistory>()
+            .with_table::<tables::AccountsHistory>()
+            .build()
+            .unwrap();
+        (temp_dir, provider)
+    }
+
+    #[test]
+    fn test_rocksdb_batch_transaction_hash_numbers() {
+        let (_temp_dir, provider) = create_rocksdb_provider();
+
+        let hash1 = B256::from([1u8; 32]);
+        let hash2 = B256::from([2u8; 32]);
+        let tx_num1 = 100u64;
+        let tx_num2 = 200u64;
+
+        // Write via RocksDBBatch (same as EitherWriter::RocksDB would use internally)
+        let mut batch = provider.batch();
+        batch.put::<tables::TransactionHashNumbers>(hash1, &tx_num1).unwrap();
+        batch.put::<tables::TransactionHashNumbers>(hash2, &tx_num2).unwrap();
+        batch.commit().unwrap();
+
+        // Read via RocksTx (same as EitherReader::RocksDB would use internally)
+        let tx = provider.tx();
+        assert_eq!(tx.get::<tables::TransactionHashNumbers>(hash1).unwrap(), Some(tx_num1));
+        assert_eq!(tx.get::<tables::TransactionHashNumbers>(hash2).unwrap(), Some(tx_num2));
+
+        // Test missing key
+        let missing_hash = B256::from([99u8; 32]);
+        assert_eq!(tx.get::<tables::TransactionHashNumbers>(missing_hash).unwrap(), None);
+    }
+
+    #[test]
+    fn test_rocksdb_batch_storage_history() {
+        let (_temp_dir, provider) = create_rocksdb_provider();
+
+        let address = Address::random();
+        let storage_key = B256::from([1u8; 32]);
+        let key = StorageShardedKey::new(address, storage_key, 1000);
+        let value = IntegerList::new([1, 5, 10, 50]).unwrap();
+
+        // Write via RocksDBBatch
+        let mut batch = provider.batch();
+        batch.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
+        batch.commit().unwrap();
+
+        // Read via RocksTx
+        let tx = provider.tx();
+        let result = tx.get::<tables::StoragesHistory>(key).unwrap();
+        assert_eq!(result, Some(value));
+
+        // Test missing key
+        let missing_key = StorageShardedKey::new(Address::random(), B256::ZERO, 0);
+        assert_eq!(tx.get::<tables::StoragesHistory>(missing_key).unwrap(), None);
+    }
+
+    #[test]
+    fn test_rocksdb_batch_account_history() {
+        let (_temp_dir, provider) = create_rocksdb_provider();
+
+        let address = Address::random();
+        let key = ShardedKey::new(address, 1000);
+        let value = IntegerList::new([1, 10, 100, 500]).unwrap();
+
+        // Write via RocksDBBatch
+        let mut batch = provider.batch();
+        batch.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
+        batch.commit().unwrap();
+
+        // Read via RocksTx
+        let tx = provider.tx();
+        let result = tx.get::<tables::AccountsHistory>(key).unwrap();
+        assert_eq!(result, Some(value));
+
+        // Test missing key
+        let missing_key = ShardedKey::new(Address::random(), 0);
+        assert_eq!(tx.get::<tables::AccountsHistory>(missing_key).unwrap(), None);
+    }
+
+    #[test]
+    fn test_rocksdb_batch_delete_transaction_hash_number() {
+        let (_temp_dir, provider) = create_rocksdb_provider();
+
+        let hash = B256::from([1u8; 32]);
+        let tx_num = 100u64;
+
+        // First write
+        provider.put::<tables::TransactionHashNumbers>(hash, &tx_num).unwrap();
+        assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), Some(tx_num));
+
+        // Delete via RocksDBBatch
+        let mut batch = provider.batch();
+        batch.delete::<tables::TransactionHashNumbers>(hash).unwrap();
+        batch.commit().unwrap();
+
+        // Verify deletion
+        assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), None);
+    }
+
+    #[test]
+    fn test_rocksdb_batch_delete_storage_history() {
+        let (_temp_dir, provider) = create_rocksdb_provider();
+
+        let address = Address::random();
+        let storage_key = B256::from([1u8; 32]);
+        let key = StorageShardedKey::new(address, storage_key, 1000);
+        let value = IntegerList::new([1, 5, 10]).unwrap();
+
+        // First write
+        provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
+        assert!(provider.get::<tables::StoragesHistory>(key.clone()).unwrap().is_some());
+
+        // Delete via RocksDBBatch
+        let mut batch = provider.batch();
+        batch.delete::<tables::StoragesHistory>(key.clone()).unwrap();
+        batch.commit().unwrap();
+
+        // Verify deletion
+        assert_eq!(provider.get::<tables::StoragesHistory>(key).unwrap(), None);
+    }
+
+    #[test]
+    fn test_rocksdb_batch_delete_account_history() {
+        let (_temp_dir, provider) = create_rocksdb_provider();
+
+        let address = Address::random();
+        let key = ShardedKey::new(address, 1000);
+        let value = IntegerList::new([1, 10, 100]).unwrap();
+
+        // First write
+        provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
+        assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
+
+        // Delete via RocksDBBatch
+        let mut batch = provider.batch();
+        batch.delete::<tables::AccountsHistory>(key.clone()).unwrap();
+        batch.commit().unwrap();
+
+        // Verify deletion
+        assert_eq!(provider.get::<tables::AccountsHistory>(key).unwrap(), None);
+    }
+}
--- a/crates/storage/provider/src/providers/blockchain_provider.rs
+++ b/crates/storage/provider/src/providers/blockchain_provider.rs
@@ -1,14 +1,15 @@
 use crate::{
    providers::{
-        ConsistentProvider, ProviderNodeTypes, StaticFileProvider, StaticFileProviderRWRefMut,
+        ConsistentProvider, ProviderNodeTypes, RocksDBProvider, StaticFileProvider,
+        StaticFileProviderRWRefMut,
    },
    AccountReader, BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
    BlockSource, CanonChainTracker, CanonStateNotifications, CanonStateSubscriptions,
    ChainSpecProvider, ChainStateBlockReader, ChangeSetReader, DatabaseProviderFactory,
    HashedPostStateProvider, HeaderProvider, ProviderError, ProviderFactory, PruneCheckpointReader,
-    ReceiptProvider, ReceiptProviderIdExt, StageCheckpointReader, StateProviderBox,
-    StateProviderFactory, StateReader, StaticFileProviderFactory, TransactionVariant,
-    TransactionsProvider, TrieReader,
+    ReceiptProvider, ReceiptProviderIdExt, RocksDBProviderFactory, StageCheckpointReader,
+    StateProviderBox, StateProviderFactory, StateReader, StaticFileProviderFactory,
+    TransactionVariant, TransactionsProvider, TrieReader,
 };
 use alloy_consensus::transaction::TransactionMeta;
 use alloy_eips::{BlockHashOrNumber, BlockId, BlockNumHash, BlockNumberOrTag};
@@ -176,6 +177,12 @@ impl<N: ProviderNodeTypes> StaticFileProviderFactory for BlockchainProvider<N> {
    }
 }

+impl<N: ProviderNodeTypes> RocksDBProviderFactory for BlockchainProvider<N> {
+    fn rocksdb_provider(&self) -> RocksDBProvider {
+        self.database.rocksdb_provider()
+    }
+}
+
 impl<N: ProviderNodeTypes> HeaderProvider for BlockchainProvider<N> {
    type Header = HeaderTy<N>;

--- a/crates/storage/provider/src/providers/database/builder.rs
+++ b/crates/storage/provider/src/providers/database/builder.rs
@@ -4,7 +4,7 @@
 //! up to the intended build target.

 use crate::{
-    providers::{NodeTypesForProvider, StaticFileProvider},
+    providers::{NodeTypesForProvider, RocksDBProvider, StaticFileProvider},
    ProviderFactory,
 };
 use reth_db::{
@@ -104,11 +104,12 @@ impl<N> ProviderFactoryBuilder<N> {
    where
        N: NodeTypesForProvider,
    {
-        let ReadOnlyConfig { db_dir, db_args, static_files_dir, watch_static_files } =
+        let ReadOnlyConfig { db_dir, db_args, static_files_dir, rocksdb_dir, watch_static_files } =
            config.into();
        self.db(Arc::new(open_db_read_only(db_dir, db_args)?))
            .chainspec(chainspec)
            .static_file(StaticFileProvider::read_only(static_files_dir, watch_static_files)?)
+            .rocksdb_provider(RocksDBProvider::builder(&rocksdb_dir).with_default_tables().build()?)
            .build_provider_factory()
            .map_err(Into::into)
    }
@@ -120,7 +121,7 @@ impl<N> Default for ProviderFactoryBuilder<N> {
    }
 }

-/// Settings for how to open the database and static files.
+/// Settings for how to open the database, static files, and `RocksDB`.
 ///
 /// The default derivation from a path assumes the path is the datadir:
 /// [`ReadOnlyConfig::from_datadir`]
@@ -132,6 +133,8 @@ pub struct ReadOnlyConfig {
    pub db_args: DatabaseArguments,
    /// The path to the static file dir
    pub static_files_dir: PathBuf,
+    /// The path to the `RocksDB` directory
+    pub rocksdb_dir: PathBuf,
    /// Whether the static files should be watched for changes.
    pub watch_static_files: bool,
 }
@@ -144,6 +147,7 @@ impl ReadOnlyConfig {
    /// ```text
    ///  -`datadir`
    ///    |__db
+    ///    |__rocksdb
    ///    |__static_files
    /// ```
    ///
@@ -151,7 +155,13 @@ impl ReadOnlyConfig {
    /// [`StaticFileProvider::read_only`]
    pub fn from_datadir(datadir: impl AsRef<Path>) -> Self {
        let datadir = datadir.as_ref();
-        Self::from_dirs(datadir.join("db"), datadir.join("static_files"))
+        Self {
+            db_dir: datadir.join("db"),
+            db_args: Default::default(),
+            static_files_dir: datadir.join("static_files"),
+            rocksdb_dir: datadir.join("rocksdb"),
+            watch_static_files: true,
+        }
    }

    /// Disables long-lived read transaction safety guarantees.
@@ -169,7 +179,8 @@ impl ReadOnlyConfig {
    ///
    /// ```text
    ///    - db
-    ///    -static_files
+    ///    - rocksdb
+    ///    - static_files
    /// ```
    ///
    /// By default this watches the static file directory for changes, see also
@@ -180,13 +191,10 @@ impl ReadOnlyConfig {
    /// If the path does not exist
    pub fn from_db_dir(db_dir: impl AsRef<Path>) -> Self {
        let db_dir = db_dir.as_ref();
-        let static_files_dir = std::fs::canonicalize(db_dir)
-            .unwrap()
-            .parent()
-            .unwrap()
-            .to_path_buf()
-            .join("static_files");
-        Self::from_dirs(db_dir, static_files_dir)
+        let datadir = std::fs::canonicalize(db_dir).unwrap().parent().unwrap().to_path_buf();
+        let static_files_dir = datadir.join("static_files");
+        let rocksdb_dir = datadir.join("rocksdb");
+        Self::from_dirs(db_dir, static_files_dir, rocksdb_dir)
    }

    /// Creates the config for the given paths.
@@ -194,11 +202,16 @@ impl ReadOnlyConfig {
    ///
    /// By default this watches the static file directory for changes, see also
    /// [`StaticFileProvider::read_only`]
-    pub fn from_dirs(db_dir: impl AsRef<Path>, static_files_dir: impl AsRef<Path>) -> Self {
+    pub fn from_dirs(
+        db_dir: impl AsRef<Path>,
+        static_files_dir: impl AsRef<Path>,
+        rocksdb_dir: impl AsRef<Path>,
+    ) -> Self {
        Self {
-            static_files_dir: static_files_dir.as_ref().into(),
            db_dir: db_dir.as_ref().into(),
            db_args: Default::default(),
+            static_files_dir: static_files_dir.as_ref().into(),
+            rocksdb_dir: rocksdb_dir.as_ref().into(),
            watch_static_files: true,
        }
    }
@@ -317,7 +330,37 @@ impl<N, Val1, Val2, Val3> TypesAnd3<N, Val1, Val2, Val3> {
    }
 }

-impl<N, DB> TypesAnd3<N, DB, Arc<N::ChainSpec>, StaticFileProvider<N::Primitives>>
+impl<N, DB, C> TypesAnd3<N, DB, Arc<C>, StaticFileProvider<N::Primitives>>
+where
+    N: NodeTypes,
+{
+    /// Configures the `RocksDB` provider.
+    pub fn rocksdb_provider(
+        self,
+        rocksdb_provider: RocksDBProvider,
+    ) -> TypesAnd4<N, DB, Arc<C>, StaticFileProvider<N::Primitives>, RocksDBProvider> {
+        TypesAnd4::new(self.val_1, self.val_2, self.val_3, rocksdb_provider)
+    }
+}
+
+/// This is staging type that contains the configured types and _four_ values.
+#[derive(Debug)]
+pub struct TypesAnd4<N, Val1, Val2, Val3, Val4> {
+    _types: PhantomData<N>,
+    val_1: Val1,
+    val_2: Val2,
+    val_3: Val3,
+    val_4: Val4,
+}
+
+impl<N, Val1, Val2, Val3, Val4> TypesAnd4<N, Val1, Val2, Val3, Val4> {
+    /// Creates a new instance with the given types and four values.
+    pub fn new(val_1: Val1, val_2: Val2, val_3: Val3, val_4: Val4) -> Self {
+        Self { _types: Default::default(), val_1, val_2, val_3, val_4 }
+    }
+}
+
+impl<N, DB> TypesAnd4<N, DB, Arc<N::ChainSpec>, StaticFileProvider<N::Primitives>, RocksDBProvider>
 where
    N: NodeTypesForProvider,
    DB: Database + DatabaseMetrics + Clone + Unpin + 'static,
@@ -326,7 +369,7 @@ where
    pub fn build_provider_factory(
        self,
    ) -> ProviderResult<ProviderFactory<NodeTypesWithDBAdapter<N, DB>>> {
-        let Self { _types, val_1, val_2, val_3 } = self;
-        ProviderFactory::new(val_1, val_2, val_3)
+        let Self { _types, val_1, val_2, val_3, val_4 } = self;
+        ProviderFactory::new(val_1, val_2, val_3, val_4)
    }
 }
--- a/crates/storage/provider/src/providers/database/mod.rs
+++ b/crates/storage/provider/src/providers/database/mod.rs
@@ -1,15 +1,15 @@
 use crate::{
    providers::{
-        state::latest::LatestStateProvider, NodeTypesForProvider, StaticFileProvider,
-        StaticFileProviderRWRefMut,
+        state::latest::LatestStateProvider, NodeTypesForProvider, RocksDBProvider,
+        StaticFileProvider, StaticFileProviderRWRefMut,
    },
    to_range,
    traits::{BlockSource, ReceiptProvider},
    BlockHashReader, BlockNumReader, BlockReader, ChainSpecProvider, DatabaseProviderFactory,
    EitherWriterDestination, HashedPostStateProvider, HeaderProvider, HeaderSyncGapProvider,
-    MetadataProvider, ProviderError, PruneCheckpointReader, StageCheckpointReader,
-    StateProviderBox, StaticFileProviderFactory, StaticFileWriter, TransactionVariant,
-    TransactionsProvider,
+    MetadataProvider, ProviderError, PruneCheckpointReader, RocksDBProviderFactory,
+    StageCheckpointReader, StateProviderBox, StaticFileProviderFactory, StaticFileWriter,
+    TransactionVariant, TransactionsProvider,
 };
 use alloy_consensus::transaction::TransactionMeta;
 use alloy_eips::BlockHashOrNumber;
@@ -72,6 +72,8 @@ pub struct ProviderFactory<N: NodeTypesWithDB> {
    storage: Arc<N::Storage>,
    /// Storage configuration settings for this node
    storage_settings: Arc<RwLock<StorageSettings>>,
+    /// `RocksDB` provider
+    rocksdb_provider: RocksDBProvider,
 }

 impl<N: NodeTypesForProvider> ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>> {
@@ -87,6 +89,7 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
        db: N::DB,
        chain_spec: Arc<N::ChainSpec>,
        static_file_provider: StaticFileProvider<N::Primitives>,
+        rocksdb_provider: RocksDBProvider,
    ) -> ProviderResult<Self> {
        // Load storage settings from database at init time. Creates a temporary provider
        // to read persisted settings, falling back to legacy defaults if none exist.
@@ -100,6 +103,7 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
            Default::default(),
            Default::default(),
            Arc::new(RwLock::new(legacy_settings)),
+            rocksdb_provider.clone(),
        )
        .storage_settings()?
        .unwrap_or(legacy_settings);
@@ -111,6 +115,7 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
            prune_modes: PruneModes::default(),
            storage: Default::default(),
            storage_settings: Arc::new(RwLock::new(storage_settings)),
+            rocksdb_provider,
        })
    }
 }
@@ -144,6 +149,12 @@ impl<N: NodeTypesWithDB> StorageSettingsCache for ProviderFactory<N> {
    }
 }

+impl<N: NodeTypesWithDB> RocksDBProviderFactory for ProviderFactory<N> {
+    fn rocksdb_provider(&self) -> RocksDBProvider {
+        self.rocksdb_provider.clone()
+    }
+}
+
 impl<N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>> ProviderFactory<N> {
    /// Create new database provider by passing a path. [`ProviderFactory`] will own the database
    /// instance.
@@ -152,11 +163,13 @@ impl<N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>> ProviderFactory<N> {
        chain_spec: Arc<N::ChainSpec>,
        args: DatabaseArguments,
        static_file_provider: StaticFileProvider<N::Primitives>,
+        rocksdb_provider: RocksDBProvider,
    ) -> RethResult<Self> {
        Self::new(
            Arc::new(init_db(path, args).map_err(RethError::msg)?),
            chain_spec,
            static_file_provider,
+            rocksdb_provider,
        )
        .map_err(RethError::Provider)
    }
@@ -178,6 +191,7 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
            self.prune_modes.clone(),
            self.storage.clone(),
            self.storage_settings.clone(),
+            self.rocksdb_provider.clone(),
        ))
    }

@@ -194,6 +208,7 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
            self.prune_modes.clone(),
            self.storage.clone(),
            self.storage_settings.clone(),
+            self.rocksdb_provider.clone(),
        )))
    }

@@ -595,8 +610,15 @@ where
    N: NodeTypesWithDB<DB: fmt::Debug, ChainSpec: fmt::Debug, Storage: fmt::Debug>,
 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let Self { db, chain_spec, static_file_provider, prune_modes, storage, storage_settings } =
-            self;
+        let Self {
+            db,
+            chain_spec,
+            static_file_provider,
+            prune_modes,
+            storage,
+            storage_settings,
+            rocksdb_provider,
+        } = self;
        f.debug_struct("ProviderFactory")
            .field("db", &db)
            .field("chain_spec", &chain_spec)
@@ -604,6 +626,7 @@ where
            .field("prune_modes", &prune_modes)
            .field("storage", &storage)
            .field("storage_settings", &*storage_settings.read())
+            .field("rocksdb_provider", &rocksdb_provider)
            .finish()
    }
 }
@@ -617,6 +640,7 @@ impl<N: NodeTypesWithDB> Clone for ProviderFactory<N> {
            prune_modes: self.prune_modes.clone(),
            storage: self.storage.clone(),
            storage_settings: self.storage_settings.clone(),
+            rocksdb_provider: self.rocksdb_provider.clone(),
        }
    }
 }
@@ -635,7 +659,7 @@ mod tests {
    use reth_chainspec::ChainSpecBuilder;
    use reth_db::{
        mdbx::DatabaseArguments,
-        test_utils::{create_test_static_files_dir, ERROR_TEMPDIR},
+        test_utils::{create_test_rocksdb_dir, create_test_static_files_dir, ERROR_TEMPDIR},
    };
    use reth_db_api::tables;
    use reth_primitives_traits::SignerRecoverable;
@@ -674,11 +698,13 @@ mod tests {
    fn provider_factory_with_database_path() {
        let chain_spec = ChainSpecBuilder::mainnet().build();
        let (_static_dir, static_dir_path) = create_test_static_files_dir();
+        let (_, rocksdb_path) = create_test_rocksdb_dir();
        let factory = ProviderFactory::<MockNodeTypesWithDB<DatabaseEnv>>::new_with_database_path(
            tempfile::TempDir::new().expect(ERROR_TEMPDIR).keep(),
            Arc::new(chain_spec),
            DatabaseArguments::new(Default::default()),
            StaticFileProvider::read_write(static_dir_path).unwrap(),
+            RocksDBProvider::builder(&rocksdb_path).build().unwrap(),
        )
        .unwrap();
        let provider = factory.provider().unwrap();
--- a/crates/storage/provider/src/providers/database/provider.rs
+++ b/crates/storage/provider/src/providers/database/provider.rs
@@ -4,6 +4,7 @@ use crate::{
    },
    providers::{
        database::{chain::ChainStorage, metrics},
+        rocksdb::RocksDBProvider,
        static_file::StaticFileWriter,
        NodeTypesForProvider, StaticFileProvider,
    },
@@ -16,10 +17,10 @@ use crate::{
    DBProvider, EitherReader, EitherWriter, EitherWriterDestination, HashingWriter, HeaderProvider,
    HeaderSyncGapProvider, HistoricalStateProvider, HistoricalStateProviderRef, HistoryWriter,
    LatestStateProvider, LatestStateProviderRef, OriginalValuesKnown, ProviderError,
-    PruneCheckpointReader, PruneCheckpointWriter, RevertsInit, StageCheckpointReader,
-    StateProviderBox, StateWriter, StaticFileProviderFactory, StatsReader, StorageReader,
-    StorageTrieWriter, TransactionVariant, TransactionsProvider, TransactionsProviderExt,
-    TrieReader, TrieWriter,
+    PruneCheckpointReader, PruneCheckpointWriter, RevertsInit, RocksDBProviderFactory,
+    StageCheckpointReader, StateProviderBox, StateWriter, StaticFileProviderFactory, StatsReader,
+    StorageReader, StorageTrieWriter, TransactionVariant, TransactionsProvider,
+    TransactionsProviderExt, TrieReader, TrieWriter,
 };
 use alloy_consensus::{
    transaction::{SignerRecoverable, TransactionMeta, TxHashRef},
@@ -164,6 +165,8 @@ pub struct DatabaseProvider<TX, N: NodeTypes> {
    storage: Arc<N::Storage>,
    /// Storage configuration settings for this node
    storage_settings: Arc<RwLock<StorageSettings>>,
+    /// `RocksDB` provider
+    rocksdb_provider: RocksDBProvider,
    /// Minimum distance from tip required for pruning
    minimum_pruning_distance: u64,
 }
@@ -251,6 +254,13 @@ impl<TX, N: NodeTypes> StaticFileProviderFactory for DatabaseProvider<TX, N> {
    }
 }

+impl<TX, N: NodeTypes> RocksDBProviderFactory for DatabaseProvider<TX, N> {
+    /// Returns the `RocksDB` provider.
+    fn rocksdb_provider(&self) -> RocksDBProvider {
+        self.rocksdb_provider.clone()
+    }
+}
+
 impl<TX: Debug + Send + Sync, N: NodeTypes<ChainSpec: EthChainSpec + 'static>> ChainSpecProvider
    for DatabaseProvider<TX, N>
 {
@@ -270,6 +280,7 @@ impl<TX: DbTxMut, N: NodeTypes> DatabaseProvider<TX, N> {
        prune_modes: PruneModes,
        storage: Arc<N::Storage>,
        storage_settings: Arc<RwLock<StorageSettings>>,
+        rocksdb_provider: RocksDBProvider,
    ) -> Self {
        Self {
            tx,
@@ -278,6 +289,7 @@ impl<TX: DbTxMut, N: NodeTypes> DatabaseProvider<TX, N> {
            prune_modes,
            storage,
            storage_settings,
+            rocksdb_provider,
            minimum_pruning_distance: MINIMUM_PRUNING_DISTANCE,
        }
    }
@@ -523,6 +535,7 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
        prune_modes: PruneModes,
        storage: Arc<N::Storage>,
        storage_settings: Arc<RwLock<StorageSettings>>,
+        rocksdb_provider: RocksDBProvider,
    ) -> Self {
        Self {
            tx,
@@ -531,6 +544,7 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
            prune_modes,
            storage,
            storage_settings,
+            rocksdb_provider,
            minimum_pruning_distance: MINIMUM_PRUNING_DISTANCE,
        }
    }
--- a/crates/storage/provider/src/providers/mod.rs
+++ b/crates/storage/provider/src/providers/mod.rs
@@ -31,10 +31,11 @@ pub use consistent::ConsistentProvider;

 // RocksDB currently only supported on Unix platforms
 // Windows support is planned for future releases
-#[cfg(all(unix, feature = "rocksdb"))]
+#[cfg_attr(all(unix, feature = "rocksdb"), path = "rocksdb/mod.rs")]
+#[cfg_attr(not(all(unix, feature = "rocksdb")), path = "rocksdb_stub.rs")]
 pub(crate) mod rocksdb;
-#[cfg(all(unix, feature = "rocksdb"))]
-pub use rocksdb::{RocksDBBuilder, RocksDBProvider, RocksDBWriteMode, RocksTx};
+
+pub use rocksdb::{RocksDBBatch, RocksDBBuilder, RocksDBProvider, RocksTx};

 /// Helper trait to bound [`NodeTypes`] so that combined with database they satisfy
 /// [`ProviderNodeTypes`].
--- a/crates/storage/provider/src/providers/rocksdb/mod.rs
+++ b/crates/storage/provider/src/providers/rocksdb/mod.rs
@@ -2,7 +2,4 @@

 mod metrics;
 mod provider;
-pub use provider::{RocksDBBuilder, RocksDBProvider, RocksDBWriteMode, RocksTx};
-
-#[cfg(test)]
-pub(crate) use provider::RocksDBBatch;
+pub use provider::{RocksDBBatch, RocksDBBuilder, RocksDBProvider, RocksTx};
--- a/crates/storage/provider/src/providers/rocksdb/provider.rs
+++ b/crates/storage/provider/src/providers/rocksdb/provider.rs
@@ -1,7 +1,7 @@
 use super::metrics::{RocksDBMetrics, RocksDBOperation};
 use reth_db_api::{
    table::{Compress, Decompress, Encode, Table},
-    DatabaseError,
+    tables, DatabaseError,
 };
 use reth_storage_errors::{
    db::{DatabaseErrorInfo, DatabaseWriteError, DatabaseWriteOperation, LogLevel},
@@ -18,7 +18,6 @@ use std::{
    sync::Arc,
    time::Instant,
 };
-use tracing::warn;

 /// Default cache size for `RocksDB` block cache (128 MB).
 const DEFAULT_CACHE_SIZE: usize = 128 << 20;
@@ -35,6 +34,11 @@ const DEFAULT_BYTES_PER_SYNC: u64 = 1_048_576;
 /// Default bloom filter bits per key (~1% false positive rate).
 const DEFAULT_BLOOM_FILTER_BITS: f64 = 10.0;

+/// Default buffer capacity for compression in batches.
+/// 4 KiB matches common block/page sizes and comfortably holds typical history values,
+/// reducing the first few reallocations without over-allocating.
+const DEFAULT_COMPRESS_BUF_CAPACITY: usize = 4096;
+
 /// Builder for [`RocksDBProvider`].
 pub struct RocksDBBuilder {
    path: PathBuf,
@@ -139,6 +143,18 @@ impl RocksDBBuilder {
        self
    }

+    /// Registers the default tables used by reth for `RocksDB` storage.
+    ///
+    /// This registers:
+    /// - [`tables::TransactionHashNumbers`] - Transaction hash to number mapping
+    /// - [`tables::AccountsHistory`] - Account history index
+    /// - [`tables::StoragesHistory`] - Storage history index
+    pub fn with_default_tables(self) -> Self {
+        self.with_table::<tables::TransactionHashNumbers>()
+            .with_table::<tables::AccountsHistory>()
+            .with_table::<tables::StoragesHistory>()
+    }
+
    /// Enables metrics.
    pub const fn with_metrics(mut self) -> Self {
        self.enable_metrics = true;
@@ -152,8 +168,10 @@ impl RocksDBBuilder {
    }

    /// Sets the log level from `DatabaseArgs` configuration.
-    pub const fn with_database_log_level(mut self, log_level: LogLevel) -> Self {
-        self.log_level = convert_log_level(log_level);
+    pub const fn with_database_log_level(mut self, log_level: Option<LogLevel>) -> Self {
+        if let Some(level) = log_level {
+            self.log_level = convert_log_level(level);
+        }
        self
    }

@@ -260,23 +278,15 @@ impl RocksDBProvider {
        RocksTx { inner, provider: self }
    }

-    /// Creates a new batch for manual commit.
+    /// Creates a new batch for atomic writes.
    ///
    /// Use [`Self::write_batch`] for closure-based atomic writes.
-    /// Use this method when the batch needs to be held by [`EitherWriter`](crate::EitherWriter).
-    ///
-    /// # Example
-    ///
-    /// ```ignore
-    /// let batch = provider.batch();
-    /// batch.put::<SomeTable>(key, &value)?;
-    /// batch.commit()?;
-    /// ```
+    /// Use this method when the batch needs to be held by [`crate::EitherWriter`].
    pub fn batch(&self) -> RocksDBBatch<'_> {
        RocksDBBatch {
            provider: self,
            inner: WriteBatchWithTransaction::<true>::default(),
-            buf: Vec::new(),
+            buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
        }
    }

@@ -375,31 +385,21 @@ impl RocksDBProvider {
    where
        F: FnOnce(&mut RocksDBBatch<'_>) -> ProviderResult<()>,
    {
-        // Note: Using "Batch" as table name for batch operations across multiple tables
        self.execute_with_operation_metric(RocksDBOperation::BatchWrite, "Batch", |this| {
-            let mut batch_handle = RocksDBBatch {
-                provider: this,
-                inner: WriteBatchWithTransaction::<true>::default(),
-                buf: Vec::new(),
-            };
-
+            let mut batch_handle = this.batch();
            f(&mut batch_handle)?;
-
-            // Take ownership of inner to prevent Drop from logging a warning
-            let batch = std::mem::take(&mut batch_handle.inner);
-            this.0.db.write(batch).map_err(|e| {
-                ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
-                    message: e.to_string().into(),
-                    code: -1,
-                }))
-            })
+            batch_handle.commit()
        })
    }
 }

 /// Handle for building a batch of operations atomically.
 ///
-/// Uses `WriteBatchWithTransaction<true>` for compatibility with `TransactionDB`.
+/// Uses `WriteBatchWithTransaction` for atomic writes without full transaction overhead.
+/// Unlike [`RocksTx`], this does NOT support read-your-writes. Use for write-only flows
+/// where you don't need to read back uncommitted data within the same operation
+/// (e.g., history index writes).
+#[must_use = "batch must be committed"]
 pub struct RocksDBBatch<'a> {
    provider: &'a RocksDBProvider,
    inner: WriteBatchWithTransaction<true>,
@@ -447,14 +447,8 @@ impl<'a> RocksDBBatch<'a> {
    /// Commits the batch to the database.
    ///
    /// This consumes the batch and writes all operations atomically to `RocksDB`.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if the write fails.
-    pub fn commit(mut self) -> ProviderResult<()> {
-        // Take ownership of inner to prevent Drop from logging a warning
-        let batch = std::mem::take(&mut self.inner);
-        self.provider.0.db.write(batch).map_err(|e| {
+    pub fn commit(self) -> ProviderResult<()> {
+        self.provider.0.db.write_opt(self.inner, &WriteOptions::default()).map_err(|e| {
            ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
                message: e.to_string().into(),
                code: -1,
@@ -462,7 +456,7 @@ impl<'a> RocksDBBatch<'a> {
        })
    }

-    /// Returns the number of operations in this batch.
+    /// Returns the number of write operations (puts + deletes) queued in this batch.
    pub fn len(&self) -> usize {
        self.inner.len()
    }
@@ -473,18 +467,6 @@ impl<'a> RocksDBBatch<'a> {
    }
 }

-impl Drop for RocksDBBatch<'_> {
-    fn drop(&mut self) {
-        if !self.inner.is_empty() {
-            warn!(
-                target: "reth::storage",
-                batch_len = %self.inner.len(),
-                "RocksDBBatch dropped without commit - data discarded"
-            );
-        }
-    }
-}
-
 /// `RocksDB` transaction wrapper providing MDBX-like semantics.
 ///
 /// Supports:
@@ -646,49 +628,6 @@ impl<T: Table> Iterator for RocksTxIter<'_, T> {
    }
 }

-/// `RocksDB` write strategy - internal implementation detail.
-///
-/// This enum encapsulates the choice between full transaction semantics
-/// and high-throughput batch writes. Use [`RocksDBWriteMode::Transaction`] for
-/// read-modify-write operations that need read-your-writes semantics.
-/// Use [`RocksDBWriteMode::Batch`] for bulk sync operations where
-/// read-your-writes is not needed.
-pub enum RocksDBWriteMode<'a> {
-    /// Full transaction with read-your-writes, rollback support.
-    /// Use for read-modify-write operations.
-    Transaction(RocksTx<'a>),
-    /// Write-only batch for maximum throughput.
-    /// Use for bulk sync operations where read-your-writes is not needed.
-    Batch(RocksDBBatch<'a>),
-}
-
-impl fmt::Debug for RocksDBWriteMode<'_> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            Self::Transaction(tx) => f.debug_tuple("Transaction").field(tx).finish(),
-            Self::Batch(batch) => f.debug_tuple("Batch").field(batch).finish(),
-        }
-    }
-}
-
-impl<'a> RocksDBWriteMode<'a> {
-    /// Puts a value into the specified table.
-    pub fn put<T: Table>(&mut self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
-        match self {
-            Self::Transaction(tx) => tx.put::<T>(key, value),
-            Self::Batch(batch) => batch.put::<T>(key, value),
-        }
-    }
-
-    /// Commits the transaction or batch.
-    pub fn commit(self) -> ProviderResult<()> {
-        match self {
-            Self::Transaction(tx) => tx.commit(),
-            Self::Batch(batch) => batch.commit(),
-        }
-    }
-}
-
 /// Converts Reth's [`LogLevel`] to `RocksDB`'s [`rocksdb::LogLevel`].
 const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
    match level {
@@ -703,10 +642,38 @@ const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use alloy_primitives::{TxHash, B256};
-    use reth_db_api::{table::Table, tables};
+    use alloy_primitives::{Address, TxHash, B256};
+    use reth_db_api::{
+        models::{sharded_key::ShardedKey, storage_sharded_key::StorageShardedKey, IntegerList},
+        table::Table,
+        tables,
+    };
    use tempfile::TempDir;

+    #[test]
+    fn test_with_default_tables_registers_required_column_families() {
+        let temp_dir = TempDir::new().unwrap();
+
+        // Build with default tables
+        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
+
+        // Should be able to write/read TransactionHashNumbers
+        let tx_hash = TxHash::from(B256::from([1u8; 32]));
+        provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
+        assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
+
+        // Should be able to write/read AccountsHistory
+        let key = ShardedKey::new(Address::ZERO, 100);
+        let value = IntegerList::default();
+        provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
+        assert!(provider.get::<tables::AccountsHistory>(key).unwrap().is_some());
+
+        // Should be able to write/read StoragesHistory
+        let key = StorageShardedKey::new(Address::ZERO, B256::ZERO, 100);
+        provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
+        assert!(provider.get::<tables::StoragesHistory>(key).unwrap().is_some());
+    }
+
    #[derive(Debug)]
    struct TestTable;

@@ -975,167 +942,4 @@ mod tests {
            assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
        }
    }
-
-    #[test]
-    fn test_write_mode_batch() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        // Create write mode using Batch
-        let batch = provider.batch();
-        let mut mode = RocksDBWriteMode::Batch(batch);
-
-        // Write via RocksDBWriteMode
-        let key = 42u64;
-        let value = b"test_via_mode".to_vec();
-        mode.put::<TestTable>(key, &value).unwrap();
-
-        // Commit via RocksDBWriteMode
-        mode.commit().unwrap();
-
-        // Verify data is visible
-        assert_eq!(provider.get::<TestTable>(key).unwrap(), Some(value));
-    }
-
-    #[test]
-    fn test_write_mode_transaction() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        // Create write mode using Transaction
-        let tx = provider.tx();
-        let mut mode = RocksDBWriteMode::Transaction(tx);
-
-        // Write via RocksDBWriteMode
-        let key = 100u64;
-        let value = b"test_via_tx_mode".to_vec();
-        mode.put::<TestTable>(key, &value).unwrap();
-
-        // Commit via RocksDBWriteMode
-        mode.commit().unwrap();
-
-        // Verify data is visible
-        assert_eq!(provider.get::<TestTable>(key).unwrap(), Some(value));
-    }
-
-    #[test]
-    fn test_batch_empty_commit() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        // Create an empty batch
-        let batch = provider.batch();
-        assert!(batch.is_empty());
-        assert_eq!(batch.len(), 0);
-
-        // Commit should succeed (no-op)
-        batch.commit().unwrap();
-    }
-
-    #[test]
-    fn test_batch_drop_without_commit_does_not_persist() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        // Create a batch and add entries but DON'T commit
-        {
-            let mut batch = provider.batch();
-            for i in 0..5u64 {
-                let value = format!("dropped_value_{i}").into_bytes();
-                batch.put::<TestTable>(i, &value).unwrap();
-            }
-            // batch dropped here without commit - should log warning
-        }
-
-        // Data should NOT be persisted
-        for i in 0..5u64 {
-            assert_eq!(
-                provider.get::<TestTable>(i).unwrap(),
-                None,
-                "Dropped batch should not persist data"
-            );
-        }
-    }
-
-    #[test]
-    fn test_write_mode_debug_formatting() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        // Test Batch variant debug output
-        let batch = provider.batch();
-        let mode_batch = RocksDBWriteMode::Batch(batch);
-        let debug_str = format!("{:?}", mode_batch);
-        assert!(debug_str.contains("Batch"), "Debug should contain 'Batch': {debug_str}");
-
-        // Test Transaction variant debug output
-        let tx = provider.tx();
-        let mode_tx = RocksDBWriteMode::Transaction(tx);
-        let debug_str = format!("{:?}", mode_tx);
-        assert!(
-            debug_str.contains("Transaction"),
-            "Debug should contain 'Transaction': {debug_str}"
-        );
-    }
-
-    #[test]
-    fn test_batch_delete_operation() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        // First, insert some data
-        let mut batch = provider.batch();
-        for i in 0..5u64 {
-            batch.put::<TestTable>(i, &vec![i as u8]).unwrap();
-        }
-        batch.commit().unwrap();
-
-        // Verify data exists
-        for i in 0..5u64 {
-            assert!(provider.get::<TestTable>(i).unwrap().is_some());
-        }
-
-        // Now delete via batch
-        let mut delete_batch = provider.batch();
-        for i in 0..5u64 {
-            delete_batch.delete::<TestTable>(i).unwrap();
-        }
-        delete_batch.commit().unwrap();
-
-        // Verify data is deleted
-        for i in 0..5u64 {
-            assert_eq!(provider.get::<TestTable>(i).unwrap(), None);
-        }
-    }
-
-    #[test]
-    fn test_batch_overwrite_existing_key() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider =
-            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
-
-        let key = 42u64;
-        let initial_value = b"initial".to_vec();
-        let updated_value = b"updated".to_vec();
-
-        // Insert initial value
-        let mut batch = provider.batch();
-        batch.put::<TestTable>(key, &initial_value).unwrap();
-        batch.commit().unwrap();
-
-        assert_eq!(provider.get::<TestTable>(key).unwrap(), Some(initial_value));
-
-        // Overwrite with new value
-        let mut batch = provider.batch();
-        batch.put::<TestTable>(key, &updated_value).unwrap();
-        batch.commit().unwrap();
-
-        assert_eq!(provider.get::<TestTable>(key).unwrap(), Some(updated_value));
-    }
 }
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Matthias Seitz	5c04d1abe1	fix: allow smaller header size	2025-12-16 17:08:53 +01:00
bigbear	4231f4b688	docs: fix incorrect API example in node-components.mdx (#20297 )	2025-12-16 15:09:29 +00:00
Léa Narzis	0b607113dc	refactor(era): make era count in era file name optional (#20292 )	2025-12-16 15:08:43 +00:00
emmmm	be4dc53b92	docs: fix --color auto option description (#20352 )	2025-12-16 15:06:04 +00:00
emmmm	4afb555d06	docs(opstack): document all rollup CLI arguments (#20374 )	2025-12-16 15:04:34 +00:00
Matthias Seitz	ab2ef99458	chore: add keccak-global (#20418 )	2025-12-16 14:59:09 +00:00
Sophia Raye	bfd4b79245	docs(trace): remove duplicate comment (#20360 )	2025-12-16 14:56:01 +00:00
Federico Gimenez	49057b1c0c	feat(storage): add with_default_tables() to register RocksDB column families at initialization (#20416 )	2025-12-16 12:59:58 +00:00
Gigi	b6772370d7	docs: fix incorrect method reference in try_recover_sealed_with_senders (#20410 )	2025-12-16 12:27:53 +00:00
Karl Yu	d72935628a	feat: add support for eth/70 eip-7975 (#20255 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-16 12:05:11 +00:00
YK	ad63b135d6	feat(storage): implement EitherWriter/EitherReader methods for RocksDB (#20408 )	2025-12-16 11:26:31 +00:00
Brian Picciano	90651ae8e8	feat(engine): Use BAL in state root validation (#20383 )	2025-12-16 11:05:51 +00:00
Matthias Seitz	bbd51862d4	chore: rm flaky bench (#20413 )	2025-12-16 09:35:38 +00:00
Arsenii Kulikov	08a16a5bde	perf: recover transactions in parallel during network import (#20385 )	2025-12-16 09:33:24 +00:00
Snezhkko	f2c39db7a2	chore(rpc): fix misleading link and comment (#20367 )	2025-12-16 09:32:25 +00:00
oooLowNeoNooo	ae9e84d6e3	fix(discv4): correct ping_interval default value in docs (#20396 )	2025-12-16 09:29:45 +00:00
theo	c51da593d1	feat(net/p2p): support fixed external addresses with DNS resolution (#20411 )	2025-12-16 09:28:31 +00:00
Matthias Seitz	0e08f9f56c	perf: remove unnecessary channels from parallel trie operations (#20406 )	2025-12-16 09:15:27 +00:00
sashass1315	7eef092110	docs(exex): sync hello-world notifications loop with code (#20403 )	2025-12-16 08:39:45 +00:00
YK	40e8241bf5	feat(storage): use RocksDBBatch in EitherWriter and related modules (#20377 )	2025-12-16 03:57:41 +00:00
dependabot[bot]	dd9ff731e4	chore(deps): bump peter-evans/create-pull-request from 7 to 8 (#20402 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-16 00:11:22 +00:00
dependabot[bot]	83f9d1837f	chore(deps): bump actions/download-artifact from 4 to 7 (#20401 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-16 00:11:00 +00:00
dependabot[bot]	68911e617b	chore(deps): bump actions/upload-artifact from 5 to 6 (#20400 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-16 00:10:40 +00:00
0xcharry	36ba6db029	chore: remove redundant .as_str() calls after to_string() (#20404 )	2025-12-16 00:10:03 +00:00
Matthias Seitz	fec4432d82	perf: defer transaction pool notifications until after lock release (#20405 )	2025-12-15 23:06:34 +00:00
Matthias Seitz	179da26305	perf: use RwLock for transaction pool listeners (#20398 ) Co-authored-by: Arsenii Kulikov <klkvrr@gmail.com>	2025-12-15 21:47:59 +00:00
Matthias Seitz	b5e7a694d2	chore: update metric once (#20371 )	2025-12-15 20:38:24 +00:00
Maxim Evtush	9489667814	fix: post-state generator to include deletions in proptest (#20276 )	2025-12-15 16:43:02 +00:00
gustavo	004877ba59	refactor(cli): cleanup repair-trie metrics (#20226 )	2025-12-15 16:41:48 +00:00
Brian Picciano	a9e36923e1	feat(trie): Proof Rewrite: Use cached branch nodes (#20075 ) Co-authored-by: YK <chiayongkang@hotmail.com> Co-authored-by: Alexey Shekhirin <5773434+shekhirin@users.noreply.github.com>	2025-12-15 15:27:04 +00:00
DaniPopes	74a3816611	ci: reduce feature powerset depth (#20379 )	2025-12-15 14:42:14 +00:00
Alexey Shekhirin	5576d4547f	revert: feat(engine): run sync state root if not enough parallelism (#20127 ) (#20378 )	2025-12-15 14:05:54 +00:00
DaniPopes	21216e2f24	perf: use indexed parallel iterators for tx recovery (#20342 )	2025-12-15 13:40:03 +00:00
YK	42c1e1afe1	feat(storage): add account history constructors to EitherWriter/EitherReader (#20366 )	2025-12-15 12:45:07 +00:00
MoNyAvA	5f7e87fa2a	docs: add blob sub-pool to tx pool docs (#20375 )	2025-12-15 12:27:54 +00:00
Matthias Seitz	1b417dacc4	chore: sanity check for u64::Max (#20373 )	2025-12-15 11:33:50 +00:00
Niven	bb952be5b5	feat(flashblocks): support eth_getBlockTransactionCount for flashblocks (#20291 ) Co-authored-by: lucas <66681646+limyeechern@users.noreply.github.com> Co-authored-by: lucas.lim <lucas.lim@okg.com>	2025-12-15 11:29:23 +00:00
Federico Magnani	f927eec880	chore: export `FlashBlockDecoder` (#20370 )	2025-12-15 11:00:46 +00:00
Tomass	9c61f5568c	fix(rpc-testing-util): use buffer_unordered in trace_block_opcode_gas_unordered (#20369 )	2025-12-15 10:38:40 +00:00
ligt	662c0486a1	feat(storage): add rocksdb provider into database provider (#20253 )	2025-12-15 10:15:57 +00:00
Matthias Seitz	997848c2a1	fix(txpool): remove stale senderinfo (#20368 )	2025-12-15 10:00:25 +00:00
Olexandr88	155bdecf3b	docs(repo): add Ethereum-specific crates section (#20363 )	2025-12-15 09:56:40 +00:00