merrge

- Updated the handling of state updates to ensure that removals are recorded before merging updates, preserving the integrity of intermediate deletions.
- Introduced a new method, `record_removals`, to track monotonic removals from EVM state updates, ensuring that once a key is marked as removed, it remains so for proof invalidation.
- Improved comments for clarity on the removal tracking process and its implications for state updates.

.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.

.github/assets/hive/run_simulator.sh

@@ -7,7 +7,7 @@ sim="${1}"
 limit="${2}"
 
 run_hive() {
-    hive --sim "${sim}" --sim.limit "${limit}" --sim.parallelism 16 --client reth 2>&1 | tee /tmp/log || true
+    hive --sim "${sim}" --sim.limit "${limit}" --sim.parallelism 8 --client reth 2>&1 | tee /tmp/log || true
 }
 
 check_log() {

.github/workflows/bench.yml

@@ -11,7 +11,6 @@ env:
   CARGO_TERM_COLOR: always
   BASELINE: base
   SEED: reth
-  RUSTC_WRAPPER: "sccache"
 
 name: bench
 jobs:
@@ -23,7 +22,6 @@ jobs:
           submodules: true
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/book.yml

@@ -10,12 +10,9 @@ on:
     types: [opened, reopened, synchronize, closed]
   merge_group:
 
-env:
-  RUSTC_WRAPPER: "sccache"
-
 jobs:
   build:
-    runs-on: depot-ubuntu-latest-8
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 90
     steps:
       - name: Checkout
@@ -36,8 +33,6 @@ jobs:
       - name: Install Rust nightly
         uses: dtolnay/rust-toolchain@nightly
 
-      - uses: mozilla-actions/sccache-action@v0.0.9
-
       - name: Build docs
         run: cd docs/vocs && bash scripts/build-cargo-docs.sh
 

.github/workflows/compact.yml

@@ -13,12 +13,11 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
-  RUSTC_WRAPPER: "sccache"
 
 name: compact-codec
 jobs:
   compact-codec:
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     strategy:
       matrix:
         bin:
@@ -27,7 +26,6 @@ jobs:
     steps:
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/e2e.yml

@@ -11,7 +11,6 @@ on:
 env:
   CARGO_TERM_COLOR: always
   SEED: rustethereumethereumrust
-  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -20,14 +19,13 @@ concurrency:
 jobs:
   test:
     name: e2e-testsuite
-    runs-on: depot-ubuntu-latest-4
+    runs-on: depot-ubuntu-latest-16
     env:
       RUST_BACKTRACE: 1
     timeout-minutes: 90
     steps:
       - uses: actions/checkout@v6
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: taiki-e/install-action@nextest
       - uses: Swatinem/rust-cache@v2
         with:

.github/workflows/hive.yml

@@ -24,8 +24,7 @@ jobs:
   prepare-hive:
     if: github.repository == 'paradigmxyz/reth'
     timeout-minutes: 45
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     steps:
       - uses: actions/checkout@v6
       - name: Checkout hive tests
@@ -45,7 +44,7 @@ jobs:
 
       - name: Restore hive assets cache
         id: cache-hive
-        uses: actions/cache@v5
+        uses: actions/cache@v4
         with:
           path: ./hive_assets
           key: hive-assets-${{ steps.hive-commit.outputs.hash }}-${{ hashFiles('.github/assets/hive/build_simulators.sh') }}
@@ -68,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
@@ -179,8 +178,7 @@ jobs:
       - prepare-reth
       - prepare-hive
     name: run ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     permissions:
       issues: write
     steps:
@@ -189,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
@@ -247,7 +245,7 @@ jobs:
   notify-on-error:
     needs: test
     if: failure()
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.github/workflows/integration.yml

@@ -14,7 +14,6 @@ on:
 env:
   CARGO_TERM_COLOR: always
   SEED: rustethereumethereumrust
-  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -24,7 +23,7 @@ jobs:
   test:
     name: test / ${{ matrix.network }}
     if: github.event_name != 'schedule'
-    runs-on: depot-ubuntu-latest-4
+    runs-on: depot-ubuntu-latest-16
     env:
       RUST_BACKTRACE: 1
     strategy:
@@ -38,7 +37,6 @@ jobs:
       - name: Install Geth
         run: .github/assets/install_geth.sh
       - uses: taiki-e/install-action@nextest
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -76,7 +74,6 @@ jobs:
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
       - uses: taiki-e/install-action@nextest
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.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
@@ -85,7 +85,7 @@ jobs:
   notify-on-error:
     needs: test
     if: failure()
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.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
@@ -58,7 +58,7 @@ jobs:
   notify-on-error:
     needs: test
     if: failure()
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.github/workflows/lint.yml

@@ -8,7 +8,6 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
-  RUSTC_WRAPPER: "sccache"
 
 jobs:
   clippy-binaries:
@@ -27,7 +26,6 @@ jobs:
       - uses: dtolnay/rust-toolchain@clippy
         with:
           components: clippy
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -42,7 +40,7 @@ jobs:
 
   clippy:
     name: clippy
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
@@ -50,7 +48,6 @@ jobs:
       - uses: dtolnay/rust-toolchain@nightly
         with:
           components: clippy
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -68,7 +65,6 @@ jobs:
         with:
           target: wasm32-wasip1
       - uses: taiki-e/install-action@cargo-hack
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -88,7 +84,6 @@ jobs:
         with:
           target: riscv32imac-unknown-none-elf
       - uses: taiki-e/install-action@cargo-hack
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -98,18 +93,17 @@ jobs:
 
   crate-checks:
     name: crate-checks (${{ matrix.partition }}/${{ matrix.total_partitions }})
-    runs-on: depot-ubuntu-latest-4
+    runs-on: depot-ubuntu-latest-16
     strategy:
       matrix:
-        partition: [1, 2, 3]
-        total_partitions: [3]
+        partition: [1, 2]
+        total_partitions: [2]
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
       - uses: taiki-e/install-action@cargo-hack
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -130,7 +124,6 @@ jobs:
       - uses: dtolnay/rust-toolchain@master
         with:
           toolchain: "1.88" # MSRV
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -140,13 +133,12 @@ jobs:
 
   docs:
     name: docs
-    runs-on: depot-ubuntu-latest-4
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@nightly
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -166,19 +158,17 @@ jobs:
       - uses: dtolnay/rust-toolchain@nightly
         with:
           components: rustfmt
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - name: Run fmt
         run: cargo fmt --all --check
 
   udeps:
     name: udeps
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@nightly
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -187,13 +177,12 @@ jobs:
 
   book:
     name: book
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@nightly
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -245,14 +234,17 @@ jobs:
 
   # Checks that selected crates can compile with power set of features
   features:
-    name: features
-    runs-on: depot-ubuntu-latest
+    name: features (${{ matrix.partition }}/${{ matrix.total_partitions }})
+    runs-on: depot-ubuntu-latest-16
+    strategy:
+      matrix:
+        partition: [1, 2]
+        total_partitions: [2]
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@clippy
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -264,7 +256,7 @@ jobs:
             --package reth-primitives-traits \
             --package reth-primitives \
             --feature-powerset \
-            --depth 2
+            --partition ${{ matrix.partition }}/${{ matrix.total_partitions }}
         env:
           RUSTFLAGS: -D warnings
 
@@ -275,9 +267,8 @@ jobs:
     steps:
       - uses: actions/checkout@v6
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: rui314/setup-mold@v1
-      - uses: taiki-e/cache-cargo-install-action@v3
+      - uses: taiki-e/cache-cargo-install-action@v2
         with:
           tool: zepter
       - name: Eagerly pull dependencies

.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

.github/workflows/release.yml

@@ -22,7 +22,6 @@ env:
   CARGO_TERM_COLOR: always
   DOCKER_IMAGE_NAME_URL: https://ghcr.io/${{ github.repository_owner }}/reth
   DOCKER_OP_IMAGE_NAME_URL: https://ghcr.io/${{ github.repository_owner }}/op-reth
-  RUSTC_WRAPPER: "sccache"
 
 jobs:
   dry-run:
@@ -52,7 +51,6 @@ jobs:
     steps:
       - uses: actions/checkout@v6
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - name: Verify crate version matches tag
         # Check that the Cargo version starts with the tag,
         # so that Cargo version 1.4.8 can be matched against both v1.4.8 and v1.4.8-rc.1
@@ -106,7 +104,6 @@ jobs:
       - uses: dtolnay/rust-toolchain@stable
         with:
           target: ${{ matrix.configs.target }}
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - name: Install cross main
         id: cross_main
         run: |
@@ -144,14 +141,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 +170,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: |

.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/

.github/workflows/stage.yml

@@ -12,7 +12,6 @@ env:
   CARGO_TERM_COLOR: always
   FROM_BLOCK: 0
   TO_BLOCK: 50000
-  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -32,7 +31,6 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/sync-era.yml

@@ -9,7 +9,6 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
-  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -42,7 +41,6 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/sync.yml

@@ -9,7 +9,6 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
-  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -42,7 +41,6 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/unit.yml

@@ -11,7 +11,6 @@ on:
 env:
   CARGO_TERM_COLOR: always
   SEED: rustethereumethereumrust
-  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -20,7 +19,7 @@ concurrency:
 jobs:
   test:
     name: test / ${{ matrix.type }} (${{ matrix.partition }}/${{ matrix.total_partitions }})
-    runs-on: depot-ubuntu-latest-4
+    runs-on: depot-ubuntu-latest-16
     env:
       RUST_BACKTRACE: 1
     strategy:
@@ -47,7 +46,6 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -93,7 +91,6 @@ jobs:
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
       - uses: taiki-e/install-action@nextest
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -101,7 +98,7 @@ jobs:
 
   doc:
     name: doc tests
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     env:
       RUST_BACKTRACE: 1
     timeout-minutes: 30
@@ -109,7 +106,6 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.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"

.github/workflows/windows.yml

@@ -9,12 +9,9 @@ on:
     branches: [main]
   merge_group:
 
-env:
-  RUSTC_WRAPPER: "sccache"
-
 jobs:
   check-reth:
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 60
 
     steps:
@@ -24,7 +21,6 @@ jobs:
         with:
           target: x86_64-pc-windows-gnu
       - uses: taiki-e/install-action@cross
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true
@@ -34,7 +30,7 @@ jobs:
         run: cargo check --target x86_64-pc-windows-gnu
 
   check-op-reth:
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     timeout-minutes: 60
 
     steps:
@@ -44,7 +40,6 @@ jobs:
         with:
           target: x86_64-pc-windows-gnu
       - uses: taiki-e/install-action@cross
-      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.gitignore

@@ -12,9 +12,6 @@ target/
 # Generated by Intellij-based IDEs.
 .idea
 
-# ck-search metadata
-.ck
-
 # Generated by MacOS
 .DS_Store
 

Cargo.toml

@@ -376,11 +376,11 @@ reth-era-utils = { path = "crates/era-utils" }
 reth-errors = { path = "crates/errors" }
 reth-eth-wire = { path = "crates/net/eth-wire" }
 reth-eth-wire-types = { path = "crates/net/eth-wire-types" }
-reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-cli = { path = "crates/ethereum/cli", default-features = false }
 reth-ethereum-consensus = { path = "crates/ethereum/consensus", default-features = false }
 reth-ethereum-engine-primitives = { path = "crates/ethereum/engine-primitives", default-features = false }
 reth-ethereum-forks = { path = "crates/ethereum/hardforks", default-features = false }
+reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-primitives = { path = "crates/ethereum/primitives", default-features = false }
 reth-ethereum = { path = "crates/ethereum/reth" }
 reth-etl = { path = "crates/etl" }
@@ -481,58 +481,57 @@ revm-primitives = { version = "21.0.2", default-features = false }
 revm-interpreter = { version = "31.1.0", default-features = false }
 revm-database-interface = { version = "8.0.5", default-features = false }
 op-revm = { version = "14.1.0", default-features = false }
-revm-inspectors = "0.33.2"
+revm-inspectors = "0.33.1"
 
 # eth
 alloy-chains = { version = "0.2.5", default-features = false }
 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-evm = { version = "0.24.1", default-features = false }
+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"
 
-alloy-consensus = { version = "1.2.1", default-features = false }
-alloy-contract = { version = "1.2.1", default-features = false }
-alloy-eips = { version = "1.2.1", default-features = false }
-alloy-genesis = { version = "1.2.1", default-features = false }
-alloy-json-rpc = { version = "1.2.1", default-features = false }
-alloy-network = { version = "1.2.1", default-features = false }
-alloy-network-primitives = { version = "1.2.1", default-features = false }
-alloy-provider = { version = "1.2.1", features = ["reqwest", "debug-api"], default-features = false }
-alloy-pubsub = { version = "1.2.1", default-features = false }
-alloy-rpc-client = { version = "1.2.1", default-features = false }
-alloy-rpc-types = { version = "1.2.1", features = ["eth"], default-features = false }
-alloy-rpc-types-admin = { version = "1.2.1", default-features = false }
-alloy-rpc-types-anvil = { version = "1.2.1", default-features = false }
-alloy-rpc-types-beacon = { version = "1.2.1", default-features = false }
-alloy-rpc-types-debug = { version = "1.2.1", default-features = false }
-alloy-rpc-types-engine = { version = "1.2.1", default-features = false }
-alloy-rpc-types-eth = { version = "1.2.1", default-features = false }
-alloy-rpc-types-mev = { version = "1.2.1", default-features = false }
-alloy-rpc-types-trace = { version = "1.2.1", default-features = false }
-alloy-rpc-types-txpool = { version = "1.2.1", default-features = false }
-alloy-serde = { version = "1.2.1", default-features = false }
-alloy-signer = { version = "1.2.1", default-features = false }
-alloy-signer-local = { version = "1.2.1", default-features = false }
-alloy-transport = { version = "1.2.1" }
-alloy-transport-http = { version = "1.2.1", features = ["reqwest-rustls-tls"], default-features = false }
-alloy-transport-ipc = { version = "1.2.1", default-features = false }
-alloy-transport-ws = { version = "1.2.1", default-features = false }
+alloy-consensus = { version = "1.1.3", default-features = false }
+alloy-contract = { version = "1.1.3", default-features = false }
+alloy-eips = { version = "1.1.3", default-features = false }
+alloy-genesis = { version = "1.1.3", default-features = false }
+alloy-json-rpc = { version = "1.1.3", default-features = false }
+alloy-network = { version = "1.1.3", default-features = false }
+alloy-network-primitives = { version = "1.1.3", default-features = false }
+alloy-provider = { version = "1.1.3", features = ["reqwest", "debug-api"], default-features = false }
+alloy-pubsub = { version = "1.1.3", default-features = false }
+alloy-rpc-client = { version = "1.1.3", default-features = false }
+alloy-rpc-types = { version = "1.1.3", features = ["eth"], default-features = false }
+alloy-rpc-types-admin = { version = "1.1.3", default-features = false }
+alloy-rpc-types-anvil = { version = "1.1.3", default-features = false }
+alloy-rpc-types-beacon = { version = "1.1.3", default-features = false }
+alloy-rpc-types-debug = { version = "1.1.3", default-features = false }
+alloy-rpc-types-engine = { version = "1.1.3", default-features = false }
+alloy-rpc-types-eth = { version = "1.1.3", default-features = false }
+alloy-rpc-types-mev = { version = "1.1.3", default-features = false }
+alloy-rpc-types-trace = { version = "1.1.3", default-features = false }
+alloy-rpc-types-txpool = { version = "1.1.3", default-features = false }
+alloy-serde = { version = "1.1.3", default-features = false }
+alloy-signer = { version = "1.1.3", default-features = false }
+alloy-signer-local = { version = "1.1.3", default-features = false }
+alloy-transport = { version = "1.1.3" }
+alloy-transport-http = { version = "1.1.3", features = ["reqwest-rustls-tls"], default-features = false }
+alloy-transport-ipc = { version = "1.1.3", default-features = false }
+alloy-transport-ws = { version = "1.1.3", default-features = false }
 
 # op
-alloy-op-evm = { version = "0.25.0", default-features = false }
+alloy-op-evm = { version = "0.24.1", default-features = false }
 alloy-op-hardforks = "0.4.4"
-op-alloy-rpc-types = { version = "0.23.1", default-features = false }
-op-alloy-rpc-types-engine = { version = "0.23.1", default-features = false }
-op-alloy-network = { version = "0.23.1", default-features = false }
-op-alloy-consensus = { version = "0.23.1", default-features = false }
-op-alloy-rpc-jsonrpsee = { version = "0.23.1", default-features = false }
+op-alloy-rpc-types = { version = "0.22.4", default-features = false }
+op-alloy-rpc-types-engine = { version = "0.22.4", default-features = false }
+op-alloy-network = { version = "0.22.4", default-features = false }
+op-alloy-consensus = { version = "0.22.4", default-features = false }
+op-alloy-rpc-jsonrpsee = { version = "0.22.4", default-features = false }
 op-alloy-flz = { version = "0.13.1", default-features = false }
 
 # misc
@@ -548,7 +547,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 +586,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"
@@ -694,7 +691,7 @@ ahash = "0.8"
 anyhow = "1.0"
 bindgen = { version = "0.71", default-features = false }
 block-padding = "0.3.2"
-cc = "1.2.15"
+cc = "=1.2.15"
 cipher = "0.4.3"
 comfy-table = "7.0"
 concat-kdf = "0.1.0"
@@ -731,7 +728,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"

Dockerfile

@@ -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

DockerfileOp

@@ -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=""

LICENSE-APACHE

@@ -186,7 +186,7 @@ APPENDIX: How to apply the Apache License to your work.
    same "printed page" as the copyright notice for easier
    identification within third-party archives.
 
-Copyright 2022-2026 Reth Contributors
+Copyright 2022-2025 Reth Contributors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

LICENSE-MIT

@@ -1,6 +1,6 @@
 The MIT License (MIT)
 
-Copyright (c) 2022-2026 Reth Contributors
+Copyright (c) 2022-2025 Reth Contributors
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

Makefile

@@ -521,3 +521,5 @@ pr:
 	make update-book-cli && \
 	cargo docs --document-private-items && \
 	make test
+
+check-features:

bin/reth-bench-compare/src/cli.rs

@@ -164,42 +164,12 @@ pub(crate) struct Args {
     #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
     pub reth_args: Vec<String>,
 
-    /// Comma-separated list of features to enable during reth compilation (applied to both builds)
+    /// Comma-separated list of features to enable during reth compilation
     ///
     /// Example: `jemalloc,asm-keccak`
     #[arg(long, value_name = "FEATURES", default_value = "jemalloc,asm-keccak")]
     pub features: String,
 
-    /// Comma-separated list of features to enable only for baseline build (overrides --features)
-    ///
-    /// Example: `--baseline-features jemalloc`
-    #[arg(long, value_name = "FEATURES")]
-    pub baseline_features: Option<String>,
-
-    /// Comma-separated list of features to enable only for feature build (overrides --features)
-    ///
-    /// Example: `--feature-features jemalloc,asm-keccak`
-    #[arg(long, value_name = "FEATURES")]
-    pub feature_features: Option<String>,
-
-    /// RUSTFLAGS to use for both baseline and feature builds
-    ///
-    /// Example: `--rustflags "-C target-cpu=native"`
-    #[arg(long, value_name = "FLAGS", default_value = "-C target-cpu=native")]
-    pub rustflags: String,
-
-    /// RUSTFLAGS to use only for baseline build (overrides --rustflags)
-    ///
-    /// Example: `--baseline-rustflags "-C target-cpu=native -C lto"`
-    #[arg(long, value_name = "FLAGS")]
-    pub baseline_rustflags: Option<String>,
-
-    /// RUSTFLAGS to use only for feature build (overrides --rustflags)
-    ///
-    /// Example: `--feature-rustflags "-C target-cpu=native -C lto"`
-    #[arg(long, value_name = "FLAGS")]
-    pub feature_rustflags: Option<String>,
-
     /// Disable automatic --debug.startup-sync-state-idle flag for specific runs.
     /// Can be "baseline", "feature", or "all".
     /// By default, the flag is passed to warmup, baseline, and feature runs.
@@ -358,6 +328,7 @@ pub(crate) async fn run_comparison(args: Args, _ctx: CliContext) -> Result<()> {
         git_manager.repo_root().to_string(),
         output_dir.clone(),
         git_manager.clone(),
+        args.features.clone(),
     )?;
     // Initialize node manager
     let mut node_manager = NodeManager::new(&args);
@@ -477,18 +448,6 @@ async fn run_compilation_phase(
         let ref_type = ref_types[i];
         let commit = &ref_commits[git_ref];
 
-        // Get per-build features and rustflags
-        let features = match ref_type {
-            "baseline" => args.baseline_features.as_ref().unwrap_or(&args.features),
-            "feature" => args.feature_features.as_ref().unwrap_or(&args.features),
-            _ => &args.features,
-        };
-        let rustflags = match ref_type {
-            "baseline" => args.baseline_rustflags.as_ref().unwrap_or(&args.rustflags),
-            "feature" => args.feature_rustflags.as_ref().unwrap_or(&args.rustflags),
-            _ => &args.rustflags,
-        };
-
         info!(
             "Compiling {} binary for reference: {} (commit: {})",
             ref_type,
@@ -500,7 +459,7 @@ async fn run_compilation_phase(
         git_manager.switch_ref(git_ref)?;
 
         // Compile reth (with caching)
-        compilation_manager.compile_reth(commit, is_optimism, features, rustflags)?;
+        compilation_manager.compile_reth(commit, is_optimism)?;
 
         info!("Completed compilation for {} reference", ref_type);
     }

bin/reth-bench-compare/src/comparison.rs

@@ -39,8 +39,7 @@ pub(crate) struct BenchmarkResults {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct CombinedLatencyRow {
     pub block_number: u64,
-    #[serde(default)]
-    pub transaction_count: Option<u64>,
+    pub transaction_count: u64,
     pub gas_used: u64,
     pub new_payload_latency: u128,
 }
@@ -49,8 +48,7 @@ pub(crate) struct CombinedLatencyRow {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct TotalGasRow {
     pub block_number: u64,
-    #[serde(default)]
-    pub transaction_count: Option<u64>,
+    pub transaction_count: u64,
     pub gas_used: u64,
     pub time: u128,
 }
@@ -127,8 +125,7 @@ pub(crate) struct ComparisonSummary {
 #[derive(Debug, Serialize)]
 pub(crate) struct BlockComparison {
     pub block_number: u64,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub transaction_count: Option<u64>,
+    pub transaction_count: u64,
     pub gas_used: u64,
     pub baseline_new_payload_latency: u128,
     pub feature_new_payload_latency: u128,

bin/reth-bench-compare/src/compilation.rs

@@ -13,6 +13,7 @@ pub(crate) struct CompilationManager {
     repo_root: String,
     output_dir: PathBuf,
     git_manager: GitManager,
+    features: String,
 }
 
 impl CompilationManager {
@@ -21,8 +22,9 @@ impl CompilationManager {
         repo_root: String,
         output_dir: PathBuf,
         git_manager: GitManager,
+        features: String,
     ) -> Result<Self> {
-        Ok(Self { repo_root, output_dir, git_manager })
+        Ok(Self { repo_root, output_dir, git_manager, features })
     }
 
     /// Detect if the RPC endpoint is an Optimism chain
@@ -66,13 +68,7 @@ impl CompilationManager {
     }
 
     /// Compile reth using cargo build and cache the binary
-    pub(crate) fn compile_reth(
-        &self,
-        commit: &str,
-        is_optimism: bool,
-        features: &str,
-        rustflags: &str,
-    ) -> Result<()> {
+    pub(crate) fn compile_reth(&self, commit: &str, is_optimism: bool) -> Result<()> {
         // Validate that current git commit matches the expected commit
         let current_commit = self.git_manager.get_current_commit()?;
         if current_commit != commit {
@@ -104,8 +100,9 @@ impl CompilationManager {
         let mut cmd = Command::new("cargo");
         cmd.arg("build").arg("--profile").arg("profiling");
 
-        cmd.arg("--features").arg(features);
-        info!("Using features: {features}");
+        // Add features
+        cmd.arg("--features").arg(&self.features);
+        info!("Using features: {}", self.features);
 
         // Add bin-specific arguments for optimism
         if is_optimism {
@@ -117,9 +114,8 @@ impl CompilationManager {
 
         cmd.current_dir(&self.repo_root);
 
-        // Set RUSTFLAGS
-        cmd.env("RUSTFLAGS", rustflags);
-        info!("Using RUSTFLAGS: {rustflags}");
+        // Set RUSTFLAGS for native CPU optimization
+        cmd.env("RUSTFLAGS", "-C target-cpu=native");
 
         // Debug log the command
         debug!("Executing cargo command: {:?}", cmd);

bin/reth-bench-compare/src/node.rs

@@ -211,11 +211,6 @@ impl NodeManager {
         cmd.arg("--");
         cmd.args(reth_args);
 
-        // Enable tracing-samply
-        if supports_samply_flags(&reth_args[0]) {
-            cmd.arg("--log.samply");
-        }
-
         // Set environment variable to disable log styling
         cmd.env("RUST_LOG_STYLE", "never");
 
@@ -408,7 +403,7 @@ impl NodeManager {
 
     /// Stop the reth node gracefully
     pub(crate) async fn stop_node(&self, child: &mut tokio::process::Child) -> Result<()> {
-        let pid = child.id().ok_or_eyre("Child process ID should be available")?;
+        let pid = child.id().expect("Child process ID should be available");
 
         // Check if the process has already exited
         match child.try_wait() {
@@ -557,16 +552,3 @@ impl NodeManager {
         Ok(())
     }
 }
-
-fn supports_samply_flags(bin: &str) -> bool {
-    let mut cmd = std::process::Command::new(bin);
-    // NOTE: The flag to check must come before --help.
-    // We pass --help as a shortcut to not execute any command.
-    cmd.args(["--log.samply", "--help"]);
-    debug!(?cmd, "Checking samply flags support");
-    let Ok(output) = cmd.output() else {
-        return false;
-    };
-    debug!(?output, "Samply flags support check");
-    output.status.success()
-}

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
 

bin/reth-bench/README.md

@@ -80,7 +80,7 @@ RUSTFLAGS="-C target-cpu=native" cargo build --profile profiling --no-default-fe
 ### Run the Benchmark:
 First, start the reth node. Here is an example that runs `reth` compiled with the `profiling` profile, runs `samply`, and configures `reth` to run with metrics enabled:
 ```bash
-samply record -p 3001 target/profiling/reth node --metrics localhost:9001 --authrpc.jwtsecret <jwt_file_path>
+samply record -p 3001 target/profiling/reth node --metrics localhost:9001 --authrpc.jwt-secret <jwt_file_path>
 ```
 
 ```bash

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

@@ -103,20 +103,14 @@ impl BenchContext {
             (bench_args.from, bench_args.to)
         };
 
-        // If `--to` are not provided, we will run the benchmark continuously,
+        // If neither `--from` nor `--to` are provided, we will run the benchmark continuously,
         // starting at the latest block.
-        let latest_block = block_provider
-            .get_block_by_number(BlockNumberOrTag::Latest)
-            .full()
-            .await?
-            .ok_or_else(|| eyre::eyre!("Failed to fetch latest block from RPC"))?;
-        let mut benchmark_mode = BenchMode::new(from, to, latest_block.into_inner().number())?;
+        let mut benchmark_mode = BenchMode::new(from, to)?;
 
         let first_block = match benchmark_mode {
-            BenchMode::Continuous(start) => {
-                block_provider.get_block_by_number(start.into()).full().await?.ok_or_else(|| {
-                    eyre::eyre!("Failed to fetch block {} from RPC for continuous mode", start)
-                })?
+            BenchMode::Continuous => {
+                // fetch Latest block
+                block_provider.get_block_by_number(BlockNumberOrTag::Latest).full().await?.unwrap()
             }
             BenchMode::Range(ref mut range) => {
                 match range.next() {
@@ -126,9 +120,7 @@ impl BenchContext {
                             .get_block_by_number(block_number.into())
                             .full()
                             .await?
-                            .ok_or_else(|| {
-                                eyre::eyre!("Failed to fetch block {} from RPC", block_number)
-                            })?
+                            .unwrap()
                     }
                     None => {
                         return Err(eyre::eyre!(

bin/reth-bench/src/bench_mode.rs

@@ -5,8 +5,9 @@ use std::ops::RangeInclusive;
 /// Whether or not the benchmark should run as a continuous stream of payloads.
 #[derive(Debug, PartialEq, Eq)]
 pub enum BenchMode {
+    // TODO: just include the start block in `Continuous`
     /// Run the benchmark as a continuous stream of payloads, until the benchmark is interrupted.
-    Continuous(u64),
+    Continuous,
     /// Run the benchmark for a specific range of blocks.
     Range(RangeInclusive<u64>),
 }
@@ -15,19 +16,18 @@ impl BenchMode {
     /// Check if the block number is in the range
     pub fn contains(&self, block_number: u64) -> bool {
         match self {
-            Self::Continuous(start) => block_number >= *start,
+            Self::Continuous => true,
             Self::Range(range) => range.contains(&block_number),
         }
     }
 
     /// Create a [`BenchMode`] from optional `from` and `to` fields.
-    pub fn new(from: Option<u64>, to: Option<u64>, latest_block: u64) -> Result<Self, eyre::Error> {
+    pub fn new(from: Option<u64>, to: Option<u64>) -> Result<Self, eyre::Error> {
         // If neither `--from` nor `--to` are provided, we will run the benchmark continuously,
         // starting at the latest block.
         match (from, to) {
             (Some(from), Some(to)) => Ok(Self::Range(from..=to)),
-            (None, None) => Ok(Self::Continuous(latest_block)),
-            (Some(start), None) => Ok(Self::Continuous(start)),
+            (None, None) => Ok(Self::Continuous),
             _ => {
                 // both or neither are allowed, everything else is ambiguous
                 Err(eyre::eyre!("`from` and `to` must be provided together, or not at all."))

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();
 }

bin/reth/Cargo.toml

@@ -81,7 +81,7 @@ backon.workspace = true
 tempfile.workspace = true
 
 [features]
-default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
+default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer"]
 
 otlp = [
     "reth-ethereum-cli/otlp",
@@ -102,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",

crates/chain-state/src/deferred_trie.rs

@@ -38,26 +38,11 @@ pub struct ComputedTrieData {
 /// Trie input bundled with its anchor hash.
 ///
 /// This is used to store the trie input and anchor hash for a block together.
-/// The `trie_input` contains the **cumulative** overlay of all in-memory ancestor blocks
-/// since the anchor, not just this block's changes. This enables O(1) overlay reuse
-/// when building child blocks with the same anchor.
-///
-/// # Invariants
-///
-/// For correctness of overlay reuse optimizations:
-/// - The `ancestors` passed to [`DeferredTrieData::pending`] must form a true ancestor chain (each
-///   entry's parent is the previous entry, oldest to newest order)
-/// - When `anchor_hash` matches the parent's `anchor_hash`, the parent's `trie_input` already
-///   contains all ancestors in that chain, enabling O(1) reuse
-/// - A given `anchor_hash` uniquely identifies a persisted base state
 #[derive(Clone, Debug)]
 pub struct AnchoredTrieInput {
     /// The persisted ancestor hash this trie input is anchored to.
     pub anchor_hash: B256,
-    /// Cumulative trie input overlay from all in-memory ancestors since the anchor.
-    /// Note: This is the merged overlay, not just this block's changes.
-    /// The per-block changes are in [`ComputedTrieData::hashed_state`] and
-    /// [`ComputedTrieData::trie_updates`].
+    /// Trie input constructed from in-memory overlays.
     pub trie_input: Arc<TrieInputSorted>,
 }
 
@@ -69,12 +54,6 @@ struct DeferredTrieMetrics {
     deferred_trie_async_ready: Counter,
     /// Number of times deferred trie data required synchronous computation (fallback path).
     deferred_trie_sync_fallback: Counter,
-    /// Number of times the parent's trie overlay was reused (O(1) fast path).
-    deferred_trie_overlay_reused: Counter,
-    /// Number of times the trie overlay was rebuilt from all ancestors (O(N) slow path).
-    deferred_trie_overlay_rebuilt: Counter,
-    /// Number of times `Arc::make_mut` triggered a clone (`strong_count` > 1).
-    deferred_trie_arc_clone_triggered: Counter,
 }
 
 static DEFERRED_TRIE_METRICS: LazyLock<DeferredTrieMetrics> =
@@ -159,15 +138,8 @@ impl DeferredTrieData {
     ///
     /// # Process
     /// 1. Sort the current block's hashed state and trie updates
-    /// 2. Try to reuse parent's overlay if anchor matches (O(1) fast path)
-    /// 3. Otherwise, merge all ancestor overlays (O(N) slow path, rare after persist/reorg)
-    /// 4. Extend the overlay with this block's sorted data
-    ///
-    /// # Complexity
-    /// - Normal case (same anchor as parent): O(1) - just clone parent's overlay
-    /// - After persist/reorg (anchor mismatch): O(N) - merge all ancestors once
-    ///
-    /// This eliminates the previous O(N²) complexity where each block re-merged all ancestors.
+    /// 2. Merge ancestor overlays (oldest -> newest, so later state takes precedence)
+    /// 3. Extend the merged overlay with this block's sorted data
     ///
     /// Used by both the async background task and the synchronous fallback path.
     ///
@@ -175,7 +147,7 @@ impl DeferredTrieData {
     /// * `hashed_state` - Unsorted hashed post-state (account/storage changes) from execution
     /// * `trie_updates` - Unsorted trie node updates from state root computation
     /// * `anchor_hash` - The persisted ancestor hash this trie input is anchored to
-    /// * `ancestors` - Deferred trie data from ancestor blocks for merging (oldest -> newest)
+    /// * `ancestors` - Deferred trie data from ancestor blocks for merging
     pub fn sort_and_build_trie_input(
         hashed_state: &HashedPostState,
         trie_updates: &TrieUpdates,
@@ -184,63 +156,28 @@ impl DeferredTrieData {
     ) -> ComputedTrieData {
         // Sort the current block's hashed state and trie updates
         let sorted_hashed_state = Arc::new(hashed_state.clone_into_sorted());
-        let sorted_trie_updates = Arc::new(trie_updates.clone_into_sorted());
+        let sorted_trie_updates = Arc::new(trie_updates.clone().into_sorted());
 
-        // Determine base overlay by checking if we can reuse parent's overlay
-        let mut overlay = if let Some(parent) = ancestors.last() {
-            let parent_data = parent.wait_cloned();
-
-            match &parent_data.anchored_trie_input {
-                // Fast path: reuse parent's already-merged overlay if anchors match.
-                // Parent's overlay already contains all ancestors merged, so we just clone
-                // the Arc-wrapped nodes and state (O(1)).
-                //
-                // IMPORTANT: We do NOT clone prefix_sets from the parent overlay.
-                // Prefix sets only need to represent the current block's changes, not
-                // cumulative ancestor changes. The incremental state root algorithms
-                // use prefix_sets to identify which trie branches changed since the
-                // last root computation - ancestors' changes are already embodied in
-                // the trie nodes. This matches the pattern in merkle_changesets.rs
-                // which explicitly uses per-block prefix sets.
-                Some(AnchoredTrieInput { anchor_hash: parent_anchor, trie_input })
-                    if *parent_anchor == anchor_hash =>
-                {
-                    DEFERRED_TRIE_METRICS.deferred_trie_overlay_reused.increment(1);
-                    TrieInputSorted::new(
-                        Arc::clone(&trie_input.nodes),
-                        Arc::clone(&trie_input.state),
-                        Default::default(), // Fresh prefix_sets - will be set by caller
-                    )
-                }
-
-                // Slow path: no matching parent overlay -> rebuild from all ancestors.
-                // This happens after persist (anchor changes) or if parent lacks anchored input.
-                // O(N) but only at persist/reorg boundaries, not per block.
-                _ => {
-                    DEFERRED_TRIE_METRICS.deferred_trie_overlay_rebuilt.increment(1);
-                    Self::merge_ancestors_into_overlay(ancestors)
-                }
+        // Merge trie data from ancestors (oldest -> newest so later state takes precedence)
+        let mut overlay = TrieInputSorted::default();
+        for ancestor in ancestors {
+            let ancestor_data = ancestor.wait_cloned();
+            {
+                let state_mut = Arc::make_mut(&mut overlay.state);
+                state_mut.extend_ref(ancestor_data.hashed_state.as_ref());
             }
-        } else {
-            // No ancestors: start from empty overlay (first block after anchor)
-            TrieInputSorted::default()
-        };
+            {
+                let nodes_mut = Arc::make_mut(&mut overlay.nodes);
+                nodes_mut.extend_ref(ancestor_data.trie_updates.as_ref());
+            }
+        }
 
-        // Extend overlay with current block's sorted data.
-        // Track if Arc::make_mut triggers a clone (strong_count > 1 means parent still holds ref).
+        // Extend overlay with current block's sorted data
         {
-            let will_clone = Arc::strong_count(&overlay.state) > 1;
-            if will_clone {
-                DEFERRED_TRIE_METRICS.deferred_trie_arc_clone_triggered.increment(1);
-            }
             let state_mut = Arc::make_mut(&mut overlay.state);
             state_mut.extend_ref(sorted_hashed_state.as_ref());
         }
         {
-            let will_clone = Arc::strong_count(&overlay.nodes) > 1;
-            if will_clone {
-                DEFERRED_TRIE_METRICS.deferred_trie_arc_clone_triggered.increment(1);
-            }
             let nodes_mut = Arc::make_mut(&mut overlay.nodes);
             nodes_mut.extend_ref(sorted_trie_updates.as_ref());
         }
@@ -253,40 +190,6 @@ impl DeferredTrieData {
         )
     }
 
-    /// Merge all ancestors into a single overlay.
-    ///
-    /// This is the slow path used when the parent's overlay cannot be reused
-    /// (e.g., after persist when anchor changes). Iterates ancestors oldest -> newest
-    /// so newer state takes precedence.
-    ///
-    /// Note: We intentionally do NOT reuse ancestor cached overlays here because
-    /// those overlays were built with a different anchor_hash. The slow path is
-    /// triggered precisely because the anchor changed, so we must rebuild from
-    /// each ancestor's per-block state changes.
-    fn merge_ancestors_into_overlay(ancestors: &[Self]) -> TrieInputSorted {
-        let mut overlay = TrieInputSorted::default();
-        for ancestor in ancestors {
-            let ancestor_data = ancestor.wait_cloned();
-            {
-                let will_clone = Arc::strong_count(&overlay.state) > 1;
-                if will_clone {
-                    DEFERRED_TRIE_METRICS.deferred_trie_arc_clone_triggered.increment(1);
-                }
-                let state_mut = Arc::make_mut(&mut overlay.state);
-                state_mut.extend_ref(ancestor_data.hashed_state.as_ref());
-            }
-            {
-                let will_clone = Arc::strong_count(&overlay.nodes) > 1;
-                if will_clone {
-                    DEFERRED_TRIE_METRICS.deferred_trie_arc_clone_triggered.increment(1);
-                }
-                let nodes_mut = Arc::make_mut(&mut overlay.nodes);
-                nodes_mut.extend_ref(ancestor_data.trie_updates.as_ref());
-            }
-        }
-        overlay
-    }
-
     /// Returns trie data, computing synchronously if the async task hasn't completed.
     ///
     /// - If the async task has completed (`Ready`), returns the cached result.
@@ -538,274 +441,4 @@ mod tests {
         let (_, account) = &overlay_state[0];
         assert_eq!(account.unwrap().nonce, 2);
     }
-
-    /// Helper to create a ready block with anchored trie input containing specific state.
-    fn ready_block_with_state(
-        anchor_hash: B256,
-        accounts: Vec<(B256, Option<Account>)>,
-    ) -> DeferredTrieData {
-        let hashed_state = Arc::new(HashedPostStateSorted::new(accounts, B256Map::default()));
-        let trie_updates = Arc::default();
-        let mut overlay = TrieInputSorted::default();
-        Arc::make_mut(&mut overlay.state).extend_ref(hashed_state.as_ref());
-
-        DeferredTrieData::ready(ComputedTrieData {
-            hashed_state,
-            trie_updates,
-            anchored_trie_input: Some(AnchoredTrieInput {
-                anchor_hash,
-                trie_input: Arc::new(overlay),
-            }),
-        })
-    }
-
-    /// Verifies that first block after anchor (no ancestors) creates empty base overlay.
-    #[test]
-    fn first_block_after_anchor_creates_empty_base() {
-        let anchor = B256::with_last_byte(1);
-        let key = B256::with_last_byte(42);
-        let account = Account { nonce: 1, balance: U256::ZERO, bytecode_hash: None };
-
-        // First block after anchor - no ancestors
-        let first_block = DeferredTrieData::pending(
-            Arc::new(HashedPostState::default().with_accounts([(key, Some(account))])),
-            Arc::new(TrieUpdates::default()),
-            anchor,
-            vec![], // No ancestors
-        );
-
-        let result = first_block.wait_cloned();
-
-        // Should have overlay with just this block's data
-        let overlay = result.anchored_trie_input.as_ref().unwrap();
-        assert_eq!(overlay.anchor_hash, anchor);
-        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
-        let (found_key, found_account) = &overlay.trie_input.state.accounts[0];
-        assert_eq!(*found_key, key);
-        assert_eq!(found_account.unwrap().nonce, 1);
-    }
-
-    /// Verifies that when parent has matching anchor, its overlay is reused (O(1) fast path).
-    #[test]
-    fn reuses_parent_overlay_when_anchor_matches() {
-        let anchor = B256::with_last_byte(1);
-        let key = B256::with_last_byte(42);
-        let account = Account { nonce: 100, balance: U256::ZERO, bytecode_hash: None };
-
-        // Create parent with anchored trie input
-        let parent = ready_block_with_state(anchor, vec![(key, Some(account))]);
-
-        // Create child with same anchor - should reuse parent's overlay
-        let child = DeferredTrieData::pending(
-            Arc::new(HashedPostState::default()),
-            Arc::new(TrieUpdates::default()),
-            anchor, // Same anchor as parent
-            vec![parent],
-        );
-
-        let result = child.wait_cloned();
-
-        // Verify parent's account is in the overlay
-        let overlay = result.anchored_trie_input.as_ref().unwrap();
-        assert_eq!(overlay.anchor_hash, anchor);
-        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
-        let (found_key, found_account) = &overlay.trie_input.state.accounts[0];
-        assert_eq!(*found_key, key);
-        assert_eq!(found_account.unwrap().nonce, 100);
-    }
-
-    /// Verifies that when anchor changes (after persist), all ancestors are rebuilt.
-    #[test]
-    fn rebuilds_overlay_when_anchor_changes() {
-        let old_anchor = B256::with_last_byte(1);
-        let new_anchor = B256::with_last_byte(2);
-        let key = B256::with_last_byte(42);
-        let account = Account { nonce: 50, balance: U256::ZERO, bytecode_hash: None };
-
-        // Create parent with OLD anchor
-        let parent = ready_block_with_state(old_anchor, vec![(key, Some(account))]);
-
-        // Create child with NEW anchor (simulates after persist)
-        let child = DeferredTrieData::pending(
-            Arc::new(HashedPostState::default()),
-            Arc::new(TrieUpdates::default()),
-            new_anchor, // Different anchor - triggers rebuild
-            vec![parent],
-        );
-
-        let result = child.wait_cloned();
-
-        // Verify result uses new anchor and still has parent's data
-        let overlay = result.anchored_trie_input.as_ref().unwrap();
-        assert_eq!(overlay.anchor_hash, new_anchor);
-        // Parent's account should still be in the overlay (from rebuild)
-        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
-        let (found_key, found_account) = &overlay.trie_input.state.accounts[0];
-        assert_eq!(*found_key, key);
-        assert_eq!(found_account.unwrap().nonce, 50);
-    }
-
-    /// Verifies that parent without `anchored_trie_input` triggers rebuild path.
-    #[test]
-    fn rebuilds_when_parent_has_no_anchored_input() {
-        let anchor = B256::with_last_byte(1);
-        let key = B256::with_last_byte(42);
-        let account = Account { nonce: 25, balance: U256::ZERO, bytecode_hash: None };
-
-        // Create parent WITHOUT anchored trie input (e.g., from without_trie_input constructor)
-        let parent_state =
-            HashedPostStateSorted::new(vec![(key, Some(account))], B256Map::default());
-        let parent = DeferredTrieData::ready(ComputedTrieData {
-            hashed_state: Arc::new(parent_state),
-            trie_updates: Arc::default(),
-            anchored_trie_input: None, // No anchored input
-        });
-
-        // Create child - should rebuild from parent's hashed_state
-        let child = DeferredTrieData::pending(
-            Arc::new(HashedPostState::default()),
-            Arc::new(TrieUpdates::default()),
-            anchor,
-            vec![parent],
-        );
-
-        let result = child.wait_cloned();
-
-        // Verify overlay is built and contains parent's data
-        let overlay = result.anchored_trie_input.as_ref().unwrap();
-        assert_eq!(overlay.anchor_hash, anchor);
-        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
-    }
-
-    /// Verifies that a chain of blocks with matching anchors builds correct cumulative overlay.
-    #[test]
-    fn chain_of_blocks_builds_cumulative_overlay() {
-        let anchor = B256::with_last_byte(1);
-        let key1 = B256::with_last_byte(1);
-        let key2 = B256::with_last_byte(2);
-        let key3 = B256::with_last_byte(3);
-
-        // Block 1: sets account at key1
-        let block1 = ready_block_with_state(
-            anchor,
-            vec![(key1, Some(Account { nonce: 1, balance: U256::ZERO, bytecode_hash: None }))],
-        );
-
-        // Block 2: adds account at key2, ancestor is block1
-        let block2_hashed = HashedPostState::default().with_accounts([(
-            key2,
-            Some(Account { nonce: 2, balance: U256::ZERO, bytecode_hash: None }),
-        )]);
-        let block2 = DeferredTrieData::pending(
-            Arc::new(block2_hashed),
-            Arc::new(TrieUpdates::default()),
-            anchor,
-            vec![block1.clone()],
-        );
-        // Compute block2's trie data
-        let block2_computed = block2.wait_cloned();
-        let block2_ready = DeferredTrieData::ready(block2_computed);
-
-        // Block 3: adds account at key3, ancestor is block2 (which includes block1)
-        let block3_hashed = HashedPostState::default().with_accounts([(
-            key3,
-            Some(Account { nonce: 3, balance: U256::ZERO, bytecode_hash: None }),
-        )]);
-        let block3 = DeferredTrieData::pending(
-            Arc::new(block3_hashed),
-            Arc::new(TrieUpdates::default()),
-            anchor,
-            vec![block1, block2_ready],
-        );
-
-        let result = block3.wait_cloned();
-
-        // Verify all three accounts are in the cumulative overlay
-        let overlay = result.anchored_trie_input.as_ref().unwrap();
-        assert_eq!(overlay.trie_input.state.accounts.len(), 3);
-
-        // Accounts should be sorted by key (B256 ordering)
-        let accounts = &overlay.trie_input.state.accounts;
-        assert!(accounts.iter().any(|(k, a)| *k == key1 && a.unwrap().nonce == 1));
-        assert!(accounts.iter().any(|(k, a)| *k == key2 && a.unwrap().nonce == 2));
-        assert!(accounts.iter().any(|(k, a)| *k == key3 && a.unwrap().nonce == 3));
-    }
-
-    /// Verifies that child block's state overwrites parent's state for the same key.
-    #[test]
-    fn child_state_overwrites_parent() {
-        let anchor = B256::with_last_byte(1);
-        let key = B256::with_last_byte(42);
-
-        // Parent sets nonce to 10
-        let parent = ready_block_with_state(
-            anchor,
-            vec![(key, Some(Account { nonce: 10, balance: U256::ZERO, bytecode_hash: None }))],
-        );
-
-        // Child overwrites nonce to 99
-        let child_hashed = HashedPostState::default().with_accounts([(
-            key,
-            Some(Account { nonce: 99, balance: U256::ZERO, bytecode_hash: None }),
-        )]);
-        let child = DeferredTrieData::pending(
-            Arc::new(child_hashed),
-            Arc::new(TrieUpdates::default()),
-            anchor,
-            vec![parent],
-        );
-
-        let result = child.wait_cloned();
-
-        // Verify child's value wins (extend_ref uses later value)
-        let overlay = result.anchored_trie_input.as_ref().unwrap();
-        // Note: extend_ref may result in duplicate keys; check the last occurrence
-        let accounts = &overlay.trie_input.state.accounts;
-        let last_account = accounts.iter().rfind(|(k, _)| *k == key).unwrap();
-        assert_eq!(last_account.1.unwrap().nonce, 99);
-    }
-
-    /// Stress test: verify O(N) behavior by building a chain of many blocks.
-    /// This test ensures the fix doesn't regress - previously this would be O(N²).
-    #[test]
-    fn long_chain_builds_in_linear_time() {
-        let anchor = B256::with_last_byte(1);
-        let num_blocks = 50; // Enough to notice O(N²) vs O(N) difference
-
-        let mut ancestors: Vec<DeferredTrieData> = Vec::new();
-
-        let start = Instant::now();
-
-        for i in 0..num_blocks {
-            let key = B256::with_last_byte(i as u8);
-            let account = Account { nonce: i as u64, balance: U256::ZERO, bytecode_hash: None };
-            let hashed = HashedPostState::default().with_accounts([(key, Some(account))]);
-
-            let block = DeferredTrieData::pending(
-                Arc::new(hashed),
-                Arc::new(TrieUpdates::default()),
-                anchor,
-                ancestors.clone(),
-            );
-
-            // Compute and add to ancestors for next iteration
-            let computed = block.wait_cloned();
-            ancestors.push(DeferredTrieData::ready(computed));
-        }
-
-        let elapsed = start.elapsed();
-
-        // With O(N) fix, 50 blocks should complete quickly (< 1 second)
-        // With O(N²), this would take significantly longer
-        assert!(
-            elapsed < Duration::from_secs(2),
-            "Chain of {num_blocks} blocks took {:?}, possible O(N²) regression",
-            elapsed
-        );
-
-        // Verify final overlay has all accounts
-        let final_result = ancestors.last().unwrap().wait_cloned();
-        let overlay = final_result.anchored_trie_input.as_ref().unwrap();
-        assert_eq!(overlay.trie_input.state.accounts.len(), num_blocks);
-    }
 }

crates/chain-state/src/in_memory.rs

@@ -86,20 +86,14 @@ impl<N: NodePrimitives> InMemoryState<N> {
     ///
     /// This tries to acquire a read lock. Drop any write locks before calling this.
     pub(crate) fn update_metrics(&self) {
-        let (count, earliest, latest) = {
-            let numbers = self.numbers.read();
-            let count = numbers.len();
-            let earliest = numbers.first_key_value().map(|(number, _)| *number);
-            let latest = numbers.last_key_value().map(|(number, _)| *number);
-            (count, earliest, latest)
-        };
-        if let Some(earliest_block_number) = earliest {
-            self.metrics.earliest_block.set(earliest_block_number as f64);
+        let numbers = self.numbers.read();
+        if let Some((earliest_block_number, _)) = numbers.first_key_value() {
+            self.metrics.earliest_block.set(*earliest_block_number as f64);
         }
-        if let Some(latest_block_number) = latest {
-            self.metrics.latest_block.set(latest_block_number as f64);
+        if let Some((latest_block_number, _)) = numbers.last_key_value() {
+            self.metrics.latest_block.set(*latest_block_number as f64);
         }
-        self.metrics.num_blocks.set(count as f64);
+        self.metrics.num_blocks.set(numbers.len() as f64);
     }
 
     /// Returns the state for a given block hash.
@@ -670,14 +664,22 @@ impl<N: NodePrimitives> BlockState<N> {
         receipts.first().map(|receipts| receipts.deref()).unwrap_or_default()
     }
 
-    /// Returns an iterator over __parent__ `BlockStates`.
+    /// Returns a vector of __parent__ `BlockStates`.
     ///
-    /// The block state order is newest to oldest (highest to lowest):
+    /// The block state order in the output vector is newest to oldest (highest to lowest):
     /// `[5,4,3,2,1]`
     ///
     /// Note: This does not include self.
-    pub fn parent_state_chain(&self) -> impl Iterator<Item = &Self> + '_ {
-        std::iter::successors(self.parent.as_deref(), |state| state.parent.as_deref())
+    pub fn parent_state_chain(&self) -> Vec<&Self> {
+        let mut parents = Vec::new();
+        let mut current = self.parent.as_deref();
+
+        while let Some(parent) = current {
+            parents.push(parent);
+            current = parent.parent.as_deref();
+        }
+
+        parents
     }
 
     /// Returns a vector of `BlockStates` representing the entire in memory chain.
@@ -688,11 +690,6 @@ impl<N: NodePrimitives> BlockState<N> {
     }
 
     /// Appends the parent chain of this [`BlockState`] to the given vector.
-    ///
-    /// Parents are appended in order from newest to oldest (highest to lowest).
-    /// This does not include self, only the parent states.
-    ///
-    /// This is a convenience method equivalent to `chain.extend(self.parent_state_chain())`.
     pub fn append_parent_chain<'a>(&'a self, chain: &mut Vec<&'a Self>) {
         chain.extend(self.parent_state_chain());
     }
@@ -1456,18 +1453,19 @@ mod tests {
         let mut test_block_builder: TestBlockBuilder = TestBlockBuilder::default();
         let chain = create_mock_state_chain(&mut test_block_builder, 4);
 
-        let parents: Vec<_> = chain[3].parent_state_chain().collect();
+        let parents = chain[3].parent_state_chain();
         assert_eq!(parents.len(), 3);
         assert_eq!(parents[0].block().recovered_block().number, 3);
         assert_eq!(parents[1].block().recovered_block().number, 2);
         assert_eq!(parents[2].block().recovered_block().number, 1);
 
-        let parents: Vec<_> = chain[2].parent_state_chain().collect();
+        let parents = chain[2].parent_state_chain();
         assert_eq!(parents.len(), 2);
         assert_eq!(parents[0].block().recovered_block().number, 2);
         assert_eq!(parents[1].block().recovered_block().number, 1);
 
-        assert_eq!(chain[0].parent_state_chain().count(), 0);
+        let parents = chain[0].parent_state_chain();
+        assert_eq!(parents.len(), 0);
     }
 
     #[test]
@@ -1478,7 +1476,8 @@ mod tests {
             create_mock_state(&mut test_block_builder, single_block_number, B256::random());
         let single_block_hash = single_block.block().recovered_block().hash();
 
-        assert_eq!(single_block.parent_state_chain().count(), 0);
+        let parents = single_block.parent_state_chain();
+        assert_eq!(parents.len(), 0);
 
         let block_state_chain = single_block.chain().collect::<Vec<_>>();
         assert_eq!(block_state_chain.len(), 1);

crates/chain-state/src/memory_overlay.rs

@@ -5,14 +5,14 @@ use reth_errors::ProviderResult;
 use reth_primitives_traits::{Account, Bytecode, NodePrimitives};
 use reth_storage_api::{
     AccountReader, BlockHashReader, BytecodeReader, HashedPostStateProvider, StateProofProvider,
-    StateProvider, StateProviderBox, StateRootProvider, StorageRootProvider,
+    StateProvider, StateRootProvider, StorageRootProvider,
 };
 use reth_trie::{
     updates::TrieUpdates, AccountProof, HashedPostState, HashedStorage, MultiProof,
     MultiProofTargets, StorageMultiProof, TrieInput,
 };
 use revm_database::BundleState;
-use std::{borrow::Cow, sync::OnceLock};
+use std::sync::OnceLock;
 
 /// A state provider that stores references to in-memory blocks along with their state as well as a
 /// reference of the historical state provider for fallback lookups.
@@ -24,11 +24,15 @@ pub struct MemoryOverlayStateProviderRef<
     /// Historical state provider for state lookups that are not found in memory blocks.
     pub(crate) historical: Box<dyn StateProvider + 'a>,
     /// The collection of executed parent blocks. Expected order is newest to oldest.
-    pub(crate) in_memory: Cow<'a, [ExecutedBlock<N>]>,
+    pub(crate) in_memory: Vec<ExecutedBlock<N>>,
     /// Lazy-loaded in-memory trie data.
     pub(crate) trie_input: OnceLock<TrieInput>,
 }
 
+/// A state provider that stores references to in-memory blocks along with their state as well as
+/// the historical state provider for fallback lookups.
+pub type MemoryOverlayStateProvider<N> = MemoryOverlayStateProviderRef<'static, N>;
+
 impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
     /// Create new memory overlay state provider.
     ///
@@ -38,7 +42,7 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
     /// - `historical` - a historical state provider for the latest ancestor block stored in the
     ///   database.
     pub fn new(historical: Box<dyn StateProvider + 'a>, in_memory: Vec<ExecutedBlock<N>>) -> Self {
-        Self { historical, in_memory: Cow::Owned(in_memory), trie_input: OnceLock::new() }
+        Self { historical, in_memory, trie_input: OnceLock::new() }
     }
 
     /// Turn this state provider into a state provider
@@ -49,14 +53,11 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
     /// Return lazy-loaded trie state aggregated from in-memory blocks.
     fn trie_input(&self) -> &TrieInput {
         self.trie_input.get_or_init(|| {
-            let mut input = TrieInput::default();
-            // Iterate from oldest to newest
-            for block in self.in_memory.iter().rev() {
-                let data = block.trie_data();
-                input.nodes.extend_from_sorted(&data.trie_updates);
-                input.state.extend_from_sorted(&data.hashed_state);
-            }
-            input
+            let bundles: Vec<_> =
+                self.in_memory.iter().rev().map(|block| block.trie_data()).collect();
+            TrieInput::from_blocks_sorted(
+                bundles.iter().map(|data| (data.hashed_state.as_ref(), data.trie_updates.as_ref())),
+            )
         })
     }
 
@@ -70,7 +71,7 @@ impl<'a, N: NodePrimitives> MemoryOverlayStateProviderRef<'a, N> {
 
 impl<N: NodePrimitives> BlockHashReader for MemoryOverlayStateProviderRef<'_, N> {
     fn block_hash(&self, number: BlockNumber) -> ProviderResult<Option<B256>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
             if block.recovered_block().number() == number {
                 return Ok(Some(block.recovered_block().hash()));
             }
@@ -89,7 +90,7 @@ impl<N: NodePrimitives> BlockHashReader for MemoryOverlayStateProviderRef<'_, N>
         let mut in_memory_hashes = Vec::with_capacity(range.size_hint().0);
 
         // iterate in ascending order (oldest to newest = low to high)
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
             let block_num = block.recovered_block().number();
             if range.contains(&block_num) {
                 in_memory_hashes.push(block.recovered_block().hash());
@@ -111,7 +112,7 @@ impl<N: NodePrimitives> BlockHashReader for MemoryOverlayStateProviderRef<'_, N>
 
 impl<N: NodePrimitives> AccountReader for MemoryOverlayStateProviderRef<'_, N> {
     fn basic_account(&self, address: &Address) -> ProviderResult<Option<Account>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
             if let Some(account) = block.execution_output.account(address) {
                 return Ok(account);
             }
@@ -215,7 +216,7 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {
         address: Address,
         storage_key: StorageKey,
     ) -> ProviderResult<Option<StorageValue>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
             if let Some(value) = block.execution_output.storage(&address, storage_key.into()) {
                 return Ok(Some(value));
             }
@@ -227,7 +228,7 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {
 
 impl<N: NodePrimitives> BytecodeReader for MemoryOverlayStateProviderRef<'_, N> {
     fn bytecode_by_hash(&self, code_hash: &B256) -> ProviderResult<Option<Bytecode>> {
-        for block in self.in_memory.iter() {
+        for block in &self.in_memory {
             if let Some(contract) = block.execution_output.bytecode(code_hash) {
                 return Ok(Some(contract));
             }
@@ -236,46 +237,3 @@ impl<N: NodePrimitives> BytecodeReader for MemoryOverlayStateProviderRef<'_, N>
         self.historical.bytecode_by_hash(code_hash)
     }
 }
-
-/// An owned state provider that stores references to in-memory blocks along with their state as
-/// well as a reference of the historical state provider for fallback lookups.
-#[expect(missing_debug_implementations)]
-pub struct MemoryOverlayStateProvider<N: NodePrimitives = reth_ethereum_primitives::EthPrimitives> {
-    /// Historical state provider for state lookups that are not found in memory blocks.
-    pub(crate) historical: StateProviderBox,
-    /// The collection of executed parent blocks. Expected order is newest to oldest.
-    pub(crate) in_memory: Vec<ExecutedBlock<N>>,
-    /// Lazy-loaded in-memory trie data.
-    pub(crate) trie_input: OnceLock<TrieInput>,
-}
-
-impl<N: NodePrimitives> MemoryOverlayStateProvider<N> {
-    /// Create new memory overlay state provider.
-    ///
-    /// ## Arguments
-    ///
-    /// - `in_memory` - the collection of executed ancestor blocks in reverse.
-    /// - `historical` - a historical state provider for the latest ancestor block stored in the
-    ///   database.
-    pub fn new(historical: StateProviderBox, in_memory: Vec<ExecutedBlock<N>>) -> Self {
-        Self { historical, in_memory, trie_input: OnceLock::new() }
-    }
-
-    /// Returns a new provider that takes the `TX` as reference
-    #[inline(always)]
-    fn as_ref(&self) -> MemoryOverlayStateProviderRef<'_, N> {
-        MemoryOverlayStateProviderRef {
-            historical: Box::new(self.historical.as_ref()),
-            in_memory: Cow::Borrowed(&self.in_memory),
-            trie_input: self.trie_input.clone(),
-        }
-    }
-
-    /// Wraps the [`Self`] in a `Box`.
-    pub fn boxed(self) -> StateProviderBox {
-        Box::new(self)
-    }
-}
-
-// Delegates all provider impls to [`MemoryOverlayStateProviderRef`]
-reth_storage_api::macros::delegate_provider_impls!(MemoryOverlayStateProvider<N> where [N: NodePrimitives]);

crates/chain-state/src/test_utils.rs

@@ -117,7 +117,7 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
             .map(|_| {
                 let tx = mock_tx(self.signer_build_account_info.nonce);
                 self.signer_build_account_info.nonce += 1;
-                self.signer_build_account_info.balance -= Self::single_tx_cost();
+                self.signer_build_account_info.balance -= signer_balance_decrease;
                 tx
             })
             .collect();

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;
 

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());
 

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) => {

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(())

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

@@ -92,8 +92,6 @@ impl Command {
             receipts_in_static_files: _,
             transaction_senders_in_static_files: _,
             storages_history_in_rocksdb: _,
-            transaction_hash_numbers_in_rocksdb: _,
-            account_history_in_rocksdb: _,
         } = settings.unwrap_or_else(StorageSettings::legacy);
 
         // Update the setting based on the key

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");

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(),
         )

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| {

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,

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,

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,

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,

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

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))]

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(())
 }
 

crates/consensus/consensus/src/lib.rs

@@ -135,7 +135,7 @@ pub enum ConsensusError {
         /// The gas limit in the block header.
         gas_limit: u64,
     },
-    /// Error when the gas limit is more than the maximum allowed.
+    /// Error when the gas the gas limit is more than the maximum allowed.
     #[error(
         "header gas limit ({gas_limit}) exceed the maximum allowed gas limit ({MAXIMUM_GAS_LIMIT_BLOCK})"
     )]

crates/consensus/debug-client/src/providers/rpc.rs

@@ -1,5 +1,5 @@
 use crate::BlockProvider;
-use alloy_provider::{ConnectionConfig, Network, Provider, ProviderBuilder, WebSocketConfig};
+use alloy_provider::{Network, Provider, ProviderBuilder};
 use alloy_transport::TransportResult;
 use futures::{Stream, StreamExt};
 use reth_node_api::Block;
@@ -25,19 +25,7 @@ impl<N: Network, PrimitiveBlock> RpcBlockProvider<N, PrimitiveBlock> {
         convert: impl Fn(N::BlockResponse) -> PrimitiveBlock + Send + Sync + 'static,
     ) -> eyre::Result<Self> {
         Ok(Self {
-            provider: Arc::new(
-                ProviderBuilder::default()
-                    .connect_with_config(
-                        rpc_url,
-                        ConnectionConfig::default().with_max_retries(u32::MAX).with_ws_config(
-                            WebSocketConfig::default()
-                                // allow larger messages/frames for big blocks
-                                .max_frame_size(Some(128 * 1024 * 1024))
-                                .max_message_size(Some(128 * 1024 * 1024)),
-                        ),
-                    )
-                    .await?,
-            ),
+            provider: Arc::new(ProviderBuilder::default().connect(rpc_url).await?),
             url: rpc_url.to_string(),
             convert: Arc::new(convert),
         })
@@ -89,8 +77,8 @@ where
                 match res {
                     Ok(block) => {
                         if tx.send((self.convert)(block)).await.is_err() {
-                            // Channel closed - receiver dropped, exit completely.
-                            return;
+                            // Channel closed.
+                            break;
                         }
                     }
                     Err(err) => {
@@ -107,7 +95,7 @@ where
             debug!(
                 target: "consensus::debug-client",
                 url=%self.url,
-                "Re-establishing block subscription",
+                "Re-estbalishing block subscription",
             );
         }
     }

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)

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");
 

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),
+    }
+}

crates/engine/invalid-block-hooks/src/witness.rs

@@ -5,7 +5,7 @@ use pretty_assertions::Comparison;
 use reth_engine_primitives::InvalidBlockHook;
 use reth_evm::{execute::Executor, ConfigureEvm};
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedHeader};
-use reth_provider::{BlockExecutionOutput, StateProvider, StateProviderBox, StateProviderFactory};
+use reth_provider::{BlockExecutionOutput, StateProvider, StateProviderFactory};
 use reth_revm::{
     database::StateProviderDatabase,
     db::{BundleState, State},
@@ -80,13 +80,13 @@ fn sort_bundle_state_for_comparison(bundle_state: &BundleState) -> BundleStateSo
                     BundleAccountSorted {
                         info: acc.info.clone(),
                         original_info: acc.original_info.clone(),
-                        storage: acc.storage.iter().map(|(k, v)| (*k, *v)).collect(),
+                        storage: BTreeMap::from_iter(acc.storage.clone()),
                         status: acc.status,
                     },
                 )
             })
             .collect(),
-        contracts: bundle_state.contracts.iter().map(|(k, v)| (*k, v.clone())).collect(),
+        contracts: BTreeMap::from_iter(bundle_state.contracts.clone()),
         reverts: bundle_state
             .reverts
             .iter()
@@ -98,7 +98,7 @@ fn sort_bundle_state_for_comparison(bundle_state: &BundleState) -> BundleStateSo
                             *addr,
                             AccountRevertSorted {
                                 account: rev.account.clone(),
-                                storage: rev.storage.iter().map(|(k, v)| (*k, *v)).collect(),
+                                storage: BTreeMap::from_iter(rev.storage.clone()),
                                 previous_status: rev.previous_status,
                                 wipe_storage: rev.wipe_storage,
                             },
@@ -114,7 +114,7 @@ fn sort_bundle_state_for_comparison(bundle_state: &BundleState) -> BundleStateSo
 
 /// Extracts execution data including codes, preimages, and hashed state from database
 fn collect_execution_data(
-    mut db: State<StateProviderDatabase<StateProviderBox>>,
+    mut db: State<StateProviderDatabase<Box<dyn StateProvider>>>,
 ) -> eyre::Result<CollectionResult> {
     let bundle_state = db.take_bundle();
     let mut codes = BTreeMap::new();
@@ -530,7 +530,9 @@ mod tests {
         // Create a State with StateProviderTest
         let state_provider = StateProviderTest::default();
         let mut state = State::builder()
-            .with_database(StateProviderDatabase::new(Box::new(state_provider) as StateProviderBox))
+            .with_database(StateProviderDatabase::new(
+                Box::new(state_provider) as Box<dyn StateProvider>
+            ))
             .with_bundle_update()
             .build();
 

crates/engine/primitives/src/config.rs

@@ -1,7 +1,5 @@
 //! Engine tree configuration.
 
-use alloy_eips::merge::EPOCH_SLOTS;
-
 /// Triggers persistence when the number of canonical blocks in memory exceeds this threshold.
 pub const DEFAULT_PERSISTENCE_THRESHOLD: u64 = 2;
 
@@ -42,7 +40,7 @@ pub const DEFAULT_RESERVED_CPU_CORES: usize = 1;
 /// Default maximum concurrency for prewarm task.
 pub const DEFAULT_PREWARM_MAX_CONCURRENCY: usize = 16;
 
-const DEFAULT_BLOCK_BUFFER_LIMIT: u32 = EPOCH_SLOTS as u32 * 2;
+const DEFAULT_BLOCK_BUFFER_LIMIT: u32 = 256;
 const DEFAULT_MAX_INVALID_HEADER_CACHE_LENGTH: u32 = 256;
 const DEFAULT_MAX_EXECUTE_BLOCK_BATCH_SIZE: usize = 4;
 const DEFAULT_CROSS_BLOCK_CACHE_SIZE: u64 = 4 * 1024 * 1024 * 1024;
@@ -91,8 +89,6 @@ pub struct TreeConfig {
     /// Whether to always compare trie updates from the state root task to the trie updates from
     /// the regular state root calculation.
     always_compare_trie_updates: bool,
-    /// Whether to disable state cache.
-    disable_state_cache: bool,
     /// Whether to disable parallel prewarming.
     disable_prewarming: bool,
     /// Whether to disable the parallel sparse trie state root algorithm.
@@ -101,7 +97,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,
@@ -147,7 +143,6 @@ impl Default for TreeConfig {
             max_execute_block_batch_size: DEFAULT_MAX_EXECUTE_BLOCK_BATCH_SIZE,
             legacy_state_root: false,
             always_compare_trie_updates: false,
-            disable_state_cache: false,
             disable_prewarming: false,
             disable_parallel_sparse_trie: false,
             state_provider_metrics: false,
@@ -178,7 +173,6 @@ impl TreeConfig {
         max_execute_block_batch_size: usize,
         legacy_state_root: bool,
         always_compare_trie_updates: bool,
-        disable_state_cache: bool,
         disable_prewarming: bool,
         disable_parallel_sparse_trie: bool,
         state_provider_metrics: bool,
@@ -203,7 +197,6 @@ impl TreeConfig {
             max_execute_block_batch_size,
             legacy_state_root,
             always_compare_trie_updates,
-            disable_state_cache,
             disable_prewarming,
             disable_parallel_sparse_trie,
             state_provider_metrics,
@@ -278,12 +271,7 @@ impl TreeConfig {
         self.disable_parallel_sparse_trie
     }
 
-    /// Returns whether or not state cache is disabled.
-    pub const fn disable_state_cache(&self) -> bool {
-        self.disable_state_cache
-    }
-
-    /// Returns whether or not parallel prewarming is disabled.
+    /// Returns whether or not parallel prewarming should be used.
     pub const fn disable_prewarming(&self) -> bool {
         self.disable_prewarming
     }
@@ -375,12 +363,6 @@ impl TreeConfig {
         self
     }
 
-    /// Setter for whether to disable state cache.
-    pub const fn without_state_cache(mut self, disable_state_cache: bool) -> Self {
-        self.disable_state_cache = disable_state_cache;
-        self
-    }
-
     /// Setter for whether to disable parallel prewarming.
     pub const fn without_prewarming(mut self, disable_prewarming: bool) -> Self {
         self.disable_prewarming = disable_prewarming;
@@ -403,12 +385,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;

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;

crates/engine/tree/Cargo.toml

@@ -16,7 +16,7 @@ reth-chain-state.workspace = true
 reth-chainspec = { workspace = true, optional = true }
 reth-consensus.workspace = true
 reth-db.workspace = true
-reth-engine-primitives = { workspace = true, 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"] }
@@ -40,7 +39,6 @@ reth-trie.workspace = true
 alloy-evm.workspace = true
 alloy-consensus.workspace = true
 alloy-eips.workspace = true
-alloy-eip7928.workspace = true
 alloy-primitives.workspace = true
 alloy-rlp.workspace = true
 alloy-rpc-types-engine.workspace = true
@@ -53,7 +51,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

crates/engine/tree/benches/state_root_task.rs

@@ -230,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();

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)?

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);
 

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<()>,
-    },
 }

crates/engine/tree/src/metrics.rs

@@ -19,8 +19,6 @@ pub(crate) struct PersistenceMetrics {
     pub(crate) remove_blocks_above_duration_seconds: Histogram,
     /// How long it took for blocks to be saved
     pub(crate) save_blocks_duration_seconds: Histogram,
-    /// How many blocks we persist at once.
-    pub(crate) save_blocks_block_count: Histogram,
     /// How long it took for blocks to be pruned
     pub(crate) prune_before_duration_seconds: Histogram,
 }

crates/engine/tree/src/persistence.rs

@@ -1,4 +1,5 @@
 use crate::metrics::PersistenceMetrics;
+use alloy_consensus::BlockHeader;
 use alloy_eips::BlockNumHash;
 use reth_chain_state::ExecutedBlock;
 use reth_errors::ProviderError;
@@ -141,25 +142,27 @@ where
         &self,
         blocks: Vec<ExecutedBlock<N::Primitives>>,
     ) -> Result<Option<BlockNumHash>, PersistenceError> {
-        let first_block = blocks.first().map(|b| b.recovered_block.num_hash());
-        let last_block = blocks.last().map(|b| b.recovered_block.num_hash());
-        let block_count = blocks.len();
-        debug!(target: "engine::persistence", ?block_count, first=?first_block, last=?last_block, "Saving range of blocks");
+        let first_block_hash = blocks.first().map(|b| b.recovered_block.num_hash());
+        let last_block_hash = blocks.last().map(|b| b.recovered_block.num_hash());
+        debug!(target: "engine::persistence", first=?first_block_hash, last=?last_block_hash, "Saving range of blocks");
 
         let start_time = Instant::now();
+        let last_block_hash_num = blocks.last().map(|block| BlockNumHash {
+            hash: block.recovered_block().hash(),
+            number: block.recovered_block().header().number(),
+        });
 
-        if last_block.is_some() {
+        if last_block_hash_num.is_some() {
             let provider_rw = self.provider.database_provider_rw()?;
 
             provider_rw.save_blocks(blocks)?;
             provider_rw.commit()?;
         }
 
-        debug!(target: "engine::persistence", first=?first_block, last=?last_block, "Saved range of blocks");
+        debug!(target: "engine::persistence", first=?first_block_hash, last=?last_block_hash, "Saved range of blocks");
 
-        self.metrics.save_blocks_block_count.record(block_count as f64);
         self.metrics.save_blocks_duration_seconds.record(start_time.elapsed());
-        Ok(last_block)
+        Ok(last_block_hash_num)
     }
 }
 

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.
@@ -66,14 +66,9 @@ impl<B: Block> BlockBuffer<B> {
     pub fn insert_block(&mut self, block: SealedBlock<B>) {
         let hash = block.hash();
 
-        match self.blocks.entry(hash) {
-            std::collections::hash_map::Entry::Occupied(_) => return,
-            std::collections::hash_map::Entry::Vacant(entry) => {
-                self.parent_to_child.entry(block.parent_hash()).or_default().insert(hash);
-                self.earliest_blocks.entry(block.number()).or_default().insert(hash);
-                entry.insert(block);
-            }
-        };
+        self.parent_to_child.entry(block.parent_hash()).or_default().insert(hash);
+        self.earliest_blocks.entry(block.number()).or_default().insert(hash);
+        self.blocks.insert(hash, block);
 
         // Add block to FIFO queue and handle eviction if needed
         if self.block_queue.len() >= self.max_blocks {

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

@@ -31,9 +31,6 @@ pub(crate) struct CachedStateProvider<S> {
 
     /// Metrics for the cached state provider
     metrics: CachedStateMetrics,
-
-    /// If prewarm enabled we populate every cache miss
-    prewarm: bool,
 }
 
 impl<S> CachedStateProvider<S>
@@ -42,32 +39,12 @@ where
 {
     /// Creates a new [`CachedStateProvider`] from an [`ExecutionCache`], state provider, and
     /// [`CachedStateMetrics`].
-    pub(crate) const fn new(
+    pub(crate) const fn new_with_caches(
         state_provider: S,
         caches: ExecutionCache,
         metrics: CachedStateMetrics,
     ) -> Self {
-        Self { state_provider, caches, metrics, prewarm: false }
-    }
-}
-
-impl<S> CachedStateProvider<S> {
-    /// Enables pre-warm mode so that every cache miss is populated.
-    ///
-    /// This is only relevant for pre-warm transaction execution with the intention to pre-populate
-    /// the cache with data for regular block execution. During regular block execution the
-    /// cache doesn't need to be populated because the actual EVM database
-    /// [`State`](revm::database::State) also caches internally during block execution and the cache
-    /// is then updated after the block with the entire [`BundleState`] output of that block which
-    /// contains all accessed accounts,code,storage. See also [`ExecutionCache::insert_state`].
-    pub(crate) const fn prewarm(mut self) -> Self {
-        self.prewarm = true;
-        self
-    }
-
-    /// Returns whether this provider should pre-warm cache misses.
-    const fn is_prewarm(&self) -> bool {
-        self.prewarm
+        Self { state_provider, caches, metrics }
     }
 }
 
@@ -146,10 +123,7 @@ impl<S: AccountReader> AccountReader for CachedStateProvider<S> {
         self.metrics.account_cache_misses.increment(1);
 
         let res = self.state_provider.basic_account(address)?;
-
-        if self.is_prewarm() {
-            self.caches.account_cache.insert(*address, res);
-        }
+        self.caches.account_cache.insert(*address, res);
         Ok(res)
     }
 }
@@ -174,19 +148,15 @@ impl<S: StateProvider> StateProvider for CachedStateProvider<S> {
         match self.caches.get_storage(&account, &storage_key) {
             (SlotStatus::NotCached, maybe_cache) => {
                 let final_res = self.state_provider.storage(account, storage_key)?;
-
-                if self.is_prewarm() {
-                    let account_cache = maybe_cache.unwrap_or_default();
-                    account_cache.insert_storage(storage_key, final_res);
-                    // we always need to insert the value to update the weights.
-                    // Note: there exists a race when the storage cache did not exist yet and two
-                    // consumers looking up the a storage value for this account for the first time,
-                    // however we can assume that this will only happen for the very first
-                    // (mostlikely the same) value, and don't expect that this
-                    // will accidentally replace an account storage cache with
-                    // additional values.
-                    self.caches.insert_storage_cache(account, account_cache);
-                }
+                let account_cache = maybe_cache.unwrap_or_default();
+                account_cache.insert_storage(storage_key, final_res);
+                // we always need to insert the value to update the weights.
+                // Note: there exists a race when the storage cache did not exist yet and two
+                // consumers looking up the a storage value for this account for the first time,
+                // however we can assume that this will only happen for the very first (mostlikely
+                // the same) value, and don't expect that this will accidentally
+                // replace an account storage cache with additional values.
+                self.caches.insert_storage_cache(account, account_cache);
 
                 self.metrics.storage_cache_misses.increment(1);
                 Ok(final_res)
@@ -213,11 +183,7 @@ impl<S: BytecodeReader> BytecodeReader for CachedStateProvider<S> {
         self.metrics.code_cache_misses.increment(1);
 
         let final_res = self.state_provider.bytecode_by_hash(code_hash)?;
-
-        if self.is_prewarm() {
-            self.caches.code_cache.insert(*code_hash, final_res.clone());
-        }
-
+        self.caches.code_cache.insert(*code_hash, final_res.clone());
         Ok(final_res)
     }
 }
@@ -819,7 +785,7 @@ mod tests {
 
         let caches = ExecutionCacheBuilder::default().build_caches(1000);
         let state_provider =
-            CachedStateProvider::new(provider, caches, CachedStateMetrics::zeroed());
+            CachedStateProvider::new_with_caches(provider, caches, CachedStateMetrics::zeroed());
 
         // check that the storage is empty
         let res = state_provider.storage(address, storage_key);
@@ -842,7 +808,7 @@ mod tests {
 
         let caches = ExecutionCacheBuilder::default().build_caches(1000);
         let state_provider =
-            CachedStateProvider::new(provider, caches, CachedStateMetrics::zeroed());
+            CachedStateProvider::new_with_caches(provider, caches, CachedStateMetrics::zeroed());
 
         // check that the storage returns the expected value
         let res = state_provider.storage(address, storage_key);

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

@@ -22,7 +22,7 @@ const NANOS_PER_SEC: u32 = 1_000_000_000;
 
 /// An atomic version of [`Duration`], using an [`AtomicU64`] to store the total nanoseconds in the
 /// duration.
-#[derive(Debug, Default)]
+#[derive(Default)]
 pub(crate) struct AtomicDuration {
     /// The nanoseconds part of the duration
     ///
@@ -59,8 +59,7 @@ impl AtomicDuration {
 }
 
 /// A wrapper of a state provider and latency metrics.
-#[derive(Debug)]
-pub struct InstrumentedStateProvider<S> {
+pub(crate) struct InstrumentedStateProvider<S> {
     /// The state provider
     state_provider: S,
 
@@ -81,12 +80,11 @@ impl<S> InstrumentedStateProvider<S>
 where
     S: StateProvider,
 {
-    /// Creates a new [`InstrumentedStateProvider`] from a state provider with the provided label
-    /// for metrics.
-    pub fn new(state_provider: S, source: &'static str) -> Self {
+    /// Creates a new [`InstrumentedStateProvider`] from a state provider
+    pub(crate) fn from_state_provider(state_provider: S) -> Self {
         Self {
             state_provider,
-            metrics: StateProviderMetrics::new_with_labels(&[("source", source)]),
+            metrics: StateProviderMetrics::default(),
             total_storage_fetch_latency: AtomicDuration::zero(),
             total_code_fetch_latency: AtomicDuration::zero(),
             total_account_fetch_latency: AtomicDuration::zero(),
@@ -136,12 +134,6 @@ impl<S> InstrumentedStateProvider<S> {
     }
 }
 
-impl<S> Drop for InstrumentedStateProvider<S> {
-    fn drop(&mut self) {
-        self.record_total_latency();
-    }
-}
-
 /// Metrics for the instrumented state provider
 #[derive(Metrics, Clone)]
 #[metrics(scope = "sync.state_provider")]

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
     }
 

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

@@ -64,7 +64,6 @@ impl EngineApiMetrics {
         &self,
         executor: E,
         mut transactions: impl Iterator<Item = Result<impl ExecutableTx<E>, BlockExecutionError>>,
-        transaction_count: usize,
         state_hook: Box<dyn OnStateHook>,
     ) -> Result<(BlockExecutionOutput<E::Receipt>, Vec<Address>), BlockExecutionError>
     where
@@ -76,7 +75,7 @@ impl EngineApiMetrics {
         // be accessible.
         let wrapper = MeteredStateHook { metrics: self.executor.clone(), inner_hook: state_hook };
 
-        let mut senders = Vec::with_capacity(transaction_count);
+        let mut senders = Vec::new();
         let mut executor = executor.with_state_hook(Some(Box::new(wrapper)));
 
         let f = || {
@@ -530,7 +529,6 @@ mod tests {
         let _result = metrics.execute_metered::<_, EmptyDB>(
             executor,
             input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
-            input.transaction_count(),
             state_hook,
         );
 
@@ -587,7 +585,6 @@ mod tests {
         let _result = metrics.execute_metered::<_, EmptyDB>(
             executor,
             input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
-            input.transaction_count(),
             state_hook,
         );
 

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,
@@ -55,7 +54,7 @@ use tracing::*;
 mod block_buffer;
 mod cached_state;
 pub mod error;
-pub mod instrumented_state;
+mod instrumented_state;
 mod invalid_headers;
 mod metrics;
 mod payload_processor;
@@ -64,6 +63,7 @@ mod persistence_state;
 pub mod precompile_cache;
 #[cfg(test)]
 mod tests;
+// TODO(alexey): compare trie updates in `insert_block_inner`
 #[expect(unused)]
 mod trie_updates;
 
@@ -426,13 +426,9 @@ where
             match self.try_recv_engine_message() {
                 Ok(Some(msg)) => {
                     debug!(target: "engine::tree", %msg, "received new engine message");
-                    match self.on_engine_message(msg) {
-                        Ok(ops::ControlFlow::Break(())) => return,
-                        Ok(ops::ControlFlow::Continue(())) => {}
-                        Err(fatal) => {
-                            error!(target: "engine::tree", %fatal, "insert block fatal error");
-                            return
-                        }
+                    if let Err(fatal) = self.on_engine_message(msg) {
+                        error!(target: "engine::tree", %fatal, "insert block fatal error");
+                        return
                     }
                 }
                 Ok(None) => {
@@ -825,8 +821,7 @@ where
         new_head_number: u64,
         current_head_number: u64,
     ) -> Vec<ExecutedBlock<N>> {
-        let mut old_blocks =
-            Vec::with_capacity((current_head_number.saturating_sub(new_head_number)) as usize);
+        let mut old_blocks = Vec::new();
 
         for block_num in (new_head_number + 1)..=current_head_number {
             if let Some(block_state) = self.canonical_in_memory_state.state_by_number(block_num) {
@@ -931,6 +926,48 @@ where
         Ok(())
     }
 
+    /// Determines if the given block is part of a fork by checking that these
+    /// conditions are true:
+    /// * walking back from the target hash to verify that the target hash is not part of an
+    ///   extension of the canonical chain.
+    /// * walking back from the current head to verify that the target hash is not already part of
+    ///   the canonical chain.
+    ///
+    /// The header is required as an arg, because we might be checking that the header is a fork
+    /// block before it's in the tree state and before it's in the database.
+    fn is_fork(&self, target: BlockWithParent) -> ProviderResult<bool> {
+        let target_hash = target.block.hash;
+        // verify that the given hash is not part of an extension of the canon chain.
+        let canonical_head = self.state.tree_state.canonical_head();
+        let mut current_hash;
+        let mut current_block = target;
+        loop {
+            if current_block.block.hash == canonical_head.hash {
+                return Ok(false)
+            }
+            // We already passed the canonical head
+            if current_block.block.number <= canonical_head.number {
+                break
+            }
+            current_hash = current_block.parent;
+
+            let Some(next_block) = self.sealed_header_by_hash(current_hash)? else { break };
+            current_block = next_block.block_with_parent();
+        }
+
+        // verify that the given hash is not already part of canonical chain stored in memory
+        if self.canonical_in_memory_state.header_by_hash(target_hash).is_some() {
+            return Ok(false)
+        }
+
+        // verify that the given hash is not already part of persisted canonical chain
+        if self.provider.block_number(target_hash)?.is_some() {
+            return Ok(false)
+        }
+
+        Ok(true)
+    }
+
     /// Invoked when we receive a new forkchoice update message. Calls into the blockchain tree
     /// to resolve chain forks and ensure that the Execution Layer is working with the latest valid
     /// chain.
@@ -1265,7 +1302,22 @@ where
             // Check if persistence has complete
             match rx.try_recv() {
                 Ok(last_persisted_hash_num) => {
-                    self.on_persistence_complete(last_persisted_hash_num, start_time)?;
+                    self.metrics.engine.persistence_duration.record(start_time.elapsed());
+                    let Some(BlockNumHash {
+                        hash: last_persisted_block_hash,
+                        number: last_persisted_block_number,
+                    }) = last_persisted_hash_num
+                    else {
+                        // if this happened, then we persisted no blocks because we sent an
+                        // empty vec of blocks
+                        warn!(target: "engine::tree", "Persistence task completed but did not persist any blocks");
+                        return Ok(())
+                    };
+
+                    debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, elapsed=?start_time.elapsed(), "Finished persisting, calling finish");
+                    self.persistence_state
+                        .finish(last_persisted_block_hash, last_persisted_block_number);
+                    self.on_new_persisted_block()?;
                 }
                 Err(TryRecvError::Closed) => return Err(TryRecvError::Closed.into()),
                 Err(TryRecvError::Empty) => {
@@ -1278,8 +1330,7 @@ where
             if let Some(new_tip_num) = self.find_disk_reorg()? {
                 self.remove_blocks(new_tip_num)
             } else if self.should_persist() {
-                let blocks_to_persist =
-                    self.get_canonical_blocks_to_persist(PersistTarget::Threshold)?;
+                let blocks_to_persist = self.get_canonical_blocks_to_persist()?;
                 self.persist_blocks(blocks_to_persist);
             }
         }
@@ -1287,72 +1338,11 @@ where
         Ok(())
     }
 
-    /// Finishes termination by persisting all remaining blocks and signaling completion.
-    ///
-    /// This blocks until all persistence is complete. Always signals completion,
-    /// even if an error occurs.
-    fn finish_termination(
-        &mut self,
-        pending_termination: oneshot::Sender<()>,
-    ) -> Result<(), AdvancePersistenceError> {
-        trace!(target: "engine::tree", "finishing termination, persisting remaining blocks");
-        let result = self.persist_until_complete();
-        let _ = pending_termination.send(());
-        result
-    }
-
-    /// Persists all remaining blocks until none are left.
-    fn persist_until_complete(&mut self) -> Result<(), AdvancePersistenceError> {
-        loop {
-            // Wait for any in-progress persistence to complete (blocking)
-            if let Some((rx, start_time, _action)) = self.persistence_state.rx.take() {
-                let result = rx.blocking_recv().map_err(|_| TryRecvError::Closed)?;
-                self.on_persistence_complete(result, start_time)?;
-            }
-
-            let blocks_to_persist = self.get_canonical_blocks_to_persist(PersistTarget::Head)?;
-
-            if blocks_to_persist.is_empty() {
-                debug!(target: "engine::tree", "persistence complete, signaling termination");
-                return Ok(())
-            }
-
-            debug!(target: "engine::tree", count = blocks_to_persist.len(), "persisting remaining blocks before shutdown");
-            self.persist_blocks(blocks_to_persist);
-        }
-    }
-
-    /// Handles a completed persistence task.
-    fn on_persistence_complete(
-        &mut self,
-        last_persisted_hash_num: Option<BlockNumHash>,
-        start_time: Instant,
-    ) -> Result<(), AdvancePersistenceError> {
-        self.metrics.engine.persistence_duration.record(start_time.elapsed());
-
-        let Some(BlockNumHash {
-            hash: last_persisted_block_hash,
-            number: last_persisted_block_number,
-        }) = last_persisted_hash_num
-        else {
-            // if this happened, then we persisted no blocks because we sent an empty vec of blocks
-            warn!(target: "engine::tree", "Persistence task completed but did not persist any blocks");
-            return Ok(())
-        };
-
-        debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, elapsed=?start_time.elapsed(), "Finished persisting, calling finish");
-        self.persistence_state.finish(last_persisted_block_hash, last_persisted_block_number);
-        self.on_new_persisted_block()?;
-        Ok(())
-    }
-
     /// Handles a message from the engine.
-    ///
-    /// Returns `ControlFlow::Break(())` if the engine should terminate.
     fn on_engine_message(
         &mut self,
         msg: FromEngine<EngineApiRequest<T, N>, N::Block>,
-    ) -> Result<ops::ControlFlow<()>, InsertBlockFatalError> {
+    ) -> Result<(), InsertBlockFatalError> {
         match msg {
             FromEngine::Event(event) => match event {
                 FromOrchestrator::BackfillSyncStarted => {
@@ -1362,13 +1352,6 @@ where
                 FromOrchestrator::BackfillSyncFinished(ctrl) => {
                     self.on_backfill_sync_finished(ctrl)?;
                 }
-                FromOrchestrator::Terminate { tx } => {
-                    debug!(target: "engine::tree", "received terminate request");
-                    if let Err(err) = self.finish_termination(tx) {
-                        error!(target: "engine::tree", %err, "Termination failed");
-                    }
-                    return Ok(ops::ControlFlow::Break(()))
-                }
             },
             FromEngine::Request(request) => {
                 match request {
@@ -1376,7 +1359,7 @@ where
                         let block_num_hash = block.recovered_block().num_hash();
                         if block_num_hash.number <= self.state.tree_state.canonical_block_number() {
                             // outdated block that can be skipped
-                            return Ok(ops::ControlFlow::Continue(()))
+                            return Ok(())
                         }
 
                         debug!(target: "engine::tree", block=?block_num_hash, "inserting already executed block");
@@ -1484,7 +1467,7 @@ where
                 }
             }
         }
-        Ok(ops::ControlFlow::Continue(()))
+        Ok(())
     }
 
     /// Invoked if the backfill sync has finished to target.
@@ -1718,10 +1701,10 @@ where
     }
 
     /// Returns a batch of consecutive canonical blocks to persist in the range
-    /// `(last_persisted_number .. target]`. The expected order is oldest -> newest.
+    /// `(last_persisted_number .. canonical_head - threshold]`. The expected
+    /// order is oldest -> newest.
     fn get_canonical_blocks_to_persist(
         &self,
-        target: PersistTarget,
     ) -> Result<Vec<ExecutedBlock<N>>, AdvancePersistenceError> {
         // We will calculate the state root using the database, so we need to be sure there are no
         // changes
@@ -1732,12 +1715,9 @@ where
         let last_persisted_number = self.persistence_state.last_persisted_block.number;
         let canonical_head_number = self.state.tree_state.canonical_block_number();
 
-        let target_number = match target {
-            PersistTarget::Head => canonical_head_number,
-            PersistTarget::Threshold => {
-                canonical_head_number.saturating_sub(self.config.memory_block_buffer_target())
-            }
-        };
+        // Persist only up to block buffer target
+        let target_number =
+            canonical_head_number.saturating_sub(self.config.memory_block_buffer_target());
 
         debug!(
             target: "engine::tree",
@@ -2527,11 +2507,14 @@ where
             Ok(Some(_)) => {}
         }
 
-        // determine whether we are on a fork chain by comparing the block number with the
-        // canonical head. This is a simple check that is sufficient for the event emission below.
-        // A block is considered a fork if its number is less than or equal to the canonical head,
-        // as this indicates there's already a canonical block at that height.
-        let is_fork = block_id.block.number <= self.state.tree_state.current_canonical_head.number;
+        // determine whether we are on a fork chain
+        let is_fork = match self.is_fork(block_id) {
+            Err(err) => {
+                let block = convert_to_block(self, input)?;
+                return Err(InsertBlockError::new(block, err.into()).into());
+            }
+            Ok(is_fork) => is_fork,
+        };
 
         let ctx = TreeCtx::new(&mut self.state, &self.canonical_in_memory_state);
 
@@ -2877,12 +2860,3 @@ pub enum InsertPayloadOk {
     /// The payload was valid and inserted into the tree.
     Inserted(BlockStatus),
 }
-
-/// Target for block persistence.
-#[derive(Debug, Clone, Copy)]
-enum PersistTarget {
-    /// Persist up to `canonical_head - memory_block_buffer_target`.
-    Threshold,
-    /// Persist all blocks up to and including the canonical head.
-    Head,
-}

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

@@ -1,541 +0,0 @@
-//! BAL (Block Access List, EIP-7928) related functionality.
-
-use alloy_consensus::constants::KECCAK_EMPTY;
-use alloy_eip7928::BlockAccessList;
-use alloy_primitives::{keccak256, Address, StorageKey, U256};
-use reth_primitives_traits::Account;
-use reth_provider::{AccountReader, ProviderError};
-use reth_trie::{HashedPostState, HashedStorage};
-use std::ops::Range;
-
-/// Returns the total number of storage slots (both changed and read-only) across all accounts in
-/// the BAL.
-pub fn total_slots(bal: &BlockAccessList) -> usize {
-    bal.iter().map(|account| account.storage_changes.len() + account.storage_reads.len()).sum()
-}
-
-/// Iterator over storage slots in a [`BlockAccessList`], with range-based filtering.
-///
-/// Iterates over all `(Address, StorageKey)` pairs representing both changed and read-only
-/// storage slots across all accounts in the BAL. For each account, changed slots are iterated
-/// first, followed by read-only slots. The iterator intelligently skips accounts and slots
-/// outside the specified range for efficient traversal.
-#[derive(Debug)]
-pub(crate) struct BALSlotIter<'a> {
-    bal: &'a BlockAccessList,
-    range: Range<usize>,
-    current_index: usize,
-    account_idx: usize,
-    /// Index within the current account's combined slots (changed + read-only).
-    /// If `slot_idx < storage_changes.len()`, we're in changed slots.
-    /// Otherwise, we're in read-only slots at index `slot_idx - storage_changes.len()`.
-    slot_idx: usize,
-}
-
-impl<'a> BALSlotIter<'a> {
-    /// Creates a new iterator over storage slots within the specified range.
-    pub(crate) fn new(bal: &'a BlockAccessList, range: Range<usize>) -> Self {
-        let mut iter = Self { bal, range, current_index: 0, account_idx: 0, slot_idx: 0 };
-        iter.skip_to_range_start();
-        iter
-    }
-
-    /// Skips to the first item within the range.
-    fn skip_to_range_start(&mut self) {
-        while self.account_idx < self.bal.len() {
-            let account = &self.bal[self.account_idx];
-            let slots_in_account = account.storage_changes.len() + account.storage_reads.len();
-
-            // Check if this account contains items in our range
-            let account_end = self.current_index + slots_in_account;
-
-            if account_end <= self.range.start {
-                // Entire account is before range, skip it
-                self.current_index = account_end;
-                self.account_idx += 1;
-                self.slot_idx = 0;
-            } else if self.current_index < self.range.start {
-                // Range starts somewhere in this account
-                let skip_slots = self.range.start - self.current_index;
-                self.slot_idx = skip_slots;
-                self.current_index = self.range.start;
-                break;
-            } else {
-                // We're at or past range start
-                break;
-            }
-        }
-    }
-}
-
-impl<'a> Iterator for BALSlotIter<'a> {
-    type Item = (Address, StorageKey);
-
-    fn next(&mut self) -> Option<Self::Item> {
-        // Check if we've exceeded the range
-        if self.current_index >= self.range.end {
-            return None;
-        }
-
-        // Find the next valid slot
-        while self.account_idx < self.bal.len() {
-            let account = &self.bal[self.account_idx];
-            let changed_len = account.storage_changes.len();
-            let total_len = changed_len + account.storage_reads.len();
-
-            if self.slot_idx < total_len {
-                let address = account.address;
-                let slot = if self.slot_idx < changed_len {
-                    // We're in changed slots
-                    account.storage_changes[self.slot_idx].slot
-                } else {
-                    // We're in read-only slots
-                    account.storage_reads[self.slot_idx - changed_len]
-                };
-
-                self.slot_idx += 1;
-                self.current_index += 1;
-
-                // Check if we've reached the end of range
-                if self.current_index > self.range.end {
-                    return None;
-                }
-
-                return Some((address, slot));
-            }
-
-            // Move to next account
-            self.account_idx += 1;
-            self.slot_idx = 0;
-        }
-
-        None
-    }
-}
-
-/// Converts a Block Access List into a [`HashedPostState`] by extracting the final state
-/// of modified accounts and storage slots.
-pub(crate) fn bal_to_hashed_post_state<P>(
-    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;
-
-        // Always fetch the account; even if we don't need the db account to construct the final
-        // `Account`, doing this fills the cache.
-        let existing_account = provider.basic_account(&address)?;
-
-        // 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
-        };
-
-        // If the account was only read then don't add it to the HashedPostState
-        if balance.is_none() &&
-            nonce.is_none() &&
-            code_hash.is_none() &&
-            account_changes.storage_changes.is_empty()
-        {
-            continue
-        }
-
-        // 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))
-            }),
-        };
-
-        let hashed_address = keccak256(address);
-        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));
-                }
-            }
-
-            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));
-    }
-
-    #[test]
-    fn test_bal_slot_iter() {
-        // Create test data with multiple accounts and slots (both changed and read-only)
-        let addr1 = Address::repeat_byte(0x01);
-        let addr2 = Address::repeat_byte(0x02);
-        let addr3 = Address::repeat_byte(0x03);
-
-        // Account 1: 2 changed slots + 1 read-only = 3 total slots (indices 0, 1, 2)
-        let account1 = AccountChanges {
-            address: addr1,
-            storage_changes: vec![
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(100)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(101)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-            ],
-            storage_reads: vec![StorageKey::from(U256::from(102))],
-            balance_changes: vec![],
-            nonce_changes: vec![],
-            code_changes: vec![],
-        };
-
-        // Account 2: 1 changed slot + 1 read-only = 2 total slots (indices 3, 4)
-        let account2 = AccountChanges {
-            address: addr2,
-            storage_changes: vec![SlotChanges {
-                slot: StorageKey::from(U256::from(200)),
-                changes: vec![StorageChange::new(0, B256::ZERO)],
-            }],
-            storage_reads: vec![StorageKey::from(U256::from(201))],
-            balance_changes: vec![],
-            nonce_changes: vec![],
-            code_changes: vec![],
-        };
-
-        // Account 3: 2 changed slots + 1 read-only = 3 total slots (indices 5, 6, 7)
-        let account3 = AccountChanges {
-            address: addr3,
-            storage_changes: vec![
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(300)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-                SlotChanges {
-                    slot: StorageKey::from(U256::from(301)),
-                    changes: vec![StorageChange::new(0, B256::ZERO)],
-                },
-            ],
-            storage_reads: vec![StorageKey::from(U256::from(302))],
-            balance_changes: vec![],
-            nonce_changes: vec![],
-            code_changes: vec![],
-        };
-
-        let bal = vec![account1, account2, account3];
-
-        // Test 1: Iterate over all slots (range 0..8)
-        let items: Vec<_> = BALSlotIter::new(&bal, 0..8).collect();
-        assert_eq!(items.len(), 8);
-        // Account 1: changed slots first (100, 101), then read-only (102)
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(100))));
-        assert_eq!(items[1], (addr1, StorageKey::from(U256::from(101))));
-        assert_eq!(items[2], (addr1, StorageKey::from(U256::from(102))));
-        // Account 2: changed slot (200), then read-only (201)
-        assert_eq!(items[3], (addr2, StorageKey::from(U256::from(200))));
-        assert_eq!(items[4], (addr2, StorageKey::from(U256::from(201))));
-        // Account 3: changed slots (300, 301), then read-only (302)
-        assert_eq!(items[5], (addr3, StorageKey::from(U256::from(300))));
-        assert_eq!(items[6], (addr3, StorageKey::from(U256::from(301))));
-        assert_eq!(items[7], (addr3, StorageKey::from(U256::from(302))));
-
-        // Test 2: Range that skips first account (range 3..6)
-        let items: Vec<_> = BALSlotIter::new(&bal, 3..6).collect();
-        assert_eq!(items.len(), 3);
-        assert_eq!(items[0], (addr2, StorageKey::from(U256::from(200))));
-        assert_eq!(items[1], (addr2, StorageKey::from(U256::from(201))));
-        assert_eq!(items[2], (addr3, StorageKey::from(U256::from(300))));
-
-        // Test 3: Range within first account (range 1..2)
-        let items: Vec<_> = BALSlotIter::new(&bal, 1..2).collect();
-        assert_eq!(items.len(), 1);
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(101))));
-
-        // Test 4: Range spanning multiple accounts (range 2..5)
-        let items: Vec<_> = BALSlotIter::new(&bal, 2..5).collect();
-        assert_eq!(items.len(), 3);
-        // Last slot from account 1 (read-only)
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(102))));
-        // Account 2 (changed + read-only)
-        assert_eq!(items[1], (addr2, StorageKey::from(U256::from(200))));
-        assert_eq!(items[2], (addr2, StorageKey::from(U256::from(201))));
-
-        // Test 5: Empty range
-        let items: Vec<_> = BALSlotIter::new(&bal, 5..5).collect();
-        assert_eq!(items.len(), 0);
-
-        // Test 6: Range beyond end (starts at index 6)
-        let items: Vec<_> = BALSlotIter::new(&bal, 6..100).collect();
-        assert_eq!(items.len(), 2);
-        assert_eq!(items[0], (addr3, StorageKey::from(U256::from(301))));
-        assert_eq!(items[1], (addr3, StorageKey::from(U256::from(302))));
-
-        // Test 7: Range that starts in read-only slots (index 2 is the read-only slot of account 1)
-        let items: Vec<_> = BALSlotIter::new(&bal, 2..4).collect();
-        assert_eq!(items.len(), 2);
-        assert_eq!(items[0], (addr1, StorageKey::from(U256::from(102))));
-        assert_eq!(items[1], (addr2, StorageKey::from(U256::from(200))));
-    }
-}

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

@@ -3,17 +3,16 @@
 use super::precompile_cache::PrecompileCacheMap;
 use crate::tree::{
     cached_state::{
-        CachedStateMetrics, CachedStateProvider, ExecutionCache as StateExecutionCache,
-        ExecutionCacheBuilder, SavedCache,
+        CachedStateMetrics, ExecutionCache as StateExecutionCache, ExecutionCacheBuilder,
+        SavedCache,
     },
     payload_processor::{
-        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, PrewarmTaskEvent},
+        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmTaskEvent},
         sparse_trie::StateRootComputeOutcome,
     },
     sparse_trie::SparseTrieTask,
     StateProviderBuilder, TreeConfig,
 };
-use alloy_eip7928::BlockAccessList;
 use alloy_eips::eip1898::BlockWithParent;
 use alloy_evm::{block::StateChangeSource, ToTxEnv};
 use alloy_primitives::B256;
@@ -22,17 +21,14 @@ use executor::WorkloadExecutor;
 use multiproof::{SparseTrieUpdate, *};
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
-use rayon::prelude::*;
+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::NodePrimitives;
-use reth_provider::{
-    BlockReader, DatabaseProviderROFactory, StateProvider, StateProviderFactory, StateReader,
-};
+use reth_provider::{BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader};
 use reth_revm::{db::BundleState, state::EvmState};
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
 use reth_trie_parallel::{
@@ -46,7 +42,6 @@ use reth_trie_sparse::{
 use reth_trie_sparse_parallel::{ParallelSparseTrie, ParallelismThresholds};
 use std::{
     collections::BTreeMap,
-    ops::Not,
     sync::{
         atomic::AtomicBool,
         mpsc::{self, channel},
@@ -56,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;
@@ -96,13 +90,6 @@ pub const SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY: usize = 1_000_000;
 /// 144MB.
 pub const SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY: usize = 1_000_000;
 
-/// Type alias for [`PayloadHandle`] returned by payload processor spawn methods.
-type IteratorPayloadHandle<Evm, I, N> = PayloadHandle<
-    WithTxEnv<TxEnvFor<Evm>, <I as ExecutableTxTuple>::Tx>,
-    <I as ExecutableTxTuple>::Error,
-    <N as NodePrimitives>::Receipt,
->;
-
 /// Entrypoint for executing the payload.
 #[derive(Debug)]
 pub struct PayloadProcessor<Evm>
@@ -119,8 +106,6 @@ where
     cross_block_cache_size: u64,
     /// Whether transactions should not be executed on prewarming task.
     disable_transaction_prewarming: bool,
-    /// Whether state cache should be disable
-    disable_state_cache: bool,
     /// Determines how to configure the evm for execution.
     evm_config: Evm,
     /// Whether precompile cache should be disabled.
@@ -164,7 +149,6 @@ where
             cross_block_cache_size: config.cross_block_cache_size(),
             disable_transaction_prewarming: config.disable_prewarming(),
             evm_config,
-            disable_state_cache: config.disable_state_cache(),
             precompile_cache_disabled: config.precompile_cache_disabled(),
             precompile_cache_map,
             sparse_state_trie: Arc::default(),
@@ -211,6 +195,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",
@@ -224,8 +209,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>
@@ -239,36 +223,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, use BAL prewarming and send BAL to multiproof
-            debug!(target: "engine::tree::payload_processor", "BAL present, using BAL prewarming");
-
-            // Send BAL message immediately to MultiProofTask
-            let _ = to_multi_proof.send(MultiProofMessage::BlockAccessList(Arc::clone(&bal)));
-
-            // Spawn with BAL prewarming
-            self.spawn_caching_with(
-                env,
-                prewarm_rx,
-                transaction_count_hint,
-                provider_builder.clone(),
-                None, // Don't send proof targets when BAL is present
-                Some(bal),
-            )
-        } else {
-            // Normal path: spawn with transaction prewarming
-            self.spawn_caching_with(
-                env,
-                prewarm_rx,
-                transaction_count_hint,
-                provider_builder.clone(),
-                Some(to_multi_proof.clone()),
-                None,
-            )
-        };
+        // We rely on the cursor factory to provide whatever DB overlay is necessary to see a
+        // consistent view of the database, including the trie tables. Because of this there is no
+        // need for an overarching prefix set to invalidate any section of the trie tables, and so
+        // we use an empty prefix set.
 
         // Create and spawn the storage proof task
         let task_ctx = ProofTaskCtx::new(multiproof_provider_factory);
@@ -285,28 +244,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();
-        let saved_cache = prewarm_handle.saved_cache.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");
-            let provider = if let Some(saved_cache) = saved_cache {
-                let (cache, metrics) = saved_cache.split();
-                Box::new(CachedStateProvider::new(provider, cache, metrics))
-                    as Box<dyn StateProvider>
-            } else {
-                Box::new(provider)
-            };
-            multi_proof_task.run(provider);
+            multi_proof_task.run();
         });
 
         // wire the sparse trie to the state root response receiver
@@ -333,14 +288,13 @@ where
         env: ExecutionEnv<Evm>,
         transactions: I,
         provider_builder: StateProviderBuilder<N, P>,
-        bal: Option<Arc<BlockAccessList>>,
-    ) -> IteratorPayloadHandle<Evm, I, N>
+    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
         let (prewarm_rx, execution_rx, size_hint) = self.spawn_tx_iterator(transactions);
         let prewarm_handle =
-            self.spawn_caching_with(env, prewarm_rx, size_hint, provider_builder, None, bal);
+            self.spawn_caching_with(env, prewarm_rx, size_hint, provider_builder, None);
         PayloadHandle {
             to_multi_proof: None,
             prewarm_handle,
@@ -361,32 +315,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;
@@ -414,8 +372,7 @@ where
         transaction_count_hint: usize,
         provider_builder: StateProviderBuilder<N, P>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-        bal: Option<Arc<BlockAccessList>>,
-    ) -> CacheTaskHandle<N::Receipt>
+    ) -> CacheTaskHandle
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
@@ -425,13 +382,14 @@ where
             transactions = mpsc::channel().1;
         }
 
-        let saved_cache = self.disable_state_cache.not().then(|| self.cache_for(env.parent_hash));
-
+        let saved_cache = self.cache_for(env.parent_hash);
+        let cache = saved_cache.cache().clone();
+        let cache_metrics = saved_cache.metrics().clone();
         // configure prewarming
         let prewarm_ctx = PrewarmContext {
             env,
             evm_config: self.evm_config.clone(),
-            saved_cache: saved_cache.clone(),
+            saved_cache,
             provider: provider_builder,
             metrics: PrewarmMetrics::default(),
             terminate_execution: Arc::new(AtomicBool::new(false)),
@@ -452,16 +410,11 @@ where
         {
             let to_prewarm_task = to_prewarm_task.clone();
             self.executor.spawn_blocking(move || {
-                let mode = if let Some(bal) = bal {
-                    PrewarmMode::BlockAccessList(bal)
-                } else {
-                    PrewarmMode::Transactions(transactions)
-                };
-                prewarm_task.run(mode, to_prewarm_task);
+                prewarm_task.run(transactions, to_prewarm_task);
             });
         }
 
-        CacheTaskHandle { saved_cache, to_prewarm_task: Some(to_prewarm_task) }
+        CacheTaskHandle { cache, to_prewarm_task: Some(to_prewarm_task), cache_metrics }
     }
 
     /// Returns the cache for the given parent hash.
@@ -594,15 +547,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
@@ -611,7 +561,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
@@ -640,19 +590,19 @@ 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()));
             }
         }
     }
 
     /// Returns a clone of the caches used by prewarming
-    pub(super) fn caches(&self) -> Option<StateExecutionCache> {
-        self.prewarm_handle.saved_cache.as_ref().map(|cache| cache.cache().clone())
+    pub(super) fn caches(&self) -> StateExecutionCache {
+        self.prewarm_handle.cache.clone()
     }
 
     /// Returns a clone of the cache metrics used by prewarming
-    pub(super) fn cache_metrics(&self) -> Option<CachedStateMetrics> {
-        self.prewarm_handle.saved_cache.as_ref().map(|cache| cache.metrics().clone())
+    pub(super) fn cache_metrics(&self) -> CachedStateMetrics {
+        self.prewarm_handle.cache_metrics.clone()
     }
 
     /// Terminates the pre-warming transaction processing.
@@ -664,14 +614,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.
@@ -683,18 +628,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.
-    saved_cache: Option<SavedCache>,
+    cache: StateExecutionCache,
+    /// Metrics for the caches
+    cache_metrics: 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.
@@ -706,25 +650,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);
     }
 }
 
@@ -777,8 +716,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()
     }
@@ -1111,17 +1048,19 @@ mod tests {
 
         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();
 

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

@@ -14,31 +14,25 @@
 use crate::tree::{
     cached_state::{CachedStateProvider, SavedCache},
     payload_processor::{
-        bal::{total_slots, BALSlotIter},
-        executor::WorkloadExecutor,
-        multiproof::MultiProofMessage,
+        executor::WorkloadExecutor, multiproof::MultiProofMessage,
         ExecutionCache as PayloadExecutionCache,
     },
     precompile_cache::{CachedPrecompile, PrecompileCacheMap},
     ExecutionEnv, StateProviderBuilder,
 };
 use alloy_consensus::transaction::TxHashRef;
-use alloy_eip7928::BlockAccessList;
 use alloy_eips::Typed2718;
 use alloy_evm::Database;
 use alloy_primitives::{keccak256, map::B256Set, B256};
 use crossbeam_channel::Sender as CrossbeamSender;
 use metrics::{Counter, Gauge, Histogram};
-use rayon::iter::{IntoParallelIterator, ParallelIterator};
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
-use reth_execution_types::ExecutionOutcome;
 use reth_metrics::Metrics;
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{AccountReader, BlockReader, StateProvider, StateProviderFactory, StateReader};
-use reth_revm::{database::StateProviderDatabase, state::EvmState};
+use reth_provider::{BlockReader, StateProviderFactory, StateReader};
+use reth_revm::{database::StateProviderDatabase, db::BundleState, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
-    ops::Range,
     sync::{
         atomic::{AtomicBool, Ordering},
         mpsc::{self, channel, Receiver, Sender},
@@ -48,14 +42,6 @@ use std::{
 };
 use tracing::{debug, debug_span, instrument, trace, warn, Span};
 
-/// Determines the prewarming mode: transaction-based or BAL-based.
-pub(super) enum PrewarmMode<Tx> {
-    /// Prewarm by executing transactions from a stream.
-    Transactions(Receiver<Tx>),
-    /// Prewarm by prefetching slots from a Block Access List.
-    BlockAccessList(Arc<BlockAccessList>),
-}
-
 /// A wrapper for transactions that includes their index in the block.
 #[derive(Clone)]
 struct IndexedTransaction<Tx> {
@@ -100,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,
 }
@@ -119,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!(
@@ -149,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();
@@ -177,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;
@@ -260,123 +248,38 @@ 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, .. }, .. } =
             self;
         let hash = env.hash;
 
-        if let Some(saved_cache) = saved_cache {
-            debug!(target: "engine::caching", parent_hash=?hash, "Updating execution cache");
-            // Perform all cache operations atomically under the lock
-            execution_cache.update_with_guard(|cached| {
-                // consumes the `SavedCache` held by the prewarming task, which releases its usage
-                // guard
-                let (caches, cache_metrics) = saved_cache.split();
-                let new_cache = SavedCache::new(hash, caches, cache_metrics);
+        debug!(target: "engine::caching", parent_hash=?hash, "Updating execution cache");
+        // Perform all cache operations atomically under the lock
+        execution_cache.update_with_guard(|cached| {
+            // consumes the `SavedCache` held by the prewarming task, which releases its usage guard
+            let (caches, cache_metrics) = saved_cache.split();
+            let new_cache = SavedCache::new(hash, caches, cache_metrics);
 
-                // Insert state into cache while holding the lock
-                // Access the BundleState through the shared ExecutionOutcome
-                if new_cache.cache().insert_state(execution_outcome.state()).is_err() {
-                    // Clear the cache on error to prevent having a polluted cache
-                    *cached = None;
-                    debug!(target: "engine::caching", "cleared execution cache on update error");
-                    return;
-                }
+            // Insert state into cache while holding the lock
+            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");
+                return;
+            }
 
-                new_cache.update_metrics();
+            new_cache.update_metrics();
 
-                // Replace the shared cache with the new one; the previous cache (if any) is
-                // dropped.
-                *cached = Some(new_cache);
-            });
+            // Replace the shared cache with the new one; the previous cache (if any) is dropped.
+            *cached = Some(new_cache);
+        });
 
-            let elapsed = start.elapsed();
-            debug!(target: "engine::caching", parent_hash=?hash, elapsed=?elapsed, "Updated execution cache");
+        let elapsed = start.elapsed();
+        debug!(target: "engine::caching", parent_hash=?hash, elapsed=?elapsed, "Updated execution cache");
 
-            metrics.cache_saving_duration.set(elapsed.as_secs_f64());
-        }
-    }
-
-    /// Runs BAL-based prewarming by spawning workers to prefetch storage slots.
-    ///
-    /// Divides the total slots across `max_concurrency` workers, each responsible for
-    /// prefetching a range of slots from the BAL.
-    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn run_bal_prewarm(
-        &self,
-        bal: Arc<BlockAccessList>,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-    ) {
-        // Only prefetch if we have a cache to populate
-        if self.ctx.saved_cache.is_none() {
-            trace!(
-                target: "engine::tree::payload_processor::prewarm",
-                "Skipping BAL prewarm - no cache available"
-            );
-            let _ =
-                actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
-            return;
-        }
-
-        let total_slots = total_slots(&bal);
-
-        trace!(
-            target: "engine::tree::payload_processor::prewarm",
-            total_slots,
-            max_concurrency = self.max_concurrency,
-            "Starting BAL prewarm"
-        );
-
-        if total_slots == 0 {
-            // No slots to prefetch, signal completion immediately
-            let _ =
-                actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
-            return;
-        }
-
-        let (done_tx, done_rx) = mpsc::channel();
-
-        // Calculate number of workers needed (at most max_concurrency)
-        let workers_needed = total_slots.min(self.max_concurrency);
-
-        // Calculate slots per worker
-        let slots_per_worker = total_slots / workers_needed;
-        let remainder = total_slots % workers_needed;
-
-        // Spawn workers with their assigned ranges
-        for i in 0..workers_needed {
-            let start = i * slots_per_worker + i.min(remainder);
-            let extra = if i < remainder { 1 } else { 0 };
-            let end = start + slots_per_worker + extra;
-
-            self.ctx.spawn_bal_worker(
-                i,
-                &self.executor,
-                Arc::clone(&bal),
-                start..end,
-                done_tx.clone(),
-            );
-        }
-
-        // Drop our handle to done_tx so we can detect completion
-        drop(done_tx);
-
-        // Wait for all workers to complete
-        let mut completed_workers = 0;
-        while done_rx.recv().is_ok() {
-            completed_workers += 1;
-        }
-
-        trace!(
-            target: "engine::tree::payload_processor::prewarm",
-            completed_workers,
-            "All BAL prewarm workers completed"
-        );
-
-        // Signal that execution has finished
-        let _ = actions_tx.send(PrewarmTaskEvent::FinishedTxExecution { executed_transactions: 0 });
+        metrics.cache_saving_duration.set(elapsed.as_secs_f64());
     }
 
     /// Executes the task.
@@ -390,24 +293,15 @@ where
         name = "prewarm and caching",
         skip_all
     )]
-    pub(super) fn run<Tx>(
+    pub(super) fn run(
         self,
-        mode: PrewarmMode<Tx>,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-    ) where
-        Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
-    {
-        // Spawn execution tasks based on mode
-        match mode {
-            PrewarmMode::Transactions(pending) => {
-                self.spawn_all(pending, actions_tx);
-            }
-            PrewarmMode::BlockAccessList(bal) => {
-                self.run_bal_prewarm(bal, actions_tx);
-            }
-        }
+        pending: mpsc::Receiver<impl ExecutableTxFor<Evm> + Clone + Send + 'static>,
+        actions_tx: Sender<PrewarmTaskEvent>,
+    ) {
+        // spawn execution tasks.
+        self.spawn_all(pending, actions_tx);
 
-        let mut final_execution_outcome = None;
+        let mut final_block_output = None;
         let mut finished_execution = false;
         while let Ok(event) = self.actions_rx.recv() {
             match event {
@@ -420,9 +314,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
@@ -436,7 +330,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
                     }
@@ -446,9 +340,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);
         }
     }
 }
@@ -462,7 +356,7 @@ where
 {
     pub(super) env: ExecutionEnv<Evm>,
     pub(super) evm_config: Evm,
-    pub(super) saved_cache: Option<SavedCache>,
+    pub(super) saved_cache: SavedCache,
     /// Provider to obtain the state
     pub(super) provider: StateProviderBuilder<N, P>,
     pub(super) metrics: PrewarmMetrics,
@@ -490,10 +384,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!(
@@ -506,15 +400,10 @@ where
         };
 
         // Use the caches to create a new provider with caching
-        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(),
-            );
-        }
+        let caches = saved_cache.cache().clone();
+        let cache_metrics = saved_cache.metrics().clone();
+        let state_provider =
+            CachedStateProvider::new_with_caches(state_provider, caches, cache_metrics);
 
         let state_provider = StateProviderDatabase::new(state_provider);
 
@@ -556,7 +445,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>,
@@ -619,8 +508,7 @@ where
                 let _enter =
                     debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm outcome", index, tx_hash=%tx.tx().tx_hash())
                         .entered();
-                let targets = multiproof_targets_from_state(res.state);
-                let storage_targets = targets.storage_targets_count();
+                let (targets, storage_targets) = multiproof_targets_from_state(res.state);
                 metrics.prefetch_storage_targets.record(storage_targets as f64);
                 let _ = sender.send(PrewarmTaskEvent::Outcome { proof_targets: Some(targets) });
                 drop(_enter);
@@ -634,181 +522,74 @@ where
     }
 
     /// Spawns a worker task for transaction execution and returns its sender channel.
-    fn spawn_workers<Tx>(
-        self,
-        workers_needed: usize,
-        task_executor: &WorkloadExecutor,
-        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
-        done_tx: Sender<()>,
-    ) -> Vec<mpsc::Sender<IndexedTransaction<Tx>>>
-    where
-        Tx: ExecutableTxFor<Evm> + Send + 'static,
-    {
-        let mut handles = Vec::with_capacity(workers_needed);
-        let mut receivers = Vec::with_capacity(workers_needed);
-
-        for _ in 0..workers_needed {
-            let (tx, rx) = mpsc::channel();
-            handles.push(tx);
-            receivers.push(rx);
-        }
-
-        // Spawn a separate task spawning workers in parallel.
-        let executor = task_executor.clone();
-        let span = Span::current();
-        task_executor.spawn_blocking(move || {
-            let _enter = span.entered();
-            for (idx, rx) in receivers.into_iter().enumerate() {
-                let ctx = self.clone();
-                let actions_tx = actions_tx.clone();
-                let done_tx = done_tx.clone();
-                let span = debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);
-                executor.spawn_blocking(move || {
-                    let _enter = span.entered();
-                    ctx.transact_batch(rx, actions_tx, done_tx);
-                });
-            }
-        });
-
-        handles
-    }
-
-    /// Spawns a worker task for BAL slot prefetching.
-    ///
-    /// The worker iterates over the specified range of slots in the BAL and ensures
-    /// each slot is loaded into the cache by accessing it through the state provider.
-    fn spawn_bal_worker(
+    fn spawn_worker<Tx>(
         &self,
         idx: usize,
         executor: &WorkloadExecutor,
-        bal: Arc<BlockAccessList>,
-        range: Range<usize>,
+        actions_tx: Sender<PrewarmTaskEvent>,
         done_tx: Sender<()>,
-    ) {
+    ) -> mpsc::Sender<IndexedTransaction<Tx>>
+    where
+        Tx: ExecutableTxFor<Evm> + Send + 'static,
+    {
+        let (tx, rx) = mpsc::channel();
         let ctx = self.clone();
-        let span = debug_span!(
-            target: "engine::tree::payload_processor::prewarm",
-            "bal prewarm worker",
-            idx,
-            range_start = range.start,
-            range_end = range.end
-        );
+        let span =
+            debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);
 
         executor.spawn_blocking(move || {
             let _enter = span.entered();
-            ctx.prefetch_bal_slots(bal, range, done_tx);
+            ctx.transact_batch(rx, actions_tx, done_tx);
         });
-    }
 
-    /// Prefetches storage slots from a BAL range into the cache.
-    ///
-    /// This iterates through the specified range of slots and accesses them via the state
-    /// provider to populate the cache.
-    #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn prefetch_bal_slots(
-        self,
-        bal: Arc<BlockAccessList>,
-        range: Range<usize>,
-        done_tx: Sender<()>,
-    ) {
-        let Self { saved_cache, provider, metrics, .. } = self;
-
-        // Build state provider
-        let state_provider = match provider.build() {
-            Ok(provider) => provider,
-            Err(err) => {
-                trace!(
-                    target: "engine::tree::payload_processor::prewarm",
-                    %err,
-                    "Failed to build state provider in BAL prewarm thread"
-                );
-                let _ = done_tx.send(());
-                return;
-            }
-        };
-
-        // Wrap with cache (guaranteed to be Some since run_bal_prewarm checks)
-        let saved_cache = saved_cache.expect("BAL prewarm should only run with cache");
-        let caches = saved_cache.cache().clone();
-        let cache_metrics = saved_cache.metrics().clone();
-        let state_provider = CachedStateProvider::new(state_provider, caches, cache_metrics);
-
-        let start = Instant::now();
-
-        // Track last seen address to avoid fetching the same account multiple times.
-        let mut last_address = None;
-
-        // Iterate through the assigned range of slots
-        for (address, slot) in BALSlotIter::new(&bal, range.clone()) {
-            // Fetch the account if this is a different address than the last one
-            if last_address != Some(address) {
-                let _ = state_provider.basic_account(&address);
-                last_address = Some(address);
-            }
-
-            // Access the slot to populate the cache
-            let _ = state_provider.storage(address, slot);
-        }
-
-        let elapsed = start.elapsed();
-
-        trace!(
-            target: "engine::tree::payload_processor::prewarm",
-            ?range,
-            elapsed_ms = elapsed.as_millis(),
-            "BAL prewarm worker completed"
-        );
-
-        // Signal completion
-        let _ = done_tx.send(());
-        metrics.bal_slot_iteration_duration.record(elapsed.as_secs_f64());
+        tx
     }
 }
 
 /// Returns a set of [`MultiProofTargets`] and the total amount of storage targets, based on the
 /// given state.
-fn multiproof_targets_from_state(state: EvmState) -> MultiProofTargets {
-    state
-        .into_par_iter()
-        .filter_map(|(address, account)| {
-            // if the account was not touched, or if the account was selfdestructed, do not
-            // fetch proofs for it
-            //
-            // Since selfdestruct can only happen in the same transaction, we can skip
-            // prefetching proofs for selfdestructed accounts
-            //
-            // See: https://eips.ethereum.org/EIPS/eip-6780
-            if !account.is_touched() || account.is_selfdestructed() {
-                return None;
+fn multiproof_targets_from_state(state: EvmState) -> (MultiProofTargets, usize) {
+    let mut targets = MultiProofTargets::with_capacity(state.len());
+    let mut storage_targets = 0;
+    for (addr, account) in state {
+        // if the account was not touched, or if the account was selfdestructed, do not
+        // fetch proofs for it
+        //
+        // Since selfdestruct can only happen in the same transaction, we can skip
+        // prefetching proofs for selfdestructed accounts
+        //
+        // See: https://eips.ethereum.org/EIPS/eip-6780
+        if !account.is_touched() || account.is_selfdestructed() {
+            continue
+        }
+
+        let mut storage_set =
+            B256Set::with_capacity_and_hasher(account.storage.len(), Default::default());
+        for (key, slot) in account.storage {
+            // do nothing if unchanged
+            if !slot.is_changed() {
+                continue
             }
 
-            let hashed_address = keccak256(address);
+            storage_set.insert(keccak256(B256::new(key.to_be_bytes())));
+        }
 
-            let storage_set: B256Set = account
-                .storage
-                .into_iter()
-                .filter(|(_, slot)| slot.is_changed())
-                .map(|(key, _)| keccak256(B256::new(key.to_be_bytes())))
-                .collect();
+        storage_targets += storage_set.len();
+        targets.insert(keccak256(addr), storage_set);
+    }
 
-            Some((hashed_address, storage_set))
-        })
-        .collect()
+    (targets, storage_targets)
 }
 
 /// 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 {
@@ -840,6 +621,4 @@ pub(crate) struct PrewarmMetrics {
     pub(crate) cache_saving_duration: Gauge,
     /// Counter for transaction execution errors during prewarming
     pub(crate) transaction_errors: Counter,
-    /// A histogram of BAL slot iteration duration during prefetching
-    pub(crate) bal_slot_iteration_duration: Histogram,
 }

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);
 

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

@@ -11,11 +11,9 @@ use crate::tree::{
     StateProviderDatabase, TreeConfig,
 };
 use alloy_consensus::transaction::Either;
-use alloy_eip7928::BlockAccessList;
 use alloy_eips::{eip1898::BlockWithParent, NumHash};
 use alloy_evm::Evm;
 use alloy_primitives::B256;
-use rayon::prelude::*;
 use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock};
 use reth_consensus::{ConsensusError, FullConsensus};
 use reth_engine_primitives::{
@@ -35,13 +33,12 @@ use reth_primitives_traits::{
 };
 use reth_provider::{
     providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockReader,
-    DatabaseProviderFactory, DatabaseProviderROFactory, ExecutionOutcome, HashedPostStateProvider,
-    ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
-    StateProviderFactory, StateReader, TrieReader,
+    DatabaseProviderFactory, ExecutionOutcome, HashedPostStateProvider, ProviderError,
+    PruneCheckpointReader, StageCheckpointReader, StateProvider, StateProviderFactory, StateReader,
+    StateRootProvider, TrieReader,
 };
 use reth_revm::db::State;
-use reth_storage_errors::db::DatabaseError;
-use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot, TrieInputSorted};
+use reth_trie::{updates::TrieUpdates, HashedPostState, TrieInputSorted};
 use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
 use revm_primitives::Address;
 use std::{
@@ -223,7 +220,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)
@@ -233,9 +230,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)
@@ -372,11 +368,10 @@ where
             )
             .into())
         };
-        let mut state_provider = ensure_ok!(provider_builder.build());
+        let state_provider = ensure_ok!(provider_builder.build());
         drop(_enter);
 
-        // Fetch parent block. This goes to memory most of the time unless the parent block is
-        // beyond the in-memory buffer.
+        // fetch parent block
         let Some(parent_block) = ensure_ok!(self.sealed_header_by_hash(parent_hash, ctx.state()))
         else {
             return Err(InsertBlockError::new(
@@ -401,17 +396,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(),
@@ -420,22 +407,25 @@ 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));
-        };
-
-        if self.config.state_provider_metrics() {
-            state_provider = Box::new(InstrumentedStateProvider::new(state_provider, "engine"));
-        }
+        let state_provider = CachedStateProvider::new_with_caches(
+            state_provider,
+            handle.caches(),
+            handle.cache_metrics(),
+        );
 
         // 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);
+            let result = self.execute_block(&state_provider, env, &input, &mut handle);
+            state_provider.record_total_latency();
+            result
+        } else {
+            self.execute_block(&state_provider, env, &input, &mut handle)
+        } {
             Ok(output) => output,
             Err(err) => return self.handle_execution_error(input, err, &parent_block),
         };
@@ -519,7 +509,7 @@ where
             }
 
             let (root, updates) = ensure_ok_post_block!(
-                self.compute_state_root_serial(block.parent_hash(), &hashed_state, ctx.state()),
+                state_provider.state_root_with_updates(hashed_state.clone()),
                 block
             );
             (root, updates, root_time.elapsed())
@@ -549,14 +539,17 @@ where
             .into())
         }
 
-        // Create ExecutionOutcome and wrap in Arc for sharing with both the caching task
-        // and the deferred trie task. This avoids cloning the expensive BundleState.
-        let execution_outcome = Arc::new(ExecutionOutcome::from((output, block_num_hash.number)));
+        // terminate prewarming task with good state output
+        handle.terminate_caching(Some(&output.state));
 
-        // Terminate prewarming task with the shared execution outcome
-        handle.terminate_caching(Some(Arc::clone(&execution_outcome)));
-
-        Ok(self.spawn_deferred_trie_task(block, execution_outcome, &ctx, hashed_state, trie_output))
+        Ok(self.spawn_deferred_trie_task(
+            block,
+            output,
+            block_num_hash.number,
+            &ctx,
+            hashed_state,
+            trie_output,
+        ))
     }
 
     /// Return sealed block header from database or in-memory state by hash.
@@ -599,10 +592,10 @@ where
         state_provider: S,
         env: ExecutionEnv<Evm>,
         input: &BlockOrPayload<T>,
-        handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
+        handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err>,
     ) -> Result<(BlockExecutionOutput<N::Receipt>, Vec<Address>), InsertBlockErrorKind>
     where
-        S: StateProvider + Send,
+        S: StateProvider,
         Err: core::error::Error + Send + Sync + 'static,
         V: PayloadValidator<T, Block = N::Block>,
         T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
@@ -611,7 +604,7 @@ where
         debug!(target: "engine::tree::payload_validator", "Executing block");
 
         let mut db = State::builder()
-            .with_database(StateProviderDatabase::new(state_provider))
+            .with_database(StateProviderDatabase::new(&state_provider))
             .with_bundle_update()
             .without_state_clear()
             .build();
@@ -643,7 +636,6 @@ where
         let (output, senders) = self.metrics.execute_metered(
             executor,
             handle.iter_transactions().map(|res| res.map_err(BlockExecutionError::other)),
-            input.transaction_count(),
             state_hook,
         )?;
         let execution_finish = Instant::now();
@@ -658,6 +650,8 @@ where
     ///
     /// Returns `Ok(_)` if computed successfully.
     /// Returns `Err(_)` if error was encountered during computation.
+    /// `Err(ProviderError::ConsistentView(_))` can be safely ignored and fallback computation
+    /// should be used instead.
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
     fn compute_state_root_parallel(
         &self,
@@ -687,36 +681,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:
@@ -811,12 +775,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,
     > {
@@ -842,14 +804,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
@@ -862,12 +822,8 @@ where
             }
             StateRootStrategy::Parallel | StateRootStrategy::Synchronous => {
                 let start = Instant::now();
-                let handle = self.payload_processor.spawn_cache_exclusive(
-                    env,
-                    txs,
-                    provider_builder,
-                    block_access_list,
-                );
+                let handle =
+                    self.payload_processor.spawn_cache_exclusive(env, txs, provider_builder);
 
                 // Record prewarming initialization duration
                 self.metrics
@@ -916,7 +872,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
@@ -1055,7 +1011,8 @@ where
     fn spawn_deferred_trie_task(
         &self,
         block: RecoveredBlock<N::Block>,
-        execution_outcome: Arc<ExecutionOutcome<N::Receipt>>,
+        output: BlockExecutionOutput<N::Receipt>,
+        block_number: u64,
         ctx: &TreeCtx<'_, N>,
         hashed_state: HashedPostState,
         trie_output: TrieUpdates,
@@ -1105,7 +1062,7 @@ where
 
         ExecutedBlock::with_deferred_trie_data(
             Arc::new(block),
-            execution_outcome,
+            Arc::new(ExecutionOutcome::from((output, block_number))),
             deferred_trie_data,
         )
     }
@@ -1286,21 +1243,4 @@ impl<T: PayloadTypes> BlockOrPayload<T> {
             Self::Block(_) => "block",
         }
     }
-
-    /// Returns the block access list if available.
-    pub const fn block_access_list(&self) -> Option<Result<BlockAccessList, alloy_rlp::Error>> {
-        // TODO decode and return `BlockAccessList`
-        None
-    }
-
-    /// Returns the number of transactions in the payload or block.
-    pub fn transaction_count(&self) -> usize
-    where
-        T::ExecutionData: ExecutionPayload,
-    {
-        match self {
-            Self::Payload(payload) => payload.transaction_count(),
-            Self::Block(block) => block.transaction_count(),
-        }
-    }
 }

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"), {

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);
-    }
 }

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

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)?;

crates/era-utils/src/history.rs

@@ -116,7 +116,7 @@ where
 /// these stages that this work has already been done. Otherwise, there might be some conflict with
 /// database integrity.
 pub fn save_stage_checkpoints<P>(
-    provider: P,
+    provider: &P,
     from: BlockNumber,
     to: BlockNumber,
     processed: u64,
@@ -170,14 +170,18 @@ where
     <P as NodePrimitivesProvider>::Primitives: NodePrimitives<BlockHeader = BH, BlockBody = BB>,
 {
     let reader = open(meta)?;
-    let iter = reader.iter().map(decode as fn(_) -> _);
+    let iter =
+        reader
+            .iter()
+            .map(Box::new(decode)
+                as Box<dyn Fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>);
     let iter = ProcessIter { iter, era: meta };
 
     process_iter(iter, writer, provider, hash_collector, block_numbers)
 }
 
 type ProcessInnerIter<R, BH, BB> =
-    Map<BlockTupleIterator<R>, fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>;
+    Map<BlockTupleIterator<R>, Box<dyn Fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>>;
 
 /// An iterator that wraps era file extraction. After the final item [`EraMeta::mark_as_processed`]
 /// is called to ensure proper cleanup.
@@ -305,7 +309,7 @@ where
         writer.append_header(&header, &hash)?;
 
         // Write bodies to database.
-        provider.append_block_bodies(vec![(header.number(), Some(&body))])?;
+        provider.append_block_bodies(vec![(header.number(), Some(body))])?;
 
         hash_collector.insert(hash, number)?;
     }

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");

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(),
-    }
-}

crates/era/src/era/types/consensus.rs

@@ -59,36 +59,11 @@
 //!     Ok(())
 //! }
 //! ```
+
 use crate::e2s::{error::E2sError, types::Entry};
 use snap::{read::FrameDecoder, write::FrameEncoder};
 use std::io::{Read, Write};
 
-/// Maximum allowed decompressed size for a signed beacon block SSZ payload.
-const MAX_DECOMPRESSED_SIGNED_BEACON_BLOCK_BYTES: usize = 256 * 1024 * 1024; // 256 MiB
-
-/// Maximum allowed decompressed size for a beacon state SSZ payload.
-const MAX_DECOMPRESSED_BEACON_STATE_BYTES: usize = 2 * 1024 * 1024 * 1024; // 2 GiB
-
-fn decompress_snappy_bounded(
-    compressed: &[u8],
-    max_decompressed_bytes: usize,
-    what: &str,
-) -> Result<Vec<u8>, E2sError> {
-    let mut decoder = FrameDecoder::new(compressed).take(max_decompressed_bytes as u64);
-    let mut decompressed = Vec::new();
-
-    Read::read_to_end(&mut decoder, &mut decompressed)
-        .map_err(|e| E2sError::SnappyDecompression(format!("Failed to decompress {what}: {e}")))?;
-
-    if decompressed.len() >= max_decompressed_bytes {
-        return Err(E2sError::SnappyDecompression(format!(
-            "Failed to decompress {what}: decompressed data exceeded limit of {max_decompressed_bytes} bytes"
-        )));
-    }
-
-    Ok(decompressed)
-}
-
 /// `CompressedSignedBeaconBlock` record type: [0x01, 0x00]
 pub const COMPRESSED_SIGNED_BEACON_BLOCK: [u8; 2] = [0x01, 0x00];
 
@@ -129,11 +104,13 @@ impl CompressedSignedBeaconBlock {
 
     /// Decompress to get the original ssz-encoded signed beacon block
     pub fn decompress(&self) -> Result<Vec<u8>, E2sError> {
-        decompress_snappy_bounded(
-            self.data.as_slice(),
-            MAX_DECOMPRESSED_SIGNED_BEACON_BLOCK_BYTES,
-            "signed beacon block",
-        )
+        let mut decoder = FrameDecoder::new(self.data.as_slice());
+        let mut decompressed = Vec::new();
+        Read::read_to_end(&mut decoder, &mut decompressed).map_err(|e| {
+            E2sError::SnappyDecompression(format!("Failed to decompress signed beacon block: {e}"))
+        })?;
+
+        Ok(decompressed)
     }
 
     /// Convert to an [`Entry`]
@@ -191,11 +168,13 @@ impl CompressedBeaconState {
 
     /// Decompress to get the original ssz-encoded beacon state
     pub fn decompress(&self) -> Result<Vec<u8>, E2sError> {
-        decompress_snappy_bounded(
-            self.data.as_slice(),
-            MAX_DECOMPRESSED_BEACON_STATE_BYTES,
-            "beacon state",
-        )
+        let mut decoder = FrameDecoder::new(self.data.as_slice());
+        let mut decompressed = Vec::new();
+        Read::read_to_end(&mut decoder, &mut decompressed).map_err(|e| {
+            E2sError::SnappyDecompression(format!("Failed to decompress beacon state: {e}"))
+        })?;
+
+        Ok(decompressed)
     }
 
     /// Convert to an [`Entry`]
@@ -281,15 +260,4 @@ mod tests {
         let result = CompressedBeaconState::from_entry(&invalid_entry);
         assert!(result.is_err());
     }
-
-    #[test]
-    fn test_bounded_decompression_rejects_oversized_output() {
-        let ssz_data = vec![42u8; 1024];
-        let compressed = CompressedBeaconState::from_ssz(&ssz_data).unwrap();
-
-        let err =
-            decompress_snappy_bounded(compressed.data.as_slice(), 100, "beacon state").unwrap_err();
-
-        assert!(format!("{err:?}").contains("exceeded limit"));
-    }
 }

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);
-    }
 }

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

@@ -252,7 +252,7 @@ impl CompressedBody {
             let mut encoder = FrameEncoder::new(&mut compressed);
 
             Write::write_all(&mut encoder, rlp_data).map_err(|e| {
-                E2sError::SnappyCompression(format!("Failed to compress body: {e}"))
+                E2sError::SnappyCompression(format!("Failed to compress header: {e}"))
             })?;
 
             encoder.flush().map_err(|e| {
@@ -339,7 +339,7 @@ impl CompressedReceipts {
             let mut encoder = FrameEncoder::new(&mut compressed);
 
             Write::write_all(&mut encoder, rlp_data).map_err(|e| {
-                E2sError::SnappyCompression(format!("Failed to compress receipts: {e}"))
+                E2sError::SnappyCompression(format!("Failed to compress header: {e}"))
             })?;
 
             encoder.flush().map_err(|e| {

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);
     }

crates/ethereum/cli/src/app.rs

@@ -1,8 +1,4 @@
-use crate::{
-    interface::{Commands, NoSubCmd},
-    Cli,
-};
-use clap::Subcommand;
+use crate::{interface::Commands, Cli};
 use eyre::{eyre, Result};
 use reth_chainspec::{ChainSpec, EthChainSpec, Hardforks};
 use reth_cli::chainspec::ChainSpecParser;
@@ -22,26 +18,20 @@ use std::{fmt, sync::Arc};
 
 /// A wrapper around a parsed CLI that handles command execution.
 #[derive(Debug)]
-pub struct CliApp<
-    Spec: ChainSpecParser,
-    Ext: clap::Args + fmt::Debug,
-    Rpc: RpcModuleValidator,
-    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
-> {
-    cli: Cli<Spec, Ext, Rpc, SubCmd>,
+pub struct CliApp<Spec: ChainSpecParser, Ext: clap::Args + fmt::Debug, Rpc: RpcModuleValidator> {
+    cli: Cli<Spec, Ext, Rpc>,
     runner: Option<CliRunner>,
     layers: Option<Layers>,
     guard: Option<FileWorkerGuard>,
 }
 
-impl<C, Ext, Rpc, SubCmd> CliApp<C, Ext, Rpc, SubCmd>
+impl<C, Ext, Rpc> CliApp<C, Ext, Rpc>
 where
     C: ChainSpecParser,
     Ext: clap::Args + fmt::Debug,
     Rpc: RpcModuleValidator,
-    SubCmd: ExtendedCommand + Subcommand + fmt::Debug,
 {
-    pub(crate) fn new(cli: Cli<C, Ext, Rpc, SubCmd>) -> Self {
+    pub(crate) fn new(cli: Cli<C, Ext, Rpc>) -> Self {
         Self { cli, runner: None, layers: Some(Layers::new()), guard: None }
     }
 
@@ -108,9 +98,9 @@ where
         self.init_tracing(&runner)?;
 
         // Install the prometheus recorder to be sure to record all metrics
-        install_prometheus_recorder();
+        let _ = install_prometheus_recorder();
 
-        run_commands_with::<C, Ext, Rpc, N, SubCmd>(self.cli, runner, components, launcher)
+        run_commands_with::<C, Ext, Rpc, N>(self.cli, runner, components, launcher)
     }
 
     /// Initializes tracing with the configured options.
@@ -127,8 +117,8 @@ where
 
 /// Run CLI commands with the provided runner, components and launcher.
 /// This is the shared implementation used by both `CliApp` and Cli methods.
-pub(crate) fn run_commands_with<C, Ext, Rpc, N, SubCmd>(
-    cli: Cli<C, Ext, Rpc, SubCmd>,
+pub(crate) fn run_commands_with<C, Ext, Rpc, N>(
+    cli: Cli<C, Ext, Rpc>,
     runner: CliRunner,
     components: impl CliComponentsBuilder<N>,
     launcher: impl AsyncFnOnce(
@@ -141,7 +131,6 @@ where
     Ext: clap::Args + fmt::Debug,
     Rpc: RpcModuleValidator,
     N: CliNodeTypes<Primitives: NodePrimitives<BlockHeader: HeaderMut>, ChainSpec: Hardforks>,
-    SubCmd: ExtendedCommand + Subcommand + fmt::Debug,
 {
     match cli.command {
         Commands::Node(command) => {
@@ -165,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))
@@ -178,19 +165,9 @@ where
         #[cfg(feature = "dev")]
         Commands::TestVectors(command) => runner.run_until_ctrl_c(command.execute()),
         Commands::ReExecute(command) => runner.run_until_ctrl_c(command.execute::<N>(components)),
-        Commands::Ext(command) => command.execute(runner),
     }
 }
 
-/// A trait for extension subcommands that can be added to the CLI.
-///
-/// Consumers implement this trait for their custom subcommands to define
-/// how they should be executed.
-pub trait ExtendedCommand {
-    /// Execute the extension command with the provided CLI runner.
-    fn execute(self, runner: CliRunner) -> Result<()>;
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ethereum/cli/src/interface.rs

@@ -37,11 +37,10 @@ pub struct Cli<
     C: ChainSpecParser = EthereumChainSpecParser,
     Ext: clap::Args + fmt::Debug = NoArgs,
     Rpc: RpcModuleValidator = DefaultRpcModuleValidator,
-    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
 > {
     /// The command to run
     #[command(subcommand)]
-    pub command: Commands<C, Ext, SubCmd>,
+    pub command: Commands<C, Ext>,
 
     /// The logging configuration for the CLI.
     #[command(flatten)]
@@ -72,18 +71,15 @@ impl Cli {
     }
 }
 
-impl<
-        C: ChainSpecParser,
-        Ext: clap::Args + fmt::Debug,
-        Rpc: RpcModuleValidator,
-        SubCmd: crate::app::ExtendedCommand + Subcommand + fmt::Debug,
-    > Cli<C, Ext, Rpc, SubCmd>
-{
+impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, Rpc: RpcModuleValidator> Cli<C, Ext, Rpc> {
     /// Configures the CLI and returns a [`CliApp`] instance.
     ///
     /// This method is used to prepare the CLI for execution by wrapping it in a
     /// [`CliApp`] that can be further configured before running.
-    pub fn configure(self) -> CliApp<C, Ext, Rpc, SubCmd> {
+    pub fn configure(self) -> CliApp<C, Ext, Rpc>
+    where
+        C: ChainSpecParser<ChainSpec = ChainSpec>,
+    {
         CliApp::new(self)
     }
 
@@ -212,10 +208,10 @@ impl<
         let _guard = self.init_tracing(&runner, Layers::new())?;
 
         // Install the prometheus recorder to be sure to record all metrics
-        install_prometheus_recorder();
+        let _ = install_prometheus_recorder();
 
         // Use the shared standalone function to avoid duplication
-        run_commands_with::<C, Ext, Rpc, N, SubCmd>(self, runner, components, launcher)
+        run_commands_with::<C, Ext, Rpc, N>(self, runner, components, launcher)
     }
 
     /// Initializes tracing with the configured options.
@@ -249,11 +245,7 @@ impl<
 
 /// Commands to be executed
 #[derive(Debug, Subcommand)]
-pub enum Commands<
-    C: ChainSpecParser,
-    Ext: clap::Args + fmt::Debug,
-    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
-> {
+pub enum Commands<C: ChainSpecParser, Ext: clap::Args + fmt::Debug> {
     /// Start the node
     #[command(name = "node")]
     Node(Box<node::NodeCommand<C, Ext>>),
@@ -299,27 +291,9 @@ pub enum Commands<
     /// Re-execute blocks in parallel to verify historical sync correctness.
     #[command(name = "re-execute")]
     ReExecute(re_execute::Command<C>),
-    /// Extension subcommands provided by consumers.
-    #[command(flatten)]
-    Ext(SubCmd),
 }
 
-/// A no-op subcommand type for when no extension subcommands are needed.
-///
-/// This is the default type parameter for `Commands` when consumers don't need
-/// to add custom subcommands.
-#[derive(Debug, Subcommand)]
-pub enum NoSubCmd {}
-
-impl crate::app::ExtendedCommand for NoSubCmd {
-    fn execute(self, _runner: CliRunner) -> eyre::Result<()> {
-        match self {}
-    }
-}
-
-impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, SubCmd: Subcommand + fmt::Debug>
-    Commands<C, Ext, SubCmd>
-{
+impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug> Commands<C, Ext> {
     /// Returns the underlying chain being used for commands
     pub fn chain_spec(&self) -> Option<&Arc<C::ChainSpec>> {
         match self {
@@ -339,7 +313,6 @@ impl<C: ChainSpecParser, Ext: clap::Args + fmt::Debug, SubCmd: Subcommand + fmt:
             Self::Config(_) => None,
             Self::Prune(cmd) => cmd.chain_spec(),
             Self::ReExecute(cmd) => cmd.chain_spec(),
-            Self::Ext(_) => None,
         }
     }
 }
@@ -563,100 +536,4 @@ mod tests {
             _ => panic!("Expected Stage command"),
         };
     }
-
-    #[test]
-    fn test_extensible_subcommands() {
-        use crate::app::ExtendedCommand;
-        use reth_cli_runner::CliRunner;
-        use reth_rpc_server_types::DefaultRpcModuleValidator;
-        use std::sync::atomic::{AtomicBool, Ordering};
-
-        #[derive(Debug, Subcommand)]
-        enum CustomCommands {
-            /// A custom hello command
-            #[command(name = "hello")]
-            Hello {
-                /// Name to greet
-                #[arg(long)]
-                name: String,
-            },
-            /// Another custom command
-            #[command(name = "goodbye")]
-            Goodbye,
-        }
-
-        static EXECUTED: AtomicBool = AtomicBool::new(false);
-
-        impl ExtendedCommand for CustomCommands {
-            fn execute(self, _runner: CliRunner) -> eyre::Result<()> {
-                match self {
-                    Self::Hello { name } => {
-                        assert_eq!(name, "world");
-                        EXECUTED.store(true, Ordering::SeqCst);
-                        Ok(())
-                    }
-                    Self::Goodbye => Ok(()),
-                }
-            }
-        }
-
-        // Test parsing the custom "hello" command
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "hello", "--name", "world"])
-        .unwrap();
-
-        match &cli.command {
-            Commands::Ext(CustomCommands::Hello { name }) => {
-                assert_eq!(name, "world");
-            }
-            _ => panic!("Expected Ext(Hello) command"),
-        }
-
-        // Test parsing the custom "goodbye" command
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "goodbye"])
-        .unwrap();
-
-        match &cli.command {
-            Commands::Ext(CustomCommands::Goodbye) => {}
-            _ => panic!("Expected Ext(Goodbye) command"),
-        }
-
-        // Test that built-in commands still work alongside custom ones
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "node"])
-        .unwrap();
-
-        match &cli.command {
-            Commands::Node(_) => {}
-            _ => panic!("Expected Node command"),
-        }
-
-        // Test executing the custom command
-        let cli = Cli::<
-            EthereumChainSpecParser,
-            NoArgs,
-            DefaultRpcModuleValidator,
-            CustomCommands,
-        >::try_parse_from(["reth", "hello", "--name", "world"])
-        .unwrap();
-
-        if let Commands::Ext(cmd) = cli.command {
-            let runner = CliRunner::try_default_runtime().unwrap();
-            cmd.execute(runner).unwrap();
-            assert!(EXECUTED.load(Ordering::SeqCst), "Custom command should have been executed");
-        }
-    }
 }

crates/ethereum/cli/src/lib.rs

@@ -14,8 +14,8 @@ pub mod app;
 pub mod chainspec;
 pub mod interface;
 
-pub use app::{CliApp, ExtendedCommand};
-pub use interface::{Cli, Commands, NoSubCmd};
+pub use app::CliApp;
+pub use interface::{Cli, Commands};
 
 #[cfg(test)]
 mod test {

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

crates/ethereum/evm/src/build.rs

@@ -5,6 +5,7 @@ use alloy_consensus::{
 };
 use alloy_eips::merge::BEACON_NONCE;
 use alloy_evm::{block::BlockExecutorFactory, eth::EthBlockExecutionCtx};
+use alloy_primitives::Bytes;
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_evm::execute::{BlockAssembler, BlockAssemblerInput, BlockExecutionError};
 use reth_execution_types::BlockExecutionResult;
@@ -16,12 +17,14 @@ use revm::context::Block as _;
 pub struct EthBlockAssembler<ChainSpec = reth_chainspec::ChainSpec> {
     /// The chainspec.
     pub chain_spec: Arc<ChainSpec>,
+    /// Extra data to use for the blocks.
+    pub extra_data: Bytes,
 }
 
 impl<ChainSpec> EthBlockAssembler<ChainSpec> {
     /// Creates a new [`EthBlockAssembler`].
-    pub const fn new(chain_spec: Arc<ChainSpec>) -> Self {
-        Self { chain_spec }
+    pub fn new(chain_spec: Arc<ChainSpec>) -> Self {
+        Self { chain_spec, extra_data: Default::default() }
     }
 }
 
@@ -107,7 +110,7 @@ where
             gas_limit: evm_env.block_env.gas_limit(),
             difficulty: evm_env.block_env.difficulty(),
             gas_used: *gas_used,
-            extra_data: ctx.extra_data,
+            extra_data: self.extra_data.clone(),
             parent_beacon_block_root: ctx.parent_beacon_block_root,
             blob_gas_used: block_blob_gas_used,
             excess_blob_gas,