refactor(storage): introduce RocksDBWriteMode for flexible write strategies

- Replaced RocksTx with RocksDBWriteMode in EitherWriter to support both transaction and batch writes. - Updated constructors and methods to utilize RocksDBWriteMode, enhancing write flexibility. - Added batch handling capabilities in RocksDBProvider, allowing manual commit of batch operations. - Improved documentation for clarity on usage and commit behavior.
2026-04-30 03:01:58 -04:00 · 2025-12-15 05:10:02 +00:00
300 changed files with 2984 additions and 8773 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,keccak-cache-global",
+        "--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",
        # 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@v6
+        uses: actions/upload-artifact@v5
        with:
          name: hive_assets
          path: ./hive_assets
@@ -187,13 +187,13 @@ jobs:
          fetch-depth: 0

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

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

  # Checks that selected crates can compile with power set of features
  features:
-    name: features
+    name: features (${{ matrix.partition }}/${{ matrix.total_partitions }})
    runs-on: depot-ubuntu-latest
+    strategy:
+      matrix:
+        partition: [1, 2]
+        total_partitions: [2]
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v6
@@ -264,7 +268,7 @@ jobs:
            --package reth-primitives-traits \
            --package reth-primitives \
            --feature-powerset \
-            --depth 2
+            --partition ${{ matrix.partition }}/${{ matrix.total_partitions }}
        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@v6
+        uses: actions/upload-artifact@v5
        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@v6
+        uses: actions/upload-artifact@v5
        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@v6
+        uses: actions/upload-artifact@v5
        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@v7
+        uses: actions/download-artifact@v6
      - 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@v6
+        uses: actions/upload-artifact@v5
        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@v7
+        uses: actions/download-artifact@v4
        with:
          name: checksum-machine-1
          path: machine-1/
      - name: Download artifacts from machine-2
-        uses: actions/download-artifact@v7
+        uses: actions/download-artifact@v4
        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@v8
+        uses: peter-evans/create-pull-request@v7
        with:
          commit-message: "chore: update superchain config"
          title: "chore: update superchain config"
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2,15 +2,6 @@
 # It is not intended for manual editing.
 version = 4

-[[package]]
-name = "addr2line"
-version = "0.25.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b5d307320b3181d6d7954e663bd7c774a838b8220fe0593c86d9fb09f498b4b"
-dependencies = [
- "gimli",
-]
-
 [[package]]
 name = "adler2"
 version = "2.0.1"
