fix: revert slow path optimization that caused regression

The slow path optimization attempted to reuse ancestor cached overlays,
but those overlays were built with a DIFFERENT anchor_hash. Since the
slow path is triggered precisely because the anchor changed (after
persist/reorg), reusing overlays from the old anchor produces incorrect
results.

Reverted to the original slow path that rebuilds from each ancestor's
per-block state changes. The Arc::make_mut tracking metrics are kept.

.config/zepter.yaml

@@ -12,7 +12,7 @@ workflows:
         # Check that `A` activates the features of `B`.
         "propagate-feature",
         # These are the features to check:
-        "--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,js-tracer,portable",
+        "--features=std,op,dev,asm-keccak,jemalloc,jemalloc-prof,tracy-allocator,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,js-tracer,portable,keccak-cache-global",
         # Do not try to add a new section to `[features]` of `A` only because `B` exposes that feature. There are edge-cases where this is still needed, but we can add them manually.
         "--left-side-feature-missing=ignore",
         # Ignore the case that `A` it outside of the workspace. Otherwise it will report errors in external dependencies that we have no influence on.

.github/actionlint.yaml

@@ -0,0 +1,7 @@
+self-hosted-runner:
+  labels:
+    - depot-ubuntu-latest
+    - depot-ubuntu-latest-2
+    - depot-ubuntu-latest-4
+    - depot-ubuntu-latest-8
+    - depot-ubuntu-latest-16

.github/assets/hive/run_simulator.sh

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

.github/workflows/bench.yml

@@ -11,17 +11,19 @@ env:
   CARGO_TERM_COLOR: always
   BASELINE: base
   SEED: reth
+  RUSTC_WRAPPER: "sccache"
 
 name: bench
 jobs:
   codspeed:
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     steps:
       - uses: actions/checkout@v6
         with:
           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,9 +10,12 @@ on:
     types: [opened, reopened, synchronize, closed]
   merge_group:
 
+env:
+  RUSTC_WRAPPER: "sccache"
+
 jobs:
   build:
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-8
     timeout-minutes: 90
     steps:
       - name: Checkout
@@ -33,6 +36,8 @@ 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,11 +13,12 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 name: compact-codec
 jobs:
   compact-codec:
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     strategy:
       matrix:
         bin:
@@ -26,6 +27,7 @@ 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,6 +11,7 @@ on:
 env:
   CARGO_TERM_COLOR: always
   SEED: rustethereumethereumrust
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -19,13 +20,14 @@ concurrency:
 jobs:
   test:
     name: e2e-testsuite
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-4
     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,7 +24,8 @@ jobs:
   prepare-hive:
     if: github.repository == 'paradigmxyz/reth'
     timeout-minutes: 45
-    runs-on: ubuntu-latest
+    runs-on:
+      group: Reth
     steps:
       - uses: actions/checkout@v6
       - name: Checkout hive tests
@@ -44,7 +45,7 @@ jobs:
 
       - name: Restore hive assets cache
         id: cache-hive
-        uses: actions/cache@v4
+        uses: actions/cache@v5
         with:
           path: ./hive_assets
           key: hive-assets-${{ steps.hive-commit.outputs.hash }}-${{ hashFiles('.github/assets/hive/build_simulators.sh') }}
@@ -67,7 +68,7 @@ jobs:
           chmod +x hive
 
       - name: Upload hive assets
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: hive_assets
           path: ./hive_assets
@@ -178,7 +179,8 @@ jobs:
       - prepare-reth
       - prepare-hive
     name: run ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
-    runs-on: ubuntu-latest
+    runs-on:
+      group: Reth
     permissions:
       issues: write
     steps:
@@ -187,13 +189,13 @@ jobs:
           fetch-depth: 0
 
       - name: Download hive assets
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: hive_assets
           path: /tmp
 
       - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: artifacts
           path: /tmp

.github/workflows/integration.yml

@@ -14,6 +14,7 @@ on:
 env:
   CARGO_TERM_COLOR: always
   SEED: rustethereumethereumrust
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -23,7 +24,7 @@ jobs:
   test:
     name: test / ${{ matrix.network }}
     if: github.event_name != 'schedule'
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-4
     env:
       RUST_BACKTRACE: 1
     strategy:
@@ -37,6 +38,7 @@ 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
@@ -74,6 +76,7 @@ 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

@@ -32,7 +32,7 @@ jobs:
     strategy:
       fail-fast: false
     name: run kurtosis
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     needs:
       - prepare-reth
     steps:
@@ -41,7 +41,7 @@ jobs:
           fetch-depth: 0
 
       - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: artifacts
           path: /tmp

.github/workflows/kurtosis.yml

@@ -30,7 +30,7 @@ jobs:
     strategy:
       fail-fast: false
     name: run kurtosis
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     needs:
       - prepare-reth
     steps:
@@ -39,7 +39,7 @@ jobs:
           fetch-depth: 0
 
       - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: artifacts
           path: /tmp

.github/workflows/lint.yml

@@ -8,11 +8,12 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 jobs:
   clippy-binaries:
     name: clippy binaries / ${{ matrix.type }}
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 30
     strategy:
       matrix:
@@ -26,6 +27,7 @@ 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
@@ -40,7 +42,7 @@ jobs:
 
   clippy:
     name: clippy
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
@@ -48,6 +50,7 @@ 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
@@ -56,7 +59,7 @@ jobs:
           RUSTFLAGS: -D warnings
 
   wasm:
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
@@ -65,6 +68,7 @@ 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
@@ -75,7 +79,7 @@ jobs:
           .github/assets/check_wasm.sh
 
   riscv:
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v6
@@ -84,6 +88,7 @@ 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
@@ -93,17 +98,18 @@ jobs:
 
   crate-checks:
     name: crate-checks (${{ matrix.partition }}/${{ matrix.total_partitions }})
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-4
     strategy:
       matrix:
-        partition: [1, 2]
-        total_partitions: [2]
+        partition: [1, 2, 3]
+        total_partitions: [3]
     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
@@ -111,7 +117,7 @@ jobs:
 
   msrv:
     name: MSRV
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 30
     strategy:
       matrix:
@@ -124,6 +130,7 @@ 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
@@ -133,12 +140,13 @@ jobs:
 
   docs:
     name: docs
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-4
     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
@@ -150,7 +158,7 @@ jobs:
 
   fmt:
     name: fmt
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@v6
@@ -158,17 +166,19 @@ 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: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     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
@@ -177,12 +187,13 @@ jobs:
 
   book:
     name: book
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     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
@@ -232,33 +243,41 @@ jobs:
       - name: Ensure no arbitrary or proptest dependency on default build
         run: cargo tree --package reth -e=features,no-dev | grep -Eq "arbitrary|proptest" && exit 1 || exit 0
 
-  # Checks that selected rates can compile with power set of features
+  # Checks that selected crates can compile with power set of features
   features:
     name: features
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     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
       - name: cargo install cargo-hack
         uses: taiki-e/install-action@cargo-hack
-      - run: make check-features
+      - run: |
+          cargo hack check \
+            --package reth-codecs \
+            --package reth-primitives-traits \
+            --package reth-primitives \
+            --feature-powerset \
+            --depth 2
         env:
           RUSTFLAGS: -D warnings
 
   # Check crates correctly propagate features
   feature-propagation:
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 20
     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@v2
+      - uses: taiki-e/cache-cargo-install-action@v3
         with:
           tool: zepter
       - name: Eagerly pull dependencies

.github/workflows/prepare-reth.yml

@@ -26,7 +26,7 @@ jobs:
   prepare-reth:
     if: github.repository == 'paradigmxyz/reth'
     timeout-minutes: 45
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     steps:
       - uses: actions/checkout@v6
       - run: mkdir artifacts
@@ -50,7 +50,7 @@ jobs:
 
       - name: Upload reth image
         id: upload
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: artifacts
           path: ./artifacts

.github/workflows/release-dist.yml

@@ -12,7 +12,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Update Homebrew formula
-        uses: dawidd6/action-homebrew-bump-formula@v6
+        uses: dawidd6/action-homebrew-bump-formula@v7
         with:
           token: ${{ secrets.HOMEBREW }}
           no_fork: true

.github/workflows/release.yml

@@ -22,6 +22,7 @@ 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:
@@ -51,6 +52,7 @@ 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
@@ -104,6 +106,7 @@ 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: |
@@ -141,14 +144,14 @@ jobs:
 
       - name: Upload artifact
         if: ${{ github.event.inputs.dry_run != 'true' }}
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz
           path: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz
 
       - name: Upload signature
         if: ${{ github.event.inputs.dry_run != 'true' }}
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz.asc
           path: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz.asc
@@ -170,7 +173,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Download artifacts
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
       - name: Generate full changelog
         id: changelog
         run: |

.github/workflows/reproducible-build.yml

@@ -42,7 +42,7 @@ jobs:
           echo "Binaries SHA256 on ${{ matrix.machine }}: $(cat checksum.sha256)"
 
       - name: Upload the hash
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: checksum-${{ matrix.machine }}
           path: |
@@ -55,12 +55,12 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Download artifacts from machine-1
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v7
         with:
           name: checksum-machine-1
           path: machine-1/
       - name: Download artifacts from machine-2
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v7
         with:
           name: checksum-machine-2
           path: machine-2/

.github/workflows/stage.yml

@@ -12,6 +12,7 @@ env:
   CARGO_TERM_COLOR: always
   FROM_BLOCK: 0
   TO_BLOCK: 50000
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -22,7 +23,7 @@ jobs:
     name: stage-run-test
     # Only run stage commands test in merge groups
     if: github.event_name == 'merge_group'
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -31,6 +32,7 @@ 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,6 +9,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -17,7 +18,7 @@ concurrency:
 jobs:
   sync:
     name: sync (${{ matrix.chain.bin }})
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -41,6 +42,7 @@ 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,6 +9,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -17,7 +18,7 @@ concurrency:
 jobs:
   sync:
     name: sync (${{ matrix.chain.bin }})
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -41,6 +42,7 @@ 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,6 +11,7 @@ on:
 env:
   CARGO_TERM_COLOR: always
   SEED: rustethereumethereumrust
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -19,7 +20,7 @@ concurrency:
 jobs:
   test:
     name: test / ${{ matrix.type }} (${{ matrix.partition }}/${{ matrix.total_partitions }})
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-4
     env:
       RUST_BACKTRACE: 1
     strategy:
@@ -46,6 +47,7 @@ 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
@@ -64,7 +66,7 @@ jobs:
 
   state:
     name: Ethereum state tests
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest-4
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -91,6 +93,7 @@ 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
@@ -98,7 +101,7 @@ jobs:
 
   doc:
     name: doc tests
-    runs-on: ubuntu-latest
+    runs-on: depot-ubuntu-latest
     env:
       RUST_BACKTRACE: 1
     timeout-minutes: 30
@@ -106,6 +109,7 @@ 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@v7
+        uses: peter-evans/create-pull-request@v8
         with:
           commit-message: "chore: update superchain config"
           title: "chore: update superchain config"

.github/workflows/windows.yml

@@ -9,9 +9,12 @@ on:
     branches: [main]
   merge_group:
 
+env:
+  RUSTC_WRAPPER: "sccache"
+
 jobs:
   check-reth:
-    runs-on: ubuntu-24.04
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 60
 
     steps:
@@ -21,6 +24,7 @@ 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
@@ -30,7 +34,7 @@ jobs:
         run: cargo check --target x86_64-pc-windows-gnu
 
   check-op-reth:
-    runs-on: ubuntu-24.04
+    runs-on: depot-ubuntu-latest
     timeout-minutes: 60
 
     steps:
@@ -40,6 +44,7 @@ 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,6 +12,9 @@ target/
 # Generated by Intellij-based IDEs.
 .idea
 
+# ck-search metadata
+.ck
+
 # Generated by MacOS
 .DS_Store
 

Cargo.toml

@@ -153,6 +153,7 @@ members = [
     "examples/custom-node-components/",
     "examples/custom-payload-builder/",
     "examples/custom-rlpx-subprotocol",
+    "examples/custom-rpc-middleware",
     "examples/custom-node",
     "examples/db-access",
     "examples/engine-api-access",
@@ -375,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" }
@@ -480,57 +481,58 @@ 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.1"
+revm-inspectors = "0.33.2"
 
 # 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-evm = { version = "0.24.1", default-features = false }
-alloy-primitives = { version = "1.4.1", default-features = false, features = ["map-foldhash"] }
+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-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
-alloy-sol-macro = "1.4.1"
-alloy-sol-types = { version = "1.4.1", default-features = false }
+alloy-sol-macro = "1.5.0"
+alloy-sol-types = { version = "1.5.0", default-features = false }
 alloy-trie = { version = "0.9.1", default-features = false }
 
 alloy-hardforks = "0.4.5"
 
-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 }
+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 }
 
 # op
-alloy-op-evm = { version = "0.24.1", default-features = false }
+alloy-op-evm = { version = "0.25.0", default-features = false }
 alloy-op-hardforks = "0.4.4"
-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-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-flz = { version = "0.13.1", default-features = false }
 
 # misc
@@ -546,6 +548,7 @@ 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"
@@ -585,6 +588,7 @@ 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"
@@ -690,7 +694,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"
@@ -727,6 +731,7 @@ 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=release
+ARG BUILD_PROFILE=maxperf
 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=release
+ARG BUILD_PROFILE=maxperf
 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-2025 Reth Contributors
+Copyright 2022-2026 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-2025 Reth Contributors
+Copyright (c) 2022-2026 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,10 +521,3 @@ pr:
 	make update-book-cli && \
 	cargo docs --document-private-items && \
 	make test
-
-check-features:
-	cargo hack check \
-		--package reth-codecs \
-		--package reth-primitives-traits \
-		--package reth-primitives \
-		--feature-powerset

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

@@ -164,12 +164,42 @@ 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
+    /// Comma-separated list of features to enable during reth compilation (applied to both builds)
     ///
     /// 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.
@@ -328,7 +358,6 @@ 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);
@@ -448,6 +477,18 @@ 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,
@@ -459,7 +500,7 @@ async fn run_compilation_phase(
         git_manager.switch_ref(git_ref)?;
 
         // Compile reth (with caching)
-        compilation_manager.compile_reth(commit, is_optimism)?;
+        compilation_manager.compile_reth(commit, is_optimism, features, rustflags)?;
 
         info!("Completed compilation for {} reference", ref_type);
     }
