wip

Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>

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

.github/workflows/book.yml

@@ -10,9 +10,12 @@ on:
     types: [opened, reopened, synchronize, closed]
   merge_group:
 
+env:
+  RUSTC_WRAPPER: "sccache"
+
 jobs:
   build:
-    runs-on: depot-ubuntu-latest-16
+    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: depot-ubuntu-latest-16
+    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: depot-ubuntu-latest-16
+    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,7 @@ jobs:
   prepare-hive:
     if: github.repository == 'paradigmxyz/reth'
     timeout-minutes: 45
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     steps:
       - uses: actions/checkout@v6
       - name: Checkout hive tests
@@ -44,7 +44,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 +67,7 @@ jobs:
           chmod +x hive
 
       - name: Upload hive assets
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: hive_assets
           path: ./hive_assets
@@ -178,7 +178,7 @@ jobs:
       - prepare-reth
       - prepare-hive
     name: run ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
-    runs-on: depot-ubuntu-latest
+    runs-on: depot-ubuntu-latest-16
     permissions:
       issues: write
     steps:
@@ -187,13 +187,13 @@ jobs:
           fetch-depth: 0
 
       - name: Download hive assets
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: hive_assets
           path: /tmp
 
       - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: artifacts
           path: /tmp
@@ -245,7 +245,7 @@ jobs:
   notify-on-error:
     needs: test
     if: failure()
-    runs-on: depot-ubuntu-latest
+    runs-on: ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.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: depot-ubuntu-latest-16
+    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

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

.github/workflows/kurtosis.yml

@@ -39,7 +39,7 @@ jobs:
           fetch-depth: 0
 
       - name: Download reth image
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
         with:
           name: artifacts
           path: /tmp
@@ -58,7 +58,7 @@ jobs:
   notify-on-error:
     needs: test
     if: failure()
-    runs-on: depot-ubuntu-latest
+    runs-on: ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.github/workflows/lint.yml

@@ -8,6 +8,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 jobs:
   clippy-binaries:
@@ -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: depot-ubuntu-latest-16
+    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
@@ -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
@@ -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: depot-ubuntu-latest-16
+    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
@@ -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: depot-ubuntu-latest-16
+    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
@@ -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: depot-ubuntu-latest-16
+    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: depot-ubuntu-latest-16
+    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
@@ -234,17 +245,14 @@ jobs:
 
   # Checks that selected crates can compile with power set of features
   features:
-    name: features (${{ matrix.partition }}/${{ matrix.total_partitions }})
-    runs-on: depot-ubuntu-latest-16
-    strategy:
-      matrix:
-        partition: [1, 2]
-        total_partitions: [2]
+    name: features
+    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
@@ -256,7 +264,7 @@ jobs:
             --package reth-primitives-traits \
             --package reth-primitives \
             --feature-powerset \
-            --partition ${{ matrix.partition }}/${{ matrix.total_partitions }}
+            --depth 2
         env:
           RUSTFLAGS: -D warnings
 
@@ -267,6 +275,7 @@ jobs:
     steps:
       - uses: actions/checkout@v6
       - uses: dtolnay/rust-toolchain@stable
+      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: rui314/setup-mold@v1
       - uses: taiki-e/cache-cargo-install-action@v2
         with:

.github/workflows/prepare-reth.yml

@@ -50,7 +50,7 @@ jobs:
 
       - name: Upload reth image
         id: upload
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: artifacts
           path: ./artifacts

.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 }}
@@ -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 }}
@@ -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 }}
@@ -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: depot-ubuntu-latest-16
+    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
@@ -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: depot-ubuntu-latest-16
+    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: depot-ubuntu-latest-16
+    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: depot-ubuntu-latest-16
+    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

Cargo.lock

@@ -97,9 +97,9 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
 
 [[package]]
 name = "alloy-chains"