@@ -186,9 +177,9 @@ dependencies = [

 [[package]]
 name = "alloy-dyn-abi"
-version = "1.5.1"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d48a9101f4a67c22fae57489f1ddf3057b8ab4a368d8eac3be088b6e9d9c9d9"
+checksum = "3fdff496dd4e98a81f4861e66f7eaf5f2488971848bb42d9c892f871730245c8"
 dependencies = [
 "alloy-json-abi",
 "alloy-primitives",
@@ -338,9 +329,9 @@ dependencies = [

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

 [[package]]
 name = "alloy-primitives"
-version = "1.5.1"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7db950a29746be9e2f2c6288c8bd7a6202a81f999ce109a2933d2379970ec0fa"
+checksum = "355bf68a433e0fd7f7d33d5a9fc2583fde70bf5c530f63b80845f8da5505cf28"
 dependencies = [
 "alloy-rlp",
 "arbitrary",
@@ -445,7 +436,6 @@ dependencies = [
 "cfg-if",
 "const-hex",
 "derive_more",
- "fixed-cache",
 "foldhash 0.2.0",
 "getrandom 0.3.4",
 "hashbrown 0.16.1",
@@ -457,7 +447,6 @@ dependencies = [
 "proptest",
 "proptest-derive 0.6.0",
 "rand 0.9.2",
- "rapidhash",
 "ruint",
 "rustc-hash",
 "serde",
@@ -792,9 +781,9 @@ dependencies = [

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

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

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

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

 [[package]]
 name = "alloy-sol-types"
-version = "1.5.1"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc1038284171df8bfd48befc0c7b78f667a7e2be162f45f07bd1c378078ebe58"
+checksum = "70319350969a3af119da6fb3e9bddb1bce66c9ea933600cb297c8b1850ad2a3c"
 dependencies = [
 "alloy-json-abi",
 "alloy-primitives",
@@ -1510,21 +1499,6 @@ dependencies = [
 "tokio",
 ]

-[[package]]
-name = "backtrace"
-version = "0.3.76"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb531853791a215d7c62a30daf0dde835f381ab5de4589cfe7c649d2cbe92bd6"
-dependencies = [
- "addr2line",
- "cfg-if",
- "libc",
- "miniz_oxide",
- "object",
- "rustc-demangle",
- "windows-link",
-]
-
 [[package]]
 name = "base-x"
 version = "0.2.11"
@@ -1964,9 +1938,9 @@ dependencies = [

 [[package]]
 name = "bumpalo"
-version = "3.19.1"
+version = "3.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510"
+checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43"

 [[package]]
 name = "byte-slice-cast"
@@ -2043,9 +2017,9 @@ dependencies = [

 [[package]]
 name = "camino"
-version = "1.2.2"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e629a66d692cb9ff1a1c664e41771b3dcaf961985a9774c0eb0bd1b51cf60a48"
+checksum = "276a59bf2b2c967788139340c9f0c5b12d7fd6630315c15c217e559de85d2609"
 dependencies = [
 "serde_core",
 ]
@@ -2156,7 +2130,7 @@ dependencies = [
 "num-traits",
 "serde",
 "wasm-bindgen",
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
@@ -2362,33 +2336,6 @@ dependencies = [
 "thiserror 1.0.69",
 ]

-[[package]]
-name = "color-eyre"
-version = "0.6.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e5920befb47832a6d61ee3a3a846565cfa39b331331e68a3b1d1116630f2f26d"
-dependencies = [
- "backtrace",
- "color-spantrace",
- "eyre",
- "indenter",
- "once_cell",
- "owo-colors",
- "tracing-error",
-]
-
-[[package]]
-name = "color-spantrace"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8b88ea9df13354b55bc7234ebcce36e6ef896aca2e42a15de9e10edce01b427"
-dependencies = [
- "once_cell",
- "owo-colors",
- "tracing-core",
- "tracing-error",
-]
-
 [[package]]
 name = "colorchoice"
 version = "1.0.4"
@@ -4038,15 +3985,6 @@ dependencies = [
 "windows-sys 0.60.2",
 ]

-[[package]]
-name = "fixed-cache"
-version = "0.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba59b6c98ba422a13f17ee1305c995cb5742bba7997f5b4d9af61b2ff0ffb213"
-dependencies = [
- "equivalent",
-]
-
 [[package]]
 name = "fixed-hash"
 version = "0.8.0"
@@ -4290,17 +4228,16 @@ checksum = "42012b0f064e01aa58b545fe3727f90f7dd4020f4a3ea735b50344965f5a57e9"

 [[package]]
 name = "generator"
-version = "0.8.8"
+version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52f04ae4152da20c76fe800fa48659201d5cf627c5149ca0b707b69d7eef6cf9"
+checksum = "605183a538e3e2a9c1038635cc5c2d194e2ee8fd0d1b66b8349fad7dbacce5a2"
 dependencies = [
 "cc",
 "cfg-if",
 "libc",
 "log",
 "rustversion",
- "windows-link",
- "windows-result 0.4.1",
+ "windows 0.61.3",
 ]

 [[package]]
@@ -4362,12 +4299,6 @@ dependencies = [
 "polyval",
 ]

-[[package]]
-name = "gimli"
-version = "0.32.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7"
-
 [[package]]
 name = "git2"
 version = "0.20.3"
@@ -5085,14 +5016,13 @@ dependencies = [

 [[package]]
 name = "insta"
-version = "1.45.0"
+version = "1.44.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b76866be74d68b1595eb8060cb9191dca9c021db2316558e52ddc5d55d41b66c"
+checksum = "b5c943d4415edd8153251b6f197de5eb1640e56d84e8d9159bea190421c73698"
 dependencies = [
 "console",
 "once_cell",
 "similar",
- "tempfile",
 ]

 [[package]]
@@ -5215,9 +5145,9 @@ dependencies = [

 [[package]]
 name = "itoa"
-version = "1.0.16"
+version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ee5b5339afb4c41626dde77b7a611bd4f2c202b897852b4bcf5d03eddc61010"
+checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c"

 [[package]]
 name = "jni"
@@ -5533,7 +5463,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55"
 dependencies = [
 "cfg-if",
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
@@ -5574,13 +5504,13 @@ dependencies = [

 [[package]]
 name = "libredox"
-version = "0.1.11"
+version = "0.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df15f6eac291ed1cf25865b1ee60399f57e7c227e7f51bdbd4c5270396a9ed50"
+checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb"
 dependencies = [
 "bitflags 2.10.0",
 "libc",
- "redox_syscall 0.6.0",
+ "redox_syscall",
 ]

 [[package]]
@@ -6089,9 +6019,9 @@ checksum = "5e0826a989adedc2a244799e823aece04662b66609d96af8dff7ac6df9a8925d"

 [[package]]
 name = "ntapi"
-version = "0.4.2"
+version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c70f219e21142367c70c0b30c6a9e3a14d55b4d12a204d897fbec83a0363f081"
+checksum = "e8a3895c6391c39d7fe7ebc444a87eb2991b2a0bc718fdabd071eec617fc68e4"
 dependencies = [
 "winapi",
 ]
@@ -6257,15 +6187,6 @@ version = "4.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33"

-[[package]]
-name = "object"
-version = "0.37.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe"
-dependencies = [
- "memchr",
-]
-
 [[package]]
 name = "once_cell"
 version = "1.21.3"
@@ -6547,12 +6468,6 @@ dependencies = [
 "num-traits",
 ]

-[[package]]
-name = "owo-colors"
-version = "4.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c6901729fa79e91a0913333229e9ca5dc725089d1c363b2f4b4760709dc4a52"
-
 [[package]]
 name = "p256"
 version = "0.13.2"
@@ -6629,9 +6544,9 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
 dependencies = [
 "cfg-if",
 "libc",
- "redox_syscall 0.5.18",
+ "redox_syscall",
 "smallvec",
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
@@ -6828,9 +6743,9 @@ dependencies = [

 [[package]]
 name = "portable-atomic"
-version = "1.12.0"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f59e70c4aef1e55797c2e8fd94a4f2a973fc972cfde0e0b05f683667b0cd39dd"
+checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483"

 [[package]]
 name = "potential_utf"
@@ -6902,7 +6817,7 @@ version = "3.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "219cb19e96be00ab2e37d6e299658a0cfa83e52429179969b0f0121b4ac46983"
 dependencies = [
- "toml_edit 0.23.10+spec-1.0.0",
+ "toml_edit 0.23.9",
 ]

 [[package]]
@@ -7294,16 +7209,6 @@ dependencies = [
 "rand_core 0.9.3",
 ]

-[[package]]
-name = "rapidhash"
-version = "4.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2988730ee014541157f48ce4dcc603940e00915edc3c7f9a8d78092256bb2493"
-dependencies = [
- "rand 0.9.2",
- "rustversion",
-]
-
 [[package]]
 name = "ratatui"
 version = "0.29.0"
@@ -7369,15 +7274,6 @@ dependencies = [
 "bitflags 2.10.0",
 ]

-[[package]]
-name = "redox_syscall"
-version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec96166dafa0886eb81fe1c0a388bece180fbef2135f97c1e2cf8302e74b43b5"
-dependencies = [
- "bitflags 2.10.0",
-]
-
 [[package]]
 name = "redox_users"
 version = "0.4.6"
@@ -7467,9 +7363,9 @@ checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2"

 [[package]]
 name = "reqwest"
-version = "0.12.26"
+version = "0.12.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b4c14b2d9afca6a60277086b0cc6a6ae0b568f6f7916c943a8cdc79f8be240f"
+checksum = "b6eff9328d40131d43bd911d42d79eb6a47312002a4daefc9e37f17e74a7701a"
 dependencies = [
 "base64 0.22.1",
 "bytes",
@@ -7601,7 +7497,6 @@ dependencies = [
 "alloy-transport-ws",
 "async-trait",
 "clap",
- "color-eyre",
 "csv",
 "eyre",
 "futures",
@@ -8341,7 +8236,6 @@ dependencies = [
 "metrics",
 "metrics-util",
 "mini-moka",
- "moka",
 "parking_lot",
 "proptest",
 "rand 0.8.5",
@@ -8375,7 +8269,6 @@ dependencies = [
 "reth-stages",
 "reth-stages-api",
 "reth-static-file",
- "reth-storage-errors",
 "reth-tasks",
 "reth-testing-utils",
 "reth-tracing",
@@ -8746,7 +8639,6 @@ dependencies = [
 "derive_more",
 "futures-util",
 "metrics",
- "rayon",
 "reth-ethereum-forks",
 "reth-ethereum-primitives",
 "reth-execution-errors",
@@ -9052,7 +8944,6 @@ dependencies = [
 "pin-project",
 "rand 0.8.5",
 "rand 0.9.2",
- "rayon",
 "reth-chainspec",
 "reth-consensus",
 "reth-discv4",
@@ -9223,7 +9114,6 @@ dependencies = [
 "fdlimit",
 "futures",
 "jsonrpsee",
- "parking_lot",
 "rayon",
 "reth-basic-payload-builder",
 "reth-chain-state",
@@ -9383,7 +9273,6 @@ dependencies = [
 "reth-rpc-eth-api",
 "reth-rpc-eth-types",
 "reth-rpc-server-types",
- "reth-stages-types",
 "reth-tasks",
 "reth-testing-utils",
 "reth-tracing",
@@ -9707,7 +9596,6 @@ dependencies = [
 "alloy-consensus",
 "alloy-genesis",
 "alloy-network",
- "alloy-op-hardforks",
 "alloy-primitives",
 "alloy-rpc-types-engine",
 "alloy-rpc-types-eth",
@@ -9748,7 +9636,6 @@ dependencies = [
 "reth-rpc-engine-api",
 "reth-rpc-eth-types",
 "reth-rpc-server-types",
- "reth-stages-types",
 "reth-tasks",
 "reth-tracing",
 "reth-transaction-pool",
@@ -10492,7 +10379,6 @@ dependencies = [
 "reth-ethereum-engine-primitives",
 "reth-ethereum-primitives",
 "reth-metrics",
- "reth-network-api",
 "reth-node-ethereum",
 "reth-payload-builder",
 "reth-payload-builder-primitives",
@@ -10679,7 +10565,6 @@ dependencies = [
 "reth-stages-api",
 "reth-static-file",
 "reth-static-file-types",
- "reth-storage-api",
 "reth-storage-errors",
 "reth-testing-utils",
 "reth-trie",
@@ -10924,7 +10809,6 @@ dependencies = [
 "tracing-appender",
 "tracing-journald",
 "tracing-logfmt",
- "tracing-samply",
 "tracing-subscriber 0.3.22",
 ]

@@ -11010,7 +10894,6 @@ dependencies = [
 "pretty_assertions",
 "proptest",
 "proptest-arbitrary-interop",
- "rand 0.9.2",
 "reth-ethereum-primitives",
 "reth-execution-errors",
 "reth-metrics",
@@ -11688,9 +11571,9 @@ dependencies = [

 [[package]]
 name = "rustls-pki-types"
-version = "1.13.2"
+version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21e6f2ab2928ca4291b86736a8bd920a277a399bba1589409d72154ff87c1282"
+checksum = "708c0f9d5f54ba0272468c1d306a52c495b31fa155e91bc25371e6df7996908c"
 dependencies = [
 "web-time",
 "zeroize",
@@ -11754,9 +11637,9 @@ dependencies = [

 [[package]]
 name = "ryu"
-version = "1.0.21"
+version = "1.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62049b2877bf12821e8f9ad256ee38fdc31db7387ec2d3b3f403024de2034aea"
+checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f"

 [[package]]
 name = "ryu-js"
@@ -12454,9 +12337,9 @@ dependencies = [

 [[package]]
 name = "syn-solidity"
-version = "1.5.1"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f6b1d2e2059056b66fec4a6bb2b79511d5e8d76196ef49c38996f4b48db7662f"
+checksum = "ff790eb176cc81bb8936aed0f7b9f14fc4670069a2d371b3e3b0ecce908b2cb3"
 dependencies = [
 "paste",
 "proc-macro2",
@@ -12914,9 +12797,9 @@ dependencies = [

 [[package]]
 name = "toml_datetime"
-version = "0.7.5+spec-1.1.0"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92e1cfed4a3038bc5a127e35a2d360f145e1f4b971b551a2ba5fd7aedf7e1347"
+checksum = "f2cdb639ebbc97961c51720f858597f7f24c4fc295327923af55b74c3c724533"
 dependencies = [
 "serde_core",
 ]
@@ -12937,21 +12820,21 @@ dependencies = [

 [[package]]
 name = "toml_edit"
-version = "0.23.10+spec-1.0.0"
+version = "0.23.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "84c8b9f757e028cee9fa244aea147aab2a9ec09d5325a9b01e0a49730c2b5269"
+checksum = "5d7cbc3b4b49633d57a0509303158ca50de80ae32c265093b24c414705807832"
 dependencies = [
 "indexmap 2.12.1",
- "toml_datetime 0.7.5+spec-1.1.0",
+ "toml_datetime 0.7.3",
 "toml_parser",
 "winnow",
 ]

 [[package]]
 name = "toml_parser"
-version = "1.0.6+spec-1.1.0"
+version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3198b4b0a8e11f09dd03e133c0280504d0801269e9afa46362ffde1cbeebf44"
+checksum = "c0cbe268d35bdb4bb5a56a2de88d0ad0eb70af5384a99d648cd4b3d04039800e"
 dependencies = [
 "winnow",
 ]
@@ -13064,9 +12947,9 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"

 [[package]]
 name = "tracing"
-version = "0.1.44"
+version = "0.1.43"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
+checksum = "2d15d90a0b5c19378952d479dc858407149d7bb45a14de0142f6c534b16fc647"
 dependencies = [
 "log",
 "pin-project-lite",
@@ -13099,24 +12982,14 @@ dependencies = [

 [[package]]
 name = "tracing-core"
-version = "0.1.36"
+version = "0.1.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
+checksum = "7a04e24fab5c89c6a36eb8558c9656f30d81de51dfa4d3b45f26b21d61fa0a6c"
 dependencies = [
 "once_cell",
 "valuable",
 ]

-[[package]]
-name = "tracing-error"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b1581020d7a273442f5b45074a6a57d5757ad0a47dac0e9f0bd57b81936f3db"
-dependencies = [
- "tracing",
- "tracing-subscriber 0.3.22",
-]
-
 [[package]]
 name = "tracing-futures"
 version = "0.2.5"
@@ -13180,22 +13053,6 @@ dependencies = [
 "web-time",
 ]

-[[package]]
-name = "tracing-samply"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c175f7ecc002b6ef04776a39f440503e4e788790ddbdbfac8259b7a069526334"
-dependencies = [
- "cfg-if",
- "itoa",
- "libc",
- "mach2",
- "memmap2",
- "smallvec",
- "tracing-core",
- "tracing-subscriber 0.3.22",
-]
-
 [[package]]
 name = "tracing-serde"
 version = "0.2.0"
@@ -13807,16 +13664,38 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

+[[package]]
+name = "windows"
+version = "0.61.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893"
+dependencies = [
+ "windows-collections 0.2.0",
+ "windows-core 0.61.2",
+ "windows-future 0.2.1",
+ "windows-link 0.1.3",
+ "windows-numerics 0.2.0",
+]
+
 [[package]]
 name = "windows"
 version = "0.62.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580"
 dependencies = [
- "windows-collections",
+ "windows-collections 0.3.2",
 "windows-core 0.62.2",
- "windows-future",
- "windows-numerics",
+ "windows-future 0.3.2",
+ "windows-numerics 0.3.1",
+]
+
+[[package]]
+name = "windows-collections"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8"
+dependencies = [
+ "windows-core 0.61.2",
 ]

 [[package]]
@@ -13840,6 +13719,19 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

+[[package]]
+name = "windows-core"
+version = "0.61.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3"
+dependencies = [
+ "windows-implement 0.60.2",
+ "windows-interface 0.59.3",
+ "windows-link 0.1.3",
+ "windows-result 0.3.4",
+ "windows-strings 0.4.2",
+]
+
 [[package]]
 name = "windows-core"
 version = "0.62.2"
@@ -13848,9 +13740,20 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
 dependencies = [
 "windows-implement 0.60.2",
 "windows-interface 0.59.3",
- "windows-link",
+ "windows-link 0.2.1",
 "windows-result 0.4.1",
- "windows-strings",
+ "windows-strings 0.5.1",
+]
+
+[[package]]
+name = "windows-future"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e"
+dependencies = [
+ "windows-core 0.61.2",
+ "windows-link 0.1.3",
+ "windows-threading 0.1.0",
 ]

 [[package]]
@@ -13860,8 +13763,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb"
 dependencies = [
 "windows-core 0.62.2",
- "windows-link",
- "windows-threading",
+ "windows-link 0.2.1",
+ "windows-threading 0.2.1",
 ]

 [[package]]
@@ -13908,12 +13811,28 @@ dependencies = [
 "syn 2.0.111",
 ]

+[[package]]
+name = "windows-link"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"

+[[package]]
+name = "windows-numerics"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1"
+dependencies = [
+ "windows-core 0.61.2",
+ "windows-link 0.1.3",
+]
+
 [[package]]
 name = "windows-numerics"
 version = "0.3.1"
@@ -13921,7 +13840,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26"
 dependencies = [
 "windows-core 0.62.2",
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
@@ -13933,13 +13852,31 @@ dependencies = [
 "windows-targets 0.52.6",
 ]

+[[package]]
+name = "windows-result"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6"
+dependencies = [
+ "windows-link 0.1.3",
+]
+
 [[package]]
 name = "windows-result"
 version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57"
+dependencies = [
+ "windows-link 0.1.3",
 ]

 [[package]]
@@ -13948,7 +13885,7 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
@@ -14002,7 +13939,7 @@ version = "0.61.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
@@ -14057,7 +13994,7 @@ version = "0.53.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
 "windows_aarch64_gnullvm 0.53.1",
 "windows_aarch64_msvc 0.53.1",
 "windows_i686_gnu 0.53.1",
@@ -14068,13 +14005,22 @@ dependencies = [
 "windows_x86_64_msvc 0.53.1",
 ]

+[[package]]
+name = "windows-threading"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6"
+dependencies = [
+ "windows-link 0.1.3",
+]
+
 [[package]]
 name = "windows-threading"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37"
 dependencies = [
- "windows-link",
+ "windows-link 0.2.1",
 ]

 [[package]]
--- 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.5.0", default-features = false, features = ["map-foldhash"] }
+alloy-primitives = { version = "1.4.1", default-features = false, features = ["map-foldhash"] }
 alloy-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
-alloy-sol-macro = "1.5.0"
-alloy-sol-types = { version = "1.5.0", default-features = false }
+alloy-sol-macro = "1.4.1"
+alloy-sol-types = { version = "1.4.1", default-features = false }
 alloy-trie = { version = "0.9.1", default-features = false }

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

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

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

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

 ARG RUSTFLAGS=""
--- a/2
+++ b/2
@@ -521,3 +521,5 @@ pr:
 	make update-book-cli && \
 	cargo docs --document-private-items && \
 	make test
+
+check-features:
--- a/bin/reth-bench-compare/src/cli.rs
+++ b/bin/reth-bench-compare/src/cli.rs
@@ -329,7 +329,6 @@ pub(crate) async fn run_comparison(args: Args, _ctx: CliContext) -> Result<()> {
        output_dir.clone(),
        git_manager.clone(),
        args.features.clone(),
-        args.profile,
    )?;
    // Initialize node manager
    let mut node_manager = NodeManager::new(&args);
--- a/bin/reth-bench-compare/src/compilation.rs
+++ b/bin/reth-bench-compare/src/compilation.rs
@@ -14,7 +14,6 @@ pub(crate) struct CompilationManager {
    output_dir: PathBuf,
    git_manager: GitManager,
    features: String,
-    enable_profiling: bool,
 }

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

    /// Detect if the RPC endpoint is an Optimism chain
@@ -102,18 +100,9 @@ impl CompilationManager {
        let mut cmd = Command::new("cargo");
        cmd.arg("build").arg("--profile").arg("profiling");

-        // Append samply feature when profiling to enable tracing span markers.
-        // NOTE: The `samply` feature must exist in the branch being compiled. If comparing
-        // against an older branch that predates the samply integration, compilation will fail
-        // or markers won't appear. In that case, omit --profile or ensure both branches
-        // include the samply feature support.
-        let features = if self.enable_profiling && !self.features.contains("samply") {
-            format!("{},samply", self.features)
-        } else {
-            self.features.clone()
-        };
-        cmd.arg("--features").arg(&features);
-        info!("Using features: {}", features);
+        // Add features
+        cmd.arg("--features").arg(&self.features);
+        info!("Using features: {}", self.features);

        // Add bin-specific arguments for optimism
        if is_optimism {
--- a/bin/reth-bench/Cargo.toml
+++ b/bin/reth-bench/Cargo.toml
@@ -58,7 +58,6 @@ tokio = { workspace = true, features = ["sync", "macros", "time", "rt-multi-thre
 # misc
 clap = { workspace = true, features = ["derive", "env"] }
 eyre.workspace = true
-color-eyre.workspace = true
 thiserror.workspace = true
 humantime.workspace = true

--- a/bin/reth-bench/src/main.rs
+++ b/bin/reth-bench/src/main.rs
@@ -23,7 +23,7 @@ use bench::BenchmarkCommand;
 use clap::Parser;
 use reth_cli_runner::CliRunner;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    /// Turn this state provider into a state provider
@@ -67,7 +71,7 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {

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

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

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

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

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

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

--- a/crates/cli/commands/src/common.rs
+++ b/crates/cli/commands/src/common.rs
@@ -23,10 +23,7 @@ use reth_node_core::{
    dirs::{ChainPath, DataDirPath},
 };
 use reth_provider::{
-    providers::{
-        BlockchainProvider, NodeTypesForProvider, RocksDBProvider, StaticFileProvider,
-        StaticFileProviderBuilder,
-    },
+    providers::{BlockchainProvider, NodeTypesForProvider, StaticFileProvider},
    ProviderFactory, StaticFileProviderFactory,
 };
 use reth_stages::{sets::DefaultStages, Pipeline, PipelineTarget};
@@ -78,12 +75,10 @@ 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());
@@ -103,32 +98,18 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
        }

        info!(target: "reth::cli", ?db_path, ?sf_path, "Opening storage");
-        let genesis_block_number = self.chain.genesis().number.unwrap_or_default();
        let (db, sfp) = match access {
            AccessRights::RW => (
                Arc::new(init_db(db_path, self.db.database_args())?),
-                StaticFileProviderBuilder::read_write(sf_path)?
-                    .with_genesis_block_number(genesis_block_number)
-                    .build()?,
+                StaticFileProvider::read_write(sf_path)?,
+            ),
+            AccessRights::RO | AccessRights::RoInconsistent => (
+                Arc::new(open_db_read_only(&db_path, self.db.database_args())?),
+                StaticFileProvider::read_only(sf_path, false)?,
            ),
-            AccessRights::RO | AccessRights::RoInconsistent => {
-                (Arc::new(open_db_read_only(&db_path, self.db.database_args())?), {
-                    let provider = StaticFileProviderBuilder::read_only(sf_path)?
-                        .with_genesis_block_number(genesis_block_number)
-                        .build()?;
-                    provider.watch_directory();
-                    provider
-                })
-            }
        };
-        // TransactionDB only support read-write mode
-        let rocksdb_provider = RocksDBProvider::builder(data_dir.rocksdb())
-            .with_default_tables()
-            .with_database_log_level(self.db.log_level)
-            .build()?;

-        let provider_factory =
-            self.create_provider_factory(&config, db, sfp, rocksdb_provider, access)?;
+        let provider_factory = self.create_provider_factory(&config, db, sfp, 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())?;
@@ -147,7 +128,6 @@ 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
@@ -158,7 +138,6 @@ 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,7 +2,6 @@ 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::{
@@ -80,10 +79,7 @@ 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,
-        ctx: CliContext,
-    ) -> eyre::Result<()> {
+    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(self) -> 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();
@@ -162,7 +158,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, ctx.task_executor.clone())?;
+                    command.execute(&tool)?;
                });
            }
            Subcommands::StaticFileHeader(command) => {
--- a/crates/cli/commands/src/db/repair_trie.rs
+++ b/crates/cli/commands/src/db/repair_trie.rs
@@ -18,7 +18,6 @@ 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,
@@ -49,37 +48,52 @@ pub struct Command {

 impl Command {
    /// Execute `db repair-trie` command
-    pub fn execute<N: ProviderNodeTypes>(
-        self,
-        tool: &DbTool<N>,
-        task_executor: TaskExecutor,
-    ) -> eyre::Result<()> {
+    pub fn execute<N: ProviderNodeTypes>(self, tool: &DbTool<N>) -> 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 = 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 = 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");

-                // Spawn the metrics server
-                if let Err(e) = MetricServer::new(config).serve().await {
-                    tracing::error!("Metrics server error: {}", e);
-                }
-            });
+                    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
+                    });
+                },
+            )?;

            Some(handle)
        } else {
@@ -301,8 +315,8 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
    if inconsistent_nodes == 0 {
        info!("No inconsistencies found");
    } else {
+        info!("Repaired {} inconsistencies, committing changes", inconsistent_nodes);
        provider_rw.commit()?;
-        info!("Repaired {} inconsistencies and committed changes", inconsistent_nodes);
    }

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

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

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

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

        info!(target: "reth::cli", hash = ?hash, "Genesis block written");
--- a/crates/cli/commands/src/init_state/without_evm.rs
+++ b/crates/cli/commands/src/init_state/without_evm.rs
@@ -79,7 +79,7 @@ where
        + StaticFileProviderFactory<Primitives: NodePrimitives<BlockHeader: Compact>>,
 {
    provider_rw.insert_block(
-        &SealedBlock::<<Provider::Primitives as NodePrimitives>::Block>::from_sealed_parts(
+        SealedBlock::<<Provider::Primitives as NodePrimitives>::Block>::from_sealed_parts(
            header.clone(),
            Default::default(),
        )
--- a/crates/cli/commands/src/p2p/mod.rs
+++ b/crates/cli/commands/src/p2p/mod.rs
@@ -72,7 +72,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                .split();
                if result.len() != 1 {
                    eyre::bail!(
-                        "Invalid number of bodies received. Expected: 1. Received: {}",
+                        "Invalid number of headers received. Expected: 1. Received: {}",
                        result.len()
                    )
                }
@@ -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.clone())
+            .external_ip_resolver(self.network.nat)
            .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, RocksDBProvider, StaticFileProvider},
+    providers::{ProviderNodeTypes, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::ExecutionStage, Stage, StageCheckpoint, UnwindInput};
@@ -42,7 +42,6 @@ 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, RocksDBProvider, StaticFileProvider},
+    providers::{ProviderNodeTypes, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::AccountHashingStage, Stage, StageCheckpoint, UnwindInput};
@@ -39,7 +39,6 @@ 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, RocksDBProvider, StaticFileProvider},
+    providers::{ProviderNodeTypes, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::StorageHashingStage, Stage, StageCheckpoint, UnwindInput};
@@ -29,7 +29,6 @@ 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, RocksDBProvider, StaticFileProvider},
+    providers::{ProviderNodeTypes, StaticFileProvider},
    DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{
@@ -62,7 +62,6 @@ 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,57 +97,6 @@ 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/config/src/config.rs
+++ b/crates/config/src/config.rs
@@ -22,6 +22,7 @@ pub const DEFAULT_BLOCK_INTERVAL: usize = 5;
 #[cfg_attr(feature = "serde", serde(default))]
 pub struct Config {
    /// Configuration for each stage in the pipeline.
+    // TODO(onbjerg): Can we make this easier to maintain when we add/remove stages?
    pub stages: StageConfig,
    /// Configuration for pruning.
    #[cfg_attr(feature = "serde", serde(default))]
--- a/crates/consensus/common/src/validation.rs
+++ b/crates/consensus/common/src/validation.rs
@@ -279,28 +279,20 @@ pub fn validate_against_parent_hash_number<H: BlockHeader>(
    header: &H,
    parent: &SealedHeader<H>,
 ) -> Result<(), ConsensusError> {
-    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() {
+    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(),
+        ))
+    }
+
    Ok(())
 }

--- a/crates/e2e-test-utils/src/setup_builder.rs
+++ b/crates/e2e-test-utils/src/setup_builder.rs
@@ -11,7 +11,6 @@ use reth_node_builder::{
    PayloadTypes,
 };
 use reth_node_core::args::{DiscoveryArgs, NetworkArgs, RpcServerArgs};
-use reth_primitives_traits::AlloyBlockHeader;
 use reth_provider::providers::BlockchainProvider;
 use reth_rpc_server_types::RpcModuleSelection;
 use reth_tasks::TaskManager;
@@ -158,8 +157,8 @@ where
                    .await?;

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

                eyre::Ok(node)
--- a/crates/e2e-test-utils/src/setup_import.rs
+++ b/crates/e2e-test-utils/src/setup_import.rs
@@ -110,7 +110,6 @@ 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
@@ -126,7 +125,6 @@ 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
@@ -313,7 +311,6 @@ 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
        {
@@ -327,9 +324,6 @@ 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");

@@ -391,9 +385,6 @@ 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");

@@ -481,15 +472,11 @@ 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/e2e-test-utils/src/testsuite/actions/engine_api.rs
+++ b/crates/e2e-test-utils/src/testsuite/actions/engine_api.rs
@@ -2,7 +2,9 @@

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

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

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

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

--- a/crates/engine/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 task.
+    /// Whether the host has enough parallelism to run state root in parallel.
    has_enough_parallelism: bool,
    /// Whether multiproof task should chunk proof targets.
    multiproof_chunking_enabled: bool,
@@ -403,12 +403,17 @@ impl TreeConfig {
        self
    }

-    /// Setter for has enough parallelism.
+    /// Setter for whether or not the host has enough parallelism to run state root in parallel.
    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,8 +22,7 @@ use reth_trie_common::HashedPostState;
 use serde::{de::DeserializeOwned, Serialize};

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

 mod error;
--- a/crates/engine/tree/Cargo.toml
+++ b/crates/engine/tree/Cargo.toml
@@ -13,10 +13,10 @@ workspace = true
 [dependencies]
 # reth
 reth-chain-state.workspace = true
-reth-chainspec.workspace = true
+reth-chainspec = { workspace = true, optional = true }
 reth-consensus.workspace = true
 reth-db.workspace = true
-reth-engine-primitives = { workspace = true, features = ["std"] }
+reth-engine-primitives.workspace = true
 reth-errors.workspace = true
 reth-execution-types.workspace = true
 reth-evm = { workspace = true, features = ["metrics"] }
@@ -29,7 +29,6 @@ reth-provider.workspace = true
 reth-prune.workspace = true
 reth-revm.workspace = true
 reth-stages-api.workspace = true
-reth-storage-errors.workspace = true
 reth-tasks.workspace = true
 reth-trie-parallel.workspace = true
 reth-trie-sparse = { workspace = true, features = ["std", "metrics"] }
@@ -53,7 +52,6 @@ futures.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "sync", "macros"] }
 mini-moka = { workspace = true, features = ["sync"] }
-moka = { workspace = true, features = ["sync"] }
 smallvec.workspace = true

 # metrics
@@ -79,6 +77,7 @@ reth-tracing = { workspace = true, optional = true }
 # reth
 reth-evm-ethereum = { workspace = true, features = ["test-utils"] }
 reth-chain-state = { workspace = true, features = ["test-utils"] }
+reth-chainspec.workspace = true
 reth-db-common.workspace = true
 reth-ethereum-consensus.workspace = true
 metrics-util = { workspace = true, features = ["debugging"] }
--- a/crates/engine/tree/benches/state_root_task.rs
+++ b/crates/engine/tree/benches/state_root_task.rs
@@ -9,7 +9,7 @@ use alloy_primitives::{Address, B256};
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
 use proptest::test_runner::TestRunner;
 use rand::Rng;
-use reth_chainspec::{ChainSpec, MAINNET};
+use reth_chainspec::ChainSpec;
 use reth_db_common::init::init_genesis;
 use reth_engine_tree::tree::{
    executor::WorkloadExecutor, precompile_cache::PrecompileCacheMap, PayloadProcessor,
@@ -220,7 +220,6 @@ fn bench_state_root(c: &mut Criterion) {
                            EthEvmConfig::new(factory.chain_spec()),
                            &TreeConfig::default(),
                            PrecompileCacheMap::default(),
-                            MAINNET.clone(),
                        );
                        let provider = BlockchainProvider::new(factory).unwrap();

@@ -231,18 +230,17 @@ fn bench_state_root(c: &mut Criterion) {
                            let mut handle = payload_processor.spawn(
                                Default::default(),
                                (
-                                    Vec::<
+                                    core::iter::empty::<
                                        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/docs/root.md
+++ b/crates/engine/tree/docs/root.md
@@ -128,12 +128,12 @@ we send them along with the state updates to the [Sparse Trie Task](#sparse-trie

 ### Finishing the calculation

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

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

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

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

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

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

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

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

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

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

--- a/crates/engine/tree/src/tree/block_buffer.rs
+++ b/crates/engine/tree/src/tree/block_buffer.rs
@@ -14,7 +14,7 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 /// * [`BlockBuffer::remove_old_blocks`] to remove old blocks that precede the finalized number.
 ///
 /// Note: Buffer is limited by number of blocks that it can contain and eviction of the block
-/// is done in FIFO order (oldest inserted block is evicted first).
+/// is done by last recently used block.
 #[derive(Debug)]
 pub struct BlockBuffer<B: Block> {
    /// All blocks in the buffer stored by their block hash.
--- a/crates/engine/tree/src/tree/cached_state.rs
+++ b/crates/engine/tree/src/tree/cached_state.rs
@@ -14,7 +14,7 @@ use reth_trie::{
    updates::TrieUpdates, AccountProof, HashedPostState, HashedStorage, MultiProof,
    MultiProofTargets, StorageMultiProof, StorageProof, TrieInput,
 };
-use revm_primitives::{hardfork::SpecId, map::DefaultHashBuilder};
+use revm_primitives::map::DefaultHashBuilder;
 use std::{sync::Arc, time::Duration};
 use tracing::{debug_span, instrument, trace};

@@ -31,9 +31,6 @@ pub(crate) struct CachedStateProvider<S> {

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

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

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

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

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

        let final_res = self.state_provider.bytecode_by_hash(code_hash)?;
-
-        if self.is_prewarm() {
-            self.caches.code_cache.insert(*code_hash, final_res.clone());
-        }
-
+        self.caches.code_cache.insert(*code_hash, final_res.clone());
        Ok(final_res)
    }
 }
@@ -442,11 +408,7 @@ impl ExecutionCache {
    ///
    /// Returns an error if the state updates are inconsistent and should be discarded.
    #[instrument(level = "debug", target = "engine::caching", skip_all)]
-    pub(crate) fn insert_state(
-        &self,
-        state_updates: &BundleState,
-        spec: &SpecId,
-    ) -> Result<(), ()> {
+    pub(crate) fn insert_state(&self, state_updates: &BundleState) -> Result<(), ()> {
        let _enter =
            debug_span!(target: "engine::tree", "contracts", len = state_updates.contracts.len())
                .entered();
@@ -471,13 +433,8 @@ impl ExecutionCache {
                continue
            }

-            // If the account was destroyed, invalidate from the account / storage caches.
-            //
-            // Post-cancun when EIP-6780 is live, an account can be destroyed only when it's created
-            // in the same transaction. This guarantees that we will not have such accounts
-            // and storage slots in our cache, because Revm doesn't go through the
-            // Database for freshly created accounts. Hence we can safely ignore invalidating them.
-            if account.was_destroyed() && !spec.is_enabled_in(revm_primitives::hardfork::CANCUN) {
+            // If the account was destroyed, invalidate from the account / storage caches
+            if account.was_destroyed() {
                // Invalidate the account cache entry if destroyed
                self.account_cache.invalidate(addr);

@@ -828,7 +785,7 @@ mod tests {

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

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

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

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

--- a/crates/engine/tree/src/tree/mod.rs
+++ b/crates/engine/tree/src/tree/mod.rs
@@ -39,7 +39,6 @@ use revm::state::EvmState;
 use state::TreeState;
 use std::{
    fmt::Debug,
-    ops,
    sync::{
        mpsc::{Receiver, RecvError, RecvTimeoutError, Sender},
        Arc,
@@ -427,13 +426,9 @@ where
            match self.try_recv_engine_message() {
                Ok(Some(msg)) => {
                    debug!(target: "engine::tree", %msg, "received new engine message");
-                    match self.on_engine_message(msg) {
-                        Ok(ops::ControlFlow::Break(())) => return,
-                        Ok(ops::ControlFlow::Continue(())) => {}
-                        Err(fatal) => {
-                            error!(target: "engine::tree", %fatal, "insert block fatal error");
-                            return
-                        }
+                    if let Err(fatal) = self.on_engine_message(msg) {
+                        error!(target: "engine::tree", %fatal, "insert block fatal error");
+                        return
                    }
                }
                Ok(None) => {
@@ -931,6 +926,48 @@ where
        Ok(())
    }

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

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

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

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

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

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

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

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

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

@@ -2877,12 +2860,3 @@ pub enum InsertPayloadOk {
    /// The payload was valid and inserted into the tree.
    Inserted(BlockStatus),
 }
-
-/// Target for block persistence.
-#[derive(Debug, Clone, Copy)]
-enum PersistTarget {
-    /// Persist up to `canonical_head - memory_block_buffer_target`.
-    Threshold,
-    /// Persist all blocks up to and including the canonical head.
-    Head,
-}
--- a/crates/engine/tree/src/tree/payload_processor/bal.rs
+++ b/crates/engine/tree/src/tree/payload_processor/bal.rs
@@ -1,318 +0,0 @@
-//! 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,7 +13,6 @@ 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;
@@ -22,15 +21,13 @@ use executor::WorkloadExecutor;
 use multiproof::{SparseTrieUpdate, *};
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
-use rayon::prelude::*;
-use reth_chainspec::EthereumHardforks;
+use rayon::iter::{ParallelBridge, ParallelIterator};
+use reth_engine_primitives::ExecutableTxIterator;
 use reth_evm::{
    execute::{ExecutableTxFor, WithTxEnv},
-    ConfigureEvm, EvmEnvFor, ExecutableTxIterator, ExecutableTxTuple, OnStateHook, SpecFor,
-    TxEnvFor,
+    ConfigureEvm, EvmEnvFor, OnStateHook, SpecFor, TxEnvFor,
 };
-use reth_execution_types::ExecutionOutcome;
-use reth_primitives_traits::{HeaderTy, NodePrimitives};
+use reth_primitives_traits::NodePrimitives;
 use reth_provider::{BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader};
 use reth_revm::{db::BundleState, state::EvmState};
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
@@ -43,7 +40,6 @@ use reth_trie_sparse::{
    ClearedSparseStateTrie, SparseStateTrie, SparseTrie,
 };
 use reth_trie_sparse_parallel::{ParallelSparseTrie, ParallelismThresholds};
-use revm_primitives::hardfork::SpecId;
 use std::{
    collections::BTreeMap,
    sync::{
@@ -55,7 +51,6 @@ use std::{
 };
 use tracing::{debug, debug_span, instrument, warn, Span};

-pub mod bal;
 mod configured_sparse_trie;
 pub mod executor;
 pub mod multiproof;
@@ -95,16 +90,9 @@ pub const SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY: usize = 1_000_000;
 /// 144MB.
 pub const SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY: usize = 1_000_000;

-/// Type alias for [`PayloadHandle`] returned by payload processor spawn methods.
-type IteratorPayloadHandle<Evm, I, N> = PayloadHandle<
-    WithTxEnv<TxEnvFor<Evm>, <I as ExecutableTxTuple>::Tx>,
-    <I as ExecutableTxTuple>::Error,
-    <N as NodePrimitives>::Receipt,
->;
-
 /// Entrypoint for executing the payload.
 #[derive(Debug)]
-pub struct PayloadProcessor<Evm, C>
+pub struct PayloadProcessor<Evm>
 where
    Evm: ConfigureEvm,
 {
@@ -120,7 +108,6 @@ where
    disable_transaction_prewarming: bool,
    /// Whether state cache should be disable
    disable_state_cache: bool,
-    chain_spec: Arc<C>,
    /// Determines how to configure the evm for execution.
    evm_config: Evm,
    /// Whether precompile cache should be disabled.
@@ -140,7 +127,7 @@ where
    prewarm_max_concurrency: usize,
 }

-impl<N, Evm, C> PayloadProcessor<Evm, C>
+impl<N, Evm> PayloadProcessor<Evm>
 where
    N: NodePrimitives,
    Evm: ConfigureEvm<Primitives = N>,
@@ -156,7 +143,6 @@ where
        evm_config: Evm,
        config: &TreeConfig,
        precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
-        chain_spec: Arc<C>,
    ) -> Self {
        Self {
            executor,
@@ -164,7 +150,6 @@ where
            trie_metrics: Default::default(),
            cross_block_cache_size: config.cross_block_cache_size(),
            disable_transaction_prewarming: config.disable_prewarming(),
-            chain_spec,
            evm_config,
            disable_state_cache: config.disable_state_cache(),
            precompile_cache_disabled: config.precompile_cache_disabled(),
@@ -176,11 +161,10 @@ where
    }
 }

-impl<N, Evm, C> PayloadProcessor<Evm, C>
+impl<N, Evm> PayloadProcessor<Evm>
 where
    N: NodePrimitives,
    Evm: ConfigureEvm<Primitives = N> + 'static,
-    C: EthereumHardforks,
 {
    /// Spawns all background tasks and returns a handle connected to the tasks.
    ///
@@ -214,6 +198,7 @@ where
    ///
    /// This returns a handle to await the final state root and to interact with the tasks (e.g.
    /// canceling)
+    #[allow(clippy::type_complexity)]
    #[instrument(
        level = "debug",
        target = "engine::tree::payload_processor",
@@ -227,8 +212,7 @@ where
        provider_builder: StateProviderBuilder<N, P>,
        multiproof_provider_factory: F,
        config: &TreeConfig,
-        bal: Option<Arc<BlockAccessList>>,
-    ) -> IteratorPayloadHandle<Evm, I, N>
+    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
        F: DatabaseProviderROFactory<Provider: TrieCursorFactory + HashedCursorFactory>
@@ -242,34 +226,11 @@ where

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

-        // Handle BAL-based optimization if available
-        let prewarm_handle = if let Some(bal) = bal {
-            // When BAL is present, 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
-            let _ = to_multi_proof.send(MultiProofMessage::BlockAccessList(bal));
-
-            // 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(),
-                Some(to_multi_proof.clone()),
-            )
-        };
+        // We rely on the cursor factory to provide whatever DB overlay is necessary to see a
+        // consistent view of the database, including the trie tables. Because of this there is no
+        // need for an overarching prefix set to invalidate any section of the trie tables, and so
+        // we use an empty prefix set.

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

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

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

        // wire the sparse trie to the state root response receiver
@@ -326,7 +291,7 @@ where
        env: ExecutionEnv<Evm>,
        transactions: I,
        provider_builder: StateProviderBuilder<N, P>,
-    ) -> IteratorPayloadHandle<Evm, I, N>
+    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
    {
@@ -353,32 +318,36 @@ where
        usize,
    ) {
        let (transactions, convert) = transactions.into();
-        let transactions = transactions.into_par_iter();
-        let transaction_count_hint = transactions.len();
+        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 (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.
+        // Spawn a task that iterates through all transactions in parallel and sends them to the
+        // main task.
+        let (tx, rx) = mpsc::channel();
        self.executor.spawn_blocking(move || {
-            transactions.enumerate().for_each_with(ooo_tx, |ooo_tx, (idx, tx)| {
+            transactions.enumerate().par_bridge().for_each_with(tx, |sender, (idx, tx)| {
                let tx = convert(tx);
                let tx = tx.map(|tx| WithTxEnv { tx_env: tx.to_tx_env(), tx: Arc::new(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));
+                let _ = sender.send((idx, tx));
            });
        });

        // Spawn a task that processes out-of-order transactions from the task above and sends them
-        // to the execution task in order.
+        // to prewarming and execution tasks.
+        let (prewarm_tx, prewarm_rx) = mpsc::channel();
+        let (execute_tx, execute_rx) = mpsc::channel();
        self.executor.spawn_blocking(move || {
            let mut next_for_execution = 0;
            let mut queue = BTreeMap::new();
-            while let Ok((idx, tx)) = ooo_rx.recv() {
+            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());
+                }
+
                if next_for_execution == idx {
                    let _ = execute_tx.send(tx);
                    next_for_execution += 1;
@@ -406,7 +375,7 @@ where
        transaction_count_hint: usize,
        provider_builder: StateProviderBuilder<N, P>,
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-    ) -> CacheTaskHandle<N::Receipt>
+    ) -> CacheTaskHandle
    where
        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
    {
@@ -546,7 +515,6 @@ where
    /// hitting the database, maintaining performance consistency.
    pub(crate) fn on_inserted_executed_block(
        &self,
-        header: &HeaderTy<N>,
        block_with_parent: BlockWithParent,
        bundle_state: &BundleState,
    ) {
@@ -573,7 +541,7 @@ where

            // Insert the block's bundle state into cache
            let new_cache = SavedCache::new(block_with_parent.block.hash, caches, cache_metrics);
-            if new_cache.cache().insert_state(bundle_state, &alloy_evm::spec(&self.chain_spec, header)).is_err() {
+            if new_cache.cache().insert_state(bundle_state).is_err() {
                *cached = None;
                debug!(target: "engine::caching", "cleared execution cache on update error");
                return;
@@ -588,15 +556,12 @@ where
 }

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

-impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
+impl<Tx, Err> PayloadHandle<Tx, Err> {
    /// Awaits the state root
    ///
    /// # Panics
@@ -634,7 +599,7 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {

        move |source: StateChangeSource, state: &EvmState| {
            if let Some(sender) = &to_multi_proof {
-                let _ = sender.send(MultiProofMessage::StateUpdate(source.into(), state.clone()));
+                let _ = sender.send(MultiProofMessage::StateUpdate(source, state.clone()));
            }
        }
    }
@@ -658,14 +623,9 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {

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

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

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

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

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

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

@@ -773,8 +725,6 @@ impl ExecutionCache {

        cache
            .as_ref()
-            // Check `is_available()` to ensure no other tasks (e.g., prewarming) currently hold
-            // a reference to this cache. We can only reuse it when we have exclusive access.
            .filter(|c| c.executed_block_hash() == parent_hash && c.is_available())
            .cloned()
    }
@@ -816,8 +766,6 @@ pub struct ExecutionEnv<Evm: ConfigureEvm> {
    pub hash: B256,
    /// Hash of the parent block.
    pub parent_hash: B256,
-    /// Spec id associated with the EVM for the block being executed.
-    pub spec_id: SpecId,
 }

 impl<Evm: ConfigureEvm> Default for ExecutionEnv<Evm>
@@ -829,7 +777,6 @@ where
            evm_env: Default::default(),
            hash: Default::default(),
            parent_hash: Default::default(),
-            spec_id: Default::default(),
        }
    }
 }
@@ -845,11 +792,10 @@ mod tests {
        precompile_cache::PrecompileCacheMap,
        StateProviderBuilder, TreeConfig,
    };
-    use alloy_consensus::Header;
    use alloy_eips::eip1898::{BlockNumHash, BlockWithParent};
    use alloy_evm::block::StateChangeSource;
    use rand::Rng;
-    use reth_chainspec::{ChainSpec, MAINNET};
+    use reth_chainspec::ChainSpec;
    use reth_db_common::init::init_genesis;
    use reth_ethereum_primitives::TransactionSigned;
    use reth_evm::OnStateHook;
@@ -945,14 +891,13 @@ mod tests {
            EthEvmConfig::new(Arc::new(ChainSpec::default())),
            &TreeConfig::default(),
            PrecompileCacheMap::default(),
-            MAINNET.clone(),
        );

-        let header = Header::default();
-        let block_hash = header.hash_slow();
+        let parent_hash = B256::from([1u8; 32]);
+        let block_hash = B256::from([10u8; 32]);
        let block_with_parent = BlockWithParent {
-            block: BlockNumHash { hash: block_hash, number: header.number },
-            parent: header.parent_hash,
+            block: BlockNumHash { hash: block_hash, number: 1 },
+            parent: parent_hash,
        };
        let bundle_state = BundleState::default();

@@ -960,7 +905,7 @@ mod tests {
        assert!(payload_processor.execution_cache.get_cache_for(block_hash).is_none());

        // Update cache with inserted block
-        payload_processor.on_inserted_executed_block(&header, block_with_parent, &bundle_state);
+        payload_processor.on_inserted_executed_block(block_with_parent, &bundle_state);

        // Cache should now exist for the block hash
        let cached = payload_processor.execution_cache.get_cache_for(block_hash);
@@ -975,27 +920,24 @@ mod tests {
            EthEvmConfig::new(Arc::new(ChainSpec::default())),
            &TreeConfig::default(),
            PrecompileCacheMap::default(),
-            MAINNET.clone(),
        );

        // Setup: populate cache with block 1
-        let header1 = Header { number: 1, ..Default::default() };
-        let block1_hash = header1.hash_slow();
+        let block1_hash = B256::from([1u8; 32]);
        payload_processor
            .execution_cache
            .update_with_guard(|slot| *slot = Some(make_saved_cache(block1_hash)));

        // Try to insert block 3 with wrong parent (should skip and keep block 1's cache)
-        let header3 =
-            Header { parent_hash: B256::from([99u8; 32]), number: 3, ..Default::default() };
-        let block3_hash = header3.hash_slow();
+        let wrong_parent = B256::from([99u8; 32]);
+        let block3_hash = B256::from([3u8; 32]);
        let block_with_parent = BlockWithParent {
-            block: BlockNumHash { hash: block3_hash, number: header3.number },
-            parent: header3.parent_hash,
+            block: BlockNumHash { hash: block3_hash, number: 3 },
+            parent: wrong_parent,
        };
        let bundle_state = BundleState::default();

-        payload_processor.on_inserted_executed_block(&header3, block_with_parent, &bundle_state);
+        payload_processor.on_inserted_executed_block(block_with_parent, &bundle_state);

        // Cache should still be for block 1 (unchanged)
        let cached = payload_processor.execution_cache.get_cache_for(block1_hash);
@@ -1111,22 +1053,23 @@ mod tests {
            EthEvmConfig::new(factory.chain_spec()),
            &TreeConfig::default(),
            PrecompileCacheMap::default(),
-            MAINNET.clone(),
        );

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

-        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 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 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,15 +1,16 @@
 //! 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, map::HashSet, B256};
+use alloy_primitives::{
+    keccak256,
+    map::{B256Set, HashSet},
+    B256,
+};
 use crossbeam_channel::{unbounded, Receiver as CrossbeamReceiver, Sender as CrossbeamSender};
 use dashmap::DashMap;
 use derive_more::derive::Deref;
 use metrics::{Gauge, Histogram};
 use reth_metrics::Metrics;
-use reth_provider::AccountReader;
 use reth_revm::state::EvmState;
 use reth_trie::{
    added_removed_keys::MultiAddedRemovedKeys, DecodedMultiProof, HashedPostState, HashedStorage,
@@ -19,35 +20,12 @@ use reth_trie_parallel::{
    proof::ParallelProof,
    proof_task::{
        AccountMultiproofInput, ProofResultContext, ProofResultMessage, ProofWorkerHandle,
+        StorageProofInput,
    },
 };
 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
@@ -104,7 +82,7 @@ pub(super) enum MultiProofMessage {
    /// Prefetch proof targets
    PrefetchProofs(MultiProofTargets),
    /// New state update from transaction execution with its source
-    StateUpdate(Source, EvmState),
+    StateUpdate(StateChangeSource, 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
@@ -115,11 +93,6 @@ 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
@@ -165,6 +138,11 @@ impl ProofSequencer {
        while let Some(pending) = self.pending_proofs.remove(&current_sequence) {
            consecutive_proofs.push(pending);
            current_sequence += 1;
+
+            // if we don't have the next number, stop collecting
+            if !self.pending_proofs.contains_key(&current_sequence) {
+                break;
+            }
        }

        self.next_to_deliver += consecutive_proofs.len() as u64;
@@ -231,10 +209,78 @@ pub(crate) fn evm_state_to_hashed_post_state(update: EvmState) -> HashedPostStat
    hashed_state
 }

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

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

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

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

@@ -664,14 +782,17 @@ impl MultiProofTask {
            available_storage_workers,
            MultiProofTargets::chunks,
            |proof_targets| {
-                self.multiproof_manager.dispatch(MultiproofInput {
-                    source: None,
-                    hashed_state_update: Default::default(),
-                    proof_targets,
-                    proof_sequence_number: self.proof_sequencer.next_sequence(),
-                    state_root_message_sender: self.tx.clone(),
-                    multi_added_removed_keys: Some(multi_added_removed_keys.clone()),
-                });
+                self.multiproof_manager.dispatch(
+                    MultiproofInput {
+                        source: None,
+                        hashed_state_update: Default::default(),
+                        proof_targets,
+                        proof_sequence_number: self.proof_sequencer.next_sequence(),
+                        state_root_message_sender: self.tx.clone(),
+                        multi_added_removed_keys: Some(multi_added_removed_keys.clone()),
+                    }
+                    .into(),
+                );
            },
        );
        self.metrics.prefetch_proof_chunks_histogram.record(num_chunks as f64);
@@ -762,19 +883,9 @@ impl MultiProofTask {
        skip(self, update),
        fields(accounts = update.len(), chunks = 0)
    )]
-    fn on_state_update(&mut self, source: Source, update: EvmState) -> u64 {
+    fn on_state_update(&mut self, source: StateChangeSource, 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);

@@ -819,14 +930,17 @@ impl MultiProofTask {
                );
                spawned_proof_targets.extend_ref(&proof_targets);

-                self.multiproof_manager.dispatch(MultiproofInput {
-                    source: Some(source),
-                    hashed_state_update,
-                    proof_targets,
-                    proof_sequence_number: self.proof_sequencer.next_sequence(),
-                    state_root_message_sender: self.tx.clone(),
-                    multi_added_removed_keys: Some(multi_added_removed_keys.clone()),
-                });
+                self.multiproof_manager.dispatch(
+                    MultiproofInput {
+                        source: Some(source),
+                        hashed_state_update,
+                        proof_targets,
+                        proof_sequence_number: self.proof_sequencer.next_sequence(),
+                        state_root_message_sender: self.tx.clone(),
+                        multi_added_removed_keys: Some(multi_added_removed_keys.clone()),
+                    }
+                    .into(),
+                );
            },
        );
        self.metrics
@@ -868,16 +982,12 @@ impl MultiProofTask {
    /// This preserves ordering without requeuing onto the channel.
    ///
    /// Returns `true` if done, `false` to continue.
-    fn process_multiproof_message<P>(
+    fn process_multiproof_message(
        &mut self,
        msg: MultiProofMessage,
        ctx: &mut MultiproofBatchCtx,
        batch_metrics: &mut MultiproofBatchMetrics,
-        provider: &P,
-    ) -> bool
-    where
-        P: AccountReader,
-    {
+    ) -> bool {
        match msg {
            // Prefetch proofs: batch consecutive prefetch requests up to target/message limits
            MultiProofMessage::PrefetchProofs(targets) => {
@@ -1036,56 +1146,6 @@ 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");
@@ -1178,10 +1238,7 @@ impl MultiProofTask {
        target = "engine::tree::payload_processor::multiproof",
        skip_all
    )]
-    pub(crate) fn run<P>(mut self, provider: P)
-    where
-        P: AccountReader,
-    {
+    pub(crate) fn run(mut self) {
        let mut ctx = MultiproofBatchCtx::new(Instant::now());
        let mut batch_metrics = MultiproofBatchMetrics::default();

@@ -1191,7 +1248,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, &provider) {
+                if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
                    break 'main;
                }
                continue;
@@ -1266,7 +1323,7 @@ impl MultiProofTask {
                        }
                    };

-                    if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics, &provider) {
+                    if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
                        break 'main;
                    }
                }
@@ -1302,9 +1359,6 @@ impl MultiProofTask {
 /// Context for multiproof message batching loop.
 ///
 /// Contains processing state that persists across loop iterations.
-///
-/// Used by `process_multiproof_message` to batch consecutive same-type messages received via
-/// `try_recv` for efficient processing.
 struct MultiproofBatchCtx {
    /// Buffers a non-matching message type encountered during batching.
    /// Processed first in next iteration to preserve ordering while allowing same-type
@@ -1320,7 +1374,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<(Source, EvmState)>,
+    accumulated_state_updates: Vec<(StateChangeSource, EvmState)>,
 }

 impl MultiproofBatchCtx {
@@ -1438,44 +1492,34 @@ 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: Source,
+    batch_source: StateChangeSource,
    batch_update: &EvmState,
-    next_source: Source,
+    next_source: StateChangeSource,
    next_update: &EvmState,
 ) -> bool {
-    if !same_source(batch_source, next_source) {
+    if !same_state_change_source(batch_source, next_source) {
        return false;
    }

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

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

    /// Get a handle to the test runtime, creating it if necessary
@@ -1532,9 +1575,8 @@ mod tests {
        let task_ctx = ProofTaskCtx::new(overlay_factory);
        let proof_handle = ProofWorkerHandle::new(rt_handle, task_ctx, 1, 1);
        let (to_sparse_trie, _receiver) = std::sync::mpsc::channel();
-        let (tx, rx) = crossbeam_channel::unbounded();

-        MultiProofTask::new(proof_handle, to_sparse_trie, Some(1), tx, rx)
+        MultiProofTask::new(proof_handle, to_sparse_trie, Some(1))
    }

    #[test]
@@ -2067,8 +2109,8 @@ mod tests {
        let source = StateChangeSource::Transaction(0);

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

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

-                task.on_state_update(source.into(), merged_update)
+                task.on_state_update(source, merged_update)
            } else {
                panic!("Expected StateUpdate message");
            };
@@ -2131,20 +2173,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.into(), create_state_update(addr_a1, 100)))
+        tx.send(MultiProofMessage::StateUpdate(source_a, create_state_update(addr_a1, 100)))
            .unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source_b.into(), create_state_update(addr_b1, 200)))
+        tx.send(MultiProofMessage::StateUpdate(source_b, create_state_update(addr_b1, 200)))
            .unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a2, 300)))
+        tx.send(MultiProofMessage::StateUpdate(source_a, 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_source(first_source, source_a.into()));
+            assert!(same_state_change_source(first_source, source_a));

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

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

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

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

-            assert!(same_source(batch_source, source_b.into()), "Batch should use matching source");
+            assert!(
+                same_state_change_source(batch_source, source_b),
+                "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));
@@ -2210,7 +2255,7 @@ mod tests {

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

        // Queue: first update dispatched immediately, next two should not merge
        let tx = task.state_root_message_sender();
-        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();
+        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();

        let mut pending_msg: Option<MultiProofMessage> = None;

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

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

            loop {
@@ -2318,7 +2360,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_source(batched_source, source.into()));
+            assert!(same_state_change_source(batched_source, source));
            assert!(batched_update.contains_key(&addr2));
            assert!(!batched_update.contains_key(&addr3));

@@ -2398,8 +2440,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.into(), state_update1)).unwrap();
-        tx.send(MultiProofMessage::StateUpdate(source.into(), state_update2)).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source, state_update1)).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source, state_update2)).unwrap();
        tx.send(MultiProofMessage::PrefetchProofs(targets3.clone())).unwrap();

        // Step 1: Receive and batch PrefetchProofs (should get targets1 + targets2)
@@ -2466,7 +2508,6 @@ 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
@@ -2498,7 +2539,7 @@ mod tests {

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

        let mut ctx = MultiproofBatchCtx::new(Instant::now());
@@ -2507,22 +2548,12 @@ 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,
-            &test_provider
-        ));
+        assert!(!task.process_multiproof_message(first, &mut ctx, &mut batch_metrics));
        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,
-            &test_provider
-        ));
+        assert!(!task.process_multiproof_message(pending, &mut ctx, &mut batch_metrics));

        // Prefetch2 should now be in pending_msg (captured by StateUpdate's batching loop).
        match ctx.pending_msg.take() {
@@ -2594,21 +2625,12 @@ 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.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::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::PrefetchProofs(prefetch2.clone())).unwrap();

        // Simulate the state-machine loop behavior
@@ -2681,44 +2703,4 @@ 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_processor/prewarm.rs
+++ b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
@@ -27,11 +27,10 @@ use alloy_primitives::{keccak256, map::B256Set, B256};
 use crossbeam_channel::Sender as CrossbeamSender;
 use metrics::{Counter, Gauge, Histogram};
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
-use reth_execution_types::ExecutionOutcome;
 use reth_metrics::Metrics;
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{BlockReader, StateProviderFactory, StateReader};
-use reth_revm::{database::StateProviderDatabase, state::EvmState};
+use reth_provider::{BlockReader, StateProviderBox, StateProviderFactory, StateReader};
+use reth_revm::{database::StateProviderDatabase, db::BundleState, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
    sync::{
@@ -87,7 +86,7 @@ where
    /// Sender to emit evm state outcome messages, if any.
    to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
    /// Receiver for events produced by tx execution
-    actions_rx: Receiver<PrewarmTaskEvent<N::Receipt>>,
+    actions_rx: Receiver<PrewarmTaskEvent>,
    /// Parent span for tracing
    parent_span: Span,
 }
@@ -106,7 +105,7 @@ where
        to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
        transaction_count_hint: usize,
        max_concurrency: usize,
-    ) -> (Self, Sender<PrewarmTaskEvent<N::Receipt>>) {
+    ) -> (Self, Sender<PrewarmTaskEvent>) {
        let (actions_tx, actions_rx) = channel();

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

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

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

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

                // Insert state into cache while holding the lock
-                // Access the BundleState through the shared ExecutionOutcome
-                if new_cache.cache().insert_state(execution_outcome.state(), &env.spec_id).is_err()
-                {
+                if new_cache.cache().insert_state(&state).is_err() {
                    // Clear the cache on error to prevent having a polluted cache
                    *cached = None;
                    debug!(target: "engine::caching", "cleared execution cache on update error");
@@ -301,12 +300,12 @@ where
    pub(super) fn run(
        self,
        pending: mpsc::Receiver<impl ExecutableTxFor<Evm> + Clone + Send + 'static>,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
+        actions_tx: Sender<PrewarmTaskEvent>,
    ) {
        // spawn execution tasks.
        self.spawn_all(pending, actions_tx);

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

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

                    finished_execution = true;

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

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

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

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

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

        let state_provider = StateProviderDatabase::new(state_provider);

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

    /// Spawns a worker task for transaction execution and returns its sender channel.
-    fn spawn_workers<Tx>(
-        self,
-        workers_needed: usize,
-        task_executor: &WorkloadExecutor,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
+    fn spawn_worker<Tx>(
+        &self,
+        idx: usize,
+        executor: &WorkloadExecutor,
+        actions_tx: Sender<PrewarmTaskEvent>,
        done_tx: Sender<()>,
-    ) -> Vec<mpsc::Sender<IndexedTransaction<Tx>>>
+    ) -> mpsc::Sender<IndexedTransaction<Tx>>
    where
        Tx: ExecutableTxFor<Evm> + Send + 'static,
    {
-        let mut handles = Vec::with_capacity(workers_needed);
-        let mut receivers = Vec::with_capacity(workers_needed);
+        let (tx, rx) = mpsc::channel();
+        let ctx = self.clone();
+        let span =
+            debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);

-        for _ in 0..workers_needed {
-            let (tx, rx) = mpsc::channel();
-            handles.push(tx);
-            receivers.push(rx);
-        }
-
-        // Spawn a separate task spawning workers in parallel.
-        let executor = task_executor.clone();
-        let span = Span::current();
-        task_executor.spawn_blocking(move || {
+        executor.spawn_blocking(move || {
            let _enter = span.entered();
-            for (idx, rx) in receivers.into_iter().enumerate() {
-                let ctx = self.clone();
-                let actions_tx = actions_tx.clone();
-                let done_tx = done_tx.clone();
-                let span = debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);
-                executor.spawn_blocking(move || {
-                    let _enter = span.entered();
-                    ctx.transact_batch(rx, actions_tx, done_tx);
-                });
-            }
+            ctx.transact_batch(rx, actions_tx, done_tx);
        });

-        handles
+        tx
    }
 }

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

 /// The events the pre-warm task can handle.
-///
-/// Generic over `R` (receipt type) to allow sharing `Arc<ExecutionOutcome<R>>` with the main
-/// execution path without cloning the expensive `BundleState`.
-pub(super) enum PrewarmTaskEvent<R> {
+pub(super) enum PrewarmTaskEvent {
    /// Forcefully terminate all remaining transaction execution.
    TerminateTransactionExecution,
    /// Forcefully terminate the task on demand and update the shared cache with the given output
    /// before exiting.
    Terminate {
-        /// The final execution outcome. Using `Arc` allows sharing with the main execution
-        /// path without cloning the expensive `BundleState`.
-        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
+        /// The final block state output.
+        block_output: Option<BundleState>,
    },
    /// The outcome of a pre-warm task
    Outcome {
--- a/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
+++ b/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
@@ -166,7 +166,8 @@ where

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

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

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

--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/crates/engine/tree/src/tree/payload_validator.rs
@@ -14,10 +14,8 @@ use alloy_consensus::transaction::Either;
 use alloy_eip7928::BlockAccessList;
 use alloy_eips::{eip1898::BlockWithParent, NumHash};
 use alloy_evm::Evm;
-use alloy_primitives::{BlockTimestamp, B256};
-use rayon::prelude::*;
+use alloy_primitives::B256;
 use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock};
-use reth_chainspec::EthereumHardforks;
 use reth_consensus::{ConsensusError, FullConsensus};
 use reth_engine_primitives::{
    ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadValidator,
@@ -35,14 +33,13 @@ use reth_primitives_traits::{
    SealedHeader, SignerRecoverable,
 };
 use reth_provider::{
-    providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockReader, ChainSpecProvider,
-    DatabaseProviderFactory, DatabaseProviderROFactory, ExecutionOutcome, HashedPostStateProvider,
-    ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
-    StateProviderFactory, StateReader, TrieReader,
+    providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockReader,
+    DatabaseProviderFactory, ExecutionOutcome, HashedPostStateProvider, ProviderError,
+    PruneCheckpointReader, StageCheckpointReader, StateProvider, StateProviderFactory, StateReader,
+    StateRootProvider, TrieReader,
 };
 use reth_revm::db::State;
-use reth_storage_errors::db::DatabaseError;
-use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot, TrieInputSorted};
+use reth_trie::{updates::TrieUpdates, HashedPostState, TrieInputSorted};
 use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
 use revm_primitives::Address;
 use std::{
@@ -109,7 +106,6 @@ impl<'a, N: NodePrimitives> TreeCtx<'a, N> {
 pub struct BasicEngineValidator<P, Evm, V>
 where
    Evm: ConfigureEvm,
-    P: ChainSpecProvider<ChainSpec: EthereumHardforks>,
 {
    /// Provider for database access.
    provider: P,
@@ -120,7 +116,7 @@ where
    /// Configuration for the tree.
    config: TreeConfig,
    /// Payload processor for state root computation.
-    payload_processor: PayloadProcessor<Evm, P::ChainSpec>,
+    payload_processor: PayloadProcessor<Evm>,
    /// Precompile cache map.
    precompile_cache_map: PrecompileCacheMap<SpecFor<Evm>>,
    /// Precompile cache metrics.
@@ -143,7 +139,6 @@ where
        + StateProviderFactory
        + StateReader
        + HashedPostStateProvider
-        + ChainSpecProvider<ChainSpec: EthereumHardforks>
        + Clone
        + 'static,
    Evm: ConfigureEvm<Primitives = N> + 'static,
@@ -164,7 +159,6 @@ where
            evm_config.clone(),
            &config,
            precompile_cache_map.clone(),
-            provider.chain_spec(),
        );
        Self {
            provider,
@@ -227,7 +221,7 @@ where
                    .map_err(NewPayloadError::other)?
                    .into();

-                let iter = Either::Left(iter.into_par_iter().map(Either::Left));
+                let iter = Either::Left(iter.into_iter().map(Either::Left));
                let convert = move |tx| {
                    let Either::Left(tx) = tx else { unreachable!() };
                    convert(tx).map(Either::Left).map_err(Either::Left)
@@ -237,9 +231,8 @@ 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_par_iter().map(Either::Right),
-                );
+                let iter =
+                    Either::Right(block.body().clone_transactions().into_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)
@@ -379,8 +372,7 @@ where
        let mut state_provider = ensure_ok!(provider_builder.build());
        drop(_enter);

-        // Fetch parent block. This goes to memory most of the time unless the parent block is
-        // beyond the in-memory buffer.
+        // fetch parent block
        let Some(parent_block) = ensure_ok!(self.sealed_header_by_hash(parent_hash, ctx.state()))
        else {
            return Err(InsertBlockError::new(
@@ -394,14 +386,7 @@ where
            .in_scope(|| self.evm_env_for(&input))
            .map_err(NewPayloadError::other)?;

-        let spec_id = alloy_evm::spec_by_timestamp_and_block_number(
-            &provider_builder.provider_factory.chain_spec(),
-            input.timestamp(),
-            input.num_hash().number,
-        );
-
-        let env =
-            ExecutionEnv { evm_env, hash: input.hash(), parent_hash: input.parent_hash(), spec_id };
+        let env = ExecutionEnv { evm_env, hash: input.hash(), parent_hash: input.parent_hash() };

        // Plan the strategy used for state root computation.
        let strategy = self.plan_state_root_computation();
@@ -412,17 +397,9 @@ where
            "Decided which state root algorithm to run"
        );

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

-        // Extract the BAL, if valid and available
-        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(),
@@ -431,22 +408,26 @@ where
            parent_hash,
            ctx.state(),
            strategy,
-            block_access_list,
        ));

        // Use cached state provider before executing, used in execution after prewarming threads
        // complete
        if let Some((caches, cache_metrics)) = handle.caches().zip(handle.cache_metrics()) {
-            state_provider =
-                Box::new(CachedStateProvider::new(state_provider, caches, cache_metrics));
+            state_provider = Box::new(CachedStateProvider::new_with_caches(
+                state_provider,
+                caches,
+                cache_metrics,
+            ));
        };

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

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

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

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

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

        let mut db = State::builder()
-            .with_database(StateProviderDatabase::new(state_provider))
+            .with_database(StateProviderDatabase::new(&state_provider))
            .with_bundle_update()
            .without_state_clear()
            .build();
@@ -668,6 +652,8 @@ where
    ///
    /// Returns `Ok(_)` if computed successfully.
    /// Returns `Err(_)` if error was encountered during computation.
+    /// `Err(ProviderError::ConsistentView(_))` can be safely ignored and fallback computation
+    /// should be used instead.
    #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
    fn compute_state_root_parallel(
        &self,
@@ -697,36 +683,6 @@ where
        ParallelStateRoot::new(factory, prefix_sets).incremental_root_with_updates()
    }

-    /// Compute state root for the given hashed post state in serial.
-    fn compute_state_root_serial(
-        &self,
-        parent_hash: B256,
-        hashed_state: &HashedPostState,
-        state: &EngineApiTreeState<N>,
-    ) -> ProviderResult<(B256, TrieUpdates)> {
-        let (mut input, block_hash) = self.compute_trie_input(parent_hash, state)?;
-
-        // Extend state overlay with current block's sorted state.
-        input.prefix_sets.extend(hashed_state.construct_prefix_sets());
-        let sorted_hashed_state = hashed_state.clone_into_sorted();
-        Arc::make_mut(&mut input.state).extend_ref(&sorted_hashed_state);
-
-        let TrieInputSorted { nodes, state, .. } = input;
-        let prefix_sets = hashed_state.construct_prefix_sets();
-
-        let factory = OverlayStateProviderFactory::new(self.provider.clone())
-            .with_block_hash(Some(block_hash))
-            .with_trie_overlay(Some(nodes))
-            .with_hashed_state_overlay(Some(state));
-
-        let provider = factory.database_provider_ro()?;
-
-        Ok(StateRoot::new(&provider, &provider)
-            .with_prefix_sets(prefix_sets.freeze())
-            .root_with_updates()
-            .map_err(Into::<DatabaseError>::into)?)
-    }
-
    /// Validates the block after execution.
    ///
    /// This performs:
@@ -821,12 +777,10 @@ 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>,
            impl core::error::Error + Send + Sync + 'static + use<N, P, Evm, V, T>,
-            N::Receipt,
        >,
        InsertBlockErrorKind,
    > {
@@ -852,14 +806,12 @@ 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
@@ -922,7 +874,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() {
+        if self.config.state_root_fallback() || !self.config.has_enough_parallelism() {
            StateRootStrategy::Synchronous
        } else if self.config.use_state_root_task() {
            StateRootStrategy::StateRootTask
@@ -1061,7 +1013,8 @@ where
    fn spawn_deferred_trie_task(
        &self,
        block: RecoveredBlock<N::Block>,
-        execution_outcome: Arc<ExecutionOutcome<N::Receipt>>,
+        output: BlockExecutionOutput<N::Receipt>,
+        block_number: u64,
        ctx: &TreeCtx<'_, N>,
        hashed_state: HashedPostState,
        trie_output: TrieUpdates,
@@ -1111,7 +1064,7 @@ where

        ExecutedBlock::with_deferred_trie_data(
            Arc::new(block),
-            execution_outcome,
+            Arc::new(ExecutionOutcome::from((output, block_number))),
            deferred_trie_data,
        )
    }
@@ -1196,7 +1149,6 @@ where
        + StateProviderFactory
        + StateReader
        + HashedPostStateProvider
-        + ChainSpecProvider<ChainSpec: EthereumHardforks>
        + Clone
        + 'static,
    N: NodePrimitives,
@@ -1238,7 +1190,6 @@ where

    fn on_inserted_executed_block(&self, block: ExecutedBlock<N>) {
        self.payload_processor.on_inserted_executed_block(
-            block.recovered_block.header(),
            block.recovered_block.block_with_parent(),
            block.execution_output.state(),
        );
@@ -1271,14 +1222,6 @@ impl<T: PayloadTypes> BlockOrPayload<T> {
        }
    }

-    /// Returns the timestamp of the block.
-    pub fn timestamp(&self) -> BlockTimestamp {
-        match self {
-            Self::Payload(payload) => payload.timestamp(),
-            Self::Block(block) => block.timestamp(),
-        }
-    }
-
    /// Returns the parent hash of the block.
    pub fn parent_hash(&self) -> B256 {
        match self {
--- a/crates/engine/tree/src/tree/precompile_cache.rs
+++ b/crates/engine/tree/src/tree/precompile_cache.rs
@@ -1,58 +1,50 @@
 //! Contains a precompile cache backed by `schnellru::LruMap` (LRU by length).

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        hash_collector.insert(hash, number)?;
    }
--- a/crates/era-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-"),
+        file_name.starts_with("mainnet-00000-00001-"),
        "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,11 +30,8 @@ pub trait EraFileFormat: Sized {

 /// Era file identifiers
 pub trait EraFileId: Clone {
-    /// 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;
+    /// Convert to standardized file name
+    fn to_file_name(&self) -> String;

    /// Get the network name
    fn network_name(&self) -> &str;
@@ -44,43 +41,6 @@ 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
@@ -194,37 +154,6 @@ 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 {
@@ -235,11 +164,3 @@ 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, EraFileType},
+    common::file_ops::EraFileId,
    e2s::types::{Entry, IndexEntry, SLOT_INDEX},
    era::types::consensus::{CompressedBeaconState, CompressedSignedBeaconBlock},
 };
@@ -163,22 +163,12 @@ 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,
-            include_era_count: false,
-        }
+        Self { network_name: network_name.into(), start_slot, slot_count, hash: None }
    }

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

-    /// 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
+    /// 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
    }
 }

 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
    }
@@ -210,13 +214,24 @@ 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();

-    fn hash(&self) -> Option<[u8; 4]> {
-        self.hash
-    }
-
-    fn include_era_count(&self) -> bool {
-        self.include_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)
+        }
    }
 }

@@ -384,40 +399,4 @@ 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, EraFileType},
+    common::file_ops::EraFileId,
    e2s::types::{Entry, IndexEntry},
    era1::types::execution::{Accumulator, BlockTuple, MAX_BLOCKS_PER_ERA1},
 };
@@ -105,10 +105,6 @@ 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 {
@@ -118,13 +114,7 @@ impl Era1Id {
        start_block: BlockNumber,
        block_count: u32,
    ) -> Self {
-        Self {
-            network_name: network_name.into(),
-            start_block,
-            block_count,
-            hash: None,
-            include_era_count: false,
-        }
+        Self { network_name: network_name.into(), start_block, block_count, hash: None }
    }

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

-    /// 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
+    // 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
    }
 }

 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
    }
@@ -155,13 +149,24 @@ impl EraFileId for Era1Id {
    fn count(&self) -> u32 {
        self.block_count
    }
-
-    fn hash(&self) -> Option<[u8; 4]> {
-        self.hash
-    }
-
-    fn include_era_count(&self) -> bool {
-        self.include_era_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)
+        }
    }
 }

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

    #[test_case::test_case(
        Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]),
-        "mainnet-00000-5ec1ffb8.era1";
+        "mainnet-00000-00001-5ec1ffb8.era1";
        "Mainnet era 0"
    )]
    #[test_case::test_case(
        Era1Id::new("mainnet", 8192, 8192).with_hash([0x5e, 0xcb, 0x9b, 0xf9]),
-        "mainnet-00001-5ecb9bf9.era1";
+        "mainnet-00001-00001-5ecb9bf9.era1";
        "Mainnet era 1"
    )]
    #[test_case::test_case(
        Era1Id::new("sepolia", 0, 8192).with_hash([0x90, 0x91, 0x84, 0x72]),
-        "sepolia-00000-90918472.era1";
+        "sepolia-00000-00001-90918472.era1";
        "Sepolia era 0"
    )]
    #[test_case::test_case(
        Era1Id::new("sepolia", 155648, 8192).with_hash([0xfa, 0x77, 0x00, 0x19]),
-        "sepolia-00019-fa770019.era1";
+        "sepolia-00019-00001-fa770019.era1";
        "Sepolia era 19"
    )]
    #[test_case::test_case(
        Era1Id::new("mainnet", 1000, 100),
-        "mainnet-00000-00000000.era1";
+        "mainnet-00000-00001-00000000.era1";
        "ID without hash"
    )]
    #[test_case::test_case(
        Era1Id::new("sepolia", 101130240, 8192).with_hash([0xab, 0xcd, 0xef, 0x12]),
-        "sepolia-12345-abcdef12.era1";
+        "sepolia-12345-00001-abcdef12.era1";
        "Large block number era 12345"
    )]
-    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) {
+    fn test_era1id_file_naming(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/Cargo.toml
+++ b/crates/ethereum/cli/Cargo.toml
@@ -38,7 +38,6 @@ tempfile.workspace = true
 default = []

 otlp = ["reth-tracing/otlp", "reth-node-core/otlp"]
-samply = ["reth-tracing/samply", "reth-node-core/samply"]

 dev = ["reth-cli-commands/arbitrary"]

--- a/crates/ethereum/cli/src/app.rs
+++ b/crates/ethereum/cli/src/app.rs
@@ -154,9 +154,7 @@ 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_command_until_exit(|ctx| command.execute::<N>(ctx))
-        }
+        Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
        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/cli/src/interface.rs
+++ b/crates/ethereum/cli/src/interface.rs
@@ -76,7 +76,10 @@ impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, Rpc: RpcModuleValidator>
    ///
    /// This method is used to prepare the CLI for execution by wrapping it in a
    /// [`CliApp`] that can be further configured before running.
-    pub fn configure(self) -> CliApp<C, Ext, Rpc> {
+    pub fn configure(self) -> CliApp<C, Ext, Rpc>
+    where
+        C: ChainSpecParser<ChainSpec = ChainSpec>,
+    {
        CliApp::new(self)
    }

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

 /// Compare the calculated receipts root with the expected receipts root, also compare
--- a/crates/ethereum/evm/src/lib.rs
+++ b/crates/ethereum/evm/src/lib.rs
@@ -19,36 +19,31 @@ 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, MAINNET};
+use reth_chainspec::{ChainSpec, EthChainSpec, EthereumHardforks, MAINNET};
 use reth_ethereum_primitives::{Block, EthPrimitives, TransactionSigned};
 use reth_evm::{
-    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEvm, EvmEnv, EvmFactory,
-    NextBlockEnvAttributes, TransactionEnv,
+    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEngineEvm, ConfigureEvm,
+    EvmEnv, EvmEnvFor, EvmFactory, ExecutableTxIterator, ExecutionCtxFor, NextBlockEnvAttributes,
+    TransactionEnv,
 };
-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,
+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,
 };
-
-pub use alloy_evm::EthEvm;

 mod config;
 use alloy_evm::eth::spec::EthExecutorSpec;
@@ -211,7 +206,6 @@ where
    }
 }

-#[cfg(feature = "std")]
 impl<ChainSpec, EvmF> ConfigureEngineEvm<ExecutionData> for EthEvmConfig<ChainSpec, EvmF>
 where
    ChainSpec: EthExecutorSpec + EthChainSpec<Header = Header> + Hardforks + 'static,
@@ -292,7 +286,7 @@ where
        &self,
        payload: &ExecutionData,
    ) -> Result<impl ExecutableTxIterator<Self>, Self::Error> {
-        let txs = payload.payload.transactions().clone();
+        let txs = payload.payload.transactions().clone().into_iter();
        let convert = |tx: Bytes| {
            let tx =
                TxTy::<Self::Primitives>::decode_2718_exact(tx.as_ref()).map_err(AnyError::new)?;
--- a/crates/ethereum/hardforks/src/display.rs
+++ b/crates/ethereum/hardforks/src/display.rs
@@ -170,7 +170,7 @@ impl DisplayHardforks {
        let mut post_merge = Vec::new();

        for (fork, condition, metadata) in hardforks {
-            let display_fork = DisplayFork {
+            let mut display_fork = DisplayFork {
                name: fork.name().to_string(),
                activated_at: condition,
                eip: None,
@@ -181,7 +181,12 @@ impl DisplayHardforks {
                ForkCondition::Block(_) => {
                    pre_merge.push(display_fork);
                }
-                ForkCondition::TTD { .. } => {
+                ForkCondition::TTD { activation_block_number, total_difficulty, fork_block } => {
+                    display_fork.activated_at = ForkCondition::TTD {
+                        activation_block_number,
+                        fork_block,
+                        total_difficulty,
+                    };
                    with_merge.push(display_fork);
                }
                ForkCondition::Timestamp(_) => {
--- a/crates/ethereum/node/Cargo.toml
+++ b/crates/ethereum/node/Cargo.toml
@@ -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, features = ["std"] }
+reth-evm-ethereum.workspace = true
 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, features = ["std"] }
+reth-engine-primitives.workspace = true
 reth-payload-primitives.workspace = true

 # ethereum
@@ -61,7 +61,6 @@ reth-node-core.workspace = true
 reth-e2e-test-utils.workspace = true
 reth-tasks.workspace = true
 reth-testing-utils.workspace = true
-reth-stages-types.workspace = true
 tempfile.workspace = true
 jsonrpsee-core.workspace = true

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

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

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

 mod blobs;
-mod custom_genesis;
 mod dev;
 mod eth;
 mod p2p;
--- a/crates/ethereum/node/tests/it/testing.rs
+++ b/crates/ethereum/node/tests/it/testing.rs
@@ -28,7 +28,6 @@ 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, features = ["std"] }
+reth-evm-ethereum.workspace = true
 reth-errors.workspace = true
 reth-chainspec.workspace = true
 reth-payload-validator.workspace = true
--- a/crates/ethereum/payload/src/lib.rs
+++ b/crates/ethereum/payload/src/lib.rs
@@ -153,9 +153,9 @@ where
    let PayloadConfig { parent_header, attributes } = config;

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

    let mut builder = evm_config
        .builder_for_next_block(
@@ -358,8 +358,7 @@ where
        return Ok(BuildOutcome::Aborted { fees: total_fees, cached_reads })
    }

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

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

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

--- a/crates/ethereum/reth/Cargo.toml
+++ b/crates/ethereum/reth/Cargo.toml
@@ -79,10 +79,7 @@ arbitrary = [
    "alloy-rpc-types-engine?/arbitrary",
    "reth-codecs?/arbitrary",
 ]
-keccak-cache-global = [
-    "reth-node-ethereum?/keccak-cache-global",
-    "reth-node-core?/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,7 +32,6 @@ 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
@@ -41,7 +40,6 @@ 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,5 +1,4 @@
 use crate::{execute::ExecutableTxFor, ConfigureEvm, EvmEnvFor, ExecutionCtxFor};
-use rayon::prelude::*;

 /// [`ConfigureEvm`] extension providing methods for executing payloads.
 pub trait ConfigureEngineEvm<ExecutionData>: ConfigureEvm {
@@ -22,7 +21,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::IntoIter, Self::Convert)> + Send + 'static {
+pub trait ExecutableTxTuple: Into<(Self::Iter, 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,
@@ -33,10 +32,8 @@ pub trait ExecutableTxTuple: Into<(Self::IntoIter, Self::Convert)> + Send + 'sta
    /// Errors that may occur while recovering or decoding transactions.
    type Error: core::error::Error + Send + Sync + 'static;

-    /// Iterator over [`ExecutableTxTuple::Tx`].
-    type IntoIter: IntoParallelIterator<Item = Self::RawTx, Iter: IndexedParallelIterator>
-        + Send
-        + 'static;
+    /// Iterator over [`ExecutableTxTuple::Tx`]
+    type Iter: Iterator<Item = Self::RawTx> + 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.
@@ -48,14 +45,14 @@ where
    RawTx: Send + Sync + 'static,
    Tx: Clone + Send + Sync + 'static,
    Err: core::error::Error + Send + Sync + 'static,
-    I: IntoParallelIterator<Item = RawTx, Iter: IndexedParallelIterator> + Send + 'static,
+    I: Iterator<Item = RawTx> + Send + 'static,
    F: Fn(RawTx) -> Result<Tx, Err> + Send + Sync + 'static,
 {
    type RawTx = RawTx;
    type Tx = Tx;
    type Error = Err;

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

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

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

 #[cfg(feature = "metrics")]
 pub mod metrics;
@@ -61,6 +59,8 @@ pub use alloy_evm::{
    *,
 };

+pub use alloy_evm::block::state_changes as state_change;
+
 /// A complete configuration of EVM for Reth.
 ///
 /// This trait encapsulates complete configuration required for transaction execution and block
--- a/crates/exex/exex/src/backfill/test_utils.rs
+++ b/crates/exex/exex/src/backfill/test_utils.rs
@@ -13,7 +13,7 @@ use reth_evm_ethereum::EthEvmConfig;
 use reth_node_api::NodePrimitives;
 use reth_primitives_traits::{Block as _, RecoveredBlock};
 use reth_provider::{
-    providers::ProviderNodeTypes, BlockWriter as _, ExecutionOutcome, LatestStateProvider,
+    providers::ProviderNodeTypes, BlockWriter as _, ExecutionOutcome, LatestStateProviderRef,
    ProviderFactory,
 };
 use reth_revm::database::StateProviderDatabase;
@@ -69,7 +69,7 @@ where

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

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

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

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

        let provider = BlockchainProvider::new(provider_factory).unwrap();
--- a/crates/exex/exex/src/notifications.rs
+++ b/crates/exex/exex/src/notifications.rs
@@ -481,12 +481,12 @@ mod tests {
            &mut rng,
            genesis_block.number + 1,
            BlockParams { parent: Some(genesis_hash), tx_count: Some(0), ..Default::default() },
-        )
-        .try_recover()?;
-        let node_head = node_head_block.num_hash();
+        );
        let provider_rw = provider_factory.provider_rw()?;
-        provider_rw.insert_block(&node_head_block)?;
+        provider_rw.insert_block(node_head_block.clone().try_recover()?)?;
        provider_rw.commit()?;
+
+        let node_head = node_head_block.num_hash();
        let exex_head =
            ExExHead { block: BlockNumHash { number: genesis_block.number, hash: genesis_hash } };

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

-            let first_block_number = committed_chain.first().number();
-            self.lowest_committed_block_height = Some(
-                self.lowest_committed_block_height
-                    .map_or(first_block_number, |lowest| lowest.min(first_block_number)),
-            );
            self.highest_committed_block_height = Some(committed_chain.tip().number());
        }
    }
--- a/crates/exex/test-utils/src/lib.rs
+++ b/crates/exex/test-utils/src/lib.rs
@@ -20,9 +20,7 @@ use futures_util::FutureExt;
 use reth_chainspec::{ChainSpec, MAINNET};
 use reth_consensus::test_utils::TestConsensus;
 use reth_db::{
-    test_utils::{
-        create_test_rocksdb_dir, create_test_rw_db, create_test_static_files_dir, TempDatabase,
-    },
+    test_utils::{create_test_rw_db, create_test_static_files_dir, TempDatabase},
    DatabaseEnv,
 };
 use reth_db_common::init::init_genesis;
@@ -52,7 +50,7 @@ use reth_node_ethereum::{
 use reth_payload_builder::noop::NoopPayloadBuilderService;
 use reth_primitives_traits::{Block as _, RecoveredBlock};
 use reth_provider::{
-    providers::{BlockchainProvider, RocksDBProvider, StaticFileProvider},
+    providers::{BlockchainProvider, StaticFileProvider},
    BlockReader, EthStorage, ProviderFactory,
 };
 use reth_tasks::TaskManager;
@@ -241,13 +239,11 @@ 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:
-    /// 10 seconds.
+    /// 10min.
    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.clone()?;
+        let resolver = self.external_ip_resolver?;
        let interval = self.resolve_external_ip_interval?;
        Some(ResolveNatInterval::interval_at(resolver, tokio::time::Instant::now(), interval))
    }
@@ -212,13 +212,13 @@ impl Discv4ConfigBuilder {
        self
    }

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

-    /// Whether to enforce expiration timestamps in messages.
+    /// Whether to enable EIP-868
    pub const fn enforce_expiration_timestamps(
        &mut self,
        enforce_expiration_timestamps: bool,
@@ -275,7 +275,10 @@ impl Discv4ConfigBuilder {
    }

    /// Configures if and how the external IP of the node should be resolved.
-    pub fn external_ip_resolver(&mut self, external_ip_resolver: Option<NatResolver>) -> &mut Self {
+    pub const 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,13 +625,10 @@ impl Discv4Service {
        self.lookup_interval = tokio::time::interval(duration);
    }

-    /// 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.
+    /// Sets the external Ip to the configured external IP if [`NatResolver::ExternalIp`].
    fn resolve_external_ip(&mut self) {
        if let Some(r) = &self.resolve_external_ip_interval &&
-            let Some(external_ip) =
-                r.resolver().clone().as_external_ip(self.local_node_record.udp_port)
+            let Some(external_ip) = r.resolver().as_external_ip()
        {
            self.set_external_ip_addr(external_ip);
        }
--- a/crates/net/ecies/src/algorithm.rs
+++ b/crates/net/ecies/src/algorithm.rs
@@ -312,6 +312,7 @@ impl ECIES {

    /// Create a new ECIES client with the given static secret key and remote peer ID.
    pub fn new_client(secret_key: SecretKey, remote_id: PeerId) -> Result<Self, ECIESError> {
+        // TODO(rand): use rng for nonce
        let mut rng = rng();
        let nonce = B256::random();
        let ephemeral_secret_key = SecretKey::new(&mut rng);
--- a/crates/net/eth-wire-types/src/broadcast.rs
+++ b/crates/net/eth-wire-types/src/broadcast.rs
@@ -169,7 +169,7 @@ impl NewPooledTransactionHashes {
                matches!(version, EthVersion::Eth67 | EthVersion::Eth66)
            }
            Self::Eth68(_) => {
-                matches!(version, EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70)
+                matches!(version, EthVersion::Eth68 | EthVersion::Eth69)
            }
        }
    }
--- a/crates/net/eth-wire-types/src/capability.rs
+++ b/crates/net/eth-wire-types/src/capability.rs
@@ -100,16 +100,6 @@ 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 {
@@ -128,26 +118,10 @@ 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_v69() ||
-            self.is_eth_v70()
+        self.is_eth_v66() || self.is_eth_v67() || self.is_eth_v68()
    }
 }

@@ -167,7 +141,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..=70)?; // Valid eth protocol versions are 66-70
+        let version = u.int_in_range(66..=69)?; // Valid eth protocol versions are 66-69
                                                // Only generate valid eth protocol name for now since it's the only supported protocol
        Ok(Self::new_static("eth", version))
    }
@@ -181,8 +155,6 @@ pub struct Capabilities {
    eth_66: bool,
    eth_67: bool,
    eth_68: bool,
-    eth_69: bool,
-    eth_70: bool,
 }

 impl Capabilities {
@@ -192,8 +164,6 @@ 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,
        }
    }
@@ -212,7 +182,7 @@ impl Capabilities {
    /// Whether the peer supports `eth` sub-protocol.
    #[inline]
    pub const fn supports_eth(&self) -> bool {
-        self.eth_70 || self.eth_69 || self.eth_68 || self.eth_67 || self.eth_66
+        self.eth_68 || self.eth_67 || self.eth_66
    }

    /// Whether this peer supports eth v66 protocol.
@@ -232,18 +202,6 @@ 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 {
@@ -266,8 +224,6 @@ 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 through 70.
+//! Implements Ethereum wire protocol for versions 66, 67, and 68.
 //! 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, GetReceipts70, NewPooledTransactionHashes66,
+    GetNodeData, GetPooledTransactions, GetReceipts, NewPooledTransactionHashes66,
    NewPooledTransactionHashes68, NodeData, PooledTransactions, Receipts, Status, StatusEth69,
    Transactions,
 };
 use crate::{
    status::StatusMessage, BlockRangeUpdate, EthNetworkPrimitives, EthVersion, NetworkPrimitives,
-    RawCapabilityMessage, Receipts69, Receipts70, SharedTransactions,
+    RawCapabilityMessage, Receipts69, SharedTransactions,
 };
 use alloc::{boxed::Box, string::String, sync::Arc};
 use alloy_primitives::{
@@ -111,29 +111,13 @@ impl<N: NetworkPrimitives> ProtocolMessage<N> {
                }
                EthMessage::NodeData(RequestPair::decode(buf)?)
            }
-            EthMessageID::GetReceipts => {
-                if version >= EthVersion::Eth70 {
-                    EthMessage::GetReceipts70(RequestPair::decode(buf)?)
-                } else {
-                    EthMessage::GetReceipts(RequestPair::decode(buf)?)
-                }
-            }
+            EthMessageID::GetReceipts => 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)?)
-                    }
+                if version < EthVersion::Eth69 {
+                    EthMessage::Receipts(RequestPair::decode(buf)?)
+                } else {
+                    // with eth69, receipts no longer include the bloom
+                    EthMessage::Receipts69(RequestPair::decode(buf)?)
                }
            }
            EthMessageID::BlockRangeUpdate => {
@@ -221,9 +205,6 @@ 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> {
@@ -278,12 +259,6 @@ 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",
@@ -296,16 +271,6 @@ 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",
@@ -335,8 +300,8 @@ impl<N: NetworkPrimitives> EthMessage<N> {
            Self::PooledTransactions(_) => EthMessageID::PooledTransactions,
            Self::GetNodeData(_) => EthMessageID::GetNodeData,
            Self::NodeData(_) => EthMessageID::NodeData,
-            Self::GetReceipts(_) | Self::GetReceipts70(_) => EthMessageID::GetReceipts,
-            Self::Receipts(_) | Self::Receipts69(_) | Self::Receipts70(_) => EthMessageID::Receipts,
+            Self::GetReceipts(_) => EthMessageID::GetReceipts,
+            Self::Receipts(_) | Self::Receipts69(_) => EthMessageID::Receipts,
            Self::BlockRangeUpdate(_) => EthMessageID::BlockRangeUpdate,
            Self::Other(msg) => EthMessageID::Other(msg.id as u8),
        }
@@ -349,7 +314,6 @@ impl<N: NetworkPrimitives> EthMessage<N> {
            Self::GetBlockBodies(_) |
                Self::GetBlockHeaders(_) |
                Self::GetReceipts(_) |
-                Self::GetReceipts70(_) |
                Self::GetPooledTransactions(_) |
                Self::GetNodeData(_)
        )
@@ -362,40 +326,11 @@ 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> {
@@ -416,10 +351,8 @@ 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),
        }
@@ -441,10 +374,8 @@ 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,42 +17,6 @@ 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)]
@@ -94,13 +58,7 @@ pub struct Receipts69<T = Receipt>(pub Vec<Vec<T>>);
 impl<T: TxReceipt> Receipts69<T> {
    /// Encodes all receipts with the bloom filter.
    ///
-    /// 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.
+    /// Note: This is an expensive operation that recalculates the bloom for each receipt.
    pub fn into_with_bloom(self) -> Receipts<T> {
        Receipts(
            self.0
@@ -117,68 +75,6 @@ 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::*;
@@ -329,70 +225,4 @@ 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/70).
+    /// The eth protocol version (e.g. eth/66 to eth/69).
    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/70).
+    /// Sets the eth protocol version (e.g., eth/66, eth/69).
    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+/eth/70 version
-/// (omits TD, includes block range).
+/// `StatusMessage` can store either the Legacy version (with TD) or the
+/// eth/69 version (omits TD).
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum StatusMessage {
@@ -546,24 +546,6 @@ 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,8 +27,6 @@ pub enum EthVersion {
    Eth68 = 68,
    /// The `eth` protocol version 69.
    Eth69 = 69,
-    /// The `eth` protocol version 70.
-    Eth70 = 70,
 }

 impl EthVersion {
@@ -57,11 +55,6 @@ 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).
@@ -103,7 +96,6 @@ 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())),
        }
    }
@@ -128,7 +120,6 @@ 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())),
        }
    }
@@ -158,7 +149,6 @@ impl From<EthVersion> for &'static str {
            EthVersion::Eth67 => "67",
            EthVersion::Eth68 => "68",
            EthVersion::Eth69 => "69",
-            EthVersion::Eth70 => "70",
        }
    }
 }
@@ -205,7 +195,7 @@ impl Decodable for ProtocolVersion {

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

@@ -215,7 +205,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!(EthVersion::Eth70, EthVersion::try_from("70").unwrap());
+        assert_eq!(Err(ParseVersionError("70".to_string())), EthVersion::try_from("70"));
    }

    #[test]
@@ -224,18 +214,12 @@ 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!(EthVersion::Eth70, "70".parse().unwrap());
+        assert_eq!(Err(ParseVersionError("70".to_string())), "70".parse::<EthVersion>());
    }

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

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

--- a/Show More
+++ b/Show More