@@ -506,8 +547,8 @@ async fn run_warmup_phase(
     // Build additional args with conditional --debug.startup-sync-state-idle flag
     let additional_args = args.build_additional_args("warmup", args.baseline_args.as_ref());
 
-    // Start reth node for warmup
-    let mut node_process =
+    // Start reth node for warmup (command is not stored for warmup phase)
+    let (mut node_process, _warmup_command) =
         node_manager.start_node(&binary_path, warmup_ref, "warmup", &additional_args).await?;
 
     // Wait for node to be ready and get its current tip
@@ -607,8 +648,8 @@ async fn run_benchmark_workflow(
         // Build additional args with conditional --debug.startup-sync-state-idle flag
         let additional_args = args.build_additional_args(ref_type, base_args_str);
 
-        // Start reth node
-        let mut node_process =
+        // Start reth node and capture the command for reporting
+        let (mut node_process, reth_command) =
             node_manager.start_node(&binary_path, git_ref, ref_type, &additional_args).await?;
 
         // Wait for node to be ready and get its current tip (wherever it is)
@@ -645,8 +686,9 @@ async fn run_benchmark_workflow(
         // Store results for comparison
         comparison_generator.add_ref_results(ref_type, &output_dir)?;
 
-        // Set the benchmark run timestamps
+        // Set the benchmark run timestamps and reth command
         comparison_generator.set_ref_timestamps(ref_type, benchmark_start, benchmark_end)?;
+        comparison_generator.set_ref_command(ref_type, reth_command)?;
 
         info!("Completed {} reference benchmark", ref_type);
     }

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

@@ -21,6 +21,8 @@ pub(crate) struct ComparisonGenerator {
     feature_ref_name: String,
     baseline_results: Option<BenchmarkResults>,
     feature_results: Option<BenchmarkResults>,
+    baseline_command: Option<String>,
+    feature_command: Option<String>,
 }
 
 /// Represents the results from a single benchmark run
@@ -37,7 +39,8 @@ pub(crate) struct BenchmarkResults {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct CombinedLatencyRow {
     pub block_number: u64,
-    pub transaction_count: u64,
+    #[serde(default)]
+    pub transaction_count: Option<u64>,
     pub gas_used: u64,
     pub new_payload_latency: u128,
 }
@@ -46,7 +49,8 @@ pub(crate) struct CombinedLatencyRow {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct TotalGasRow {
     pub block_number: u64,
-    pub transaction_count: u64,
+    #[serde(default)]
+    pub transaction_count: Option<u64>,
     pub gas_used: u64,
     pub time: u128,
 }
@@ -57,7 +61,6 @@ pub(crate) struct TotalGasRow {
 /// - `mean_new_payload_latency_ms`: arithmetic mean latency across blocks.
 /// - `median_new_payload_latency_ms`: p50 latency across blocks.
 /// - `p90_new_payload_latency_ms` / `p99_new_payload_latency_ms`: tail latencies across blocks.
-/// - `std_dev_new_payload_latency_ms`: standard deviation of latency across blocks.
 #[derive(Debug, Clone, Serialize)]
 pub(crate) struct BenchmarkSummary {
     pub total_blocks: u64,
@@ -67,7 +70,6 @@ pub(crate) struct BenchmarkSummary {
     pub median_new_payload_latency_ms: f64,
     pub p90_new_payload_latency_ms: f64,
     pub p99_new_payload_latency_ms: f64,
-    pub std_dev_new_payload_latency_ms: f64,
     pub gas_per_second: f64,
     pub blocks_per_second: f64,
     pub min_block_number: u64,
@@ -91,6 +93,7 @@ pub(crate) struct RefInfo {
     pub summary: BenchmarkSummary,
     pub start_timestamp: Option<DateTime<Utc>>,
     pub end_timestamp: Option<DateTime<Utc>>,
+    pub reth_command: Option<String>,
 }
 
 /// Summary of the comparison between references.
@@ -98,7 +101,6 @@ pub(crate) struct RefInfo {
 /// Percent deltas are `(feature - baseline) / baseline * 100`:
 /// - `new_payload_latency_p50_change_percent` / p90 / p99: percent changes of the respective
 ///   per-block percentiles.
-/// - `std_dev_change_percent`: percent change in standard deviation of newPayload latency.
 /// - `per_block_latency_change_mean_percent` / `per_block_latency_change_median_percent` are the
 ///   mean and median of per-block percent deltas (feature vs baseline), capturing block-level
 ///   drift.
@@ -117,7 +119,6 @@ pub(crate) struct ComparisonSummary {
     pub new_payload_latency_p50_change_percent: f64,
     pub new_payload_latency_p90_change_percent: f64,
     pub new_payload_latency_p99_change_percent: f64,
-    pub std_dev_change_percent: f64,
     pub gas_per_second_change_percent: f64,
     pub blocks_per_second_change_percent: f64,
 }
@@ -126,7 +127,8 @@ pub(crate) struct ComparisonSummary {
 #[derive(Debug, Serialize)]
 pub(crate) struct BlockComparison {
     pub block_number: u64,
-    pub transaction_count: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub transaction_count: Option<u64>,
     pub gas_used: u64,
     pub baseline_new_payload_latency: u128,
     pub feature_new_payload_latency: u128,
@@ -146,6 +148,8 @@ impl ComparisonGenerator {
             feature_ref_name: args.feature_ref.clone(),
             baseline_results: None,
             feature_results: None,
+            baseline_command: None,
+            feature_command: None,
         }
     }
 
@@ -210,6 +214,21 @@ impl ComparisonGenerator {
         Ok(())
     }
 
+    /// Set the reth command for a reference
+    pub(crate) fn set_ref_command(&mut self, ref_type: &str, command: String) -> Result<()> {
+        match ref_type {
+            "baseline" => {
+                self.baseline_command = Some(command);
+            }
+            "feature" => {
+                self.feature_command = Some(command);
+            }
+            _ => return Err(eyre!("Unknown reference type: {}", ref_type)),
+        }
+
+        Ok(())
+    }
+
     /// Generate the final comparison report
     pub(crate) async fn generate_comparison_report(&self) -> Result<()> {
         info!("Generating comparison report...");
@@ -234,12 +253,14 @@ impl ComparisonGenerator {
                 summary: baseline.summary.clone(),
                 start_timestamp: baseline.start_timestamp,
                 end_timestamp: baseline.end_timestamp,
+                reth_command: self.baseline_command.clone(),
             },
             feature: RefInfo {
                 ref_name: feature.ref_name.clone(),
                 summary: feature.summary.clone(),
                 start_timestamp: feature.start_timestamp,
                 end_timestamp: feature.end_timestamp,
+                reth_command: self.feature_command.clone(),
             },
             comparison_summary,
             per_block_comparisons,
@@ -339,9 +360,6 @@ impl ComparisonGenerator {
         let mean_new_payload_latency_ms: f64 =
             latencies_ms.iter().sum::<f64>() / total_blocks as f64;
 
-        let std_dev_new_payload_latency_ms =
-            calculate_std_dev(&latencies_ms, mean_new_payload_latency_ms);
-
         let mut sorted_latencies_ms = latencies_ms;
         sorted_latencies_ms.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal));
         let median_new_payload_latency_ms = percentile(&sorted_latencies_ms, 0.5);
@@ -372,7 +390,6 @@ impl ComparisonGenerator {
             median_new_payload_latency_ms,
             p90_new_payload_latency_ms,
             p99_new_payload_latency_ms,
-            std_dev_new_payload_latency_ms,
             gas_per_second,
             blocks_per_second,
             min_block_number,
@@ -440,10 +457,6 @@ impl ComparisonGenerator {
                 baseline.p99_new_payload_latency_ms,
                 feature.p99_new_payload_latency_ms,
             ),
-            std_dev_change_percent: calc_percent_change(
-                baseline.std_dev_new_payload_latency_ms,
-                feature.std_dev_new_payload_latency_ms,
-            ),
             gas_per_second_change_percent: calc_percent_change(
                 baseline.gas_per_second,
                 feature.gas_per_second,
@@ -574,7 +587,6 @@ impl ComparisonGenerator {
             "  NewPayload Latency p99:           {:+.2}%",
             summary.new_payload_latency_p99_change_percent
         );
-        println!("  NewPayload Latency std dev:       {:+.2}%", summary.std_dev_change_percent);
         println!(
             "  Gas/Second:                      {:+.2}%",
             summary.gas_per_second_change_percent
@@ -597,12 +609,11 @@ impl ComparisonGenerator {
         );
         println!("  NewPayload latency (ms):");
         println!(
-            "    mean: {:.2}, p50: {:.2}, p90: {:.2}, p99: {:.2}, std dev: {:.2}",
+            "    mean: {:.2}, p50: {:.2}, p90: {:.2}, p99: {:.2}",
             baseline.mean_new_payload_latency_ms,
             baseline.median_new_payload_latency_ms,
             baseline.p90_new_payload_latency_ms,
-            baseline.p99_new_payload_latency_ms,
-            baseline.std_dev_new_payload_latency_ms
+            baseline.p99_new_payload_latency_ms
         );
         if let (Some(start), Some(end)) =
             (&report.baseline.start_timestamp, &report.baseline.end_timestamp)
@@ -613,6 +624,9 @@ impl ComparisonGenerator {
                 end.format("%Y-%m-%d %H:%M:%S UTC")
             );
         }
+        if let Some(ref cmd) = report.baseline.reth_command {
+            println!("  Command: {}", cmd);
+        }
         println!();
 
         println!("Feature Summary:");
@@ -627,12 +641,11 @@ impl ComparisonGenerator {
         );
         println!("  NewPayload latency (ms):");
         println!(
-            "    mean: {:.2}, p50: {:.2}, p90: {:.2}, p99: {:.2}, std dev: {:.2}",
+            "    mean: {:.2}, p50: {:.2}, p90: {:.2}, p99: {:.2}",
             feature.mean_new_payload_latency_ms,
             feature.median_new_payload_latency_ms,
             feature.p90_new_payload_latency_ms,
-            feature.p99_new_payload_latency_ms,
-            feature.std_dev_new_payload_latency_ms
+            feature.p99_new_payload_latency_ms
         );
         if let (Some(start), Some(end)) =
             (&report.feature.start_timestamp, &report.feature.end_timestamp)
@@ -643,6 +656,9 @@ impl ComparisonGenerator {
                 end.format("%Y-%m-%d %H:%M:%S UTC")
             );
         }
+        if let Some(ref cmd) = report.feature.reth_command {
+            println!("  Command: {}", cmd);
+        }
         println!();
     }
 }

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

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

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

@@ -211,6 +211,11 @@ 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");
 
@@ -240,19 +245,24 @@ impl NodeManager {
     }
 
     /// Start a reth node using the specified binary path and return the process handle
+    /// along with the formatted reth command string for reporting.
     pub(crate) async fn start_node(
         &mut self,
         binary_path: &std::path::Path,
         _git_ref: &str,
         ref_type: &str,
         additional_args: &[String],
-    ) -> Result<tokio::process::Child> {
+    ) -> Result<(tokio::process::Child, String)> {
         // Store the binary path for later use (e.g., in unwind_to_block)
         self.binary_path = Some(binary_path.to_path_buf());
 
         let binary_path_str = binary_path.to_string_lossy();
         let (reth_args, _) = self.build_reth_args(&binary_path_str, additional_args, ref_type);
 
+        // Format the reth command string for reporting
+        let reth_command = shlex::try_join(reth_args.iter().map(|s| s.as_str()))
+            .wrap_err("Failed to format reth command string")?;
+
         // Log additional arguments if any
         if !self.additional_reth_args.is_empty() {
             info!("Using common additional reth arguments: {:?}", self.additional_reth_args);
@@ -346,7 +356,7 @@ impl NodeManager {
         // Give the node a moment to start up
         sleep(Duration::from_secs(5)).await;
 
-        Ok(child)
+        Ok((child, reth_command))
     }
 
     /// Wait for the node to be ready and return its current tip
@@ -398,7 +408,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().expect("Child process ID should be available");
+        let pid = child.id().ok_or_eyre("Child process ID should be available")?;
 
         // Check if the process has already exited
         match child.try_wait() {
@@ -547,3 +557,16 @@ 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,6 +58,7 @@ 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.jwt-secret <jwt_file_path>
+samply record -p 3001 target/profiling/reth node --metrics localhost:9001 --authrpc.jwtsecret <jwt_file_path>
 ```
 
 ```bash
@@ -143,5 +143,5 @@ To reproduce the benchmark, first re-set the node to the block that the benchmar
 - **RPC Configuration**: The RPC endpoints should be accessible and configured correctly, specifically the RPC endpoint must support `eth_getBlockByNumber` and support fetching full transactions. The benchmark will make one RPC query per block as fast as possible, so ensure the RPC endpoint does not rate limit or block requests after a certain volume.
 - **Reproducibility**: Ensure that the node is at the same state before attempting to retry a benchmark. The `new-payload-fcu` command specifically will commit to the database, so the node must be rolled back using `reth stage unwind` to reproducibly retry benchmarks.
 - **Profiling tools**: If you are collecting CPU profiles, tools like [`samply`](https://github.com/mstange/samply) and [`perf`](https://perf.wiki.kernel.org/index.php/Main_Page) can be useful for analyzing node performance.
-- **Benchmark Data**: `reth-bench` additionally contains a `--benchmark.output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
+- **Benchmark Data**: `reth-bench` additionally contains a `--output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
 - **Platform Information**: To ensure accurate and reproducible benchmarking, document the platform details, including hardware specifications, OS version, and any other relevant information before publishing any benchmarks.

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

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

bin/reth-bench/src/bench_mode.rs

@@ -5,9 +5,8 @@ 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,
+    Continuous(u64),
     /// Run the benchmark for a specific range of blocks.
     Range(RangeInclusive<u64>),
 }
@@ -16,18 +15,19 @@ impl BenchMode {
     /// Check if the block number is in the range
     pub fn contains(&self, block_number: u64) -> bool {
         match self {
-            Self::Continuous => true,
+            Self::Continuous(start) => block_number >= *start,
             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>) -> Result<Self, eyre::Error> {
+    pub fn new(from: Option<u64>, to: Option<u64>, latest_block: u64) -> Result<Self, eyre::Error> {
         // If neither `--from` nor `--to` are provided, we will run the benchmark continuously,
         // starting at the latest block.
         match (from, to) {
             (Some(from), Some(to)) => Ok(Self::Range(from..=to)),
-            (None, None) => Ok(Self::Continuous),
+            (None, None) => Ok(Self::Continuous(latest_block)),
+            (Some(start), None) => Ok(Self::Continuous(start)),
             _ => {
                 // 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() {
+fn main() -> eyre::Result<()> {
     // Enable backtraces unless a RUST_BACKTRACE value has already been explicitly provided.
     if std::env::var_os("RUST_BACKTRACE").is_none() {
         unsafe {
@@ -31,12 +31,11 @@ fn main() {
         }
     }
 
+    color_eyre::install()?;
+
     // Run until either exit or sigint or sigterm
-    let runner = CliRunner::try_default_runtime().unwrap();
-    runner
-        .run_command_until_exit(|ctx| {
-            let command = BenchmarkCommand::parse();
-            command.execute(ctx)
-        })
-        .unwrap();
+    let runner = CliRunner::try_default_runtime()?;
+    runner.run_command_until_exit(|ctx| BenchmarkCommand::parse().execute(ctx))?;
+
+    Ok(())
 }

bin/reth/Cargo.toml

@@ -81,7 +81,7 @@ backon.workspace = true
 tempfile.workspace = true
 
 [features]
-default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer"]
+default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
 
 otlp = [
     "reth-ethereum-cli/otlp",
@@ -102,7 +102,10 @@ 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,11 +38,26 @@ 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,
-    /// Trie input constructed from in-memory overlays.
+    /// 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`].
     pub trie_input: Arc<TrieInputSorted>,
 }
 
@@ -54,6 +69,12 @@ 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> =
@@ -138,8 +159,15 @@ impl DeferredTrieData {
     ///
     /// # Process
     /// 1. Sort the current block's hashed state and trie updates
-    /// 2. Merge ancestor overlays (oldest -> newest, so later state takes precedence)
-    /// 3. Extend the merged overlay with this block's sorted data
+    /// 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.
     ///
     /// Used by both the async background task and the synchronous fallback path.
     ///
@@ -147,7 +175,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
+    /// * `ancestors` - Deferred trie data from ancestor blocks for merging (oldest -> newest)
     pub fn sort_and_build_trie_input(
         hashed_state: &HashedPostState,
         trie_updates: &TrieUpdates,
@@ -156,28 +184,63 @@ 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());
 
-        // 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());
-            }
-            {
-                let nodes_mut = Arc::make_mut(&mut overlay.nodes);
-                nodes_mut.extend_ref(ancestor_data.trie_updates.as_ref());
-            }
-        }
+        // 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();
 
-        // Extend overlay with current block's sorted data
+            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)
+                }
+            }
+        } else {
+            // No ancestors: start from empty overlay (first block after anchor)
+            TrieInputSorted::default()
+        };
+
+        // Extend overlay with current block's sorted data.
+        // Track if Arc::make_mut triggers a clone (strong_count > 1 means parent still holds ref).
         {
+            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());
         }
@@ -190,6 +253,40 @@ 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.
@@ -441,4 +538,274 @@ 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

@@ -18,7 +18,7 @@ use reth_primitives_traits::{
 };
 use reth_storage_api::StateProviderBox;
 use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, TrieInputSorted};
-use std::{collections::BTreeMap, sync::Arc, time::Instant};
+use std::{collections::BTreeMap, ops::Deref, sync::Arc, time::Instant};
 use tokio::sync::{broadcast, watch};
 
 /// Size of the broadcast channel used to notify canonical state events.
@@ -86,14 +86,20 @@ 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 numbers = self.numbers.read();
-        if let Some((earliest_block_number, _)) = numbers.first_key_value() {
-            self.metrics.earliest_block.set(*earliest_block_number as f64);
+        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);
         }
-        if let Some((latest_block_number, _)) = numbers.last_key_value() {
-            self.metrics.latest_block.set(*latest_block_number as f64);
+        if let Some(latest_block_number) = latest {
+            self.metrics.latest_block.set(latest_block_number as f64);
         }
-        self.metrics.num_blocks.set(numbers.len() as f64);
+        self.metrics.num_blocks.set(count as f64);
     }
 
     /// Returns the state for a given block hash.
@@ -634,6 +640,8 @@ impl<N: NodePrimitives> BlockState<N> {
     /// We assume that the `Receipts` in the executed block `ExecutionOutcome`
     /// has only one element corresponding to the executed block associated to
     /// the state.
+    ///
+    /// This clones the vector of receipts. To avoid it, use [`Self::executed_block_receipts_ref`].
     pub fn executed_block_receipts(&self) -> Vec<N::Receipt> {
         let receipts = self.receipts();
 
@@ -646,22 +654,30 @@ impl<N: NodePrimitives> BlockState<N> {
         receipts.first().cloned().unwrap_or_default()
     }
 
-    /// Returns a vector of __parent__ `BlockStates`.
+    /// Returns a slice of `Receipt` of executed block that determines the state.
+    /// We assume that the `Receipts` in the executed block `ExecutionOutcome`
+    /// has only one element corresponding to the executed block associated to
+    /// the state.
+    pub fn executed_block_receipts_ref(&self) -> &[N::Receipt] {
+        let receipts = self.receipts();
+
+        debug_assert!(
+            receipts.len() <= 1,
+            "Expected at most one block's worth of receipts, found {}",
+            receipts.len()
+        );
+
+        receipts.first().map(|receipts| receipts.deref()).unwrap_or_default()
+    }
+
+    /// Returns an iterator over __parent__ `BlockStates`.
     ///
-    /// The block state order in the output vector is newest to oldest (highest to lowest):
+    /// The block state order is newest to oldest (highest to lowest):
     /// `[5,4,3,2,1]`
     ///
     /// Note: This does not include self.
-    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
+    pub fn parent_state_chain(&self) -> impl Iterator<Item = &Self> + '_ {
+        std::iter::successors(self.parent.as_deref(), |state| state.parent.as_deref())
     }
 
     /// Returns a vector of `BlockStates` representing the entire in memory chain.
@@ -672,6 +688,11 @@ 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());
     }
@@ -1435,19 +1456,18 @@ mod tests {
         let mut test_block_builder: TestBlockBuilder = TestBlockBuilder::default();
         let chain = create_mock_state_chain(&mut test_block_builder, 4);
 
-        let parents = chain[3].parent_state_chain();
+        let parents: Vec<_> = chain[3].parent_state_chain().collect();
         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 = chain[2].parent_state_chain();
+        let parents: Vec<_> = chain[2].parent_state_chain().collect();
         assert_eq!(parents.len(), 2);
         assert_eq!(parents[0].block().recovered_block().number, 2);
         assert_eq!(parents[1].block().recovered_block().number, 1);
 
-        let parents = chain[0].parent_state_chain();
-        assert_eq!(parents.len(), 0);
+        assert_eq!(chain[0].parent_state_chain().count(), 0);
     }
 
     #[test]
@@ -1458,8 +1478,7 @@ mod tests {
             create_mock_state(&mut test_block_builder, single_block_number, B256::random());
         let single_block_hash = single_block.block().recovered_block().hash();
 
-        let parents = single_block.parent_state_chain();
-        assert_eq!(parents.len(), 0);
+        assert_eq!(single_block.parent_state_chain().count(), 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, StateRootProvider, StorageRootProvider,
+    StateProvider, StateProviderBox, StateRootProvider, StorageRootProvider,
 };
 use reth_trie::{
     updates::TrieUpdates, AccountProof, HashedPostState, HashedStorage, MultiProof,
     MultiProofTargets, StorageMultiProof, TrieInput,
 };
 use revm_database::BundleState;
-use std::sync::OnceLock;
+use std::{borrow::Cow, 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,15 +24,11 @@ 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: Vec<ExecutedBlock<N>>,
+    pub(crate) in_memory: Cow<'a, [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.
     ///
@@ -42,7 +38,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, trie_input: OnceLock::new() }
+        Self { historical, in_memory: Cow::Owned(in_memory), trie_input: OnceLock::new() }
     }
 
     /// Turn this state provider into a state provider
@@ -53,11 +49,14 @@ 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 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())),
-            )
+            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
         })
     }
 
@@ -71,7 +70,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 {
+        for block in self.in_memory.iter() {
             if block.recovered_block().number() == number {
                 return Ok(Some(block.recovered_block().hash()));
             }
@@ -90,7 +89,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 {
+        for block in self.in_memory.iter() {
             let block_num = block.recovered_block().number();
             if range.contains(&block_num) {
                 in_memory_hashes.push(block.recovered_block().hash());
@@ -112,7 +111,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 {
+        for block in self.in_memory.iter() {
             if let Some(account) = block.execution_output.account(address) {
                 return Ok(account);
             }
@@ -216,7 +215,7 @@ impl<N: NodePrimitives> StateProvider for MemoryOverlayStateProviderRef<'_, N> {
         address: Address,
         storage_key: StorageKey,
     ) -> ProviderResult<Option<StorageValue>> {
-        for block in &self.in_memory {
+        for block in self.in_memory.iter() {
             if let Some(value) = block.execution_output.storage(&address, storage_key.into()) {
                 return Ok(Some(value));
             }
@@ -228,7 +227,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 {
+        for block in self.in_memory.iter() {
             if let Some(contract) = block.execution_output.bytecode(code_hash) {
                 return Ok(Some(contract));
             }
@@ -237,3 +236,46 @@ 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 -= signer_balance_decrease;
+                self.signer_build_account_info.balance -= Self::single_tx_cost();
                 tx
             })
             .collect();

crates/chainspec/src/spec.rs

@@ -80,6 +80,8 @@ 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(),
@@ -968,7 +970,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 + Sync {
+pub trait ChainSpecProvider: Debug + Send {
     /// The chain spec type.
     type ChainSpec: EthChainSpec + 'static;
 

crates/cli/commands/src/common.rs

@@ -23,7 +23,10 @@ use reth_node_core::{
     dirs::{ChainPath, DataDirPath},
 };
 use reth_provider::{
-    providers::{BlockchainProvider, NodeTypesForProvider, StaticFileProvider},
+    providers::{
+        BlockchainProvider, NodeTypesForProvider, RocksDBProvider, StaticFileProvider,
+        StaticFileProviderBuilder,
+    },
     ProviderFactory, StaticFileProviderFactory,
 };
 use reth_stages::{sets::DefaultStages, Pipeline, PipelineTarget};
@@ -75,10 +78,12 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
         let data_dir = self.datadir.clone().resolve_datadir(self.chain.chain());
         let db_path = data_dir.db();
         let sf_path = data_dir.static_files();
+        let rocksdb_path = data_dir.rocksdb();
 
         if access.is_read_write() {
             reth_fs_util::create_dir_all(&db_path)?;
             reth_fs_util::create_dir_all(&sf_path)?;
+            reth_fs_util::create_dir_all(&rocksdb_path)?;
         }
 
         let config_path = self.config.clone().unwrap_or_else(|| data_dir.config());
@@ -98,18 +103,32 @@ 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())?),
-                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)?,
+                StaticFileProviderBuilder::read_write(sf_path)?
+                    .with_genesis_block_number(genesis_block_number)
+                    .build()?,
             ),
+            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, access)?;
+        let provider_factory =
+            self.create_provider_factory(&config, db, sfp, rocksdb_provider, access)?;
         if access.is_read_write() {
             debug!(target: "reth::cli", chain=%self.chain.chain(), genesis=?self.chain.genesis_hash(), "Initializing genesis");
             init_genesis_with_settings(&provider_factory, self.static_files.to_settings())?;
@@ -128,6 +147,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
         config: &Config,
         db: Arc<DatabaseEnv>,
         static_file_provider: StaticFileProvider<N::Primitives>,
+        rocksdb_provider: RocksDBProvider,
         access: AccessRights,
     ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>>
     where
@@ -138,6 +158,7 @@ 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/get.rs

@@ -8,12 +8,17 @@ use reth_db::{
     RawDupSort,
 };
 use reth_db_api::{
-    table::{Decompress, DupSort, Table},
-    tables, RawKey, RawTable, Receipts, TableViewer, Transactions,
+    cursor::{DbCursorRO, DbDupCursorRO},
+    database::Database,
+    table::{Compress, Decompress, DupSort, Table},
+    tables,
+    transaction::DbTx,
+    RawKey, RawTable, Receipts, TableViewer, Transactions,
 };
 use reth_db_common::DbTool;
 use reth_node_api::{HeaderTy, ReceiptTy, TxTy};
 use reth_node_builder::NodeTypesWithDB;
+use reth_primitives_traits::ValueWithSubKey;
 use reth_provider::{providers::ProviderNodeTypes, StaticFileProviderFactory};
 use reth_static_file_types::StaticFileSegment;
 use tracing::error;
@@ -39,6 +44,14 @@ enum Subcommand {
         #[arg(value_parser = maybe_json_value_parser)]
         subkey: Option<String>,
 
+        /// Optional end key for range query (exclusive upper bound)
+        #[arg(value_parser = maybe_json_value_parser)]
+        end_key: Option<String>,
+
+        /// Optional end subkey for range query (exclusive upper bound)
+        #[arg(value_parser = maybe_json_value_parser)]
+        end_subkey: Option<String>,
+
         /// Output bytes instead of human-readable decoded value
         #[arg(long)]
         raw: bool,
@@ -61,8 +74,8 @@ impl Command {
     /// Execute `db get` command
     pub fn execute<N: ProviderNodeTypes>(self, tool: &DbTool<N>) -> eyre::Result<()> {
         match self.subcommand {
-            Subcommand::Mdbx { table, key, subkey, raw } => {
-                table.view(&GetValueViewer { tool, key, subkey, raw })?
+            Subcommand::Mdbx { table, key, subkey, end_key, end_subkey, raw } => {
+                table.view(&GetValueViewer { tool, key, subkey, end_key, end_subkey, raw })?
             }
             Subcommand::StaticFile { segment, key, raw } => {
                 let (key, mask): (u64, _) = match segment {
@@ -154,6 +167,8 @@ struct GetValueViewer<'a, N: NodeTypesWithDB> {
     tool: &'a DbTool<N>,
     key: String,
     subkey: Option<String>,
+    end_key: Option<String>,
+    end_subkey: Option<String>,
     raw: bool,
 }
 
@@ -163,53 +178,158 @@ impl<N: ProviderNodeTypes> TableViewer<()> for GetValueViewer<'_, N> {
     fn view<T: Table>(&self) -> Result<(), Self::Error> {
         let key = table_key::<T>(&self.key)?;
 
-        let content = if self.raw {
-            self.tool
-                .get::<RawTable<T>>(RawKey::from(key))?
-                .map(|content| hex::encode_prefixed(content.raw_value()))
-        } else {
-            self.tool.get::<T>(key)?.as_ref().map(serde_json::to_string_pretty).transpose()?
-        };
+        // A non-dupsort table cannot have subkeys. The `subkey` arg becomes the `end_key`. First we
+        // check that `end_key` and `end_subkey` weren't previously given, as that wouldn't be
+        // valid.
+        if self.end_key.is_some() || self.end_subkey.is_some() {
+            return Err(eyre::eyre!("Only END_KEY can be given for non-DUPSORT tables"));
+        }
 
-        match content {
-            Some(content) => {
-                println!("{content}");
-            }
-            None => {
-                error!(target: "reth::cli", "No content for the given table key.");
-            }
-        };
+        let end_key = self.subkey.clone();
+
+        // Check if we're doing a range query
+        if let Some(ref end_key_str) = end_key {
+            let end_key = table_key::<T>(end_key_str)?;
+
+            // Use walk_range to iterate over the range
+            self.tool.provider_factory.db_ref().view(|tx| {
+                let mut cursor = tx.cursor_read::<T>()?;
+                let walker = cursor.walk_range(key..end_key)?;
+
+                for result in walker {
+                    let (k, v) = result?;
+                    let json_val = if self.raw {
+                        let raw_key = RawKey::from(k);
+                        serde_json::json!({
+                            "key": hex::encode_prefixed(raw_key.raw_key()),
+                            "val": hex::encode_prefixed(v.compress().as_ref()),
+                        })
+                    } else {
+                        serde_json::json!({
+                            "key": &k,
+                            "val": &v,
+                        })
+                    };
+
+                    println!("{}", serde_json::to_string_pretty(&json_val)?);
+                }
+
+                Ok::<_, eyre::Report>(())
+            })??;
+        } else {
+            // Single key lookup
+            let content = if self.raw {
+                self.tool
+                    .get::<RawTable<T>>(RawKey::from(key))?
+                    .map(|content| hex::encode_prefixed(content.raw_value()))
+            } else {
+                self.tool.get::<T>(key)?.as_ref().map(serde_json::to_string_pretty).transpose()?
+            };
+
+            match content {
+                Some(content) => {
+                    println!("{content}");
+                }
+                None => {
+                    error!(target: "reth::cli", "No content for the given table key.");
+                }
+            };
+        }
 
         Ok(())
     }
 
-    fn view_dupsort<T: DupSort>(&self) -> Result<(), Self::Error> {
+    fn view_dupsort<T: DupSort>(&self) -> Result<(), Self::Error>
+    where
+        T::Value: reth_primitives_traits::ValueWithSubKey<SubKey = T::SubKey>,
+    {
         // get a key for given table
         let key = table_key::<T>(&self.key)?;
 
-        // process dupsort table
-        let subkey = table_subkey::<T>(self.subkey.as_deref())?;
-
-        let content = if self.raw {
-            self.tool
-                .get_dup::<RawDupSort<T>>(RawKey::from(key), RawKey::from(subkey))?
-                .map(|content| hex::encode_prefixed(content.raw_value()))
-        } else {
-            self.tool
-                .get_dup::<T>(key, subkey)?
+        // Check if we're doing a range query
+        if let Some(ref end_key_str) = self.end_key {
+            let end_key = table_key::<T>(end_key_str)?;
+            let start_subkey = table_subkey::<T>(Some(
+                self.subkey.as_ref().expect("must have been given if end_key is given").as_str(),
+            ))?;
+            let end_subkey_parsed = self
+                .end_subkey
                 .as_ref()
-                .map(serde_json::to_string_pretty)
-                .transpose()?
-        };
+                .map(|s| table_subkey::<T>(Some(s.as_str())))
+                .transpose()?;
 
-        match content {
-            Some(content) => {
-                println!("{content}");
-            }
-            None => {
-                error!(target: "reth::cli", "No content for the given table subkey.");
-            }
-        };
+            self.tool.provider_factory.db_ref().view(|tx| {
+                let mut cursor = tx.cursor_dup_read::<T>()?;
+
+                // Seek to the starting key. If there is actually a key at the starting key then
+                // seek to the subkey within it.
+                if let Some((decoded_key, _)) = cursor.seek(key.clone())? &&
+                    decoded_key == key
+                {
+                    cursor.seek_by_key_subkey(key.clone(), start_subkey.clone())?;
+                }
+
+                // Get the current position to start iteration
+                let mut current = cursor.current()?;
+
+                while let Some((decoded_key, decoded_value)) = current {
+                    // Extract the subkey using the ValueWithSubKey trait
+                    let decoded_subkey = decoded_value.get_subkey();
+
+                    // Check if we've reached the end (exclusive)
+                    if (&decoded_key, Some(&decoded_subkey)) >=
+                        (&end_key, end_subkey_parsed.as_ref())
+                    {
+                        break;
+                    }
+
+                    // Output the entry with both key and subkey
+                    let json_val = if self.raw {
+                        let raw_key = RawKey::from(decoded_key.clone());
+                        serde_json::json!({
+                            "key": hex::encode_prefixed(raw_key.raw_key()),
+                            "val": hex::encode_prefixed(decoded_value.compress().as_ref()),
+                        })
+                    } else {
+                        serde_json::json!({
+                            "key": &decoded_key,
+                            "val": &decoded_value,
+                        })
+                    };
+
+                    println!("{}", serde_json::to_string_pretty(&json_val)?);
+
+                    // Move to next entry
+                    current = cursor.next()?;
+                }
+
+                Ok::<_, eyre::Report>(())
+            })??;
+        } else {
+            // Single key/subkey lookup
+            let subkey = table_subkey::<T>(self.subkey.as_deref())?;
+
+            let content = if self.raw {
+                self.tool
+                    .get_dup::<RawDupSort<T>>(RawKey::from(key), RawKey::from(subkey))?
+                    .map(|content| hex::encode_prefixed(content.raw_value()))
+            } else {
+                self.tool
+                    .get_dup::<T>(key, subkey)?
+                    .as_ref()
+                    .map(serde_json::to_string_pretty)
+                    .transpose()?
+            };
+
+            match content {
+                Some(content) => {
+                    println!("{content}");
+                }
+                None => {
+                    error!(target: "reth::cli", "No content for the given table subkey.");
+                }
+            };
+        }
         Ok(())
     }
 }

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

@@ -2,6 +2,7 @@ use crate::common::{AccessRights, CliNodeTypes, Environment, EnvironmentArgs};
 use clap::{Parser, Subcommand};
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
+use reth_cli_runner::CliContext;
 use reth_db::version::{get_db_version, DatabaseVersionError, DB_VERSION};
 use reth_db_common::DbTool;
 use std::{
@@ -79,7 +80,10 @@ macro_rules! db_exec {
 
 impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C> {
     /// Execute `db` command
-    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(self) -> eyre::Result<()> {
+    pub async fn execute<N: CliNodeTypes<ChainSpec = C::ChainSpec>>(
+        self,
+        ctx: CliContext,
+    ) -> eyre::Result<()> {
         let data_dir = self.env.datadir.clone().resolve_datadir(self.env.chain.chain());
         let db_path = data_dir.db();
         let static_files_path = data_dir.static_files();
@@ -158,7 +162,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> Command<C>
                 let access_rights =
                     if command.dry_run { AccessRights::RO } else { AccessRights::RW };
                 db_exec!(self.env, tool, N, access_rights, {
-                    command.execute(&tool)?;
+                    command.execute(&tool, ctx.task_executor.clone())?;
                 });
             }
             Subcommands::StaticFileHeader(command) => {

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

@@ -18,6 +18,7 @@ use reth_node_metrics::{
 };
 use reth_provider::{providers::ProviderNodeTypes, ChainSpecProvider, StageCheckpointReader};
 use reth_stages::StageId;
+use reth_tasks::TaskExecutor;
 use reth_trie::{
     verify::{Output, Verifier},
     Nibbles,
@@ -48,52 +49,37 @@ pub struct Command {
 
 impl Command {
     /// Execute `db repair-trie` command
-    pub fn execute<N: ProviderNodeTypes>(self, tool: &DbTool<N>) -> eyre::Result<()> {
+    pub fn execute<N: ProviderNodeTypes>(
+        self,
+        tool: &DbTool<N>,
+        task_executor: TaskExecutor,
+    ) -> eyre::Result<()> {
         // Set up metrics server if requested
         let _metrics_handle = if let Some(listen_addr) = self.metrics {
-            // Spawn an OS thread with a single-threaded tokio runtime for the metrics server
             let chain_name = tool.provider_factory.chain_spec().chain().to_string();
+            let executor = task_executor.clone();
 
-            let handle = std::thread::Builder::new().name("metrics-server".to_string()).spawn(
-                move || {
-                    // Create a single-threaded tokio runtime
-                    let runtime = tokio::runtime::Builder::new_current_thread()
-                        .enable_all()
-                        .build()
-                        .expect("Failed to create tokio runtime for metrics server");
+            let handle = task_executor.spawn_critical("metrics server", async move {
+                let config = MetricServerConfig::new(
+                    listen_addr,
+                    VersionInfo {
+                        version: version_metadata().cargo_pkg_version.as_ref(),
+                        build_timestamp: version_metadata().vergen_build_timestamp.as_ref(),
+                        cargo_features: version_metadata().vergen_cargo_features.as_ref(),
+                        git_sha: version_metadata().vergen_git_sha.as_ref(),
+                        target_triple: version_metadata().vergen_cargo_target_triple.as_ref(),
+                        build_profile: version_metadata().build_profile_name.as_ref(),
+                    },
+                    ChainSpecInfo { name: chain_name },
+                    executor,
+                    Hooks::builder().build(),
+                );
 
-                    let handle = runtime.handle().clone();
-                    runtime.block_on(async move {
-                        let task_manager = reth_tasks::TaskManager::new(handle.clone());
-                        let task_executor = task_manager.executor();
-
-                        let config = MetricServerConfig::new(
-                            listen_addr,
-                            VersionInfo {
-                                version: version_metadata().cargo_pkg_version.as_ref(),
-                                build_timestamp: version_metadata().vergen_build_timestamp.as_ref(),
-                                cargo_features: version_metadata().vergen_cargo_features.as_ref(),
-                                git_sha: version_metadata().vergen_git_sha.as_ref(),
-                                target_triple: version_metadata()
-                                    .vergen_cargo_target_triple
-                                    .as_ref(),
-                                build_profile: version_metadata().build_profile_name.as_ref(),
-                            },
-                            ChainSpecInfo { name: chain_name },
-                            task_executor,
-                            Hooks::builder().build(),
-                        );
-
-                        // Spawn the metrics server
-                        if let Err(e) = MetricServer::new(config).serve().await {
-                            tracing::error!("Metrics server error: {}", e);
-                        }
-
-                        // Block forever to keep the runtime alive
-                        std::future::pending::<()>().await
-                    });
-                },
-            )?;
+                // Spawn the metrics server
+                if let Err(e) = MetricServer::new(config).serve().await {
+                    tracing::error!("Metrics server error: {}", e);
+                }
+            });
 
             Some(handle)
         } else {
@@ -315,8 +301,8 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
     if inconsistent_nodes == 0 {
         info!("No inconsistencies found");
     } else {
-        info!("Repaired {} inconsistencies, committing changes", inconsistent_nodes);
         provider_rw.commit()?;
+        info!("Repaired {} inconsistencies and committed changes", inconsistent_nodes);
     }
 
     Ok(())

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

@@ -92,6 +92,8 @@ 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,8 +1,9 @@
 //! 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::{EthChainSpec, EthereumHardforks};
+use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
 use reth_provider::BlockHashReader;
 use std::sync::Arc;
@@ -22,8 +23,9 @@ 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(0)?
+            .block_hash(genesis_block_number)?
             .ok_or_else(|| eyre::eyre!("Genesis hash not found."))?;
 
         info!(target: "reth::cli", hash = ?hash, "Genesis block written");

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

@@ -110,7 +110,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> InitStateC
                 static_file_provider.commit()?;
             } else if last_block_number > 0 && last_block_number < header.number() {
                 return Err(eyre::eyre!(
-                    "Data directory should be empty when calling init-state with --without-evm-history."
+                    "Data directory should be empty when calling init-state with --without-evm."
                 ));
             }
         }

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 headers received. Expected: 1. Received: {}",
+                        "Invalid number of bodies 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)
+            .external_ip_resolver(self.network.nat.clone())
             .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, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
     DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::ExecutionStage, Stage, StageCheckpoint, UnwindInput};
@@ -42,6 +42,7 @@ where
                 Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
             )?,
             to,
             from,

crates/cli/commands/src/stage/dump/hashing_account.rs

@@ -6,7 +6,7 @@ use reth_db_api::{database::Database, table::TableImporter, tables};
 use reth_db_common::DbTool;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
     DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::AccountHashingStage, Stage, StageCheckpoint, UnwindInput};
@@ -39,6 +39,7 @@ pub(crate) async fn dump_hashing_account_stage<N: ProviderNodeTypes<DB = Arc<Dat
                 Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
             )?,
             to,
             from,

crates/cli/commands/src/stage/dump/hashing_storage.rs

@@ -5,7 +5,7 @@ use reth_db_api::{database::Database, table::TableImporter, tables};
 use reth_db_common::DbTool;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
     DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{stages::StorageHashingStage, Stage, StageCheckpoint, UnwindInput};
@@ -29,6 +29,7 @@ pub(crate) async fn dump_hashing_storage_stage<N: ProviderNodeTypes<DB = Arc<Dat
                 Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
             )?,
             to,
             from,

crates/cli/commands/src/stage/dump/merkle.rs

@@ -12,7 +12,7 @@ use reth_evm::ConfigureEvm;
 use reth_exex::ExExManagerHandle;
 use reth_node_core::dirs::{ChainPath, DataDirPath};
 use reth_provider::{
-    providers::{ProviderNodeTypes, StaticFileProvider},
+    providers::{ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
     DatabaseProviderFactory, ProviderFactory,
 };
 use reth_stages::{
@@ -62,6 +62,7 @@ where
                 Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).build()?,
             )?,
             to,
             from,

crates/cli/runner/src/lib.rs

@@ -97,6 +97,57 @@ impl CliRunner {
         command_res
     }
 
+    /// Executes a command in a blocking context with access to `CliContext`.
+    ///
+    /// See [`Runtime::spawn_blocking`](tokio::runtime::Runtime::spawn_blocking).
+    pub fn run_blocking_command_until_exit<F, E>(
+        self,
+        command: impl FnOnce(CliContext) -> F + Send + 'static,
+    ) -> Result<(), E>
+    where
+        F: Future<Output = Result<(), E>> + Send + 'static,
+        E: Send + Sync + From<std::io::Error> + From<reth_tasks::PanickedTaskError> + 'static,
+    {
+        let AsyncCliRunner { context, mut task_manager, tokio_runtime } =
+            AsyncCliRunner::new(self.tokio_runtime);
+
+        // Spawn the command on the blocking thread pool
+        let handle = tokio_runtime.handle().clone();
+        let command_handle =
+            tokio_runtime.handle().spawn_blocking(move || handle.block_on(command(context)));
+
+        // Wait for the command to complete or ctrl-c
+        let command_res = tokio_runtime.block_on(run_to_completion_or_panic(
+            &mut task_manager,
+            run_until_ctrl_c(
+                async move { command_handle.await.expect("Failed to join blocking task") },
+            ),
+        ));
+
+        if command_res.is_err() {
+            error!(target: "reth::cli", "shutting down due to error");
+        } else {
+            debug!(target: "reth::cli", "shutting down gracefully");
+            task_manager.graceful_shutdown_with_timeout(Duration::from_secs(5));
+        }
+
+        // Shutdown the runtime on a separate thread
+        let (tx, rx) = mpsc::channel();
+        std::thread::Builder::new()
+            .name("tokio-runtime-shutdown".to_string())
+            .spawn(move || {
+                drop(tokio_runtime);
+                let _ = tx.send(());
+            })
+            .unwrap();
+
+        let _ = rx.recv_timeout(Duration::from_secs(5)).inspect_err(|err| {
+            debug!(target: "reth::cli", %err, "tokio runtime shutdown timed out");
+        });
+
+        command_res
+    }
+
     /// Executes a regular future until completion or until external signal received.
     pub fn run_until_ctrl_c<F, E>(self, fut: F) -> Result<(), E>
     where

crates/config/src/config.rs

@@ -22,7 +22,6 @@ 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))]
@@ -531,8 +530,12 @@ impl PruneConfig {
         self.segments.receipts.is_some() || !self.segments.receipts_log_filter.is_empty()
     }
 
-    /// Merges another `PruneConfig` into this one, taking values from the other config if and only
-    /// if the corresponding value in this config is not set.
+    /// Merges values from `other` into `self`.
+    /// - `Option<PruneMode>` fields: set from `other` only if `self` is `None`.
+    /// - `block_interval`: set from `other` only if `self.block_interval ==
+    ///   DEFAULT_BLOCK_INTERVAL`.
+    /// - `merkle_changesets`: always set from `other`.
+    /// - `receipts_log_filter`: set from `other` only if `self` is empty and `other` is non-empty.
     pub fn merge(&mut self, other: Self) {
         let Self {
             block_interval,
@@ -561,7 +564,7 @@ impl PruneConfig {
         self.segments.account_history = self.segments.account_history.or(account_history);
         self.segments.storage_history = self.segments.storage_history.or(storage_history);
         self.segments.bodies_history = self.segments.bodies_history.or(bodies_history);
-        // Merkle changesets is not optional, so we just replace it if provided
+        // Merkle changesets is not optional; always take the value from `other`
         self.segments.merkle_changesets = merkle_changesets;
 
         if self.segments.receipts_log_filter.0.is_empty() && !receipts_log_filter.0.is_empty() {

crates/consensus/common/src/validation.rs

@@ -279,20 +279,28 @@ pub fn validate_against_parent_hash_number<H: BlockHeader>(
     header: &H,
     parent: &SealedHeader<H>,
 ) -> Result<(), ConsensusError> {
-    // Parent number is consistent.
-    if parent.number() + 1 != header.number() {
-        return Err(ConsensusError::ParentBlockNumberMismatch {
-            parent_block_number: parent.number(),
-            block_number: header.number(),
-        })
-    }
-
     if parent.hash() != header.parent_hash() {
         return Err(ConsensusError::ParentHashMismatch(
             GotExpected { got: header.parent_hash(), expected: parent.hash() }.into(),
         ))
     }
 
+    let Some(parent_number) = parent.number().checked_add(1) else {
+        // parent block already reached the maximum
+        return Err(ConsensusError::ParentBlockNumberMismatch {
+            parent_block_number: parent.number(),
+            block_number: u64::MAX,
+        })
+    };
+
+    // Parent number is consistent.
+    if parent_number != header.number() {
+        return Err(ConsensusError::ParentBlockNumberMismatch {
+            parent_block_number: parent.number(),
+            block_number: header.number(),
+        })
+    }
+
     Ok(())
 }
 
@@ -327,7 +335,7 @@ pub fn validate_against_parent_eip1559_base_fee<ChainSpec: EthChainSpec + Ethere
     Ok(())
 }
 
-/// Validates the timestamp against the parent to make sure it is in the past.
+/// Validates that the block timestamp is greater than the parent block timestamp.
 #[inline]
 pub fn validate_against_parent_timestamp<H: BlockHeader>(
     header: &H,

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 the gas limit is more than the maximum allowed.
+    /// Error when 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::{Network, Provider, ProviderBuilder};
+use alloy_provider::{ConnectionConfig, Network, Provider, ProviderBuilder, WebSocketConfig};
 use alloy_transport::TransportResult;
 use futures::{Stream, StreamExt};
 use reth_node_api::Block;
@@ -25,7 +25,19 @@ 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(rpc_url).await?),
+            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?,
+            ),
             url: rpc_url.to_string(),
             convert: Arc::new(convert),
         })
@@ -77,8 +89,8 @@ where
                 match res {
                     Ok(block) => {
                         if tx.send((self.convert)(block)).await.is_err() {
-                            // Channel closed.
-                            break;
+                            // Channel closed - receiver dropped, exit completely.
+                            return;
                         }
                     }
                     Err(err) => {
@@ -95,7 +107,7 @@ where
             debug!(
                 target: "consensus::debug-client",
                 url=%self.url,
-                "Re-estbalishing block subscription",
+                "Re-establishing block subscription",
             );
         }
     }

crates/e2e-test-utils/src/setup_builder.rs

@@ -11,6 +11,7 @@ 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;
@@ -157,8 +158,8 @@ where
                     .await?;
 
                 let node = NodeTestContext::new(node, self.attributes_generator).await?;
-
-                let genesis = node.block_hash(0);
+                let genesis_number = self.chain_spec.genesis_header().number();
+                let genesis = node.block_hash(genesis_number);
                 node.update_forkchoice(genesis, genesis).await?;
 
                 eyre::Ok(node)

crates/e2e-test-utils/src/setup_import.rs

@@ -110,6 +110,7 @@ pub async fn setup_engine_with_chain_import(
         // Create database path and static files path
         let db_path = datadir.join("db");
         let static_files_path = datadir.join("static_files");
+        let rocksdb_dir_path = datadir.join("rocksdb");
 
         // Initialize the database using init_db (same as CLI import command)
         // Use the same database arguments as the node will use
@@ -125,6 +126,7 @@ pub async fn setup_engine_with_chain_import(
             db.clone(),
             chain_spec.clone(),
             reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())?,
+            reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
         )?;
 
         // Initialize genesis if needed
@@ -311,6 +313,7 @@ mod tests {
         std::fs::create_dir_all(&datadir).unwrap();
         let db_path = datadir.join("db");
         let static_files_path = datadir.join("static_files");
+        let rocksdb_dir_path = datadir.join("rocksdb");
 
         // Import the chain
         {
@@ -324,6 +327,9 @@ mod tests {
                 chain_spec.clone(),
                 reth_provider::providers::StaticFileProvider::read_write(static_files_path.clone())
                     .unwrap(),
+                reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path.clone())
+                    .build()
+                    .unwrap(),
             )
             .expect("failed to create provider factory");
 
@@ -385,6 +391,9 @@ mod tests {
                 chain_spec.clone(),
                 reth_provider::providers::StaticFileProvider::read_only(static_files_path, false)
                     .unwrap(),
+                reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
+                    .build()
+                    .unwrap(),
             )
             .expect("failed to create provider factory");
 
@@ -472,11 +481,15 @@ mod tests {
         // Create static files path
         let static_files_path = datadir.join("static_files");
 
+        // Create rocksdb path
+        let rocksdb_dir_path = datadir.join("rocksdb");
+
         // Create a provider factory
         let provider_factory: ProviderFactory<MockNodeTypesWithDB> = ProviderFactory::new(
             db.clone(),
             chain_spec.clone(),
             reth_provider::providers::StaticFileProvider::read_write(static_files_path).unwrap(),
+            reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path).build().unwrap(),
         )
         .expect("failed to create provider factory");
 

crates/e2e-test-utils/src/testsuite/actions/engine_api.rs

@@ -2,9 +2,7 @@
 
 use crate::testsuite::{Action, Environment};
 use alloy_primitives::B256;
-use alloy_rpc_types_engine::{
-    ExecutionPayloadV1, ExecutionPayloadV2, ExecutionPayloadV3, PayloadStatusEnum,
-};
+use alloy_rpc_types_engine::{ExecutionPayloadV3, PayloadStatusEnum};
 use alloy_rpc_types_eth::{Block, Header, Receipt, Transaction, TransactionRequest};
 use eyre::Result;
 use futures_util::future::BoxFuture;
@@ -131,7 +129,10 @@ where
             })?;
 
             // Convert block to ExecutionPayloadV3
-            let payload = block_to_payload_v3(block.clone());
+            let payload = ExecutionPayloadV3::from_block_unchecked(
+                block.hash(),
+                &block.map_transactions(|tx| tx.inner).into_consensus(),
+            );
 
             // Send the payload to the target node
             let target_engine = env.node_clients[self.node_idx].engine.http_client();
@@ -327,32 +328,3 @@ 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, StateProviderFactory};
+use reth_provider::{BlockExecutionOutput, StateProvider, StateProviderBox, 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: BTreeMap::from_iter(acc.storage.clone()),
+                        storage: acc.storage.iter().map(|(k, v)| (*k, *v)).collect(),
                         status: acc.status,
                     },
                 )
             })
             .collect(),
-        contracts: BTreeMap::from_iter(bundle_state.contracts.clone()),
+        contracts: bundle_state.contracts.iter().map(|(k, v)| (*k, v.clone())).collect(),
         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: BTreeMap::from_iter(rev.storage.clone()),
+                                storage: rev.storage.iter().map(|(k, v)| (*k, *v)).collect(),
                                 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<Box<dyn StateProvider>>>,
+    mut db: State<StateProviderDatabase<StateProviderBox>>,
 ) -> eyre::Result<CollectionResult> {
     let bundle_state = db.take_bundle();
     let mut codes = BTreeMap::new();
@@ -530,9 +530,7 @@ 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 Box<dyn StateProvider>
-            ))
+            .with_database(StateProviderDatabase::new(Box::new(state_provider) as StateProviderBox))
             .with_bundle_update()
             .build();
 

crates/engine/primitives/src/config.rs

@@ -1,5 +1,7 @@
 //! 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;
 
@@ -40,7 +42,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 = 256;
+const DEFAULT_BLOCK_BUFFER_LIMIT: u32 = EPOCH_SLOTS as u32 * 2;
 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;
@@ -89,6 +91,8 @@ 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.
@@ -97,7 +101,7 @@ pub struct TreeConfig {
     state_provider_metrics: bool,
     /// Cross-block cache size in bytes.
     cross_block_cache_size: u64,
-    /// Whether the host has enough parallelism to run state root in parallel.
+    /// Whether the host has enough parallelism to run state root task.
     has_enough_parallelism: bool,
     /// Whether multiproof task should chunk proof targets.
     multiproof_chunking_enabled: bool,
@@ -143,6 +147,7 @@ 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,
@@ -173,6 +178,7 @@ 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,
@@ -197,6 +203,7 @@ 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,
@@ -271,7 +278,12 @@ impl TreeConfig {
         self.disable_parallel_sparse_trie
     }
 
-    /// Returns whether or not parallel prewarming should be used.
+    /// 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.
     pub const fn disable_prewarming(&self) -> bool {
         self.disable_prewarming
     }
@@ -363,6 +375,12 @@ 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;
@@ -385,17 +403,12 @@ impl TreeConfig {
         self
     }
 
-    /// Setter for whether or not the host has enough parallelism to run state root in parallel.
+    /// Setter for has enough parallelism.
     pub const fn with_has_enough_parallelism(mut self, has_enough_parallelism: bool) -> Self {
         self.has_enough_parallelism = has_enough_parallelism;
         self
     }
 
-    /// Whether or not the host has enough parallelism to run state root in parallel.
-    pub const fn has_enough_parallelism(&self) -> bool {
-        self.has_enough_parallelism
-    }
-
     /// Setter for state provider metrics.
     pub const fn with_state_provider_metrics(mut self, state_provider_metrics: bool) -> Self {
         self.state_provider_metrics = state_provider_metrics;

crates/engine/primitives/src/lib.rs

@@ -22,7 +22,8 @@ use reth_trie_common::HashedPostState;
 use serde::{de::DeserializeOwned, Serialize};
 
 // Re-export [`ExecutionPayload`] moved to `reth_payload_primitives`
-pub use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator};
+#[cfg(feature = "std")]
+pub use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator, ExecutableTxTuple};
 pub use reth_payload_primitives::ExecutionPayload;
 
 mod error;

crates/engine/tree/Cargo.toml

@@ -16,7 +16,7 @@ reth-chain-state.workspace = true
 reth-chainspec = { workspace = true, optional = true }
 reth-consensus.workspace = true
 reth-db.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-errors.workspace = true
 reth-execution-types.workspace = true
 reth-evm = { workspace = true, features = ["metrics"] }
@@ -29,6 +29,7 @@ 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"] }
@@ -39,6 +40,7 @@ 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
@@ -51,6 +53,7 @@ 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,17 +230,18 @@ fn bench_state_root(c: &mut Criterion) {
                             let mut handle = payload_processor.spawn(
                                 Default::default(),
                                 (
-                                    core::iter::empty::<
+                                    Vec::<
                                         Result<
                                             Recovered<TransactionSigned>,
                                             core::convert::Infallible,
                                         >,
-                                    >(),
+                                    >::new(),
                                     std::convert::identity,
                                 ),
                                 StateProviderBuilder::new(provider.clone(), genesis_hash, None),
                                 OverlayStateProviderFactory::new(provider),
                                 &TreeConfig::default(),
+                                None,
                             );
 
                             let mut state_hook = handle.state_hook();

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::FinishStateUpdates` message
+Once all transactions are executed, the [Engine](#engine) sends a `StateRootMessage::FinishedStateUpdates` 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::FinishStateUpdates` was sent)
+1. Are all updates received? (`StateRootMessage::FinishedStateUpdates` 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 + Sync {
+pub trait BackfillSync: Send {
     /// Performs a backfill action.
     fn on_action(&mut self, action: BackfillAction);
 

crates/engine/tree/src/chain.rs

@@ -219,10 +219,19 @@ pub enum HandlerEvent<T> {
 }
 
 /// Internal events issued by the [`ChainOrchestrator`].
-#[derive(Clone, Debug)]
+#[derive(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,6 +19,8 @@ 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,5 +1,4 @@
 use crate::metrics::PersistenceMetrics;
-use alloy_consensus::BlockHeader;
 use alloy_eips::BlockNumHash;
 use reth_chain_state::ExecutedBlock;
 use reth_errors::ProviderError;
@@ -142,27 +141,25 @@ where
         &self,
         blocks: Vec<ExecutedBlock<N::Primitives>>,
     ) -> Result<Option<BlockNumHash>, PersistenceError> {
-        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 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 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_hash_num.is_some() {
+        if last_block.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_hash, last=?last_block_hash, "Saved range of blocks");
+        debug!(target: "engine::persistence", first=?first_block, last=?last_block, "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_hash_num)
+        Ok(last_block)
     }
 }
 

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 by last recently used block.
+/// is done in FIFO order (oldest inserted block is evicted first).
 #[derive(Debug)]
 pub struct BlockBuffer<B: Block> {
     /// All blocks in the buffer stored by their block hash.
@@ -66,9 +66,14 @@ impl<B: Block> BlockBuffer<B> {
     pub fn insert_block(&mut self, block: SealedBlock<B>) {
         let hash = block.hash();
 
-        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);
+        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);
+            }
+        };
 
         // 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,6 +31,9 @@ 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>
@@ -39,12 +42,32 @@ where
 {
     /// Creates a new [`CachedStateProvider`] from an [`ExecutionCache`], state provider, and
     /// [`CachedStateMetrics`].
-    pub(crate) const fn new_with_caches(
+    pub(crate) const fn new(
         state_provider: S,
         caches: ExecutionCache,
         metrics: CachedStateMetrics,
     ) -> Self {
-        Self { state_provider, caches, metrics }
+        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
     }
 }
 
@@ -123,7 +146,10 @@ impl<S: AccountReader> AccountReader for CachedStateProvider<S> {
         self.metrics.account_cache_misses.increment(1);
 
         let res = self.state_provider.basic_account(address)?;
-        self.caches.account_cache.insert(*address, res);
+
+        if self.is_prewarm() {
+            self.caches.account_cache.insert(*address, res);
+        }
         Ok(res)
     }
 }
@@ -148,15 +174,19 @@ 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)?;
-                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);
+
+                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);
+                }
 
                 self.metrics.storage_cache_misses.increment(1);
                 Ok(final_res)
@@ -183,7 +213,11 @@ 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)?;
-        self.caches.code_cache.insert(*code_hash, final_res.clone());
+
+        if self.is_prewarm() {
+            self.caches.code_cache.insert(*code_hash, final_res.clone());
+        }
+
         Ok(final_res)
     }
 }
@@ -785,7 +819,7 @@ mod tests {
 
         let caches = ExecutionCacheBuilder::default().build_caches(1000);
         let state_provider =
-            CachedStateProvider::new_with_caches(provider, caches, CachedStateMetrics::zeroed());
+            CachedStateProvider::new(provider, caches, CachedStateMetrics::zeroed());
 
         // check that the storage is empty
         let res = state_provider.storage(address, storage_key);
@@ -808,7 +842,7 @@ mod tests {
 
         let caches = ExecutionCacheBuilder::default().build_caches(1000);
         let state_provider =
-            CachedStateProvider::new_with_caches(provider, caches, CachedStateMetrics::zeroed());
+            CachedStateProvider::new(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(Default)]
+#[derive(Debug, Default)]
 pub(crate) struct AtomicDuration {
     /// The nanoseconds part of the duration
     ///
@@ -59,7 +59,8 @@ impl AtomicDuration {
 }
 
 /// A wrapper of a state provider and latency metrics.
-pub(crate) struct InstrumentedStateProvider<S> {
+#[derive(Debug)]
+pub struct InstrumentedStateProvider<S> {
     /// The state provider
     state_provider: S,
 
@@ -80,11 +81,12 @@ impl<S> InstrumentedStateProvider<S>
 where
     S: StateProvider,
 {
-    /// Creates a new [`InstrumentedStateProvider`] from a state provider
-    pub(crate) fn from_state_provider(state_provider: S) -> Self {
+    /// Creates a new [`InstrumentedStateProvider`] from a state provider with the provided label
+    /// for metrics.
+    pub fn new(state_provider: S, source: &'static str) -> Self {
         Self {
             state_provider,
-            metrics: StateProviderMetrics::default(),
+            metrics: StateProviderMetrics::new_with_labels(&[("source", source)]),
             total_storage_fetch_latency: AtomicDuration::zero(),
             total_code_fetch_latency: AtomicDuration::zero(),
             total_account_fetch_latency: AtomicDuration::zero(),
@@ -134,6 +136,12 @@ 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,6 +48,7 @@ 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

@@ -63,7 +63,8 @@ impl EngineApiMetrics {
     pub(crate) fn execute_metered<E, DB>(
         &self,
         executor: E,
-        transactions: impl Iterator<Item = Result<impl ExecutableTx<E>, BlockExecutionError>>,
+        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
@@ -75,31 +76,46 @@ impl EngineApiMetrics {
         // be accessible.
         let wrapper = MeteredStateHook { metrics: self.executor.clone(), inner_hook: state_hook };
 
-        let mut senders = Vec::new();
+        let mut senders = Vec::with_capacity(transaction_count);
         let mut executor = executor.with_state_hook(Some(Box::new(wrapper)));
 
         let f = || {
+            let start = Instant::now();
             debug_span!(target: "engine::tree", "pre execution")
                 .entered()
                 .in_scope(|| executor.apply_pre_execution_changes())?;
+            self.executor.pre_execution_histogram.record(start.elapsed());
+
             let exec_span = debug_span!(target: "engine::tree", "execution").entered();
-            for tx in transactions {
+            loop {
+                let start = Instant::now();
+                let Some(tx) = transactions.next() else { break };
+                self.executor.transaction_wait_histogram.record(start.elapsed());
+
                 let tx = tx?;
+                senders.push(*tx.signer());
+
                 let span =
                     debug_span!(target: "engine::tree", "execute tx", tx_hash=?tx.tx().tx_hash());
                 let enter = span.entered();
                 trace!(target: "engine::tree", "Executing transaction");
-                senders.push(*tx.signer());
+                let start = Instant::now();
                 let gas_used = executor.execute_transaction(tx)?;
+                self.executor.transaction_execution_histogram.record(start.elapsed());
 
                 // record the tx gas used
                 enter.record("gas_used", gas_used);
             }
             drop(exec_span);
-            debug_span!(target: "engine::tree", "finish")
+
+            let start = Instant::now();
+            let result = debug_span!(target: "engine::tree", "finish")
                 .entered()
                 .in_scope(|| executor.finish())
-                .map(|(evm, result)| (evm.into_db(), result))
+                .map(|(evm, result)| (evm.into_db(), result));
+            self.executor.post_execution_histogram.record(start.elapsed());
+
+            result
         };
 
         // Use metered to execute and track timing/gas metrics
@@ -514,6 +530,7 @@ mod tests {
         let _result = metrics.execute_metered::<_, EmptyDB>(
             executor,
             input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
+            input.transaction_count(),
             state_hook,
         );
 
@@ -570,6 +587,7 @@ 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,6 +39,7 @@ use revm::state::EvmState;
 use state::TreeState;
 use std::{
     fmt::Debug,
+    ops,
     sync::{
         mpsc::{Receiver, RecvError, RecvTimeoutError, Sender},
         Arc,
@@ -54,7 +55,7 @@ use tracing::*;
 mod block_buffer;
 mod cached_state;
 pub mod error;
-mod instrumented_state;
+pub mod instrumented_state;
 mod invalid_headers;
 mod metrics;
 mod payload_processor;
@@ -63,7 +64,6 @@ 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,9 +426,13 @@ where
             match self.try_recv_engine_message() {
                 Ok(Some(msg)) => {
                     debug!(target: "engine::tree", %msg, "received new engine message");
-                    if let Err(fatal) = self.on_engine_message(msg) {
-                        error!(target: "engine::tree", %fatal, "insert block fatal error");
-                        return
+                    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
+                        }
                     }
                 }
                 Ok(None) => {
@@ -821,7 +825,8 @@ where
         new_head_number: u64,
         current_head_number: u64,
     ) -> Vec<ExecutedBlock<N>> {
-        let mut old_blocks = Vec::new();
+        let mut old_blocks =
+            Vec::with_capacity((current_head_number.saturating_sub(new_head_number)) as usize);
 
         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) {
@@ -926,48 +931,6 @@ 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.
@@ -1302,22 +1265,7 @@ where
             // Check if persistence has complete
             match rx.try_recv() {
                 Ok(last_persisted_hash_num) => {
-                    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()?;
+                    self.on_persistence_complete(last_persisted_hash_num, start_time)?;
                 }
                 Err(TryRecvError::Closed) => return Err(TryRecvError::Closed.into()),
                 Err(TryRecvError::Empty) => {
@@ -1330,7 +1278,8 @@ 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()?;
+                let blocks_to_persist =
+                    self.get_canonical_blocks_to_persist(PersistTarget::Threshold)?;
                 self.persist_blocks(blocks_to_persist);
             }
         }
@@ -1338,11 +1287,72 @@ 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<(), InsertBlockFatalError> {
+    ) -> Result<ops::ControlFlow<()>, InsertBlockFatalError> {
         match msg {
             FromEngine::Event(event) => match event {
                 FromOrchestrator::BackfillSyncStarted => {
@@ -1352,6 +1362,13 @@ 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 {
@@ -1359,7 +1376,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(())
+                            return Ok(ops::ControlFlow::Continue(()))
                         }
 
                         debug!(target: "engine::tree", block=?block_num_hash, "inserting already executed block");
@@ -1467,7 +1484,7 @@ where
                 }
             }
         }
-        Ok(())
+        Ok(ops::ControlFlow::Continue(()))
     }
 
     /// Invoked if the backfill sync has finished to target.
@@ -1701,10 +1718,10 @@ where
     }
 
     /// Returns a batch of consecutive canonical blocks to persist in the range
-    /// `(last_persisted_number .. canonical_head - threshold]`. The expected
-    /// order is oldest -> newest.
+    /// `(last_persisted_number .. target]`. 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
@@ -1715,9 +1732,12 @@ where
         let last_persisted_number = self.persistence_state.last_persisted_block.number;
         let canonical_head_number = self.state.tree_state.canonical_block_number();
 
-        // Persist only up to block buffer target
-        let target_number =
-            canonical_head_number.saturating_sub(self.config.memory_block_buffer_target());
+        let target_number = match target {
+            PersistTarget::Head => canonical_head_number,
+            PersistTarget::Threshold => {
+                canonical_head_number.saturating_sub(self.config.memory_block_buffer_target())
+            }
+        };
 
         debug!(
             target: "engine::tree",
@@ -2507,14 +2527,11 @@ where
             Ok(Some(_)) => {}
         }
 
-        // 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,
-        };
+        // 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;
 
         let ctx = TreeCtx::new(&mut self.state, &self.canonical_in_memory_state);
 
@@ -2860,3 +2877,12 @@ 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

@@ -0,0 +1,541 @@
+//! 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,16 +3,17 @@
 use super::precompile_cache::PrecompileCacheMap;
 use crate::tree::{
     cached_state::{
-        CachedStateMetrics, ExecutionCache as StateExecutionCache, ExecutionCacheBuilder,
-        SavedCache,
+        CachedStateMetrics, CachedStateProvider, ExecutionCache as StateExecutionCache,
+        ExecutionCacheBuilder, SavedCache,
     },
     payload_processor::{
-        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmTaskEvent},
+        prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, 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;
@@ -21,14 +22,17 @@ use executor::WorkloadExecutor;
 use multiproof::{SparseTrieUpdate, *};
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
-use rayon::iter::{ParallelBridge, ParallelIterator};
-use reth_engine_primitives::ExecutableTxIterator;
+use rayon::prelude::*;
 use reth_evm::{
     execute::{ExecutableTxFor, WithTxEnv},
-    ConfigureEvm, EvmEnvFor, OnStateHook, SpecFor, TxEnvFor,
+    ConfigureEvm, EvmEnvFor, ExecutableTxIterator, ExecutableTxTuple, OnStateHook, SpecFor,
+    TxEnvFor,
 };
+use reth_execution_types::ExecutionOutcome;
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader};
+use reth_provider::{
+    BlockReader, DatabaseProviderROFactory, StateProvider, StateProviderFactory, StateReader,
+};
 use reth_revm::{db::BundleState, state::EvmState};
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
 use reth_trie_parallel::{
@@ -42,6 +46,7 @@ use reth_trie_sparse::{
 use reth_trie_sparse_parallel::{ParallelSparseTrie, ParallelismThresholds};
 use std::{
     collections::BTreeMap,
+    ops::Not,
     sync::{
         atomic::AtomicBool,
         mpsc::{self, channel},
@@ -51,6 +56,7 @@ use std::{
 };
 use tracing::{debug, debug_span, instrument, warn, Span};
 
+pub mod bal;
 mod configured_sparse_trie;
 pub mod executor;
 pub mod multiproof;
@@ -90,6 +96,13 @@ 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>
@@ -106,6 +119,8 @@ 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.
@@ -149,6 +164,7 @@ 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(),
@@ -195,7 +211,6 @@ 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",
@@ -209,7 +224,8 @@ where
         provider_builder: StateProviderBuilder<N, P>,
         multiproof_provider_factory: F,
         config: &TreeConfig,
-    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
+        bal: Option<Arc<BlockAccessList>>,
+    ) -> IteratorPayloadHandle<Evm, I, N>
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
         F: DatabaseProviderROFactory<Provider: TrieCursorFactory + HashedCursorFactory>
@@ -223,11 +239,36 @@ where
 
         let span = Span::current();
         let (to_sparse_trie, sparse_trie_rx) = channel();
+        let (to_multi_proof, from_multi_proof) = crossbeam_channel::unbounded();
 
-        // 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.
+        // 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,
+            )
+        };
 
         // Create and spawn the storage proof task
         let task_ctx = ProofTaskCtx::new(multiproof_provider_factory);
@@ -244,24 +285,28 @@ 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();
-            multi_proof_task.run();
+            // 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);
         });
 
         // wire the sparse trie to the state root response receiver
@@ -288,13 +333,14 @@ where
         env: ExecutionEnv<Evm>,
         transactions: I,
         provider_builder: StateProviderBuilder<N, P>,
-    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
+        bal: Option<Arc<BlockAccessList>>,
+    ) -> IteratorPayloadHandle<Evm, I, N>
     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);
+            self.spawn_caching_with(env, prewarm_rx, size_hint, provider_builder, None, bal);
         PayloadHandle {
             to_multi_proof: None,
             prewarm_handle,
@@ -315,36 +361,32 @@ where
         usize,
     ) {
         let (transactions, convert) = transactions.into();
-        let transactions = transactions.into_iter();
-        // Get the transaction count for prewarming task
-        // Use upper bound if available (more accurate), otherwise use lower bound
-        let (lower, upper) = transactions.size_hint();
-        let transaction_count_hint = upper.unwrap_or(lower);
+        let transactions = transactions.into_par_iter();
+        let transaction_count_hint = transactions.len();
 
-        // Spawn a task that iterates through all transactions in parallel and sends them to the
-        // main task.
-        let (tx, rx) = mpsc::channel();
+        let (ooo_tx, ooo_rx) = mpsc::channel();
+        let (prewarm_tx, prewarm_rx) = mpsc::channel();
+        let (execute_tx, execute_rx) = mpsc::channel();
+
+        // Spawn a task that `convert`s all transactions in parallel and sends them out-of-order.
         self.executor.spawn_blocking(move || {
-            transactions.enumerate().par_bridge().for_each_with(tx, |sender, (idx, tx)| {
+            transactions.enumerate().for_each_with(ooo_tx, |ooo_tx, (idx, tx)| {
                 let tx = convert(tx);
                 let tx = tx.map(|tx| WithTxEnv { tx_env: tx.to_tx_env(), tx: Arc::new(tx) });
-                let _ = sender.send((idx, tx));
+                // Only send Ok(_) variants to prewarming task.
+                if let Ok(tx) = &tx {
+                    let _ = prewarm_tx.send(tx.clone());
+                }
+                let _ = ooo_tx.send((idx, tx));
             });
         });
 
         // Spawn a task that processes out-of-order transactions from the task above and sends them
-        // to prewarming and execution tasks.
-        let (prewarm_tx, prewarm_rx) = mpsc::channel();
-        let (execute_tx, execute_rx) = mpsc::channel();
+        // to the execution task in order.
         self.executor.spawn_blocking(move || {
             let mut next_for_execution = 0;
             let mut queue = BTreeMap::new();
-            while let Ok((idx, tx)) = rx.recv() {
-                // only send Ok(_) variants to prewarming task
-                if let Ok(tx) = &tx {
-                    let _ = prewarm_tx.send(tx.clone());
-                }
-
+            while let Ok((idx, tx)) = ooo_rx.recv() {
                 if next_for_execution == idx {
                     let _ = execute_tx.send(tx);
                     next_for_execution += 1;
@@ -372,7 +414,8 @@ where
         transaction_count_hint: usize,
         provider_builder: StateProviderBuilder<N, P>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-    ) -> CacheTaskHandle
+        bal: Option<Arc<BlockAccessList>>,
+    ) -> CacheTaskHandle<N::Receipt>
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
@@ -382,14 +425,13 @@ where
             transactions = mpsc::channel().1;
         }
 
-        let saved_cache = self.cache_for(env.parent_hash);
-        let cache = saved_cache.cache().clone();
-        let cache_metrics = saved_cache.metrics().clone();
+        let saved_cache = self.disable_state_cache.not().then(|| self.cache_for(env.parent_hash));
+
         // configure prewarming
         let prewarm_ctx = PrewarmContext {
             env,
             evm_config: self.evm_config.clone(),
-            saved_cache,
+            saved_cache: saved_cache.clone(),
             provider: provider_builder,
             metrics: PrewarmMetrics::default(),
             terminate_execution: Arc::new(AtomicBool::new(false)),
@@ -410,11 +452,16 @@ where
         {
             let to_prewarm_task = to_prewarm_task.clone();
             self.executor.spawn_blocking(move || {
-                prewarm_task.run(transactions, to_prewarm_task);
+                let mode = if let Some(bal) = bal {
+                    PrewarmMode::BlockAccessList(bal)
+                } else {
+                    PrewarmMode::Transactions(transactions)
+                };
+                prewarm_task.run(mode, to_prewarm_task);
             });
         }
 
-        CacheTaskHandle { cache, to_prewarm_task: Some(to_prewarm_task), cache_metrics }
+        CacheTaskHandle { saved_cache, to_prewarm_task: Some(to_prewarm_task) }
     }
 
     /// Returns the cache for the given parent hash.
@@ -547,12 +594,15 @@ 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> {
+pub struct PayloadHandle<Tx, Err, R> {
     /// 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,
+    prewarm_handle: CacheTaskHandle<R>,
     /// Stream of block transactions
     transactions: mpsc::Receiver<Result<Tx, Err>>,
     /// Receiver for the state root
@@ -561,7 +611,7 @@ pub struct PayloadHandle<Tx, Err> {
     _span: Span,
 }
 
-impl<Tx, Err> PayloadHandle<Tx, Err> {
+impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
     /// Awaits the state root
     ///
     /// # Panics
@@ -590,19 +640,19 @@ impl<Tx, Err> PayloadHandle<Tx, Err> {
 
         move |source: StateChangeSource, state: &EvmState| {
             if let Some(sender) = &to_multi_proof {
-                let _ = sender.send(MultiProofMessage::StateUpdate(source, state.clone()));
+                let _ = sender.send(MultiProofMessage::StateUpdate(source.into(), state.clone()));
             }
         }
     }
 
     /// Returns a clone of the caches used by prewarming
-    pub(super) fn caches(&self) -> StateExecutionCache {
-        self.prewarm_handle.cache.clone()
+    pub(super) fn caches(&self) -> Option<StateExecutionCache> {
+        self.prewarm_handle.saved_cache.as_ref().map(|cache| cache.cache().clone())
     }
 
     /// Returns a clone of the cache metrics used by prewarming
-    pub(super) fn cache_metrics(&self) -> CachedStateMetrics {
-        self.prewarm_handle.cache_metrics.clone()
+    pub(super) fn cache_metrics(&self) -> Option<CachedStateMetrics> {
+        self.prewarm_handle.saved_cache.as_ref().map(|cache| cache.metrics().clone())
     }
 
     /// Terminates the pre-warming transaction processing.
@@ -614,9 +664,14 @@ impl<Tx, Err> PayloadHandle<Tx, Err> {
 
     /// Terminates the entire caching task.
     ///
-    /// 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)
+    /// 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)
     }
 
     /// Returns iterator yielding transactions from the stream.
@@ -628,17 +683,18 @@ impl<Tx, Err> PayloadHandle<Tx, Err> {
 }
 
 /// 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 {
+pub(crate) struct CacheTaskHandle<R> {
     /// The shared cache the task operates with.
-    cache: StateExecutionCache,
-    /// Metrics for the caches
-    cache_metrics: CachedStateMetrics,
+    saved_cache: Option<SavedCache>,
     /// Channel to the spawned prewarm task if any
-    to_prewarm_task: Option<std::sync::mpsc::Sender<PrewarmTaskEvent>>,
+    to_prewarm_task: Option<std::sync::mpsc::Sender<PrewarmTaskEvent<R>>>,
 }
 
-impl CacheTaskHandle {
+impl<R: Send + Sync + 'static> CacheTaskHandle<R> {
     /// Terminates the pre-warming transaction processing.
     ///
     /// Note: This does not terminate the task yet.
@@ -650,20 +706,25 @@ impl CacheTaskHandle {
 
     /// Terminates the entire pre-warming task.
     ///
-    /// If the [`BundleState`] is provided it will update the shared cache.
-    pub(super) fn terminate_caching(&mut self, block_output: Option<&BundleState>) {
+    /// 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 let Some(tx) = self.to_prewarm_task.take() {
-            // Only clone when we have an active task and a state to send
-            let event = PrewarmTaskEvent::Terminate { block_output: block_output.cloned() };
+            let event = PrewarmTaskEvent::Terminate { execution_outcome };
             let _ = tx.send(event);
         }
     }
 }
 
-impl Drop for CacheTaskHandle {
+impl<R> Drop for CacheTaskHandle<R> {
     fn drop(&mut self) {
-        // Ensure we always terminate on drop
-        self.terminate_caching(None);
+        // 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 });
+        }
     }
 }
 
@@ -716,6 +777,8 @@ 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()
     }
@@ -1048,19 +1111,17 @@ mod tests {
 
         let provider_factory = BlockchainProvider::new(factory).unwrap();
 
-        let mut handle =
-            payload_processor.spawn(
-                Default::default(),
-                (
-                    core::iter::empty::<
-                        Result<Recovered<TransactionSigned>, core::convert::Infallible>,
-                    >(),
-                    std::convert::identity,
-                ),
-                StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
-                OverlayStateProviderFactory::new(provider_factory),
-                &TreeConfig::default(),
-            );
+        let mut handle = payload_processor.spawn(
+            Default::default(),
+            (
+                Vec::<Result<Recovered<TransactionSigned>, core::convert::Infallible>>::new(),
+                std::convert::identity,
+            ),
+            StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
+            OverlayStateProviderFactory::new(provider_factory),
+            &TreeConfig::default(),
+            None, // No BAL for test
+        );
 
         let mut state_hook = handle.state_hook();
 

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

@@ -14,25 +14,31 @@
 use crate::tree::{
     cached_state::{CachedStateProvider, SavedCache},
     payload_processor::{
-        executor::WorkloadExecutor, multiproof::MultiProofMessage,
+        bal::{total_slots, BALSlotIter},
+        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::{BlockReader, StateProviderFactory, StateReader};
-use reth_revm::{database::StateProviderDatabase, db::BundleState, state::EvmState};
+use reth_provider::{AccountReader, BlockReader, StateProvider, StateProviderFactory, StateReader};
+use reth_revm::{database::StateProviderDatabase, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
+    ops::Range,
     sync::{
         atomic::{AtomicBool, Ordering},
         mpsc::{self, channel, Receiver, Sender},
@@ -42,6 +48,14 @@ 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> {
@@ -86,7 +100,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>,
+    actions_rx: Receiver<PrewarmTaskEvent<N::Receipt>>,
     /// Parent span for tracing
     parent_span: Span,
 }
@@ -105,7 +119,7 @@ where
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
         transaction_count_hint: usize,
         max_concurrency: usize,
-    ) -> (Self, Sender<PrewarmTaskEvent>) {
+    ) -> (Self, Sender<PrewarmTaskEvent<N::Receipt>>) {
         let (actions_tx, actions_rx) = channel();
 
         trace!(
@@ -135,8 +149,11 @@ 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>)
-    where
+    fn spawn_all<Tx>(
+        &self,
+        pending: mpsc::Receiver<Tx>,
+        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
+    ) where
         Tx: ExecutableTxFor<Evm> + Clone + Send + 'static,
     {
         let executor = self.executor.clone();
@@ -160,12 +177,7 @@ where
             };
 
             // Initialize worker handles container
-            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()));
-            }
+            let handles = ctx.clone().spawn_workers(workers_needed, &executor, actions_tx.clone(), done_tx.clone());
 
             // Distribute transactions to workers
             let mut tx_index = 0usize;
@@ -248,38 +260,123 @@ 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, state: BundleState) {
+    fn save_cache(self, execution_outcome: Arc<ExecutionOutcome<N::Receipt>>) {
         let start = Instant::now();
 
         let Self { execution_cache, ctx: PrewarmContext { env, metrics, saved_cache, .. }, .. } =
             self;
         let hash = env.hash;
 
-        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);
+        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);
 
-            // 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;
-            }
+                // 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;
+                }
 
-            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());
+            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 });
     }
 
     /// Executes the task.
@@ -293,15 +390,24 @@ where
         name = "prewarm and caching",
         skip_all
     )]
-    pub(super) fn run(
+    pub(super) fn run<Tx>(
         self,
-        pending: mpsc::Receiver<impl ExecutableTxFor<Evm> + Clone + Send + 'static>,
-        actions_tx: Sender<PrewarmTaskEvent>,
-    ) {
-        // spawn execution tasks.
-        self.spawn_all(pending, actions_tx);
+        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);
+            }
+        }
 
-        let mut final_block_output = None;
+        let mut final_execution_outcome = None;
         let mut finished_execution = false;
         while let Ok(event) = self.actions_rx.recv() {
             match event {
@@ -314,9 +420,9 @@ where
                     // completed executing a set of transactions
                     self.send_multi_proof_targets(proof_targets);
                 }
-                PrewarmTaskEvent::Terminate { block_output } => {
+                PrewarmTaskEvent::Terminate { execution_outcome } => {
                     trace!(target: "engine::tree::payload_processor::prewarm", "Received termination signal");
-                    final_block_output = Some(block_output);
+                    final_execution_outcome = Some(execution_outcome);
 
                     if finished_execution {
                         // all tasks are done, we can exit, which will save caches and exit
@@ -330,7 +436,7 @@ where
 
                     finished_execution = true;
 
-                    if final_block_output.is_some() {
+                    if final_execution_outcome.is_some() {
                         // all tasks are done, we can exit, which will save caches and exit
                         break
                     }
@@ -340,9 +446,9 @@ where
 
         debug!(target: "engine::tree::payload_processor::prewarm", "Completed prewarm execution");
 
-        // save caches and finish
-        if let Some(Some(state)) = final_block_output {
-            self.save_cache(state);
+        // save caches and finish using the shared ExecutionOutcome
+        if let Some(Some(execution_outcome)) = final_execution_outcome {
+            self.save_cache(execution_outcome);
         }
     }
 }
@@ -356,7 +462,7 @@ where
 {
     pub(super) env: ExecutionEnv<Evm>,
     pub(super) evm_config: Evm,
-    pub(super) saved_cache: SavedCache,
+    pub(super) saved_cache: Option<SavedCache>,
     /// Provider to obtain the state
     pub(super) provider: StateProviderBuilder<N, P>,
     pub(super) metrics: PrewarmMetrics,
@@ -384,10 +490,10 @@ where
             metrics,
             terminate_execution,
             precompile_cache_disabled,
-            mut precompile_cache_map,
+            precompile_cache_map,
         } = self;
 
-        let state_provider = match provider.build() {
+        let mut state_provider = match provider.build() {
             Ok(provider) => provider,
             Err(err) => {
                 trace!(
@@ -400,10 +506,15 @@ where
         };
 
         // Use the caches to create a new provider with caching
-        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);
+        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 state_provider = StateProviderDatabase::new(state_provider);
 
@@ -445,7 +556,7 @@ where
     fn transact_batch<Tx>(
         self,
         txs: mpsc::Receiver<IndexedTransaction<Tx>>,
-        sender: Sender<PrewarmTaskEvent>,
+        sender: Sender<PrewarmTaskEvent<N::Receipt>>,
         done_tx: Sender<()>,
     ) where
         Tx: ExecutableTxFor<Evm>,
@@ -508,7 +619,8 @@ where
                 let _enter =
                     debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm outcome", index, tx_hash=%tx.tx().tx_hash())
                         .entered();
-                let (targets, storage_targets) = multiproof_targets_from_state(res.state);
+                let targets = multiproof_targets_from_state(res.state);
+                let storage_targets = targets.storage_targets_count();
                 metrics.prefetch_storage_targets.record(storage_targets as f64);
                 let _ = sender.send(PrewarmTaskEvent::Outcome { proof_targets: Some(targets) });
                 drop(_enter);
@@ -522,74 +634,181 @@ where
     }
 
     /// Spawns a worker task for transaction execution and returns its sender channel.
-    fn spawn_worker<Tx>(
-        &self,
-        idx: usize,
-        executor: &WorkloadExecutor,
-        actions_tx: Sender<PrewarmTaskEvent>,
+    fn spawn_workers<Tx>(
+        self,
+        workers_needed: usize,
+        task_executor: &WorkloadExecutor,
+        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
         done_tx: Sender<()>,
-    ) -> mpsc::Sender<IndexedTransaction<Tx>>
+    ) -> Vec<mpsc::Sender<IndexedTransaction<Tx>>>
     where
         Tx: ExecutableTxFor<Evm> + Send + 'static,
     {
-        let (tx, rx) = mpsc::channel();
+        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(
+        &self,
+        idx: usize,
+        executor: &WorkloadExecutor,
+        bal: Arc<BlockAccessList>,
+        range: Range<usize>,
+        done_tx: Sender<()>,
+    ) {
         let ctx = self.clone();
-        let span =
-            debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm worker", idx);
+        let span = debug_span!(
+            target: "engine::tree::payload_processor::prewarm",
+            "bal prewarm worker",
+            idx,
+            range_start = range.start,
+            range_end = range.end
+        );
 
         executor.spawn_blocking(move || {
             let _enter = span.entered();
-            ctx.transact_batch(rx, actions_tx, done_tx);
+            ctx.prefetch_bal_slots(bal, range, done_tx);
         });
+    }
 
-        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());
     }
 }
 
 /// 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, 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
+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;
             }
 
-            storage_set.insert(keccak256(B256::new(key.to_be_bytes())));
-        }
+            let hashed_address = keccak256(address);
 
-        storage_targets += storage_set.len();
-        targets.insert(keccak256(addr), storage_set);
-    }
+            let storage_set: B256Set = account
+                .storage
+                .into_iter()
+                .filter(|(_, slot)| slot.is_changed())
+                .map(|(key, _)| keccak256(B256::new(key.to_be_bytes())))
+                .collect();
 
-    (targets, storage_targets)
+            Some((hashed_address, storage_set))
+        })
+        .collect()
 }
 
 /// The events the pre-warm task can handle.
-pub(super) enum PrewarmTaskEvent {
+///
+/// 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> {
     /// 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 block state output.
-        block_output: Option<BundleState>,
+        /// 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 outcome of a pre-warm task
     Outcome {
@@ -621,4 +840,6 @@ 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,8 +166,7 @@ where
 
     // Update storage slots with new values and calculate storage roots.
     let span = tracing::Span::current();
-    let (tx, rx) = mpsc::channel();
-    state
+    let results: Vec<_> = state
         .storages
         .into_iter()
         .map(|(address, storage)| (address, storage, trie.take_storage_trie(&address)))
@@ -217,13 +216,7 @@ where
 
             SparseStateTrieResult::Ok((address, storage_trie))
         })
-        .for_each_init(
-            || tx.clone(),
-            |tx, result| {
-                let _ = tx.send(result);
-            },
-        );
-    drop(tx);
+        .collect();
 
     // 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.
@@ -235,7 +228,7 @@ where
     let _enter =
         tracing::debug_span!(target: "engine::tree::payload_processor::sparse_trie", "account trie")
             .entered();
-    for result in rx {
+    for result in results {
         let (address, storage_trie) = result?;
         trie.insert_storage_trie(address, storage_trie);
 

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

@@ -11,9 +11,11 @@ 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::{
@@ -33,12 +35,13 @@ use reth_primitives_traits::{
 };
 use reth_provider::{
     providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockReader,
-    DatabaseProviderFactory, ExecutionOutcome, HashedPostStateProvider, ProviderError,
-    PruneCheckpointReader, StageCheckpointReader, StateProvider, StateProviderFactory, StateReader,
-    StateRootProvider, TrieReader,
+    DatabaseProviderFactory, DatabaseProviderROFactory, ExecutionOutcome, HashedPostStateProvider,
+    ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
+    StateProviderFactory, StateReader, TrieReader,
 };
 use reth_revm::db::State;
-use reth_trie::{updates::TrieUpdates, HashedPostState, TrieInputSorted};
+use reth_storage_errors::db::DatabaseError;
+use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot, TrieInputSorted};
 use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
 use revm_primitives::Address;
 use std::{
@@ -220,7 +223,7 @@ where
                     .map_err(NewPayloadError::other)?
                     .into();
 
-                let iter = Either::Left(iter.into_iter().map(Either::Left));
+                let iter = Either::Left(iter.into_par_iter().map(Either::Left));
                 let convert = move |tx| {
                     let Either::Left(tx) = tx else { unreachable!() };
                     convert(tx).map(Either::Left).map_err(Either::Left)
@@ -230,8 +233,9 @@ where
                 Ok((iter, Box::new(convert) as Box<dyn Fn(_) -> _ + Send + Sync + 'static>))
             }
             BlockOrPayload::Block(block) => {
-                let iter =
-                    Either::Right(block.body().clone_transactions().into_iter().map(Either::Right));
+                let iter = Either::Right(
+                    block.body().clone_transactions().into_par_iter().map(Either::Right),
+                );
                 let convert = move |tx: Either<_, N::SignedTx>| {
                     let Either::Right(tx) = tx else { unreachable!() };
                     tx.try_into_recovered().map(Either::Right).map_err(Either::Right)
@@ -368,10 +372,11 @@ where
             )
             .into())
         };
-        let state_provider = ensure_ok!(provider_builder.build());
+        let mut state_provider = ensure_ok!(provider_builder.build());
         drop(_enter);
 
-        // fetch parent block
+        // Fetch parent block. This goes to memory most of the time unless the parent block is
+        // beyond the in-memory buffer.
         let Some(parent_block) = ensure_ok!(self.sealed_header_by_hash(parent_hash, ctx.state()))
         else {
             return Err(InsertBlockError::new(
@@ -396,9 +401,17 @@ where
             "Decided which state root algorithm to run"
         );
 
-        // use prewarming background task
+        // Get an iterator over the transactions in the payload
         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(),
@@ -407,25 +420,22 @@ where
             parent_hash,
             ctx.state(),
             strategy,
+            block_access_list,
         ));
 
         // Use cached state provider before executing, used in execution after prewarming threads
         // complete
-        let state_provider = CachedStateProvider::new_with_caches(
-            state_provider,
-            handle.caches(),
-            handle.cache_metrics(),
-        );
+        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"));
+        }
 
         // Execute the block and handle any execution errors
-        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)
-        } {
+        let (output, senders) = match self.execute_block(state_provider, env, &input, &mut handle) {
             Ok(output) => output,
             Err(err) => return self.handle_execution_error(input, err, &parent_block),
         };
@@ -509,7 +519,7 @@ where
             }
 
             let (root, updates) = ensure_ok_post_block!(
-                state_provider.state_root_with_updates(hashed_state.clone()),
+                self.compute_state_root_serial(block.parent_hash(), &hashed_state, ctx.state()),
                 block
             );
             (root, updates, root_time.elapsed())
@@ -539,17 +549,14 @@ where
             .into())
         }
 
-        // terminate prewarming task with good state output
-        handle.terminate_caching(Some(&output.state));
+        // 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)));
 
-        Ok(self.spawn_deferred_trie_task(
-            block,
-            output,
-            block_num_hash.number,
-            &ctx,
-            hashed_state,
-            trie_output,
-        ))
+        // 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))
     }
 
     /// Return sealed block header from database or in-memory state by hash.
@@ -592,10 +599,10 @@ where
         state_provider: S,
         env: ExecutionEnv<Evm>,
         input: &BlockOrPayload<T>,
-        handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err>,
+        handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
     ) -> Result<(BlockExecutionOutput<N::Receipt>, Vec<Address>), InsertBlockErrorKind>
     where
-        S: StateProvider,
+        S: StateProvider + Send,
         Err: core::error::Error + Send + Sync + 'static,
         V: PayloadValidator<T, Block = N::Block>,
         T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
@@ -604,7 +611,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();
@@ -636,6 +643,7 @@ 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();
@@ -650,8 +658,6 @@ 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,
@@ -681,6 +687,36 @@ 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:
@@ -775,10 +811,12 @@ 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,
     > {
@@ -804,12 +842,14 @@ where
                     .record(trie_input_start.elapsed().as_secs_f64());
 
                 let spawn_start = Instant::now();
+
                 let handle = self.payload_processor.spawn(
                     env,
                     txs,
                     provider_builder,
                     multiproof_provider_factory,
                     &self.config,
+                    block_access_list,
                 );
 
                 // record prewarming initialization duration
@@ -822,8 +862,12 @@ where
             }
             StateRootStrategy::Parallel | StateRootStrategy::Synchronous => {
                 let start = Instant::now();
-                let handle =
-                    self.payload_processor.spawn_cache_exclusive(env, txs, provider_builder);
+                let handle = self.payload_processor.spawn_cache_exclusive(
+                    env,
+                    txs,
+                    provider_builder,
+                    block_access_list,
+                );
 
                 // Record prewarming initialization duration
                 self.metrics
@@ -872,7 +916,7 @@ where
     /// Note: Use state root task only if prefix sets are empty, otherwise proof generation is
     /// too expensive because it requires walking all paths in every proof.
     const fn plan_state_root_computation(&self) -> StateRootStrategy {
-        if self.config.state_root_fallback() || !self.config.has_enough_parallelism() {
+        if self.config.state_root_fallback() {
             StateRootStrategy::Synchronous
         } else if self.config.use_state_root_task() {
             StateRootStrategy::StateRootTask
@@ -1011,8 +1055,7 @@ where
     fn spawn_deferred_trie_task(
         &self,
         block: RecoveredBlock<N::Block>,
-        output: BlockExecutionOutput<N::Receipt>,
-        block_number: u64,
+        execution_outcome: Arc<ExecutionOutcome<N::Receipt>>,
         ctx: &TreeCtx<'_, N>,
         hashed_state: HashedPostState,
         trie_output: TrieUpdates,
@@ -1062,7 +1105,7 @@ where
 
         ExecutedBlock::with_deferred_trie_data(
             Arc::new(block),
-            Arc::new(ExecutionOutcome::from((output, block_number))),
+            execution_outcome,
             deferred_trie_data,
         )
     }
@@ -1243,4 +1286,21 @@ 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,50 +1,58 @@
 //! Contains a precompile cache backed by `schnellru::LruMap` (LRU by length).
 
 use alloy_primitives::Bytes;
-use parking_lot::Mutex;
+use dashmap::DashMap;
+use moka::policy::EvictionPolicy;
 use reth_evm::precompiles::{DynPrecompile, Precompile, PrecompileInput};
 use revm::precompile::{PrecompileId, PrecompileOutput, PrecompileResult};
 use revm_primitives::Address;
-use schnellru::LruMap;
-use std::{
-    collections::HashMap,
-    hash::{Hash, Hasher},
-    sync::Arc,
-};
+use std::{hash::Hash, 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>(HashMap<Address, PrecompileCache<S>>)
+pub struct PrecompileCacheMap<S>(Arc<DashMap<Address, PrecompileCache<S>>>)
 where
-    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone;
+    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static;
 
 impl<S> PrecompileCacheMap<S>
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
-    pub(crate) fn cache_for_address(&mut self, address: Address) -> PrecompileCache<S> {
+    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.
         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>(Arc<Mutex<LruMap<CacheKey<S>, CacheEntry>>>)
+pub struct PrecompileCache<S>(
+    moka::sync::Cache<Bytes, CacheEntry<S>, alloy_primitives::map::DefaultHashBuilder>,
+)
 where
-    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone;
+    S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static;
 
 impl<S> Default for PrecompileCache<S>
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
     fn default() -> Self {
-        Self(Arc::new(Mutex::new(LruMap::new(schnellru::ByLength::new(MAX_CACHE_SIZE)))))
+        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()),
+        )
     }
 }
 
@@ -52,63 +60,31 @@ impl<S> PrecompileCache<S>
 where
     S: Eq + Hash + std::fmt::Debug + Send + Sync + Clone + 'static,
 {
-    fn get(&self, key: &CacheKeyRef<'_, S>) -> Option<CacheEntry> {
-        self.0.lock().get(key).cloned()
+    fn get(&self, input: &[u8], spec: S) -> Option<CacheEntry<S>> {
+        self.0.get(input).filter(|e| e.spec == spec)
     }
 
     /// Inserts the given key and value into the cache, returning the new cache size.
-    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);
+    fn insert(&self, input: Bytes, value: CacheEntry<S>) -> usize {
+        self.0.insert(input, value);
+        self.0.entry_count() as usize
     }
 }
 
 /// Cache entry, precompile successful output.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CacheEntry(PrecompileOutput);
+pub struct CacheEntry<S> {
+    output: PrecompileOutput,
+    spec: S,
+}
 
-impl CacheEntry {
+impl<S> CacheEntry<S> {
     const fn gas_used(&self) -> u64 {
-        self.0.gas_used
+        self.output.gas_used
     }
 
     fn to_precompile_result(&self) -> PrecompileResult {
-        Ok(self.0.clone())
+        Ok(self.output.clone())
     }
 }
 
@@ -190,9 +166,7 @@ where
     }
 
     fn call(&self, input: PrecompileInput<'_>) -> PrecompileResult {
-        let key = CacheKeyRef::new(self.spec_id.clone(), input.data);
-
-        if let Some(entry) = &self.cache.get(&key) {
+        if let Some(entry) = &self.cache.get(input.data, self.spec_id.clone()) {
             self.increment_by_one_precompile_cache_hits();
             if input.gas >= entry.gas_used() {
                 return entry.to_precompile_result()
@@ -204,8 +178,10 @@ where
 
         match &result {
             Ok(output) => {
-                let key = CacheKey::new(self.spec_id.clone(), Bytes::copy_from_slice(calldata));
-                let size = self.cache.insert(key, CacheEntry(output.clone()));
+                let size = self.cache.insert(
+                    Bytes::copy_from_slice(calldata),
+                    CacheEntry { output: output.clone(), spec: self.spec_id.clone() },
+                );
                 self.set_precompile_cache_size_metric(size as f64);
                 self.increment_by_one_precompile_cache_misses();
             }
@@ -246,31 +222,12 @@ 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 {
@@ -293,12 +250,11 @@ mod tests {
             reverted: false,
         };
 
-        let key = CacheKey::new(SpecId::PRAGUE, b"test_input".into());
-        let expected = CacheEntry(output);
-        cache.cache.insert(key, expected.clone());
+        let input = b"test_input";
+        let expected = CacheEntry { output, spec: SpecId::PRAGUE };
+        cache.cache.insert(input.into(), expected.clone());
 
-        let key = CacheKeyRef::new(SpecId::PRAGUE, b"test_input");
-        let actual = cache.cache.get(&key).unwrap();
+        let actual = cache.cache.get(input, SpecId::PRAGUE).unwrap();
 
         assert_eq!(actual, expected);
     }
@@ -312,7 +268,7 @@ mod tests {
         let address1 = Address::repeat_byte(1);
         let address2 = Address::repeat_byte(2);
 
-        let mut cache_map = PrecompileCacheMap::default();
+        let 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,
-        TreeConfig,
+        PersistTarget, TreeConfig,
     },
 };
 
@@ -285,7 +285,8 @@ impl TestHarness {
         let fcu_state = self.fcu_state(block_hash);
 
         let (tx, rx) = oneshot::channel();
-        self.tree
+        let _ = self
+            .tree
             .on_engine_message(FromEngine::Request(
                 BeaconEngineMessage::ForkchoiceUpdated {
                     state: fcu_state,
@@ -498,7 +499,7 @@ fn test_tree_persist_block_batch() {
 
     // process the message
     let msg = test_harness.tree.try_recv_engine_message().unwrap().unwrap();
-    test_harness.tree.on_engine_message(msg).unwrap();
+    let _ = 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();
@@ -577,7 +578,7 @@ async fn test_engine_request_during_backfill() {
         .with_backfill_state(BackfillSyncState::Active);
 
     let (tx, rx) = oneshot::channel();
-    test_harness
+    let _ = test_harness
         .tree
         .on_engine_message(FromEngine::Request(
             BeaconEngineMessage::ForkchoiceUpdated {
@@ -658,7 +659,7 @@ async fn test_holesky_payload() {
         TestHarness::new(HOLESKY.clone()).with_backfill_state(BackfillSyncState::Active);
 
     let (tx, rx) = oneshot::channel();
-    test_harness
+    let _ = test_harness
         .tree
         .on_engine_message(FromEngine::Request(
             BeaconEngineMessage::NewPayload {
@@ -883,7 +884,8 @@ 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().unwrap();
+    let blocks_to_persist =
+        test_harness.tree.get_canonical_blocks_to_persist(PersistTarget::Threshold).unwrap();
 
     let expected_blocks_to_persist_length: usize =
         (canonical_head_number - memory_block_buffer_target - last_persisted_block_number)
@@ -902,7 +904,8 @@ 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().unwrap();
+    let blocks_to_persist =
+        test_harness.tree.get_canonical_blocks_to_persist(PersistTarget::Threshold).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
@@ -981,7 +984,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());
-    test_harness.tree.on_engine_message(FromEngine::Event(backfill_finished)).unwrap();
+    let _ = test_harness.tree.on_engine_message(FromEngine::Event(backfill_finished)).unwrap();
 
     let event = test_harness.from_tree_rx.recv().await.unwrap();
     match event {
@@ -991,7 +994,7 @@ async fn test_engine_tree_live_sync_transition_required_blocks_requested() {
         _ => panic!("Unexpected event: {event:#?}"),
     }
 
-    test_harness
+    let _ = test_harness
         .tree
         .on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain
             .last()
@@ -1047,7 +1050,7 @@ async fn test_fcu_with_canonical_ancestor_updates_latest_block() {
 
     // Send FCU to the canonical ancestor
     let (tx, rx) = oneshot::channel();
-    test_harness
+    let _ = test_harness
         .tree
         .on_engine_message(FromEngine::Request(
             BeaconEngineMessage::ForkchoiceUpdated {
@@ -1943,4 +1946,53 @@ 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
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-engine-tree.workspace = true
 reth-evm.workspace = true
 reth-revm.workspace = true

crates/era-utils/src/export.rs

@@ -150,6 +150,12 @@ where
         let era1_id = Era1Id::new(&config.network, start_block, block_count as u32)
             .with_hash(historical_root);
 
+        let era1_id = if config.max_blocks_per_file == MAX_BLOCKS_PER_ERA1 as u64 {
+            era1_id
+        } else {
+            era1_id.with_era_count()
+        };
+
         debug!("Final file name {}", era1_id.to_file_name());
         let file_path = config.dir.join(era1_id.to_file_name());
         let file = std::fs::File::create(&file_path)?;

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,18 +170,14 @@ where
     <P as NodePrimitivesProvider>::Primitives: NodePrimitives<BlockHeader = BH, BlockBody = BB>,
 {
     let reader = open(meta)?;
-    let iter =
-        reader
-            .iter()
-            .map(Box::new(decode)
-                as Box<dyn Fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>);
+    let iter = reader.iter().map(decode as fn(_) -> _);
     let iter = ProcessIter { iter, era: meta };
 
     process_iter(iter, writer, provider, hash_collector, block_numbers)
 }
 
 type ProcessInnerIter<R, BH, BB> =
-    Map<BlockTupleIterator<R>, Box<dyn Fn(Result<BlockTuple, E2sError>) -> eyre::Result<(BH, BB)>>>;
+    Map<BlockTupleIterator<R>, 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.
@@ -252,7 +248,7 @@ where
 
 /// Extracts block headers and bodies from `iter` and appends them using `writer` and `provider`.
 ///
-/// Adds on to `total_difficulty` and collects hash to height using `hash_collector`.
+/// Collects hash to height using `hash_collector`.
 ///
 /// Skips all blocks below the [`start_bound`] of `block_numbers` and stops when reaching past the
 /// [`end_bound`] or the end of the file.
@@ -309,7 +305,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-00001-"),
+        file_name.starts_with("mainnet-00000-"),
         "File should have correct prefix with era format"
     );
     assert!(file_name.ends_with(".era1"), "File should have correct extension");

crates/era/src/common/file_ops.rs

@@ -30,8 +30,11 @@ pub trait EraFileFormat: Sized {
 
 /// Era file identifiers
 pub trait EraFileId: Clone {
-    /// Convert to standardized file name
-    fn to_file_name(&self) -> String;
+    /// File type for this identifier
+    const FILE_TYPE: EraFileType;
+
+    /// Number of items, slots for `era`, blocks for `era1`, per era
+    const ITEMS_PER_ERA: u64;
 
     /// Get the network name
     fn network_name(&self) -> &str;
@@ -41,6 +44,43 @@ pub trait EraFileId: Clone {
 
     /// Get the count of items
     fn count(&self) -> u32;
+
+    /// Get the optional hash identifier
+    fn hash(&self) -> Option<[u8; 4]>;
+
+    /// Whether to include era count in filename
+    fn include_era_count(&self) -> bool;
+
+    /// Calculate era number
+    fn era_number(&self) -> u64 {
+        self.start_number() / Self::ITEMS_PER_ERA
+    }
+
+    /// Calculate the number of eras spanned per file.
+    ///
+    /// If the user can decide how many slots/blocks per era file there are, we need to calculate
+    /// it. Most of the time it should be 1, but it can never be more than 2 eras per file
+    /// as there is a maximum of 8192 slots/blocks per era file.
+    fn era_count(&self) -> u64 {
+        if self.count() == 0 {
+            return 0;
+        }
+        let first_era = self.era_number();
+        let last_number = self.start_number() + self.count() as u64 - 1;
+        let last_era = last_number / Self::ITEMS_PER_ERA;
+        last_era - first_era + 1
+    }
+
+    /// Convert to standardized file name.
+    fn to_file_name(&self) -> String {
+        Self::FILE_TYPE.format_filename(
+            self.network_name(),
+            self.era_number(),
+            self.hash(),
+            self.include_era_count(),
+            self.era_count(),
+        )
+    }
 }
 
 /// [`StreamReader`] for reading era-format files
@@ -154,6 +194,37 @@ impl EraFileType {
         }
     }
 
+    /// Generate era file name.
+    ///
+    /// Standard format: `<config-name>-<era-number>-<short-historical-root>.<ext>`
+    /// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
+    ///
+    /// With era count (for custom exports):
+    /// `<config-name>-<era-number>-<era-count>-<short-historical-root>.<ext>`
+    pub fn format_filename(
+        &self,
+        network_name: &str,
+        era_number: u64,
+        hash: Option<[u8; 4]>,
+        include_era_count: bool,
+        era_count: u64,
+    ) -> String {
+        let hash = format_hash(hash);
+
+        if include_era_count {
+            format!(
+                "{}-{:05}-{:05}-{}{}",
+                network_name,
+                era_number,
+                era_count,
+                hash,
+                self.extension()
+            )
+        } else {
+            format!("{}-{:05}-{}{}", network_name, era_number, hash, self.extension())
+        }
+    }
+
     /// Detect file type from URL
     /// By default, it assumes `Era` type
     pub fn from_url(url: &str) -> Self {
@@ -164,3 +235,11 @@ impl EraFileType {
         }
     }
 }
+
+/// Format hash as hex string, or placeholder if none
+pub fn format_hash(hash: Option<[u8; 4]>) -> String {
+    match hash {
+        Some(h) => format!("{:02x}{:02x}{:02x}{:02x}", h[0], h[1], h[2], h[3]),
+        None => "00000000".to_string(),
+    }
+}

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

@@ -59,11 +59,36 @@
 //!     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];
 
@@ -104,13 +129,11 @@ impl CompressedSignedBeaconBlock {
 
     /// Decompress to get the original ssz-encoded signed beacon block
     pub fn decompress(&self) -> Result<Vec<u8>, E2sError> {
-        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)
+        decompress_snappy_bounded(
+            self.data.as_slice(),
+            MAX_DECOMPRESSED_SIGNED_BEACON_BLOCK_BYTES,
+            "signed beacon block",
+        )
     }
 
     /// Convert to an [`Entry`]
@@ -168,13 +191,11 @@ impl CompressedBeaconState {
 
     /// Decompress to get the original ssz-encoded beacon state
     pub fn decompress(&self) -> Result<Vec<u8>, E2sError> {
-        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)
+        decompress_snappy_bounded(
+            self.data.as_slice(),
+            MAX_DECOMPRESSED_BEACON_STATE_BYTES,
+            "beacon state",
+        )
     }
 
     /// Convert to an [`Entry`]
@@ -260,4 +281,15 @@ 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,
+    common::file_ops::{EraFileId, EraFileType},
     e2s::types::{Entry, IndexEntry, SLOT_INDEX},
     era::types::consensus::{CompressedBeaconState, CompressedSignedBeaconBlock},
 };
@@ -163,12 +163,22 @@ pub struct EraId {
     /// Optional hash identifier for this file
     /// First 4 bytes of the last historical root in the last state in the era file
     pub hash: Option<[u8; 4]>,
+
+    /// Whether to include era count in filename
+    /// It is used for custom exports when we don't use the max number of items per file
+    include_era_count: bool,
 }
 
 impl EraId {
     /// Create a new [`EraId`]
     pub fn new(network_name: impl Into<String>, start_slot: u64, slot_count: u32) -> Self {
-        Self { network_name: network_name.into(), start_slot, slot_count, hash: None }
+        Self {
+            network_name: network_name.into(),
+            start_slot,
+            slot_count,
+            hash: None,
+            include_era_count: false,
+        }
     }
 
     /// Add a hash identifier to  [`EraId`]
@@ -177,32 +187,18 @@ impl EraId {
         self
     }
 
-    /// Calculate which era number the file starts at
-    pub const fn era_number(&self) -> u64 {
-        self.start_slot / SLOTS_PER_HISTORICAL_ROOT
-    }
-
-    // Helper function to calculate the number of eras per era1 file,
-    // If the user can decide how many blocks per era1 file there are, we need to calculate it.
-    // Most of the time it should be 1, but it can never be more than 2 eras per file
-    // as there is a maximum of 8192 blocks per era1 file.
-    const fn calculate_era_count(&self) -> u64 {
-        if self.slot_count == 0 {
-            return 0;
-        }
-
-        let first_era = self.era_number();
-
-        // Calculate the actual last slot number in the range
-        let last_slot = self.start_slot + self.slot_count as u64 - 1;
-        // Find which era the last block belongs to
-        let last_era = last_slot / SLOTS_PER_HISTORICAL_ROOT;
-        // Count how many eras we span
-        last_era - first_era + 1
+    /// Include era count in filename, for custom slot-per-file exports
+    pub const fn with_era_count(mut self) -> Self {
+        self.include_era_count = true;
+        self
     }
 }
 
 impl EraFileId for EraId {
+    const FILE_TYPE: EraFileType = EraFileType::Era;
+
+    const ITEMS_PER_ERA: u64 = SLOTS_PER_HISTORICAL_ROOT;
+
     fn network_name(&self) -> &str {
         &self.network_name
     }
@@ -214,24 +210,13 @@ impl EraFileId for EraId {
     fn count(&self) -> u32 {
         self.slot_count
     }
-    /// Convert to file name following the era file naming:
-    /// `<config-name>-<era-number>-<era-count>-<short-historical-root>.era`
-    /// <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
-    /// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md>
-    fn to_file_name(&self) -> String {
-        let era_number = self.era_number();
-        let era_count = self.calculate_era_count();
 
-        if let Some(hash) = self.hash {
-            format!(
-                "{}-{:05}-{:05}-{:02x}{:02x}{:02x}{:02x}.era",
-                self.network_name, era_number, era_count, hash[0], hash[1], hash[2], hash[3]
-            )
-        } else {
-            // era spec format with placeholder hash when no hash available
-            // Format: `<config-name>-<era-number>-<era-count>-00000000.era`
-            format!("{}-{:05}-{:05}-00000000.era", self.network_name, era_number, era_count)
-        }
+    fn hash(&self) -> Option<[u8; 4]> {
+        self.hash
+    }
+
+    fn include_era_count(&self) -> bool {
+        self.include_era_count
     }
 }
 
@@ -399,4 +384,40 @@ mod tests {
         let parsed_offset = index.offsets[0];
         assert_eq!(parsed_offset, -1024);
     }
+
+    #[test_case::test_case(
+        EraId::new("mainnet", 0, 8192).with_hash([0x4b, 0x36, 0x3d, 0xb9]),
+        "mainnet-00000-4b363db9.era";
+        "Mainnet era 0"
+    )]
+    #[test_case::test_case(
+        EraId::new("mainnet", 8192, 8192).with_hash([0x40, 0xcf, 0x2f, 0x3c]),
+        "mainnet-00001-40cf2f3c.era";
+        "Mainnet era 1"
+    )]
+    #[test_case::test_case(
+        EraId::new("mainnet", 0, 8192),
+        "mainnet-00000-00000000.era";
+        "Without hash"
+    )]
+    fn test_era_id_file_naming(id: EraId, expected_file_name: &str) {
+        let actual_file_name = id.to_file_name();
+        assert_eq!(actual_file_name, expected_file_name);
+    }
+
+    // File naming with era-count, for custom exports
+    #[test_case::test_case(
+        EraId::new("mainnet", 0, 8192).with_hash([0x4b, 0x36, 0x3d, 0xb9]).with_era_count(),
+        "mainnet-00000-00001-4b363db9.era";
+        "Mainnet era 0 with count"
+    )]
+    #[test_case::test_case(
+        EraId::new("mainnet", 8000, 500).with_hash([0xab, 0xcd, 0xef, 0x12]).with_era_count(),
+        "mainnet-00000-00002-abcdef12.era";
+        "Spanning two eras with count"
+    )]
+    fn test_era_id_file_naming_with_era_count(id: EraId, expected_file_name: &str) {
+        let actual_file_name = id.to_file_name();
+        assert_eq!(actual_file_name, expected_file_name);
+    }
 }

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 header: {e}"))
+                E2sError::SnappyCompression(format!("Failed to compress body: {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 header: {e}"))
+                E2sError::SnappyCompression(format!("Failed to compress receipts: {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,
+    common::file_ops::{EraFileId, EraFileType},
     e2s::types::{Entry, IndexEntry},
     era1::types::execution::{Accumulator, BlockTuple, MAX_BLOCKS_PER_ERA1},
 };
@@ -105,6 +105,10 @@ pub struct Era1Id {
     /// Optional hash identifier for this file
     /// First 4 bytes of the last historical root in the last state in the era file
     pub hash: Option<[u8; 4]>,
+
+    /// Whether to include era count in filename
+    /// It is used for custom exports when we don't use the max number of items per file
+    pub include_era_count: bool,
 }
 
 impl Era1Id {
@@ -114,7 +118,13 @@ impl Era1Id {
         start_block: BlockNumber,
         block_count: u32,
     ) -> Self {
-        Self { network_name: network_name.into(), start_block, block_count, hash: None }
+        Self {
+            network_name: network_name.into(),
+            start_block,
+            block_count,
+            hash: None,
+            include_era_count: false,
+        }
     }
 
     /// Add a hash identifier to  [`Era1Id`]
@@ -123,21 +133,17 @@ impl Era1Id {
         self
     }
 
-    // Helper function to calculate the number of eras per era1 file,
-    // If the user can decide how many blocks per era1 file there are, we need to calculate it.
-    // Most of the time it should be 1, but it can never be more than 2 eras per file
-    // as there is a maximum of 8192 blocks per era1 file.
-    const fn calculate_era_count(&self, first_era: u64) -> u64 {
-        // Calculate the actual last block number in the range
-        let last_block = self.start_block + self.block_count as u64 - 1;
-        // Find which era the last block belongs to
-        let last_era = last_block / MAX_BLOCKS_PER_ERA1 as u64;
-        // Count how many eras we span
-        last_era - first_era + 1
+    /// Include era count in filename, for custom block-per-file exports
+    pub const fn with_era_count(mut self) -> Self {
+        self.include_era_count = true;
+        self
     }
 }
 
 impl EraFileId for Era1Id {
+    const FILE_TYPE: EraFileType = EraFileType::Era1;
+
+    const ITEMS_PER_ERA: u64 = MAX_BLOCKS_PER_ERA1 as u64;
     fn network_name(&self) -> &str {
         &self.network_name
     }
@@ -149,24 +155,13 @@ impl EraFileId for Era1Id {
     fn count(&self) -> u32 {
         self.block_count
     }
-    /// Convert to file name following the era file naming:
-    /// `<config-name>-<era-number>-<era-count>-<short-historical-root>.era(1)`
-    /// <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md#file-name>
-    /// See also <https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md>
-    fn to_file_name(&self) -> String {
-        // Find which era the first block belongs to
-        let era_number = self.start_block / MAX_BLOCKS_PER_ERA1 as u64;
-        let era_count = self.calculate_era_count(era_number);
-        if let Some(hash) = self.hash {
-            format!(
-                "{}-{:05}-{:05}-{:02x}{:02x}{:02x}{:02x}.era1",
-                self.network_name, era_number, era_count, hash[0], hash[1], hash[2], hash[3]
-            )
-        } else {
-            // era spec format with placeholder hash when no hash available
-            // Format: `<config-name>-<era-number>-<era-count>-00000000.era1`
-            format!("{}-{:05}-{:05}-00000000.era1", self.network_name, era_number, era_count)
-        }
+
+    fn hash(&self) -> Option<[u8; 4]> {
+        self.hash
+    }
+
+    fn include_era_count(&self) -> bool {
+        self.include_era_count
     }
 }
 
@@ -314,35 +309,51 @@ mod tests {
 
     #[test_case::test_case(
         Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]),
-        "mainnet-00000-00001-5ec1ffb8.era1";
+        "mainnet-00000-5ec1ffb8.era1";
         "Mainnet era 0"
     )]
     #[test_case::test_case(
         Era1Id::new("mainnet", 8192, 8192).with_hash([0x5e, 0xcb, 0x9b, 0xf9]),
-        "mainnet-00001-00001-5ecb9bf9.era1";
+        "mainnet-00001-5ecb9bf9.era1";
         "Mainnet era 1"
     )]
     #[test_case::test_case(
         Era1Id::new("sepolia", 0, 8192).with_hash([0x90, 0x91, 0x84, 0x72]),
-        "sepolia-00000-00001-90918472.era1";
+        "sepolia-00000-90918472.era1";
         "Sepolia era 0"
     )]
     #[test_case::test_case(
         Era1Id::new("sepolia", 155648, 8192).with_hash([0xfa, 0x77, 0x00, 0x19]),
-        "sepolia-00019-00001-fa770019.era1";
+        "sepolia-00019-fa770019.era1";
         "Sepolia era 19"
     )]
     #[test_case::test_case(
         Era1Id::new("mainnet", 1000, 100),
-        "mainnet-00000-00001-00000000.era1";
+        "mainnet-00000-00000000.era1";
         "ID without hash"
     )]
     #[test_case::test_case(
         Era1Id::new("sepolia", 101130240, 8192).with_hash([0xab, 0xcd, 0xef, 0x12]),
-        "sepolia-12345-00001-abcdef12.era1";
+        "sepolia-12345-abcdef12.era1";
         "Large block number era 12345"
     )]
-    fn test_era1id_file_naming(id: Era1Id, expected_file_name: &str) {
+    fn test_era1_id_file_naming(id: Era1Id, expected_file_name: &str) {
+        let actual_file_name = id.to_file_name();
+        assert_eq!(actual_file_name, expected_file_name);
+    }
+
+    // File naming with era-count, for custom exports
+    #[test_case::test_case(
+        Era1Id::new("mainnet", 0, 8192).with_hash([0x5e, 0xc1, 0xff, 0xb8]).with_era_count(),
+        "mainnet-00000-00001-5ec1ffb8.era1";
+        "Mainnet era 0 with count"
+    )]
+    #[test_case::test_case(
+        Era1Id::new("mainnet", 8000, 500).with_hash([0xab, 0xcd, 0xef, 0x12]).with_era_count(),
+        "mainnet-00000-00002-abcdef12.era1";
+        "Spanning two eras with count"
+    )]
+    fn test_era1_id_file_naming_with_era_count(id: Era1Id, expected_file_name: &str) {
         let actual_file_name = id.to_file_name();
         assert_eq!(actual_file_name, expected_file_name);
     }

crates/ethereum/cli/src/app.rs

@@ -1,4 +1,8 @@
-use crate::{interface::Commands, Cli};
+use crate::{
+    interface::{Commands, NoSubCmd},
+    Cli,
+};
+use clap::Subcommand;
 use eyre::{eyre, Result};
 use reth_chainspec::{ChainSpec, EthChainSpec, Hardforks};
 use reth_cli::chainspec::ChainSpecParser;
@@ -18,20 +22,26 @@ 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> {
-    cli: Cli<Spec, Ext, Rpc>,
+pub struct CliApp<
+    Spec: ChainSpecParser,
+    Ext: clap::Args + fmt::Debug,
+    Rpc: RpcModuleValidator,
+    SubCmd: Subcommand + fmt::Debug = NoSubCmd,
+> {
+    cli: Cli<Spec, Ext, Rpc, SubCmd>,
     runner: Option<CliRunner>,
     layers: Option<Layers>,
     guard: Option<FileWorkerGuard>,
 }
 
-impl<C, Ext, Rpc> CliApp<C, Ext, Rpc>
+impl<C, Ext, Rpc, SubCmd> CliApp<C, Ext, Rpc, SubCmd>
 where
     C: ChainSpecParser,
     Ext: clap::Args + fmt::Debug,
     Rpc: RpcModuleValidator,
+    SubCmd: ExtendedCommand + Subcommand + fmt::Debug,
 {
-    pub(crate) fn new(cli: Cli<C, Ext, Rpc>) -> Self {
+    pub(crate) fn new(cli: Cli<C, Ext, Rpc, SubCmd>) -> Self {
         Self { cli, runner: None, layers: Some(Layers::new()), guard: None }
     }
 
@@ -98,9 +108,9 @@ where
         self.init_tracing(&runner)?;
 
         // Install the prometheus recorder to be sure to record all metrics
-        let _ = install_prometheus_recorder();
+        install_prometheus_recorder();
 
-        run_commands_with::<C, Ext, Rpc, N>(self.cli, runner, components, launcher)
+        run_commands_with::<C, Ext, Rpc, N, SubCmd>(self.cli, runner, components, launcher)
     }
 
     /// Initializes tracing with the configured options.
@@ -117,8 +127,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>(
-    cli: Cli<C, Ext, Rpc>,
+pub(crate) fn run_commands_with<C, Ext, Rpc, N, SubCmd>(
+    cli: Cli<C, Ext, Rpc, SubCmd>,
     runner: CliRunner,
     components: impl CliComponentsBuilder<N>,
     launcher: impl AsyncFnOnce(
@@ -131,6 +141,7 @@ 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) => {
@@ -154,7 +165,9 @@ where
         Commands::ImportEra(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
         Commands::ExportEra(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
         Commands::DumpGenesis(command) => runner.run_blocking_until_ctrl_c(command.execute()),
-        Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
+        Commands::Db(command) => {
+            runner.run_blocking_command_until_exit(|ctx| command.execute::<N>(ctx))
+        }
         Commands::Download(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
         Commands::Stage(command) => {
             runner.run_command_until_exit(|ctx| command.execute::<N, _>(ctx, components))
@@ -165,9 +178,19 @@ 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::*;