-version = "0.2.21"
+version = "0.2.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b9ebac8ff9c2f07667e1803dc777304337e160ce5153335beb45e8ec0751808"
+checksum = "35d744058a9daa51a8cf22a3009607498fcf82d3cf4c5444dd8056cdf651f471"
 dependencies = [
  "alloy-primitives",
  "alloy-rlp",
@@ -238,6 +238,18 @@ dependencies = [
  "thiserror 2.0.17",
 ]
 
+[[package]]
+name = "alloy-eip7928"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "926b2c0d34e641cf8b17bf54ce50fda16715b9f68ad878fa6128bae410c6f890"
+dependencies = [
+ "alloy-primitives",
+ "alloy-rlp",
+ "borsh",
+ "serde",
+]
+
 [[package]]
 name = "alloy-eips"
 version = "1.1.3"
@@ -266,9 +278,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-evm"
-version = "0.24.2"
+version = "0.25.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "01be36ba6f5e6e62563b369e03ca529eac46aea50677f84655084b4750816574"
+checksum = "e6ccc4c702c840148af1ce784cc5c6ed9274a020ef32417c5b1dbeab8c317673"
 dependencies = [
  "alloy-consensus",
  "alloy-eips",
@@ -317,9 +329,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-json-abi"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5513d5e6bd1cba6bdcf5373470f559f320c05c8c59493b6e98912fbe6733943f"
+checksum = "6bfca3dbbcb7498f0f60e67aff2ad6aff57032e22eb2fd03189854be11a22c03"
 dependencies = [
  "alloy-primitives",
  "alloy-sol-type-parser",
@@ -383,9 +395,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-op-evm"
-version = "0.24.2"
+version = "0.25.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "231262d7e06000f3fb642d32d38ca75e09e78e04977c10be0a07a5ee2c869cfd"
+checksum = "0f640da852f93ddaa3b9a602b7ca41d80e0023f77a67b68aaaf511c32f1fe0ce"
 dependencies = [
  "alloy-consensus",
  "alloy-eips",
@@ -414,9 +426,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-primitives"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "355bf68a433e0fd7f7d33d5a9fc2583fde70bf5c530f63b80845f8da5505cf28"
+checksum = "5c850e6ccbd34b8a463a1e934ffc8fc00e1efc5e5489f2ad82d7797949f3bd4e"
 dependencies = [
  "alloy-rlp",
  "arbitrary",
@@ -435,6 +447,7 @@ dependencies = [
  "proptest",
  "proptest-derive 0.6.0",
  "rand 0.9.2",
+ "rapidhash",
  "ruint",
  "rustc-hash",
  "serde",
@@ -769,9 +782,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-sol-macro"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3ce480400051b5217f19d6e9a82d9010cdde20f1ae9c00d53591e4a1afbb312"
+checksum = "b2218e3aeb3ee665d117fdf188db0d5acfdc3f7b7502c827421cb78f26a2aec0"
 dependencies = [
  "alloy-sol-macro-expander",
  "alloy-sol-macro-input",
@@ -783,9 +796,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-sol-macro-expander"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d792e205ed3b72f795a8044c52877d2e6b6e9b1d13f431478121d8d4eaa9028"
+checksum = "b231cb8cc48e66dd1c6e11a1402f3ac86c3667cbc13a6969a0ac030ba7bb8c88"
 dependencies = [
  "alloy-sol-macro-input",
  "const-hex",
@@ -801,9 +814,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-sol-macro-input"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bd1247a8f90b465ef3f1207627547ec16940c35597875cdc09c49d58b19693c"
+checksum = "49a522d79929c1bf0152b07567a38f7eaed3ab149e53e7528afa78ff11994668"
 dependencies = [
  "const-hex",
  "dunce",
@@ -817,9 +830,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-sol-type-parser"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "954d1b2533b9b2c7959652df3076954ecb1122a28cc740aa84e7b0a49f6ac0a9"
+checksum = "0475c459859c8d9428af6ff3736614655a57efda8cc435a3b8b4796fa5ac1dd0"
 dependencies = [
  "serde",
  "winnow",
@@ -827,9 +840,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-sol-types"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70319350969a3af119da6fb3e9bddb1bce66c9ea933600cb297c8b1850ad2a3c"
+checksum = "35287d9d821d5f26011bcd8d9101340898f761c9933cf50fca689bb7ed62fdeb"
 dependencies = [
  "alloy-json-abi",
  "alloy-primitives",
@@ -1369,9 +1382,9 @@ dependencies = [
 
 [[package]]
 name = "async-compression"
-version = "0.4.34"
+version = "0.4.36"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e86f6d3dc9dc4352edeea6b8e499e13e3f5dc3b964d7ca5fd411415a3498473"
+checksum = "98ec5f6c2f8bc326c994cb9e241cc257ddaba9afa8555a43cffbb5dd86efaa37"
 dependencies = [
  "compression-codecs",
  "compression-core",
@@ -1529,9 +1542,9 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
 
 [[package]]
 name = "base64ct"
-version = "1.8.0"
+version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55248b47b0caf0546f7988906588779981c43bb1bc9d0c44087278f80cdb44ba"
+checksum = "0e050f626429857a27ddccb31e0aca21356bfa709c04041aefddac081a8f068a"
 
 [[package]]
 name = "bech32"
@@ -2377,9 +2390,9 @@ dependencies = [
 
 [[package]]
 name = "compression-codecs"
-version = "0.4.33"
+version = "0.4.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "302266479cb963552d11bd042013a58ef1adc56768016c8b82b4199488f2d4ad"
+checksum = "b0f7ac3e5b97fdce45e8922fb05cae2c37f7bbd63d30dd94821dacfd8f3f2bf2"
 dependencies = [
  "brotli",
  "compression-core",
@@ -3993,9 +4006,9 @@ checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
 
 [[package]]
 name = "flate2"
-version = "1.1.7"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2152dbcb980c05735e2a651d96011320a949eb31a0c8b38b72645ce97dec676"
+checksum = "bfe33edd8e85a12a67454e37f8c75e730830d83e313556ab9ebf9ee7fbeb3bfb"
 dependencies = [
  "crc32fast",
  "miniz_oxide",
@@ -5462,9 +5475,9 @@ checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de"
 
 [[package]]
 name = "libp2p-identity"
-version = "0.2.12"
+version = "0.2.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3104e13b51e4711ff5738caa1fb54467c8604c2e94d607e27745bcf709068774"
+checksum = "f0c7892c221730ba55f7196e98b0b8ba5e04b4155651736036628e9f73ed6fc3"
 dependencies = [
  "asn1_der",
  "bs58",
@@ -6199,9 +6212,9 @@ checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
 
 [[package]]
 name = "op-alloy"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3b13412d297c1f9341f678b763750b120a73ffe998fa54a94d6eda98449e7ca"
+checksum = "e9b8fee21003dd4f076563de9b9d26f8c97840157ef78593cd7f262c5ca99848"
 dependencies = [
  "op-alloy-consensus",
  "op-alloy-network",
@@ -6212,9 +6225,9 @@ dependencies = [
 
 [[package]]
 name = "op-alloy-consensus"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "726da827358a547be9f1e37c2a756b9e3729cb0350f43408164794b370cad8ae"
+checksum = "736381a95471d23e267263cfcee9e1d96d30b9754a94a2819148f83379de8a86"
 dependencies = [
  "alloy-consensus",
  "alloy-eips",
@@ -6238,9 +6251,9 @@ checksum = "a79f352fc3893dcd670172e615afef993a41798a1d3fc0db88a3e60ef2e70ecc"
 
 [[package]]
 name = "op-alloy-network"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f63f27e65be273ec8fcb0b6af0fd850b550979465ab93423705ceb3dfddbd2ab"
+checksum = "4034183dca6bff6632e7c24c92e75ff5f0eabb58144edb4d8241814851334d47"
 dependencies = [
  "alloy-consensus",
  "alloy-network",
@@ -6254,9 +6267,9 @@ dependencies = [
 
 [[package]]
 name = "op-alloy-provider"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a71456699aa256dc20119736422ad9a44da8b9585036117afb936778122093b9"
+checksum = "6753d90efbaa8ea8bcb89c1737408ca85fa60d7adb875049d3f382c063666f86"
 dependencies = [
  "alloy-network",
  "alloy-primitives",
@@ -6269,9 +6282,9 @@ dependencies = [
 
 [[package]]
 name = "op-alloy-rpc-jsonrpsee"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ef9114426b16172254555aad34a8ea96c01895e40da92f5d12ea680a1baeaa7"
+checksum = "c1c820ef9c802ebc732281a940bfb6ac2345af4d9fff041cbb64b4b546676686"
 dependencies = [
  "alloy-primitives",
  "jsonrpsee",
@@ -6279,9 +6292,9 @@ dependencies = [
 
 [[package]]
 name = "op-alloy-rpc-types"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "562dd4462562c41f9fdc4d860858c40e14a25df7f983ae82047f15f08fce4d19"
+checksum = "ddd87c6b9e5b6eee8d6b76f41b04368dca0e9f38d83338e5b00e730c282098a4"
 dependencies = [
  "alloy-consensus",
  "alloy-eips",
@@ -6299,9 +6312,9 @@ dependencies = [
 
 [[package]]
 name = "op-alloy-rpc-types-engine"
-version = "0.22.4"
+version = "0.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8f24b8cb66e4b33e6c9e508bf46b8ecafc92eadd0b93fedd306c0accb477657"
+checksum = "77727699310a18cdeed32da3928c709e2704043b6584ed416397d5da65694efc"
 dependencies = [
  "alloy-consensus",
  "alloy-eips",
@@ -6315,6 +6328,7 @@ dependencies = [
  "ethereum_ssz_derive",
  "op-alloy-consensus",
  "serde",
+ "sha2",
  "snap",
  "thiserror 2.0.17",
 ]
@@ -7196,6 +7210,16 @@ dependencies = [
  "rand_core 0.9.3",
 ]
 
+[[package]]
+name = "rapidhash"
+version = "4.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d8e65c75143ce5d47c55b510297eeb1182f3c739b6043c537670e9fc18612dae"
+dependencies = [
+ "rand 0.9.2",
+ "rustversion",
+]
+
 [[package]]
 name = "ratatui"
 version = "0.29.0"
@@ -7350,9 +7374,9 @@ checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2"
 
 [[package]]
 name = "reqwest"
-version = "0.12.24"
+version = "0.12.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d0946410b9f7b082a427e4ef5c8ff541a88b357bc6c637c40db3a68ac70a36f"
+checksum = "b6eff9328d40131d43bd911d42d79eb6a47312002a4daefc9e37f17e74a7701a"
 dependencies = [
  "base64 0.22.1",
  "bytes",
@@ -8207,6 +8231,7 @@ name = "reth-engine-tree"
 version = "1.9.3"
 dependencies = [
  "alloy-consensus",
+ "alloy-eip7928",
  "alloy-eips",
  "alloy-evm",
  "alloy-primitives",
@@ -8222,6 +8247,7 @@ dependencies = [
  "metrics",
  "metrics-util",
  "mini-moka",
+ "moka",
  "parking_lot",
  "proptest",
  "rand 0.8.5",
@@ -8255,6 +8281,7 @@ dependencies = [
  "reth-stages",
  "reth-stages-api",
  "reth-static-file",
+ "reth-storage-errors",
  "reth-tasks",
  "reth-testing-utils",
  "reth-tracing",
@@ -8625,6 +8652,7 @@ dependencies = [
  "derive_more",
  "futures-util",
  "metrics",
+ "rayon",
  "reth-ethereum-forks",
  "reth-ethereum-primitives",
  "reth-execution-errors",
@@ -8930,6 +8958,7 @@ dependencies = [
  "pin-project",
  "rand 0.8.5",
  "rand 0.9.2",
+ "rayon",
  "reth-chainspec",
  "reth-consensus",
  "reth-discv4",
@@ -9100,6 +9129,7 @@ dependencies = [
  "fdlimit",
  "futures",
  "jsonrpsee",
+ "parking_lot",
  "rayon",
  "reth-basic-payload-builder",
  "reth-chain-state",
@@ -9231,6 +9261,7 @@ dependencies = [
  "alloy-sol-types",
  "eyre",
  "futures",
+ "jsonrpsee-core",
  "rand 0.9.2",
  "reth-chainspec",
  "reth-db",
@@ -9258,6 +9289,7 @@ dependencies = [
  "reth-rpc-eth-api",
  "reth-rpc-eth-types",
  "reth-rpc-server-types",
+ "reth-stages-types",
  "reth-tasks",
  "reth-testing-utils",
  "reth-tracing",
@@ -9266,6 +9298,7 @@ dependencies = [
  "serde",
  "serde_json",
  "similar-asserts",
+ "tempfile",
  "tokio",
 ]
 
@@ -9620,6 +9653,7 @@ dependencies = [
  "reth-rpc-engine-api",
  "reth-rpc-eth-types",
  "reth-rpc-server-types",
+ "reth-stages-types",
  "reth-tasks",
  "reth-tracing",
  "reth-transaction-pool",
@@ -10156,6 +10190,7 @@ dependencies = [
  "reth-db-api",
  "reth-engine-primitives",
  "reth-errors",
+ "reth-ethereum-engine-primitives",
  "reth-ethereum-primitives",
  "reth-evm",
  "reth-evm-ethereum",
@@ -10218,6 +10253,9 @@ dependencies = [
  "reth-network-peers",
  "reth-rpc-eth-api",
  "reth-trie-common",
+ "serde",
+ "serde_json",
+ "tokio",
 ]
 
 [[package]]
@@ -10282,6 +10320,7 @@ dependencies = [
  "reth-rpc-server-types",
  "reth-storage-api",
  "reth-tasks",
+ "reth-tokio-util",
  "reth-tracing",
  "reth-transaction-pool",
  "serde",
@@ -10358,6 +10397,7 @@ dependencies = [
  "reth-ethereum-engine-primitives",
  "reth-ethereum-primitives",
  "reth-metrics",
+ "reth-network-api",
  "reth-node-ethereum",
  "reth-payload-builder",
  "reth-payload-builder-primitives",
@@ -10544,6 +10584,7 @@ dependencies = [
  "reth-stages-api",
  "reth-static-file",
  "reth-static-file-types",
+ "reth-storage-api",
  "reth-storage-errors",
  "reth-testing-utils",
  "reth-trie",
@@ -10873,6 +10914,7 @@ dependencies = [
  "pretty_assertions",
  "proptest",
  "proptest-arbitrary-interop",
+ "rand 0.9.2",
  "reth-ethereum-primitives",
  "reth-execution-errors",
  "reth-metrics",
@@ -11175,9 +11217,9 @@ dependencies = [
 
 [[package]]
 name = "revm-inspectors"
-version = "0.33.1"
+version = "0.33.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c93974333e7acc4b2dc024b10def99707f7375a4d53db7a7f8351722d25673f"
+checksum = "01def7351cd9af844150b8e88980bcd11304f33ce23c3d7c25f2a8dab87c1345"
 dependencies = [
  "alloy-primitives",
  "alloy-rpc-types-eth",
@@ -12059,9 +12101,9 @@ dependencies = [
 
 [[package]]
 name = "simd-adler32"
-version = "0.3.7"
+version = "0.3.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d66dc143e6b11c1eddc06d5c423cfc97062865baf299914ab64caa38182078fe"
+checksum = "e320a6c5ad31d271ad523dcf3ad13e2767ad8b1cb8f047f75a8aeaf8da139da2"
 
 [[package]]
 name = "similar"
@@ -12316,9 +12358,9 @@ dependencies = [
 
 [[package]]
 name = "syn-solidity"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff790eb176cc81bb8936aed0f7b9f14fc4670069a2d371b3e3b0ecce908b2cb3"
+checksum = "60ceeb7c95a4536de0c0e1649bd98d1a72a4bb9590b1f3e45a8a0bfdb7c188c0"
 dependencies = [
  "paste",
  "proc-macro2",
@@ -12883,9 +12925,9 @@ dependencies = [
 
 [[package]]
 name = "tower-http"
-version = "0.6.7"
+version = "0.6.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9cf146f99d442e8e68e585f5d798ccd3cad9a7835b917e09728880a862706456"
+checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8"
 dependencies = [
  "async-compression",
  "base64 0.22.1",

Cargo.toml

@@ -376,11 +376,11 @@ reth-era-utils = { path = "crates/era-utils" }
 reth-errors = { path = "crates/errors" }
 reth-eth-wire = { path = "crates/net/eth-wire" }
 reth-eth-wire-types = { path = "crates/net/eth-wire-types" }
+reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-cli = { path = "crates/ethereum/cli", default-features = false }
 reth-ethereum-consensus = { path = "crates/ethereum/consensus", default-features = false }
 reth-ethereum-engine-primitives = { path = "crates/ethereum/engine-primitives", default-features = false }
 reth-ethereum-forks = { path = "crates/ethereum/hardforks", default-features = false }
-reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-primitives = { path = "crates/ethereum/primitives", default-features = false }
 reth-ethereum = { path = "crates/ethereum/reth" }
 reth-etl = { path = "crates/etl" }
@@ -481,17 +481,18 @@ 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"
@@ -525,13 +526,13 @@ alloy-transport-ipc = { version = "1.1.3", default-features = false }
 alloy-transport-ws = { version = "1.1.3", default-features = false }
 
 # op
-alloy-op-evm = { version = "0.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
@@ -586,6 +587,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"

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

Makefile

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

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

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,9 @@ asm-keccak = [
     "reth-ethereum-cli/asm-keccak",
     "reth-node-ethereum/asm-keccak",
 ]
-
+keccak-cache-global = [
+    "reth-node-ethereum/keccak-cache-global",
+]
 jemalloc = [
     "reth-cli-util/jemalloc",
     "reth-node-core/jemalloc",

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

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

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

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

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/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/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/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/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;
@@ -426,9 +427,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) => {
@@ -1302,22 +1307,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 +1320,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 +1329,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 +1404,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 +1418,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 +1526,7 @@ where
                 }
             }
         }
-        Ok(())
+        Ok(ops::ControlFlow::Continue(()))
     }
 
     /// Invoked if the backfill sync has finished to target.
@@ -1701,10 +1760,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 +1774,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",
@@ -2860,3 +2922,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,318 @@
+//! BAL (Block Access List, EIP-7928) related functionality.
+
+use alloy_consensus::constants::KECCAK_EMPTY;
+use alloy_eip7928::BlockAccessList;
+use alloy_primitives::{keccak256, U256};
+use reth_primitives_traits::Account;
+use reth_provider::{AccountReader, ProviderError};
+use reth_trie::{HashedPostState, HashedStorage};
+
+/// Converts a Block Access List into a [`HashedPostState`] by extracting the final state
+/// of modified accounts and storage slots.
+pub fn bal_to_hashed_post_state<P>(
+    bal: &BlockAccessList,
+    provider: &P,
+) -> Result<HashedPostState, ProviderError>
+where
+    P: AccountReader,
+{
+    let mut hashed_state = HashedPostState::with_capacity(bal.len());
+
+    for account_changes in bal {
+        let address = account_changes.address;
+        let hashed_address = keccak256(address);
+
+        // Get the latest balance (last balance change if any)
+        let balance = account_changes.balance_changes.last().map(|change| change.post_balance);
+
+        // Get the latest nonce (last nonce change if any)
+        let nonce = account_changes.nonce_changes.last().map(|change| change.new_nonce);
+
+        // Get the latest code (last code change if any)
+        let code_hash = if let Some(code_change) = account_changes.code_changes.last() {
+            if code_change.new_code.is_empty() {
+                Some(Some(KECCAK_EMPTY))
+            } else {
+                Some(Some(keccak256(&code_change.new_code)))
+            }
+        } else {
+            None
+        };
+
+        // Only fetch account from provider if we're missing any field
+        let existing_account = if balance.is_none() || nonce.is_none() || code_hash.is_none() {
+            provider.basic_account(&address)?
+        } else {
+            None
+        };
+
+        // Build the final account state
+        let account = Account {
+            balance: balance.unwrap_or_else(|| {
+                existing_account.as_ref().map(|acc| acc.balance).unwrap_or(U256::ZERO)
+            }),
+            nonce: nonce
+                .unwrap_or_else(|| existing_account.as_ref().map(|acc| acc.nonce).unwrap_or(0)),
+            bytecode_hash: code_hash.unwrap_or_else(|| {
+                existing_account.as_ref().and_then(|acc| acc.bytecode_hash).or(Some(KECCAK_EMPTY))
+            }),
+        };
+
+        hashed_state.accounts.insert(hashed_address, Some(account));
+
+        // Process storage changes
+        if !account_changes.storage_changes.is_empty() {
+            let mut storage_map = HashedStorage::new(false);
+
+            for slot_changes in &account_changes.storage_changes {
+                let hashed_slot = keccak256(slot_changes.slot);
+
+                // Get the last change for this slot
+                if let Some(last_change) = slot_changes.changes.last() {
+                    storage_map
+                        .storage
+                        .insert(hashed_slot, U256::from_be_bytes(last_change.new_value.0));
+                }
+            }
+
+            if !storage_map.storage.is_empty() {
+                hashed_state.storages.insert(hashed_address, storage_map);
+            }
+        }
+    }
+
+    Ok(hashed_state)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_eip7928::{
+        AccountChanges, BalanceChange, CodeChange, NonceChange, SlotChanges, StorageChange,
+    };
+    use alloy_primitives::{Address, Bytes, StorageKey, B256};
+    use reth_revm::test_utils::StateProviderTest;
+
+    #[test]
+    fn test_bal_to_hashed_post_state_basic() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(100))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        assert_eq!(result.accounts.len(), 1);
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        assert!(account_opt.is_some());
+
+        let account = account_opt.as_ref().unwrap();
+        assert_eq!(account.balance, U256::from(100));
+        assert_eq!(account.nonce, 1);
+        assert_eq!(account.bytecode_hash, Some(KECCAK_EMPTY));
+    }
+
+    #[test]
+    fn test_bal_with_storage_changes() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let slot = StorageKey::random();
+        let value = B256::random();
+
+        let slot_changes = SlotChanges { slot, changes: vec![StorageChange::new(0, value)] };
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![slot_changes],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(500))],
+            nonce_changes: vec![NonceChange::new(0, 2)],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        assert!(result.storages.contains_key(&hashed_address));
+
+        let storage = result.storages.get(&hashed_address).unwrap();
+        let hashed_slot = keccak256(slot);
+
+        let stored_value = storage.storage.get(&hashed_slot).unwrap();
+        assert_eq!(*stored_value, U256::from_be_bytes(value.0));
+    }
+
+    #[test]
+    fn test_bal_with_code_change() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let code = Bytes::from(vec![0x60, 0x80, 0x60, 0x40]); // Some bytecode
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![CodeChange::new(0, code.clone())],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        let expected_code_hash = keccak256(&code);
+        assert_eq!(account.bytecode_hash, Some(expected_code_hash));
+    }
+
+    #[test]
+    fn test_bal_with_empty_code() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let empty_code = Bytes::default();
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![CodeChange::new(0, empty_code)],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        assert_eq!(account.bytecode_hash, Some(KECCAK_EMPTY));
+    }
+
+    #[test]
+    fn test_bal_multiple_changes_takes_last() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+
+        // Multiple balance changes - should take the last one
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![
+                BalanceChange::new(0, U256::from(100)),
+                BalanceChange::new(1, U256::from(200)),
+                BalanceChange::new(2, U256::from(300)),
+            ],
+            nonce_changes: vec![
+                NonceChange::new(0, 1),
+                NonceChange::new(1, 2),
+                NonceChange::new(2, 3),
+            ],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        // Should have the last values
+        assert_eq!(account.balance, U256::from(300));
+        assert_eq!(account.nonce, 3);
+    }
+
+    #[test]
+    fn test_bal_uses_provider_for_missing_fields() {
+        let mut provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let code_hash = B256::random();
+        let existing_account =
+            Account { balance: U256::from(999), nonce: 42, bytecode_hash: Some(code_hash) };
+        provider.insert_account(address, existing_account, None, Default::default());
+
+        // Only change balance, nonce and code should come from provider
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(1500))],
+            nonce_changes: vec![],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let account_opt = result.accounts.get(&hashed_address).unwrap();
+        let account = account_opt.as_ref().unwrap();
+
+        // Balance should be updated
+        assert_eq!(account.balance, U256::from(1500));
+        // Nonce and bytecode_hash should come from provider
+        assert_eq!(account.nonce, 42);
+        assert_eq!(account.bytecode_hash, Some(code_hash));
+    }
+
+    #[test]
+    fn test_bal_multiple_storage_changes_per_slot() {
+        let provider = StateProviderTest::default();
+
+        let address = Address::random();
+        let slot = StorageKey::random();
+
+        // Multiple changes to the same slot - should take the last one
+        let slot_changes = SlotChanges {
+            slot,
+            changes: vec![
+                StorageChange::new(0, B256::from(U256::from(100).to_be_bytes::<32>())),
+                StorageChange::new(1, B256::from(U256::from(200).to_be_bytes::<32>())),
+                StorageChange::new(2, B256::from(U256::from(300).to_be_bytes::<32>())),
+            ],
+        };
+
+        let account_changes = AccountChanges {
+            address,
+            storage_changes: vec![slot_changes],
+            storage_reads: vec![],
+            balance_changes: vec![BalanceChange::new(0, U256::from(100))],
+            nonce_changes: vec![NonceChange::new(0, 1)],
+            code_changes: vec![],
+        };
+
+        let bal = vec![account_changes];
+        let result = bal_to_hashed_post_state(&bal, &provider).unwrap();
+
+        let hashed_address = keccak256(address);
+        let storage = result.storages.get(&hashed_address).unwrap();
+        let hashed_slot = keccak256(slot);
+
+        let stored_value = storage.storage.get(&hashed_slot).unwrap();
+
+        // Should have the last value
+        assert_eq!(*stored_value, U256::from(300));
+    }
+}

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

@@ -13,6 +13,7 @@ use crate::tree::{
     sparse_trie::SparseTrieTask,
     StateProviderBuilder, TreeConfig,
 };
+use alloy_eip7928::BlockAccessList;
 use alloy_eips::eip1898::BlockWithParent;
 use alloy_evm::{block::StateChangeSource, ToTxEnv};
 use alloy_primitives::B256;
@@ -21,12 +22,13 @@ 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_revm::{db::BundleState, state::EvmState};
@@ -49,8 +51,9 @@ use std::{
     },
     time::Instant,
 };
-use tracing::{debug, debug_span, instrument, warn, Span};
+use tracing::{debug, debug_span, error, instrument, warn, Span};
 
+pub mod bal;
 mod configured_sparse_trie;
 pub mod executor;
 pub mod multiproof;
@@ -90,6 +93,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 +116,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 +161,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 +208,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 +221,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>
@@ -249,19 +262,45 @@ where
         // wire the multiproof task to the prewarm task
         let to_multi_proof = Some(multi_proof_task.state_root_message_sender());
 
-        let prewarm_handle = self.spawn_caching_with(
-            env,
-            prewarm_rx,
-            transaction_count_hint,
-            provider_builder,
-            to_multi_proof.clone(),
-        );
+        // Handle BAL-based optimization if available
+        let prewarm_handle = if let Some(bal) = bal {
+            // When BAL is present, skip spawning prewarm tasks entirely and send BAL to multiproof
+            debug!(target: "engine::tree::payload_processor", "BAL present, skipping prewarm tasks");
+
+            // Send BAL message immediately to MultiProofTask
+            if let Some(ref sender) = to_multi_proof &&
+                let Err(err) = sender.send(MultiProofMessage::BlockAccessList(bal))
+            {
+                // In this case state root validation will simply fail
+                error!(target: "engine::tree::payload_processor", ?err, "Failed to send BAL to MultiProofTask");
+            }
+
+            // Spawn minimal cache-only task without prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                transaction_count_hint,
+                provider_builder.clone(),
+                None, // Don't send proof targets when BAL is present
+            )
+        } else {
+            // Normal path: spawn with full prewarming
+            self.spawn_caching_with(
+                env,
+                prewarm_rx,
+                transaction_count_hint,
+                provider_builder.clone(),
+                to_multi_proof.clone(),
+            )
+        };
 
         // spawn multi-proof task
         let parent_span = span.clone();
         self.executor.spawn_blocking(move || {
             let _enter = parent_span.entered();
-            multi_proof_task.run();
+            // Build a state provider for the multiproof task
+            let provider = provider_builder.build().expect("failed to build provider");
+            multi_proof_task.run(provider);
         });
 
         // wire the sparse trie to the state root response receiver
@@ -288,7 +327,7 @@ where
         env: ExecutionEnv<Evm>,
         transactions: I,
         provider_builder: StateProviderBuilder<N, P>,
-    ) -> PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>
+    ) -> IteratorPayloadHandle<Evm, I, N>
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
@@ -315,36 +354,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 +407,7 @@ where
         transaction_count_hint: usize,
         provider_builder: StateProviderBuilder<N, P>,
         to_multi_proof: Option<CrossbeamSender<MultiProofMessage>>,
-    ) -> CacheTaskHandle
+    ) -> CacheTaskHandle<N::Receipt>
     where
         P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
     {
@@ -382,9 +417,15 @@ 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, cache, cache_metrics) = if self.disable_state_cache {
+            (None, None, None)
+        } else {
+            let saved_cache = self.cache_for(env.parent_hash);
+            let cache = saved_cache.cache().clone();
+            let cache_metrics = saved_cache.metrics().clone();
+            (Some(saved_cache), Some(cache), Some(cache_metrics))
+        };
+
         // configure prewarming
         let prewarm_ctx = PrewarmContext {
             env,
@@ -547,12 +588,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 +605,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,18 +634,18 @@ 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 {
+    pub(super) fn caches(&self) -> Option<StateExecutionCache> {
         self.prewarm_handle.cache.clone()
     }
 
     /// Returns a clone of the cache metrics used by prewarming
-    pub(super) fn cache_metrics(&self) -> CachedStateMetrics {
+    pub(super) fn cache_metrics(&self) -> Option<CachedStateMetrics> {
         self.prewarm_handle.cache_metrics.clone()
     }
 
@@ -614,9 +658,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 +677,20 @@ 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,
+    cache: Option<StateExecutionCache>,
     /// Metrics for the caches
-    cache_metrics: CachedStateMetrics,
+    cache_metrics: Option<CachedStateMetrics>,
     /// 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 +702,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 +773,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 +1107,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

@@ -27,10 +27,11 @@ use alloy_primitives::{keccak256, map::B256Set, B256};
 use crossbeam_channel::Sender as CrossbeamSender;
 use metrics::{Counter, Gauge, Histogram};
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
+use reth_execution_types::ExecutionOutcome;
 use reth_metrics::Metrics;
 use reth_primitives_traits::NodePrimitives;
 use reth_provider::{BlockReader, StateProviderFactory, StateReader};
-use reth_revm::{database::StateProviderDatabase, db::BundleState, state::EvmState};
+use reth_revm::{database::StateProviderDatabase, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
     sync::{
@@ -86,7 +87,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 +106,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 +136,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();
@@ -248,38 +252,43 @@ 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());
+        }
     }
 
     /// Executes the task.
@@ -296,12 +305,12 @@ where
     pub(super) fn run(
         self,
         pending: mpsc::Receiver<impl ExecutableTxFor<Evm> + Clone + Send + 'static>,
-        actions_tx: Sender<PrewarmTaskEvent>,
+        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
     ) {
         // spawn execution tasks.
         self.spawn_all(pending, 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 +323,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 +339,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 +349,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 +365,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 +393,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 +409,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 +459,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>,
@@ -526,7 +540,7 @@ where
         &self,
         idx: usize,
         executor: &WorkloadExecutor,
-        actions_tx: Sender<PrewarmTaskEvent>,
+        actions_tx: Sender<PrewarmTaskEvent<N::Receipt>>,
         done_tx: Sender<()>,
     ) -> mpsc::Sender<IndexedTransaction<Tx>>
     where
@@ -582,14 +596,18 @@ fn multiproof_targets_from_state(state: EvmState) -> (MultiProofTargets, usize)
 }
 
 /// 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 {

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,23 @@ 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 +520,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 +550,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,7 +600,7 @@ 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,
@@ -604,7 +612,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();
@@ -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
@@ -872,7 +912,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 +1051,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 +1101,7 @@ where
 
         ExecutedBlock::with_deferred_trie_data(
             Arc::new(block),
-            Arc::new(ExecutionOutcome::from((output, block_number))),
+            execution_outcome,
             deferred_trie_data,
         )
     }
@@ -1243,4 +1282,10 @@ 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
+    }
 }

crates/engine/tree/src/tree/precompile_cache.rs

@@ -1,50 +1,56 @@
 //! Contains a precompile cache backed by `schnellru::LruMap` (LRU by length).
 
 use alloy_primitives::Bytes;
-use parking_lot::Mutex;
+use dashmap::DashMap;
 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)
+                .build_with_hasher(Default::default()),
+        )
     }
 }
 
@@ -52,63 +58,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 +164,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 +176,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 +220,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 +248,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 +266,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/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/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/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

@@ -154,7 +154,9 @@ where
         Commands::ImportEra(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
         Commands::ExportEra(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
         Commands::DumpGenesis(command) => runner.run_blocking_until_ctrl_c(command.execute()),
-        Commands::Db(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
+        Commands::Db(command) => {
+            runner.run_blocking_command_until_exit(|ctx| command.execute::<N>(ctx))
+        }
         Commands::Download(command) => runner.run_blocking_until_ctrl_c(command.execute::<N>()),
         Commands::Stage(command) => {
             runner.run_command_until_exit(|ctx| command.execute::<N, _>(ctx, components))

crates/ethereum/evm/src/build.rs

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

crates/ethereum/evm/src/lib.rs

@@ -19,32 +19,37 @@ extern crate alloc;
 
 use alloc::{borrow::Cow, sync::Arc};
 use alloy_consensus::Header;
-use alloy_eips::Decodable2718;
-pub use alloy_evm::EthEvm;
 use alloy_evm::{
     eth::{EthBlockExecutionCtx, EthBlockExecutorFactory},
     EthEvmFactory, FromRecoveredTx, FromTxWithEncoded,
 };
-use alloy_primitives::{Bytes, U256};
-use alloy_rpc_types_engine::ExecutionData;
 use core::{convert::Infallible, fmt::Debug};
-use reth_chainspec::{ChainSpec, EthChainSpec, EthereumHardforks, MAINNET};
+use reth_chainspec::{ChainSpec, EthChainSpec, MAINNET};
 use reth_ethereum_primitives::{Block, EthPrimitives, TransactionSigned};
 use reth_evm::{
-    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEngineEvm, ConfigureEvm,
-    EvmEnv, EvmEnvFor, EvmFactory, ExecutableTxIterator, ExecutionCtxFor, NextBlockEnvAttributes,
-    TransactionEnv,
+    eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEvm, EvmEnv, EvmFactory,
+    NextBlockEnvAttributes, TransactionEnv,
 };
-use reth_primitives_traits::{
-    constants::MAX_TX_GAS_LIMIT_OSAKA, SealedBlock, SealedHeader, SignedTransaction, TxTy,
-};
-use reth_storage_errors::any::AnyError;
-use revm::{
-    context::{BlockEnv, CfgEnv},
-    context_interface::block::BlobExcessGasAndPrice,
-    primitives::hardfork::SpecId,
+use reth_primitives_traits::{SealedBlock, SealedHeader};
+use revm::{context::BlockEnv, primitives::hardfork::SpecId};
+
+#[cfg(feature = "std")]
+use reth_evm::{ConfigureEngineEvm, ExecutableTxIterator};
+#[allow(unused_imports)]
+use {
+    alloy_eips::Decodable2718,
+    alloy_primitives::{Bytes, U256},
+    alloy_rpc_types_engine::ExecutionData,
+    reth_chainspec::EthereumHardforks,
+    reth_evm::{EvmEnvFor, ExecutionCtxFor},
+    reth_primitives_traits::{constants::MAX_TX_GAS_LIMIT_OSAKA, SignedTransaction, TxTy},
+    reth_storage_errors::any::AnyError,
+    revm::context::CfgEnv,
+    revm::context_interface::block::BlobExcessGasAndPrice,
 };
 
+pub use alloy_evm::EthEvm;
+
 mod config;
 use alloy_evm::eth::spec::EthExecutorSpec;
 pub use config::{revm_spec, revm_spec_by_timestamp_and_block_number};
@@ -116,12 +121,6 @@ impl<ChainSpec, EvmFactory> EthEvmConfig<ChainSpec, EvmFactory> {
     pub const fn chain_spec(&self) -> &Arc<ChainSpec> {
         self.executor_factory.spec()
     }
-
-    /// Sets the extra data for the block assembler.
-    pub fn with_extra_data(mut self, extra_data: Bytes) -> Self {
-        self.block_assembler.extra_data = extra_data;
-        self
-    }
 }
 
 impl<ChainSpec, EvmF> ConfigureEvm for EthEvmConfig<ChainSpec, EvmF>
@@ -193,6 +192,7 @@ where
             parent_beacon_block_root: block.header().parent_beacon_block_root,
             ommers: &block.body().ommers,
             withdrawals: block.body().withdrawals.as_ref().map(Cow::Borrowed),
+            extra_data: block.header().extra_data.clone(),
         })
     }
 
@@ -206,10 +206,12 @@ where
             parent_beacon_block_root: attributes.parent_beacon_block_root,
             ommers: &[],
             withdrawals: attributes.withdrawals.map(Cow::Owned),
+            extra_data: attributes.extra_data,
         })
     }
 }
 
+#[cfg(feature = "std")]
 impl<ChainSpec, EvmF> ConfigureEngineEvm<ExecutionData> for EthEvmConfig<ChainSpec, EvmF>
 where
     ChainSpec: EthExecutorSpec + EthChainSpec<Header = Header> + Hardforks + 'static,
@@ -282,6 +284,7 @@ where
             parent_beacon_block_root: payload.sidecar.parent_beacon_block_root(),
             ommers: &[],
             withdrawals: payload.payload.withdrawals().map(|w| Cow::Owned(w.clone().into())),
+            extra_data: payload.payload.as_v1().extra_data.clone(),
         })
     }
 
@@ -289,7 +292,7 @@ where
         &self,
         payload: &ExecutionData,
     ) -> Result<impl ExecutableTxIterator<Self>, Self::Error> {
-        let txs = payload.payload.transactions().clone().into_iter();
+        let txs = payload.payload.transactions().clone();
         let convert = |tx: Bytes| {
             let tx =
                 TxTy::<Self::Primitives>::decode_2718_exact(tx.as_ref()).map_err(AnyError::new)?;

crates/ethereum/hardforks/src/display.rs

@@ -170,7 +170,7 @@ impl DisplayHardforks {
         let mut post_merge = Vec::new();
 
         for (fork, condition, metadata) in hardforks {
-            let mut display_fork = DisplayFork {
+            let display_fork = DisplayFork {
                 name: fork.name().to_string(),
                 activated_at: condition,
                 eip: None,
@@ -181,12 +181,7 @@ impl DisplayHardforks {
                 ForkCondition::Block(_) => {
                     pre_merge.push(display_fork);
                 }
-                ForkCondition::TTD { activation_block_number, total_difficulty, fork_block } => {
-                    display_fork.activated_at = ForkCondition::TTD {
-                        activation_block_number,
-                        fork_block,
-                        total_difficulty,
-                    };
+                ForkCondition::TTD { .. } => {
                     with_merge.push(display_fork);
                 }
                 ForkCondition::Timestamp(_) => {

crates/ethereum/node/Cargo.toml

@@ -24,7 +24,7 @@ reth-provider.workspace = true
 reth-transaction-pool.workspace = true
 reth-network.workspace = true
 reth-evm.workspace = true
-reth-evm-ethereum.workspace = true
+reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-rpc.workspace = true
 reth-rpc-api.workspace = true
 reth-rpc-eth-api.workspace = true
@@ -35,7 +35,7 @@ reth-chainspec.workspace = true
 reth-revm = { workspace = true, features = ["std"] }
 reth-rpc-eth-types.workspace = true
 reth-engine-local.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-payload-primitives.workspace = true
 
 # ethereum
@@ -61,6 +61,9 @@ reth-node-core.workspace = true
 reth-e2e-test-utils.workspace = true
 reth-tasks.workspace = true
 reth-testing-utils.workspace = true
+reth-stages-types.workspace = true
+tempfile.workspace = true
+jsonrpsee-core.workspace = true
 
 alloy-primitives.workspace = true
 alloy-provider.workspace = true
@@ -86,6 +89,9 @@ asm-keccak = [
     "reth-node-core/asm-keccak",
     "revm/asm-keccak",
 ]
+keccak-cache-global = [
+    "alloy-primitives/keccak-cache-global",
+]
 js-tracer = [
     "reth-node-builder/js-tracer",
     "reth-rpc/js-tracer",
@@ -104,4 +110,5 @@ test-utils = [
     "reth-evm/test-utils",
     "reth-primitives-traits/test-utils",
     "reth-evm-ethereum/test-utils",
+    "reth-stages-types/test-utils",
 ]

crates/ethereum/node/src/node.rs

@@ -32,15 +32,15 @@ use reth_node_builder::{
         EngineValidatorBuilder, EthApiBuilder, EthApiCtx, Identity, PayloadValidatorBuilder,
         RethRpcAddOns, RpcAddOns, RpcHandle,
     },
-    BuilderContext, DebugNode, Node, NodeAdapter, PayloadBuilderConfig,
+    BuilderContext, DebugNode, Node, NodeAdapter,
 };
 use reth_payload_primitives::PayloadTypes;
 use reth_provider::{providers::ProviderFactoryBuilder, EthStorage};
 use reth_rpc::{
     eth::core::{EthApiFor, EthRpcConverterFor},
-    ValidationApi,
+    TestingApi, ValidationApi,
 };
-use reth_rpc_api::servers::BlockSubmissionValidationApiServer;
+use reth_rpc_api::servers::{BlockSubmissionValidationApiServer, TestingApiServer};
 use reth_rpc_builder::{config::RethRpcServerConfig, middleware::RethRpcMiddleware};
 use reth_rpc_eth_api::{
     helpers::{
@@ -118,13 +118,14 @@ impl EthereumNode {
     /// use reth_chainspec::ChainSpecBuilder;
     /// use reth_db::open_db_read_only;
     /// use reth_node_ethereum::EthereumNode;
-    /// use reth_provider::providers::StaticFileProvider;
+    /// use reth_provider::providers::{RocksDBProvider, StaticFileProvider};
     /// use std::sync::Arc;
     ///
     /// let factory = EthereumNode::provider_factory_builder()
     ///     .db(Arc::new(open_db_read_only("db", Default::default()).unwrap()))
     ///     .chainspec(ChainSpecBuilder::mainnet().build().into())
     ///     .static_file(StaticFileProvider::read_only("db/static_files", false).unwrap())
+    ///     .rocksdb_provider(RocksDBProvider::builder("db/rocksdb").build().unwrap())
     ///     .build_provider_factory();
     /// ```
     pub fn provider_factory_builder() -> ProviderFactoryBuilder<Self> {
@@ -313,6 +314,17 @@ where
                     .modules
                     .merge_if_module_configured(RethRpcModule::Eth, eth_config.into_rpc())?;
 
+                // testing_buildBlockV1: only wire when the hidden testing module is explicitly
+                // requested on any transport. Default stays disabled to honor security guidance.
+                let testing_api = TestingApi::new(
+                    container.registry.eth_api().clone(),
+                    container.registry.evm_config().clone(),
+                )
+                .into_rpc();
+                container
+                    .modules
+                    .merge_if_module_configured(RethRpcModule::Testing, testing_api)?;
+
                 Ok(())
             })
             .await
@@ -426,9 +438,7 @@ where
     type EVM = EthEvmConfig<Types::ChainSpec>;
 
     async fn build_evm(self, ctx: &BuilderContext<Node>) -> eyre::Result<Self::EVM> {
-        let evm_config = EthEvmConfig::new(ctx.chain_spec())
-            .with_extra_data(ctx.payload_builder_config().extra_data_bytes());
-        Ok(evm_config)
+        Ok(EthEvmConfig::new(ctx.chain_spec()))
     }
 }
 

crates/ethereum/node/src/payload.rs

@@ -54,7 +54,8 @@ where
             evm_config,
             EthereumBuilderConfig::new()
                 .with_gas_limit(gas_limit)
-                .with_max_blobs_per_block(conf.max_blobs_per_block()),
+                .with_max_blobs_per_block(conf.max_blobs_per_block())
+                .with_extra_data(conf.extra_data_bytes()),
         ))
     }
 }

crates/ethereum/node/tests/e2e/custom_genesis.rs

@@ -0,0 +1,100 @@
+use crate::utils::eth_payload_attributes;
+use alloy_genesis::Genesis;
+use alloy_primitives::B256;
+use reth_chainspec::{ChainSpecBuilder, MAINNET};
+use reth_e2e_test_utils::{setup, transaction::TransactionTestContext};
+use reth_node_ethereum::EthereumNode;
+use reth_provider::{HeaderProvider, StageCheckpointReader};
+use reth_stages_types::StageId;
+use std::sync::Arc;
+
+/// Tests that a node can initialize and advance with a custom genesis block number.
+#[tokio::test]
+async fn can_run_eth_node_with_custom_genesis_number() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    // Create genesis with custom block number (e.g., 1000)
+    let mut genesis: Genesis =
+        serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
+    genesis.number = Some(1000);
+    genesis.parent_hash = Some(B256::random());
+
+    let chain_spec = Arc::new(
+        ChainSpecBuilder::default()
+            .chain(MAINNET.chain)
+            .genesis(genesis)
+            .cancun_activated()
+            .build(),
+    );
+
+    let (mut nodes, _tasks, wallet) =
+        setup::<EthereumNode>(1, chain_spec, false, eth_payload_attributes).await?;
+
+    let mut node = nodes.pop().unwrap();
+
+    // Verify stage checkpoints are initialized to genesis block number (1000)
+    for stage in StageId::ALL {
+        let checkpoint = node.inner.provider.get_stage_checkpoint(stage)?;
+        assert!(checkpoint.is_some(), "Stage {:?} checkpoint should exist", stage);
+        assert_eq!(
+            checkpoint.unwrap().block_number,
+            1000,
+            "Stage {:?} checkpoint should be at genesis block 1000",
+            stage
+        );
+    }
+
+    // Advance the chain (block 1001)
+    let raw_tx = TransactionTestContext::transfer_tx_bytes(1, wallet.inner).await;
+    let tx_hash = node.rpc.inject_tx(raw_tx).await?;
+    let payload = node.advance_block().await?;
+
+    let block_hash = payload.block().hash();
+    let block_number = payload.block().number;
+
+    // Verify we're at block 1001 (genesis + 1)
+    assert_eq!(block_number, 1001, "Block number should be 1001 after advancing from genesis 1000");
+
+    // Assert the block has been committed
+    node.assert_new_block(tx_hash, block_hash, block_number).await?;
+
+    Ok(())
+}
+
+/// Tests that block queries respect custom genesis boundaries.
+#[tokio::test]
+async fn custom_genesis_block_query_boundaries() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let genesis_number = 5000u64;
+
+    let mut genesis: Genesis =
+        serde_json::from_str(include_str!("../assets/genesis.json")).unwrap();
+    genesis.number = Some(genesis_number);
+    genesis.parent_hash = Some(B256::random());
+
+    let chain_spec = Arc::new(
+        ChainSpecBuilder::default()
+            .chain(MAINNET.chain)
+            .genesis(genesis)
+            .cancun_activated()
+            .build(),
+    );
+
+    let (mut nodes, _tasks, _wallet) =
+        setup::<EthereumNode>(1, chain_spec, false, eth_payload_attributes).await?;
+
+    let node = nodes.pop().unwrap();
+
+    // Query genesis block should succeed
+    let genesis_header = node.inner.provider.header_by_number(genesis_number)?;
+    assert!(genesis_header.is_some(), "Genesis block at {} should exist", genesis_number);
+
+    // Query blocks before genesis should return None
+    for block_num in [0, 1, genesis_number - 1] {
+        let header = node.inner.provider.header_by_number(block_num)?;
+        assert!(header.is_none(), "Block {} before genesis should not exist", block_num);
+    }
+
+    Ok(())
+}

crates/ethereum/node/tests/e2e/eth.rs

@@ -7,6 +7,7 @@ use reth_e2e_test_utils::{
 use reth_node_builder::{NodeBuilder, NodeHandle};
 use reth_node_core::{args::RpcServerArgs, node_config::NodeConfig};
 use reth_node_ethereum::EthereumNode;
+use reth_provider::BlockNumReader;
 use reth_tasks::TaskManager;
 use std::sync::Arc;
 
@@ -127,3 +128,55 @@ async fn test_failed_run_eth_node_with_no_auth_engine_api_over_ipc_opts() -> eyr
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_engine_graceful_shutdown() -> eyre::Result<()> {
+    reth_tracing::init_test_tracing();
+
+    let (mut nodes, _tasks, wallet) = setup::<EthereumNode>(
+        1,
+        Arc::new(
+            ChainSpecBuilder::default()
+                .chain(MAINNET.chain)
+                .genesis(serde_json::from_str(include_str!("../assets/genesis.json")).unwrap())
+                .cancun_activated()
+                .build(),
+        ),
+        false,
+        eth_payload_attributes,
+    )
+    .await?;
+
+    let mut node = nodes.pop().unwrap();
+
+    let raw_tx = TransactionTestContext::transfer_tx_bytes(1, wallet.inner).await;
+    let tx_hash = node.rpc.inject_tx(raw_tx).await?;
+    let payload = node.advance_block().await?;
+    node.assert_new_block(tx_hash, payload.block().hash(), payload.block().number).await?;
+
+    // Get block number before shutdown
+    let block_before = node.inner.provider.best_block_number()?;
+    assert_eq!(block_before, 1, "Expected 1 block before shutdown");
+
+    // Verify block is NOT yet persisted to database
+    let db_block_before = node.inner.provider.last_block_number()?;
+    assert_eq!(db_block_before, 0, "Block should not be persisted yet");
+
+    // Trigger graceful shutdown
+    let done_rx = node
+        .inner
+        .add_ons_handle
+        .engine_shutdown
+        .shutdown()
+        .expect("shutdown should return receiver");
+
+    tokio::time::timeout(std::time::Duration::from_secs(2), done_rx)
+        .await
+        .expect("shutdown timed out")
+        .expect("shutdown completion channel should not be closed");
+
+    let db_block = node.inner.provider.last_block_number()?;
+    assert_eq!(db_block, 1, "Database should have persisted block 1");
+
+    Ok(())
+}

crates/ethereum/node/tests/e2e/main.rs

@@ -1,6 +1,7 @@
 #![allow(missing_docs)]
 
 mod blobs;
+mod custom_genesis;
 mod dev;
 mod eth;
 mod p2p;

crates/ethereum/node/tests/it/main.rs

@@ -2,5 +2,6 @@
 
 mod builder;
 mod exex;
+mod testing;
 
 const fn main() {}

crates/ethereum/node/tests/it/testing.rs

@@ -0,0 +1,85 @@
+//! E2E tests for the testing RPC namespace.
+
+use alloy_primitives::{Address, B256};
+use alloy_rpc_types_engine::ExecutionPayloadEnvelopeV4;
+use jsonrpsee_core::client::ClientT;
+use reth_db::test_utils::create_test_rw_db;
+use reth_ethereum_engine_primitives::EthPayloadAttributes;
+use reth_node_builder::{NodeBuilder, NodeConfig};
+use reth_node_core::{
+    args::DatadirArgs,
+    dirs::{DataDirPath, MaybePlatformPath},
+};
+use reth_node_ethereum::{node::EthereumAddOns, EthereumNode};
+use reth_rpc_api::TestingBuildBlockRequestV1;
+use reth_rpc_server_types::{RethRpcModule, RpcModuleSelection};
+use reth_tasks::TaskManager;
+use std::str::FromStr;
+use tempfile::tempdir;
+use tokio::sync::oneshot;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn testing_rpc_build_block_works() -> eyre::Result<()> {
+    let tasks = TaskManager::current();
+    let mut rpc_args = reth_node_core::args::RpcServerArgs::default().with_http();
+    rpc_args.http_api = Some(RpcModuleSelection::from_iter([RethRpcModule::Testing]));
+    let tempdir = tempdir().expect("temp datadir");
+    let datadir_args = DatadirArgs {
+        datadir: MaybePlatformPath::<DataDirPath>::from_str(tempdir.path().to_str().unwrap())
+            .expect("valid datadir"),
+        static_files_path: Some(tempdir.path().join("static")),
+        rocksdb_path: Some(tempdir.path().join("rocksdb")),
+    };
+    let config = NodeConfig::test().with_datadir_args(datadir_args).with_rpc(rpc_args);
+    let db = create_test_rw_db();
+
+    let (tx, rx): (
+        oneshot::Sender<eyre::Result<ExecutionPayloadEnvelopeV4>>,
+        oneshot::Receiver<eyre::Result<ExecutionPayloadEnvelopeV4>>,
+    ) = oneshot::channel();
+
+    let builder = NodeBuilder::new(config)
+        .with_database(db)
+        .with_launch_context(tasks.executor())
+        .with_types::<EthereumNode>()
+        .with_components(EthereumNode::components())
+        .with_add_ons(EthereumAddOns::default())
+        .on_rpc_started(move |ctx, handles| {
+            let Some(client) = handles.rpc.http_client() else { return Ok(()) };
+
+            let chain = ctx.config().chain.clone();
+            let parent_block_hash = chain.genesis_hash();
+            let payload_attributes = EthPayloadAttributes {
+                timestamp: chain.genesis().timestamp + 1,
+                prev_randao: B256::ZERO,
+                suggested_fee_recipient: Address::ZERO,
+                withdrawals: None,
+                parent_beacon_block_root: None,
+            };
+
+            let request = TestingBuildBlockRequestV1 {
+                parent_block_hash,
+                payload_attributes,
+                transactions: vec![],
+                extra_data: None,
+            };
+
+            tokio::spawn(async move {
+                let res: eyre::Result<ExecutionPayloadEnvelopeV4> =
+                    client.request("testing_buildBlockV1", [request]).await.map_err(Into::into);
+                let _ = tx.send(res);
+            });
+
+            Ok(())
+        });
+
+    // Launch the node with the default engine launcher.
+    let launcher = builder.engine_api_launcher();
+    let _node = builder.launch_with(launcher).await?;
+
+    // Wait for the testing RPC call to return.
+    let res = rx.await.expect("testing_buildBlockV1 response");
+    assert!(res.is_ok(), "testing_buildBlockV1 failed: {:?}", res.err());
+
+    Ok(())
+}

crates/ethereum/payload/Cargo.toml

@@ -24,7 +24,7 @@ reth-payload-builder-primitives.workspace = true
 reth-payload-primitives.workspace = true
 reth-basic-payload-builder.workspace = true
 reth-evm.workspace = true
-reth-evm-ethereum.workspace = true
+reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-errors.workspace = true
 reth-chainspec.workspace = true
 reth-payload-validator.workspace = true

crates/ethereum/payload/src/config.rs

@@ -1,4 +1,5 @@
 use alloy_eips::eip1559::ETHEREUM_BLOCK_GAS_LIMIT_30M;
+use alloy_primitives::Bytes;
 use reth_primitives_traits::constants::GAS_LIMIT_BOUND_DIVISOR;
 
 /// Settings for the Ethereum builder.
@@ -13,6 +14,8 @@ pub struct EthereumBuilderConfig {
     ///
     /// If `None`, defaults to the protocol maximum.
     pub max_blobs_per_block: Option<u64>,
+    /// Extra data for built blocks.
+    pub extra_data: Bytes,
 }
 
 impl Default for EthereumBuilderConfig {
@@ -28,6 +31,7 @@ impl EthereumBuilderConfig {
             desired_gas_limit: ETHEREUM_BLOCK_GAS_LIMIT_30M,
             await_payload_on_missing: true,
             max_blobs_per_block: None,
+            extra_data: Bytes::new(),
         }
     }
 
@@ -49,6 +53,12 @@ impl EthereumBuilderConfig {
         self.max_blobs_per_block = max_blobs_per_block;
         self
     }
+
+    /// Set the extra data for built blocks.
+    pub fn with_extra_data(mut self, extra_data: Bytes) -> Self {
+        self.extra_data = extra_data;
+        self
+    }
 }
 
 impl EthereumBuilderConfig {

crates/ethereum/payload/src/lib.rs

@@ -168,6 +168,7 @@ where
                 gas_limit: builder_config.gas_limit(parent_header.gas_limit),
                 parent_beacon_block_root: attributes.parent_beacon_block_root(),
                 withdrawals: Some(attributes.withdrawals().clone()),
+                extra_data: builder_config.extra_data,
             },
         )
         .map_err(PayloadBuilderError::other)?;

crates/ethereum/reth/Cargo.toml

@@ -79,7 +79,9 @@ arbitrary = [
     "alloy-rpc-types-engine?/arbitrary",
     "reth-codecs?/arbitrary",
 ]
-
+keccak-cache-global = [
+    "reth-node-ethereum?/keccak-cache-global",
+]
 test-utils = [
     "reth-chainspec/test-utils",
     "reth-consensus?/test-utils",

crates/evm/evm/Cargo.toml

@@ -32,6 +32,7 @@ auto_impl.workspace = true
 derive_more.workspace = true
 futures-util.workspace = true
 metrics = { workspace = true, optional = true }
+rayon = { workspace = true, optional = true }
 
 [dev-dependencies]
 reth-ethereum-primitives.workspace = true
@@ -40,6 +41,7 @@ reth-ethereum-forks.workspace = true
 [features]
 default = ["std"]
 std = [
+    "dep:rayon",
     "reth-primitives-traits/std",
     "alloy-eips/std",
     "alloy-primitives/std",

crates/evm/evm/src/engine.rs

@@ -1,4 +1,5 @@
 use crate::{execute::ExecutableTxFor, ConfigureEvm, EvmEnvFor, ExecutionCtxFor};
+use rayon::prelude::*;
 
 /// [`ConfigureEvm`] extension providing methods for executing payloads.
 pub trait ConfigureEngineEvm<ExecutionData>: ConfigureEvm {
@@ -21,7 +22,7 @@ pub trait ConfigureEngineEvm<ExecutionData>: ConfigureEvm {
 /// A helper trait representing a pair of a "raw" transactions iterator and a closure that can be
 /// used to convert them to an executable transaction. This tuple is used in the engine to
 /// parallelize heavy work like decoding or recovery.
-pub trait ExecutableTxTuple: Into<(Self::Iter, Self::Convert)> + Send + 'static {
+pub trait ExecutableTxTuple: Into<(Self::IntoIter, Self::Convert)> + Send + 'static {
     /// Raw transaction that can be converted to an [`ExecutableTxTuple::Tx`]
     ///
     /// This can be any type that can be converted to an [`ExecutableTxTuple::Tx`]. For example,
@@ -32,8 +33,10 @@ pub trait ExecutableTxTuple: Into<(Self::Iter, Self::Convert)> + Send + 'static
     /// Errors that may occur while recovering or decoding transactions.
     type Error: core::error::Error + Send + Sync + 'static;
 
-    /// Iterator over [`ExecutableTxTuple::Tx`]
-    type Iter: Iterator<Item = Self::RawTx> + Send + 'static;
+    /// Iterator over [`ExecutableTxTuple::Tx`].
+    type IntoIter: IntoParallelIterator<Item = Self::RawTx, Iter: IndexedParallelIterator>
+        + Send
+        + 'static;
     /// Closure that can be used to convert a [`ExecutableTxTuple::RawTx`] to a
     /// [`ExecutableTxTuple::Tx`]. This might involve heavy work like decoding or recovery
     /// and will be parallelized in the engine.
@@ -45,14 +48,14 @@ where
     RawTx: Send + Sync + 'static,
     Tx: Clone + Send + Sync + 'static,
     Err: core::error::Error + Send + Sync + 'static,
-    I: Iterator<Item = RawTx> + Send + 'static,
+    I: IntoParallelIterator<Item = RawTx, Iter: IndexedParallelIterator> + Send + 'static,
     F: Fn(RawTx) -> Result<Tx, Err> + Send + Sync + 'static,
 {
     type RawTx = RawTx;
     type Tx = Tx;
     type Error = Err;
 
-    type Iter = I;
+    type IntoIter = I;
     type Convert = F;
 }
 

crates/evm/evm/src/lib.rs

@@ -28,7 +28,7 @@ use alloy_evm::{
     block::{BlockExecutorFactory, BlockExecutorFor},
     precompiles::PrecompilesMap,
 };
-use alloy_primitives::{Address, B256};
+use alloy_primitives::{Address, Bytes, B256};
 use core::{error::Error, fmt::Debug};
 use execute::{BasicBlockExecutor, BlockAssembler, BlockBuilder};
 use reth_execution_errors::BlockExecutionError;
@@ -44,8 +44,10 @@ pub mod execute;
 mod aliases;
 pub use aliases::*;
 
+#[cfg(feature = "std")]
 mod engine;
-pub use engine::{ConfigureEngineEvm, ExecutableTxIterator};
+#[cfg(feature = "std")]
+pub use engine::{ConfigureEngineEvm, ExecutableTxIterator, ExecutableTxTuple};
 
 #[cfg(feature = "metrics")]
 pub mod metrics;
@@ -501,6 +503,8 @@ pub struct NextBlockEnvAttributes {
     pub parent_beacon_block_root: Option<B256>,
     /// Withdrawals
     pub withdrawals: Option<Withdrawals>,
+    /// Optional extra data.
+    pub extra_data: Bytes,
 }
 
 /// Abstraction over transaction environment.

crates/exex/test-utils/src/lib.rs

@@ -20,7 +20,9 @@ use futures_util::FutureExt;
 use reth_chainspec::{ChainSpec, MAINNET};
 use reth_consensus::test_utils::TestConsensus;
 use reth_db::{
-    test_utils::{create_test_rw_db, create_test_static_files_dir, TempDatabase},
+    test_utils::{
+        create_test_rocksdb_dir, create_test_rw_db, create_test_static_files_dir, TempDatabase,
+    },
     DatabaseEnv,
 };
 use reth_db_common::init::init_genesis;
@@ -50,7 +52,7 @@ use reth_node_ethereum::{
 use reth_payload_builder::noop::NoopPayloadBuilderService;
 use reth_primitives_traits::{Block as _, RecoveredBlock};
 use reth_provider::{
-    providers::{BlockchainProvider, StaticFileProvider},
+    providers::{BlockchainProvider, RocksDBProvider, StaticFileProvider},
     BlockReader, EthStorage, ProviderFactory,
 };
 use reth_tasks::TaskManager;
@@ -239,11 +241,13 @@ pub async fn test_exex_context_with_chain_spec(
     let consensus = Arc::new(TestConsensus::default());
 
     let (static_dir, _) = create_test_static_files_dir();
+    let (rocksdb_dir, _) = create_test_rocksdb_dir();
     let db = create_test_rw_db();
     let provider_factory = ProviderFactory::<NodeTypesWithDBAdapter<TestNode, _>>::new(
         db,
         chain_spec.clone(),
         StaticFileProvider::read_write(static_dir.keep()).expect("static file provider"),
+        RocksDBProvider::builder(rocksdb_dir.keep()).build().unwrap(),
     )?;
 
     let genesis_hash = init_genesis(&provider_factory)?;

crates/net/discv4/src/config.rs

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

crates/net/discv4/src/lib.rs

@@ -625,10 +625,13 @@ impl Discv4Service {
         self.lookup_interval = tokio::time::interval(duration);
     }
 
-    /// Sets the external Ip to the configured external IP if [`NatResolver::ExternalIp`].
+    /// Sets the external Ip to the configured external IP if [`NatResolver::ExternalIp`] or
+    /// [`NatResolver::ExternalAddr`]. In the case of [`NatResolver::ExternalAddr`], it will return
+    /// the first IP address found for the domain associated with the discv4 UDP port.
     fn resolve_external_ip(&mut self) {
         if let Some(r) = &self.resolve_external_ip_interval &&
-            let Some(external_ip) = r.resolver().as_external_ip()
+            let Some(external_ip) =
+                r.resolver().clone().as_external_ip(self.local_node_record.udp_port)
         {
             self.set_external_ip_addr(external_ip);
         }

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

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

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

@@ -100,6 +100,16 @@ impl Capability {
         Self::eth(EthVersion::Eth68)
     }
 
+    /// Returns the [`EthVersion::Eth69`] capability.
+    pub const fn eth_69() -> Self {
+        Self::eth(EthVersion::Eth69)
+    }
+
+    /// Returns the [`EthVersion::Eth70`] capability.
+    pub const fn eth_70() -> Self {
+        Self::eth(EthVersion::Eth70)
+    }
+
     /// Whether this is eth v66 protocol.
     #[inline]
     pub fn is_eth_v66(&self) -> bool {
@@ -118,10 +128,26 @@ impl Capability {
         self.name == "eth" && self.version == 68
     }
 
+    /// Whether this is eth v69.
+    #[inline]
+    pub fn is_eth_v69(&self) -> bool {
+        self.name == "eth" && self.version == 69
+    }
+
+    /// Whether this is eth v70.
+    #[inline]
+    pub fn is_eth_v70(&self) -> bool {
+        self.name == "eth" && self.version == 70
+    }
+
     /// Whether this is any eth version.
     #[inline]
     pub fn is_eth(&self) -> bool {
-        self.is_eth_v66() || self.is_eth_v67() || self.is_eth_v68()
+        self.is_eth_v66() ||
+            self.is_eth_v67() ||
+            self.is_eth_v68() ||
+            self.is_eth_v69() ||
+            self.is_eth_v70()
     }
 }
 
@@ -141,7 +167,7 @@ impl From<EthVersion> for Capability {
 #[cfg(any(test, feature = "arbitrary"))]
 impl<'a> arbitrary::Arbitrary<'a> for Capability {
     fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
-        let version = u.int_in_range(66..=69)?; // Valid eth protocol versions are 66-69
+        let version = u.int_in_range(66..=70)?; // Valid eth protocol versions are 66-70
                                                 // Only generate valid eth protocol name for now since it's the only supported protocol
         Ok(Self::new_static("eth", version))
     }
@@ -155,6 +181,8 @@ pub struct Capabilities {
     eth_66: bool,
     eth_67: bool,
     eth_68: bool,
+    eth_69: bool,
+    eth_70: bool,
 }
 
 impl Capabilities {
@@ -164,6 +192,8 @@ impl Capabilities {
             eth_66: value.iter().any(Capability::is_eth_v66),
             eth_67: value.iter().any(Capability::is_eth_v67),
             eth_68: value.iter().any(Capability::is_eth_v68),
+            eth_69: value.iter().any(Capability::is_eth_v69),
+            eth_70: value.iter().any(Capability::is_eth_v70),
             inner: value,
         }
     }
@@ -182,7 +212,7 @@ impl Capabilities {
     /// Whether the peer supports `eth` sub-protocol.
     #[inline]
     pub const fn supports_eth(&self) -> bool {
-        self.eth_68 || self.eth_67 || self.eth_66
+        self.eth_70 || self.eth_69 || self.eth_68 || self.eth_67 || self.eth_66
     }
 
     /// Whether this peer supports eth v66 protocol.
@@ -202,6 +232,18 @@ impl Capabilities {
     pub const fn supports_eth_v68(&self) -> bool {
         self.eth_68
     }
+
+    /// Whether this peer supports eth v69 protocol.
+    #[inline]
+    pub const fn supports_eth_v69(&self) -> bool {
+        self.eth_69
+    }
+
+    /// Whether this peer supports eth v70 protocol.
+    #[inline]
+    pub const fn supports_eth_v70(&self) -> bool {
+        self.eth_70
+    }
 }
 
 impl From<Vec<Capability>> for Capabilities {
@@ -224,6 +266,8 @@ impl Decodable for Capabilities {
             eth_66: inner.iter().any(Capability::is_eth_v66),
             eth_67: inner.iter().any(Capability::is_eth_v67),
             eth_68: inner.iter().any(Capability::is_eth_v68),
+            eth_69: inner.iter().any(Capability::is_eth_v69),
+            eth_70: inner.iter().any(Capability::is_eth_v70),
             inner,
         })
     }

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

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

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

@@ -17,6 +17,42 @@ pub struct GetReceipts(
     pub Vec<B256>,
 );
 
+/// Eth/70 `GetReceipts` request payload that supports partial receipt queries.
+///
+/// When used with eth/70, the request id is carried by the surrounding
+/// [`crate::message::RequestPair`], and the on-wire shape is the flattened list
+/// `firstBlockReceiptIndex, [blockhash₁, ...]`.
+///
+/// See also [eip-7975](https://eips.ethereum.org/EIPS/eip-7975)
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
+pub struct GetReceipts70 {
+    /// Index into the receipts of the first requested block hash.
+    pub first_block_receipt_index: u64,
+    /// The block hashes to request receipts for.
+    pub block_hashes: Vec<B256>,
+}
+
+impl alloy_rlp::Encodable for GetReceipts70 {
+    fn encode(&self, out: &mut dyn alloy_rlp::BufMut) {
+        self.first_block_receipt_index.encode(out);
+        self.block_hashes.encode(out);
+    }
+
+    fn length(&self) -> usize {
+        self.first_block_receipt_index.length() + self.block_hashes.length()
+    }
+}
+
+impl alloy_rlp::Decodable for GetReceipts70 {
+    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
+        let first_block_receipt_index = u64::decode(buf)?;
+        let block_hashes = Vec::<B256>::decode(buf)?;
+        Ok(Self { first_block_receipt_index, block_hashes })
+    }
+}
+
 /// The response to [`GetReceipts`], containing receipt lists that correspond to each block
 /// requested.
 #[derive(Clone, Debug, PartialEq, Eq, Default)]
@@ -58,7 +94,13 @@ pub struct Receipts69<T = Receipt>(pub Vec<Vec<T>>);
 impl<T: TxReceipt> Receipts69<T> {
     /// Encodes all receipts with the bloom filter.
     ///
-    /// Note: This is an expensive operation that recalculates the bloom for each receipt.
+    /// Eth/69 omits bloom filters on the wire, while some internal callers
+    /// (and legacy APIs) still operate on [`Receipts`] with
+    /// [`ReceiptWithBloom`]. This helper reconstructs the bloom locally from
+    /// each receipt's logs so the older API can be used on top of eth/69 data.
+    ///
+    /// Note: This is an expensive operation that recalculates the bloom for
+    /// every receipt.
     pub fn into_with_bloom(self) -> Receipts<T> {
         Receipts(
             self.0
@@ -75,6 +117,68 @@ impl<T: TxReceipt> From<Receipts69<T>> for Receipts<T> {
     }
 }
 
+/// Eth/70 `Receipts` response payload.
+///
+/// This is used in conjunction with [`crate::message::RequestPair`] to encode the full wire
+/// message `[request-id, lastBlockIncomplete, [[receipt₁, receipt₂], ...]]`.
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
+pub struct Receipts70<T = Receipt> {
+    /// Whether the receipts list for the last block is incomplete.
+    pub last_block_incomplete: bool,
+    /// Receipts grouped by block.
+    pub receipts: Vec<Vec<T>>,
+}
+
+impl<T> alloy_rlp::Encodable for Receipts70<T>
+where
+    T: alloy_rlp::Encodable,
+{
+    fn encode(&self, out: &mut dyn alloy_rlp::BufMut) {
+        self.last_block_incomplete.encode(out);
+        self.receipts.encode(out);
+    }
+
+    fn length(&self) -> usize {
+        self.last_block_incomplete.length() + self.receipts.length()
+    }
+}
+
+impl<T> alloy_rlp::Decodable for Receipts70<T>
+where
+    T: alloy_rlp::Decodable,
+{
+    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
+        let last_block_incomplete = bool::decode(buf)?;
+        let receipts = Vec::<Vec<T>>::decode(buf)?;
+        Ok(Self { last_block_incomplete, receipts })
+    }
+}
+
+impl<T: TxReceipt> Receipts70<T> {
+    /// Encodes all receipts with the bloom filter.
+    ///
+    /// Just like eth/69, eth/70 does not transmit bloom filters over the wire.
+    /// When higher layers still expect the older bloom-bearing [`Receipts`]
+    /// type, this helper converts the eth/70 payload into that shape by
+    /// recomputing the bloom locally from the contained receipts.
+    ///
+    /// Note: This is an expensive operation that recalculates the bloom for
+    /// every receipt.
+    pub fn into_with_bloom(self) -> Receipts<T> {
+        // Reuse the eth/69 helper, since both variants carry the same
+        // receipt list shape (only eth/70 adds request metadata).
+        Receipts69(self.receipts).into_with_bloom()
+    }
+}
+
+impl<T: TxReceipt> From<Receipts70<T>> for Receipts<T> {
+    fn from(receipts: Receipts70<T>) -> Self {
+        receipts.into_with_bloom()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -225,4 +329,70 @@ mod tests {
         let encoded = alloy_rlp::encode(&request);
         assert_eq!(encoded, data);
     }
+
+    #[test]
+    fn encode_get_receipts70_inline_shape() {
+        let req = RequestPair {
+            request_id: 1111,
+            message: GetReceipts70 {
+                first_block_receipt_index: 0,
+                block_hashes: vec![
+                    hex!("00000000000000000000000000000000000000000000000000000000deadc0de").into(),
+                    hex!("00000000000000000000000000000000000000000000000000000000feedbeef").into(),
+                ],
+            },
+        };
+
+        let mut out = vec![];
+        req.encode(&mut out);
+
+        let mut buf = out.as_slice();
+        let header = alloy_rlp::Header::decode(&mut buf).unwrap();
+        let payload_start = buf.len();
+        let request_id = u64::decode(&mut buf).unwrap();
+        let first_block_receipt_index = u64::decode(&mut buf).unwrap();
+        let block_hashes = Vec::<B256>::decode(&mut buf).unwrap();
+
+        assert!(buf.is_empty(), "buffer not fully consumed");
+        assert_eq!(request_id, 1111);
+        assert_eq!(first_block_receipt_index, 0);
+        assert_eq!(block_hashes.len(), 2);
+        // ensure payload length matches header
+        assert_eq!(payload_start - buf.len(), header.payload_length);
+
+        let mut buf = out.as_slice();
+        let decoded = RequestPair::<GetReceipts70>::decode(&mut buf).unwrap();
+        assert!(buf.is_empty(), "buffer not fully consumed on decode");
+        assert_eq!(decoded, req);
+    }
+
+    #[test]
+    fn encode_receipts70_inline_shape() {
+        let payload: Receipts70<Receipt> =
+            Receipts70 { last_block_incomplete: true, receipts: vec![vec![Receipt::default()]] };
+
+        let resp = RequestPair { request_id: 7, message: payload };
+
+        let mut out = vec![];
+        resp.encode(&mut out);
+
+        let mut buf = out.as_slice();
+        let header = alloy_rlp::Header::decode(&mut buf).unwrap();
+        let payload_start = buf.len();
+        let request_id = u64::decode(&mut buf).unwrap();
+        let last_block_incomplete = bool::decode(&mut buf).unwrap();
+        let receipts = Vec::<Vec<Receipt>>::decode(&mut buf).unwrap();
+
+        assert!(buf.is_empty(), "buffer not fully consumed");
+        assert_eq!(payload_start - buf.len(), header.payload_length);
+        assert_eq!(request_id, 7);
+        assert!(last_block_incomplete);
+        assert_eq!(receipts.len(), 1);
+        assert_eq!(receipts[0].len(), 1);
+
+        let mut buf = out.as_slice();
+        let decoded = RequestPair::<Receipts70>::decode(&mut buf).unwrap();
+        assert!(buf.is_empty(), "buffer not fully consumed on decode");
+        assert_eq!(decoded, resp);
+    }
 }

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

@@ -13,7 +13,7 @@ use reth_codecs_derive::add_arbitrary_tests;
 /// unsupported fields are stripped out.
 #[derive(Clone, Debug, PartialEq, Eq, Copy)]
 pub struct UnifiedStatus {
-    /// The eth protocol version (e.g. eth/66 to eth/69).
+    /// The eth protocol version (e.g. eth/66 to eth/70).
     pub version: EthVersion,
     /// The chain ID identifying the peer’s network.
     pub chain: Chain,
@@ -157,7 +157,7 @@ impl StatusBuilder {
         self.status
     }
 
-    /// Sets the eth protocol version (e.g., eth/66, eth/69).
+    /// Sets the eth protocol version (e.g., eth/66, eth/70).
     pub const fn version(mut self, version: EthVersion) -> Self {
         self.status.version = version;
         self
@@ -378,8 +378,8 @@ impl Debug for StatusEth69 {
     }
 }
 
-/// `StatusMessage` can store either the Legacy version (with TD) or the
-/// eth/69 version (omits TD).
+/// `StatusMessage` can store either the Legacy version (with TD), or the eth/69+/eth/70 version
+/// (omits TD, includes block range).
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum StatusMessage {
@@ -546,6 +546,24 @@ mod tests {
         assert_eq!(unified_status, roundtripped_unified_status);
     }
 
+    #[test]
+    fn roundtrip_eth70() {
+        let unified_status = UnifiedStatus::builder()
+            .version(EthVersion::Eth70)
+            .chain(Chain::mainnet())
+            .genesis(MAINNET_GENESIS_HASH)
+            .forkid(ForkId { hash: ForkHash([0xb7, 0x15, 0x07, 0x7d]), next: 0 })
+            .blockhash(b256!("0xfeb27336ca7923f8fab3bd617fcb6e75841538f71c1bcfc267d7838489d9e13d"))
+            .total_difficulty(None)
+            .earliest_block(Some(1))
+            .latest_block(Some(2))
+            .build();
+
+        let status_message = unified_status.into_message();
+        let roundtripped_unified_status = UnifiedStatus::from_message(status_message);
+        assert_eq!(unified_status, roundtripped_unified_status);
+    }
+
     #[test]
     fn encode_eth69_status_message() {
         let expected = hex!("f8544501a0d4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3c684b715077d8083ed14f2840112a880a0feb27336ca7923f8fab3bd617fcb6e75841538f71c1bcfc267d7838489d9e13d");

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

@@ -27,6 +27,8 @@ pub enum EthVersion {
     Eth68 = 68,
     /// The `eth` protocol version 69.
     Eth69 = 69,
+    /// The `eth` protocol version 70.
+    Eth70 = 70,
 }
 
 impl EthVersion {
@@ -55,6 +57,11 @@ impl EthVersion {
     pub const fn is_eth69(&self) -> bool {
         matches!(self, Self::Eth69)
     }
+
+    /// Returns true if the version is eth/70
+    pub const fn is_eth70(&self) -> bool {
+        matches!(self, Self::Eth70)
+    }
 }
 
 /// RLP encodes `EthVersion` as a single byte (66-69).
@@ -96,6 +103,7 @@ impl TryFrom<&str> for EthVersion {
             "67" => Ok(Self::Eth67),
             "68" => Ok(Self::Eth68),
             "69" => Ok(Self::Eth69),
+            "70" => Ok(Self::Eth70),
             _ => Err(ParseVersionError(s.to_string())),
         }
     }
@@ -120,6 +128,7 @@ impl TryFrom<u8> for EthVersion {
             67 => Ok(Self::Eth67),
             68 => Ok(Self::Eth68),
             69 => Ok(Self::Eth69),
+            70 => Ok(Self::Eth70),
             _ => Err(ParseVersionError(u.to_string())),
         }
     }
@@ -149,6 +158,7 @@ impl From<EthVersion> for &'static str {
             EthVersion::Eth67 => "67",
             EthVersion::Eth68 => "68",
             EthVersion::Eth69 => "69",
+            EthVersion::Eth70 => "70",
         }
     }
 }
@@ -195,7 +205,7 @@ impl Decodable for ProtocolVersion {
 
 #[cfg(test)]
 mod tests {
-    use super::{EthVersion, ParseVersionError};
+    use super::EthVersion;
     use alloy_rlp::{Decodable, Encodable, Error as RlpError};
     use bytes::BytesMut;
 
@@ -205,7 +215,7 @@ mod tests {
         assert_eq!(EthVersion::Eth67, EthVersion::try_from("67").unwrap());
         assert_eq!(EthVersion::Eth68, EthVersion::try_from("68").unwrap());
         assert_eq!(EthVersion::Eth69, EthVersion::try_from("69").unwrap());
-        assert_eq!(Err(ParseVersionError("70".to_string())), EthVersion::try_from("70"));
+        assert_eq!(EthVersion::Eth70, EthVersion::try_from("70").unwrap());
     }
 
     #[test]
@@ -214,12 +224,18 @@ mod tests {
         assert_eq!(EthVersion::Eth67, "67".parse().unwrap());
         assert_eq!(EthVersion::Eth68, "68".parse().unwrap());
         assert_eq!(EthVersion::Eth69, "69".parse().unwrap());
-        assert_eq!(Err(ParseVersionError("70".to_string())), "70".parse::<EthVersion>());
+        assert_eq!(EthVersion::Eth70, "70".parse().unwrap());
     }
 
     #[test]
     fn test_eth_version_rlp_encode() {
-        let versions = [EthVersion::Eth66, EthVersion::Eth67, EthVersion::Eth68, EthVersion::Eth69];
+        let versions = [
+            EthVersion::Eth66,
+            EthVersion::Eth67,
+            EthVersion::Eth68,
+            EthVersion::Eth69,
+            EthVersion::Eth70,
+        ];
 
         for version in versions {
             let mut encoded = BytesMut::new();
@@ -236,7 +252,7 @@ mod tests {
             (67_u8, Ok(EthVersion::Eth67)),
             (68_u8, Ok(EthVersion::Eth68)),
             (69_u8, Ok(EthVersion::Eth69)),
-            (70_u8, Err(RlpError::Custom("invalid eth version"))),
+            (70_u8, Ok(EthVersion::Eth70)),
             (65_u8, Err(RlpError::Custom("invalid eth version"))),
         ];
 

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

@@ -418,6 +418,8 @@ mod tests {
             Capability::new_static("eth", 66),
             Capability::new_static("eth", 67),
             Capability::new_static("eth", 68),
+            Capability::new_static("eth", 69),
+            Capability::new_static("eth", 70),
         ]
         .into();
 
@@ -425,6 +427,8 @@ mod tests {
         assert!(capabilities.supports_eth_v66());
         assert!(capabilities.supports_eth_v67());
         assert!(capabilities.supports_eth_v68());
+        assert!(capabilities.supports_eth_v69());
+        assert!(capabilities.supports_eth_v70());
     }
 
     #[test]

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

@@ -260,10 +260,11 @@ mod tests {
 
         assert_eq!(hello_encoded.len(), hello.length());
     }
+    //TODO: add test for eth70 here once we have fully support it
 
     #[test]
-    fn test_default_protocols_include_eth69() {
-        // ensure that the default protocol list includes Eth69 as the latest version
+    fn test_default_protocols_still_include_eth69() {
+        // ensure that older eth/69 remains advertised for compatibility
         let secret_key = SecretKey::new(&mut rand_08::thread_rng());
         let id = pk2id(&secret_key.public_key(SECP256K1));
         let hello = HelloMessageWithProtocols::builder(id).build();

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

@@ -101,8 +101,9 @@ where
             .or(Err(P2PStreamError::HandshakeError(P2PHandshakeError::Timeout)))?
             .ok_or(P2PStreamError::HandshakeError(P2PHandshakeError::NoResponse))??;
 
-        // let's check the compressed length first, we will need to check again once confirming
-        // that it contains snappy-compressed data (this will be the case for all non-p2p messages).
+        // Check that the uncompressed message length does not exceed the max payload size.
+        // Note: The first message (Hello/Disconnect) is not snappy compressed. We will check the
+        // decompressed length again for subsequent messages after the handshake.
         if first_message_bytes.len() > MAX_PAYLOAD_SIZE {
             return Err(P2PStreamError::MessageTooBig {
                 message_size: first_message_bytes.len(),

crates/net/nat/src/lib.rs

@@ -19,7 +19,7 @@ pub use net_if::{NetInterfaceError, DEFAULT_NET_IF_NAME};
 use std::{
     fmt,
     future::{poll_fn, Future},
-    net::{AddrParseError, IpAddr},
+    net::{AddrParseError, IpAddr, ToSocketAddrs},
     pin::Pin,
     str::FromStr,
     task::{Context, Poll},
@@ -38,7 +38,7 @@ const EXTERNAL_IP_APIS: &[&str] =
     &["https://ipinfo.io/ip", "https://icanhazip.com", "https://ifconfig.me"];
 
 /// All builtin resolvers.
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Default, Hash)]
+#[derive(Debug, Clone, Eq, PartialEq, Default, Hash)]
 #[cfg_attr(feature = "serde", derive(SerializeDisplay, DeserializeFromStr))]
 pub enum NatResolver {
     /// Resolve with any available resolver.
@@ -50,6 +50,14 @@ pub enum NatResolver {
     PublicIp,
     /// Use the given [`IpAddr`]
     ExternalIp(IpAddr),
+    /// Use the given domain name as the external address to expose to peers.
+    /// This is behaving essentially the same as [`NatResolver::ExternalIp`], but supports domain
+    /// names. Domain names are resolved to IP addresses using the OS's resolver. The first IP
+    /// address found is used.
+    /// This may be useful in docker bridge networks where containers are usually queried by DNS
+    /// instead of direct IP addresses.
+    /// Note: the domain shouldn't include a port number. Only the IP address is resolved.
+    ExternalAddr(String),
     /// Resolve external IP via the network interface.
     NetIf,
     /// Resolve nothing
@@ -62,10 +70,17 @@ impl NatResolver {
         external_addr_with(self).await
     }
 
-    /// Returns the external ip, if it is [`NatResolver::ExternalIp`]
-    pub const fn as_external_ip(self) -> Option<IpAddr> {
+    /// Returns the fixed ip, if it is [`NatResolver::ExternalIp`] or [`NatResolver::ExternalAddr`].
+    ///
+    /// In the case of [`NatResolver::ExternalAddr`], it will return the first IP address found for
+    /// the domain.
+    pub fn as_external_ip(self, port: u16) -> Option<IpAddr> {
         match self {
             Self::ExternalIp(ip) => Some(ip),
+            Self::ExternalAddr(domain) => format!("{domain}:{port}")
+                .to_socket_addrs()
+                .ok()
+                .and_then(|mut addrs| addrs.next().map(|addr| addr.ip())),
             _ => None,
         }
     }
@@ -78,6 +93,7 @@ impl fmt::Display for NatResolver {
             Self::Upnp => f.write_str("upnp"),
             Self::PublicIp => f.write_str("publicip"),
             Self::ExternalIp(ip) => write!(f, "extip:{ip}"),
+            Self::ExternalAddr(domain) => write!(f, "extaddr:{domain}"),
             Self::NetIf => f.write_str("netif"),
             Self::None => f.write_str("none"),
         }
@@ -106,12 +122,15 @@ impl FromStr for NatResolver {
             "publicip" | "public-ip" => Self::PublicIp,
             "netif" => Self::NetIf,
             s => {
-                let Some(ip) = s.strip_prefix("extip:") else {
+                if let Some(ip) = s.strip_prefix("extip:") {
+                    Self::ExternalIp(ip.parse()?)
+                } else if let Some(domain) = s.strip_prefix("extaddr:") {
+                    Self::ExternalAddr(domain.to_string())
+                } else {
                     return Err(ParseNatResolverError::UnknownVariant(format!(
                         "Unknown Nat Resolver: {s}"
-                    )))
-                };
-                Self::ExternalIp(ip.parse()?)
+                    )));
+                }
             }
         };
         Ok(r)
@@ -180,7 +199,7 @@ impl ResolveNatInterval {
     ///    `None` if the attempt was unsuccessful.
     pub fn poll_tick(&mut self, cx: &mut Context<'_>) -> Poll<Option<IpAddr>> {
         if self.interval.poll_tick(cx).is_ready() {
-            self.future = Some(Box::pin(self.resolver.external_addr()));
+            self.future = Some(Box::pin(self.resolver.clone().external_addr()));
         }
 
         if let Some(mut fut) = self.future.take() {
@@ -212,6 +231,9 @@ pub async fn external_addr_with(resolver: NatResolver) -> Option<IpAddr> {
                 );
             })
             .ok(),
+        NatResolver::ExternalAddr(domain) => {
+            domain.to_socket_addrs().ok().and_then(|mut addrs| addrs.next().map(|addr| addr.ip()))
+        }
         NatResolver::None => None,
     }
 }
@@ -245,7 +267,7 @@ async fn resolve_external_ip_url(url: &str) -> Option<IpAddr> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::net::Ipv4Addr;
+    use std::net::{Ipv4Addr, Ipv6Addr};
 
     #[tokio::test]
     #[ignore]
@@ -267,6 +289,18 @@ mod tests {
         dbg!(ip);
     }
 
+    #[test]
+    fn as_external_ip_test() {
+        let resolver = NatResolver::ExternalAddr("localhost".to_string());
+        let ip = resolver.as_external_ip(30303).expect("localhost should be resolvable");
+
+        if ip.is_ipv4() {
+            assert_eq!(ip, IpAddr::V4(Ipv4Addr::LOCALHOST));
+        } else {
+            assert_eq!(ip, IpAddr::V6(Ipv6Addr::LOCALHOST));
+        }
+    }
+
     #[test]
     fn test_from_str() {
         assert_eq!(NatResolver::Any, "any".parse().unwrap());
@@ -275,6 +309,6 @@ mod tests {
         let ip = NatResolver::ExternalIp(IpAddr::V4(Ipv4Addr::UNSPECIFIED));
         let s = "extip:0.0.0.0";
         assert_eq!(ip, s.parse().unwrap());
-        assert_eq!(ip.to_string().as_str(), s);
+        assert_eq!(ip.to_string(), s);
     }
 }

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

@@ -3,8 +3,8 @@
 use reth_eth_wire_types::{
     message::RequestPair, BlockBodies, BlockHeaders, Capabilities, DisconnectReason, EthMessage,
     EthNetworkPrimitives, EthVersion, GetBlockBodies, GetBlockHeaders, GetNodeData,
-    GetPooledTransactions, GetReceipts, NetworkPrimitives, NodeData, PooledTransactions, Receipts,
-    Receipts69, UnifiedStatus,
+    GetPooledTransactions, GetReceipts, GetReceipts70, NetworkPrimitives, NodeData,
+    PooledTransactions, Receipts, Receipts69, Receipts70, UnifiedStatus,
 };
 use reth_ethereum_forks::ForkId;
 use reth_network_p2p::error::{RequestError, RequestResult};
@@ -238,6 +238,15 @@ pub enum PeerRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
         /// The channel to send the response for receipts.
         response: oneshot::Sender<RequestResult<Receipts69<N::Receipt>>>,
     },
+    /// Requests receipts from the peer using eth/70 (supports `firstBlockReceiptIndex`).
+    ///
+    /// The response should be sent through the channel.
+    GetReceipts70 {
+        /// The request for receipts.
+        request: GetReceipts70,
+        /// The channel to send the response for receipts.
+        response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
+    },
 }
 
 // === impl PeerRequest ===
@@ -257,6 +266,7 @@ impl<N: NetworkPrimitives> PeerRequest<N> {
             Self::GetNodeData { response, .. } => response.send(Err(err)).ok(),
             Self::GetReceipts { response, .. } => response.send(Err(err)).ok(),
             Self::GetReceipts69 { response, .. } => response.send(Err(err)).ok(),
+            Self::GetReceipts70 { response, .. } => response.send(Err(err)).ok(),
         };
     }
 
@@ -281,6 +291,9 @@ impl<N: NetworkPrimitives> PeerRequest<N> {
             Self::GetReceipts { request, .. } | Self::GetReceipts69 { request, .. } => {
                 EthMessage::GetReceipts(RequestPair { request_id, message: request.clone() })
             }
+            Self::GetReceipts70 { request, .. } => {
+                EthMessage::GetReceipts70(RequestPair { request_id, message: request.clone() })
+            }
         }
     }
 

crates/net/network/Cargo.toml

@@ -66,6 +66,7 @@ tracing.workspace = true
 rustc-hash.workspace = true
 thiserror.workspace = true
 parking_lot.workspace = true
+rayon.workspace = true
 rand.workspace = true
 rand_08.workspace = true
 secp256k1 = { workspace = true, features = ["global-context", "std", "recovery"] }

crates/net/network/src/config.rs

@@ -433,7 +433,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
     pub fn external_ip_resolver(mut self, resolver: NatResolver) -> Self {
         self.discovery_v4_builder
             .get_or_insert_with(Discv4Config::builder)
-            .external_ip_resolver(Some(resolver));
+            .external_ip_resolver(Some(resolver.clone()));
         self.nat = Some(resolver);
         self
     }
@@ -484,7 +484,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
     }
 
     // Disable nat
-    pub const fn disable_nat(mut self) -> Self {
+    pub fn disable_nat(mut self) -> Self {
         self.nat = None;
         self
     }
@@ -579,7 +579,7 @@ impl<N: NetworkPrimitives> NetworkConfigBuilder<N> {
     }
 
     /// Sets the NAT resolver for external IP.
-    pub const fn add_nat(mut self, nat: Option<NatResolver>) -> Self {
+    pub fn add_nat(mut self, nat: Option<NatResolver>) -> Self {
         self.nat = nat;
         self
     }