fix: allow smaller header size

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

--- a/.github/workflows/compact.yml
+++ b/.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
--- a/.github/workflows/e2e.yml
+++ b/.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:
--- a/.github/workflows/hive.yml
+++ b/.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
--- a/.github/workflows/integration.yml
+++ b/.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
--- a/.github/workflows/kurtosis-op.yml
+++ b/.github/workflows/kurtosis-op.yml
@@ -41,7 +41,7 @@ jobs:
          fetch-depth: 0

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

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

      - name: Upload reth image
        id: upload
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
        with:
          name: artifacts
          path: ./artifacts
--- a/.github/workflows/release-dist.yml
+++ b/.github/workflows/release-dist.yml
@@ -12,7 +12,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Update Homebrew formula
-        uses: dawidd6/action-homebrew-bump-formula@v6
+        uses: dawidd6/action-homebrew-bump-formula@v7
        with:
          token: ${{ secrets.HOMEBREW }}
          no_fork: true
--- a/.github/workflows/release.yml
+++ b/.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: |
--- a/.github/workflows/reproducible-build.yml
+++ b/.github/workflows/reproducible-build.yml
@@ -42,7 +42,7 @@ jobs:
          echo "Binaries SHA256 on ${{ matrix.machine }}: $(cat checksum.sha256)"

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

      - name: Create Pull Request
-        uses: peter-evans/create-pull-request@v7
+        uses: peter-evans/create-pull-request@v8
        with:
          commit-message: "chore: update superchain config"
          title: "chore: update superchain config"
--- a/.github/workflows/windows.yml
+++ b/.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
--- a/Cargo.lock
+++ b/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",
@@ -3671,6 +3684,17 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "example-custom-rpc-middleware"
+version = "0.0.0"
+dependencies = [
+ "clap",
+ "jsonrpsee",
+ "reth-ethereum",
+ "tower",
+ "tracing",
+]
+
 [[package]]
 name = "example-db-access"
 version = "0.0.0"
@@ -3982,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",
@@ -5451,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",
@@ -6188,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",
@@ -6201,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",
@@ -6227,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",
@@ -6243,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",
@@ -6258,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",
@@ -6268,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",
@@ -6288,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",
@@ -6304,6 +6328,7 @@ dependencies = [
 "ethereum_ssz_derive",
 "op-alloy-consensus",
 "serde",
+ "sha2",
 "snap",
 "thiserror 2.0.17",
 ]
@@ -7185,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"
@@ -7339,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",
@@ -8196,6 +8231,7 @@ name = "reth-engine-tree"
 version = "1.9.3"
 dependencies = [
 "alloy-consensus",
+ "alloy-eip7928",
 "alloy-eips",
 "alloy-evm",
 "alloy-primitives",
@@ -8614,6 +8650,7 @@ dependencies = [
 "derive_more",
 "futures-util",
 "metrics",
+ "rayon",
 "reth-ethereum-forks",
 "reth-ethereum-primitives",
 "reth-execution-errors",
@@ -8919,6 +8956,7 @@ dependencies = [
 "pin-project",
 "rand 0.8.5",
 "rand 0.9.2",
+ "rayon",
 "reth-chainspec",
 "reth-consensus",
 "reth-discv4",
@@ -9220,6 +9258,7 @@ dependencies = [
 "alloy-sol-types",
 "eyre",
 "futures",
+ "jsonrpsee-core",
 "rand 0.9.2",
 "reth-chainspec",
 "reth-db",
@@ -9255,6 +9294,7 @@ dependencies = [
 "serde",
 "serde_json",
 "similar-asserts",
+ "tempfile",
 "tokio",
 ]

@@ -10145,6 +10185,7 @@ dependencies = [
 "reth-db-api",
 "reth-engine-primitives",
 "reth-errors",
+ "reth-ethereum-engine-primitives",
 "reth-ethereum-primitives",
 "reth-evm",
 "reth-evm-ethereum",
@@ -10207,6 +10248,9 @@ dependencies = [
 "reth-network-peers",
 "reth-rpc-eth-api",
 "reth-trie-common",
+ "serde",
+ "serde_json",
+ "tokio",
 ]

 [[package]]
@@ -10271,6 +10315,7 @@ dependencies = [
 "reth-rpc-server-types",
 "reth-storage-api",
 "reth-tasks",
+ "reth-tokio-util",
 "reth-tracing",
 "reth-transaction-pool",
 "serde",
@@ -10862,6 +10907,7 @@ dependencies = [
 "pretty_assertions",
 "proptest",
 "proptest-arbitrary-interop",
+ "rand 0.9.2",
 "reth-ethereum-primitives",
 "reth-execution-errors",
 "reth-metrics",
@@ -10919,7 +10965,6 @@ dependencies = [
 "alloy-rlp",
 "proptest",
 "proptest-arbitrary-interop",
- "rayon",
 "reth-chainspec",
 "reth-db",
 "reth-db-api",
@@ -11165,9 +11210,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",
@@ -12049,9 +12094,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"
@@ -12306,9 +12351,9 @@ dependencies = [

 [[package]]
 name = "syn-solidity"
-version = "1.4.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff790eb176cc81bb8936aed0f7b9f14fc4670069a2d371b3e3b0ecce908b2cb3"
+checksum = "60ceeb7c95a4536de0c0e1649bd98d1a72a4bb9590b1f3e45a8a0bfdb7c188c0"
 dependencies = [
 "paste",
 "proc-macro2",
@@ -12873,9 +12918,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",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -153,6 +153,7 @@ members = [
    "examples/custom-node-components/",
    "examples/custom-payload-builder/",
    "examples/custom-rlpx-subprotocol",
+    "examples/custom-rpc-middleware",
    "examples/custom-node",
    "examples/db-access",
    "examples/engine-api-access",
@@ -375,11 +376,11 @@ reth-era-utils = { path = "crates/era-utils" }
 reth-errors = { path = "crates/errors" }
 reth-eth-wire = { path = "crates/net/eth-wire" }
 reth-eth-wire-types = { path = "crates/net/eth-wire-types" }
+reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-cli = { path = "crates/ethereum/cli", default-features = false }
 reth-ethereum-consensus = { path = "crates/ethereum/consensus", default-features = false }
 reth-ethereum-engine-primitives = { path = "crates/ethereum/engine-primitives", default-features = false }
 reth-ethereum-forks = { path = "crates/ethereum/hardforks", default-features = false }
-reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-primitives = { path = "crates/ethereum/primitives", default-features = false }
 reth-ethereum = { path = "crates/ethereum/reth" }
 reth-etl = { path = "crates/etl" }
@@ -480,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"
@@ -524,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
--- a/2
+++ b/2
@@ -521,5 +521,3 @@ pr:
 	make update-book-cli && \
 	cargo docs --document-private-items && \
 	make test
-
-check-features:
--- a/bin/reth-bench/README.md
+++ b/bin/reth-bench/README.md
@@ -80,7 +80,7 @@ RUSTFLAGS="-C target-cpu=native" cargo build --profile profiling --no-default-fe
 ### Run the Benchmark:
 First, start the reth node. Here is an example that runs `reth` compiled with the `profiling` profile, runs `samply`, and configures `reth` to run with metrics enabled:
 ```bash
-samply record -p 3001 target/profiling/reth node --metrics localhost:9001 --authrpc.jwt-secret <jwt_file_path>
+samply record -p 3001 target/profiling/reth node --metrics localhost:9001 --authrpc.jwtsecret <jwt_file_path>
 ```

 ```bash
@@ -143,5 +143,5 @@ To reproduce the benchmark, first re-set the node to the block that the benchmar
 - **RPC Configuration**: The RPC endpoints should be accessible and configured correctly, specifically the RPC endpoint must support `eth_getBlockByNumber` and support fetching full transactions. The benchmark will make one RPC query per block as fast as possible, so ensure the RPC endpoint does not rate limit or block requests after a certain volume.
 - **Reproducibility**: Ensure that the node is at the same state before attempting to retry a benchmark. The `new-payload-fcu` command specifically will commit to the database, so the node must be rolled back using `reth stage unwind` to reproducibly retry benchmarks.
 - **Profiling tools**: If you are collecting CPU profiles, tools like [`samply`](https://github.com/mstange/samply) and [`perf`](https://perf.wiki.kernel.org/index.php/Main_Page) can be useful for analyzing node performance.
- **Benchmark Data**: `reth-bench` additionally contains a `--benchmark.output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
+- **Benchmark Data**: `reth-bench` additionally contains a `--output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
 - **Platform Information**: To ensure accurate and reproducible benchmarking, document the platform details, including hardware specifications, OS version, and any other relevant information before publishing any benchmarks.
--- a/bin/reth/Cargo.toml
+++ b/bin/reth/Cargo.toml
@@ -81,7 +81,7 @@ backon.workspace = true
 tempfile.workspace = true

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -327,7 +335,7 @@ pub fn validate_against_parent_eip1559_base_fee<ChainSpec: EthChainSpec + Ethere
    Ok(())
 }

-/// Validates the timestamp against the parent to make sure it is in the past.
+/// Validates that the block timestamp is greater than the parent block timestamp.
 #[inline]
 pub fn validate_against_parent_timestamp<H: BlockHeader>(
    header: &H,
--- a/crates/consensus/debug-client/src/providers/rpc.rs
+++ b/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),
        })
--- a/crates/e2e-test-utils/src/setup_import.rs
+++ b/crates/e2e-test-utils/src/setup_import.rs
@@ -110,6 +110,7 @@ pub async fn setup_engine_with_chain_import(
        // Create database path and static files path
        let db_path = datadir.join("db");
        let static_files_path = datadir.join("static_files");
+        let rocksdb_dir_path = datadir.join("rocksdb");

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

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

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

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

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

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

--- a/crates/engine/primitives/src/config.rs
+++ b/crates/engine/primitives/src/config.rs
@@ -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;
--- a/crates/engine/primitives/src/lib.rs
+++ b/crates/engine/primitives/src/lib.rs
@@ -22,7 +22,8 @@ use reth_trie_common::HashedPostState;
 use serde::{de::DeserializeOwned, Serialize};

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

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

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

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

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

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

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

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

        // Spawn a task that processes out-of-order transactions from the task above and sends them
-        // to prewarming and execution tasks.
-        let (prewarm_tx, prewarm_rx) = mpsc::channel();
-        let (execute_tx, execute_rx) = mpsc::channel();
+        // to the execution task in order.
        self.executor.spawn_blocking(move || {
            let mut next_for_execution = 0;
            let mut queue = BTreeMap::new();
-            while let Ok((idx, tx)) = rx.recv() {
-                // only send Ok(_) variants to prewarming task
-                if let Ok(tx) = &tx {
-                    let _ = prewarm_tx.send(tx.clone());
-                }
-
+            while let Ok((idx, tx)) = ooo_rx.recv() {
                if next_for_execution == idx {
                    let _ = execute_tx.send(tx);
                    next_for_execution += 1;
@@ -382,9 +410,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,
@@ -590,18 +624,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()
    }

@@ -631,9 +665,9 @@ impl<Tx, Err> PayloadHandle<Tx, Err> {
 #[derive(Debug)]
 pub(crate) struct CacheTaskHandle {
    /// 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>>,
 }
@@ -1048,19 +1082,17 @@ mod tests {

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

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

        let mut state_hook = handle.state_hook();

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

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

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

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

@@ -982,12 +1024,16 @@ impl MultiProofTask {
    /// This preserves ordering without requeuing onto the channel.
    ///
    /// Returns `true` if done, `false` to continue.
-    fn process_multiproof_message(
+    fn process_multiproof_message<P>(
        &mut self,
        msg: MultiProofMessage,
        ctx: &mut MultiproofBatchCtx,
        batch_metrics: &mut MultiproofBatchMetrics,
-    ) -> bool {
+        provider: &P,
+    ) -> bool
+    where
+        P: AccountReader,
+    {
        match msg {
            // Prefetch proofs: batch consecutive prefetch requests up to target/message limits
            MultiProofMessage::PrefetchProofs(targets) => {
@@ -1045,7 +1091,7 @@ impl MultiProofTask {
                let storage_targets =
                    merged_targets.values().map(|slots| slots.len()).sum::<usize>();
                batch_metrics.prefetch_proofs_requested += self.on_prefetch_proof(merged_targets);
-                debug!(
+                trace!(
                    target: "engine::tree::payload_processor::multiproof",
                    account_targets,
                    storage_targets,
@@ -1135,7 +1181,7 @@ impl MultiProofTask {
                let batch_len = merged_update.len();
                batch_metrics.state_update_proofs_requested +=
                    self.on_state_update(batch_source, merged_update);
-                debug!(
+                trace!(
                    target: "engine::tree::payload_processor::multiproof",
                    ?batch_source,
                    len = batch_len,
@@ -1146,6 +1192,56 @@ impl MultiProofTask {

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

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

            if let Some(msg) = ctx.pending_msg.take() {
-                if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics) {
+                if self.process_multiproof_message(msg, &mut ctx, &mut batch_metrics, &provider) {
                    break 'main;
                }
                continue;
@@ -1271,7 +1370,7 @@ impl MultiProofTask {
                            // Convert ProofResultMessage to SparseTrieUpdate
                            match proof_result.result {
                                Ok(proof_result_data) => {
-                                    debug!(
+                                    trace!(
                                        target: "engine::tree::payload_processor::multiproof",
                                        sequence = proof_result.sequence_number,
                                        total_proofs = batch_metrics.proofs_processed,
@@ -1323,7 +1422,7 @@ impl MultiProofTask {
                        }
                    };

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

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

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

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

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

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

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

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

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

        let mut pending_msg: Option<MultiProofMessage> = None;

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

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

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

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

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

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

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

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

        let mut pending_msg: Option<MultiProofMessage> = None;

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

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

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

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

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

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

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

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

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

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

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

        // Simulate the state-machine loop behavior
@@ -2703,4 +2833,44 @@ mod tests {
            _ => panic!("Prefetch2 was lost!"),
        }
    }
+
+    /// Verifies that BAL messages are processed correctly and generate state updates.
+    #[test]
+    fn test_bal_message_processing() {
+        let test_provider_factory = create_test_provider_factory();
+        let test_provider = test_provider_factory.latest().unwrap();
+        let mut task = create_test_state_root_task(test_provider_factory);
+
+        // Create a simple BAL with one account change
+        let account_address = Address::random();
+        let account_changes = AccountChanges {
+            address: account_address,
+            balance_changes: vec![BalanceChange::new(0, U256::from(1000))],
+            nonce_changes: vec![],
+            code_changes: vec![],
+            storage_changes: vec![],
+            storage_reads: vec![],
+        };
+
+        let bal = Arc::new(vec![account_changes]);
+
+        let mut ctx = MultiproofBatchCtx::new(Instant::now());
+        let mut batch_metrics = MultiproofBatchMetrics::default();
+
+        let should_finish = task.process_multiproof_message(
+            MultiProofMessage::BlockAccessList(bal),
+            &mut ctx,
+            &mut batch_metrics,
+            &test_provider,
+        );
+
+        // BAL should mark updates as finished
+        assert!(ctx.updates_finished_time.is_some());
+
+        // Should have dispatched state update proofs
+        assert!(batch_metrics.state_update_proofs_requested > 0);
+
+        // Should need to wait for the results of those proofs to arrive
+        assert!(!should_finish, "Should continue waiting for proofs");
+    }
 }
--- a/crates/engine/tree/src/tree/payload_processor/prewarm.rs
+++ b/crates/engine/tree/src/tree/payload_processor/prewarm.rs
@@ -29,7 +29,7 @@ use metrics::{Counter, Gauge, Histogram};
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
 use reth_metrics::Metrics;
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{BlockReader, StateProviderFactory, StateReader};
+use reth_provider::{BlockReader, StateProviderBox, StateProviderFactory, StateReader};
 use reth_revm::{database::StateProviderDatabase, db::BundleState, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
@@ -255,31 +255,35 @@ where
            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
+                if new_cache.cache().insert_state(&state).is_err() {
+                    // Clear the cache on error to prevent having a polluted cache
+                    *cached = None;
+                    debug!(target: "engine::caching", "cleared execution cache on update error");
+                    return;
+                }

-            new_cache.update_metrics();
+                new_cache.update_metrics();

-            // Replace the shared cache with the new one; the previous cache (if any) is dropped.
-            *cached = Some(new_cache);
-        });
+                // Replace the shared cache with the new one; the previous cache (if any) is
+                // dropped.
+                *cached = Some(new_cache);
+            });

-        let elapsed = start.elapsed();
-        debug!(target: "engine::caching", parent_hash=?hash, elapsed=?elapsed, "Updated execution cache");
+            let elapsed = start.elapsed();
+            debug!(target: "engine::caching", parent_hash=?hash, elapsed=?elapsed, "Updated execution cache");

-        metrics.cache_saving_duration.set(elapsed.as_secs_f64());
+            metrics.cache_saving_duration.set(elapsed.as_secs_f64());
+        }
    }

    /// Executes the task.
@@ -356,7 +360,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,
@@ -400,10 +404,13 @@ 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);
+        let state_provider: StateProviderBox = if let Some(saved_cache) = saved_cache {
+            let caches = saved_cache.cache().clone();
+            let cache_metrics = saved_cache.metrics().clone();
+            Box::new(CachedStateProvider::new_with_caches(state_provider, caches, cache_metrics))
+        } else {
+            state_provider
+        };

        let state_provider = StateProviderDatabase::new(state_provider);

--- a/crates/engine/tree/src/tree/payload_validator.rs
+++ b/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::{
@@ -220,7 +222,7 @@ where
                    .map_err(NewPayloadError::other)?
                    .into();

-                let iter = Either::Left(iter.into_iter().map(Either::Left));
+                let iter = Either::Left(iter.into_par_iter().map(Either::Left));
                let convert = move |tx| {
                    let Either::Left(tx) = tx else { unreachable!() };
                    convert(tx).map(Either::Left).map_err(Either::Left)
@@ -230,8 +232,9 @@ where
                Ok((iter, Box::new(convert) as Box<dyn Fn(_) -> _ + Send + Sync + 'static>))
            }
            BlockOrPayload::Block(block) => {
-                let iter =
-                    Either::Right(block.body().clone_transactions().into_iter().map(Either::Right));
+                let iter = Either::Right(
+                    block.body().clone_transactions().into_par_iter().map(Either::Right),
+                );
                let convert = move |tx: Either<_, N::SignedTx>| {
                    let Either::Right(tx) = tx else { unreachable!() };
                    tx.try_into_recovered().map(Either::Right).map_err(Either::Right)
@@ -368,7 +371,7 @@ where
            )
            .into())
        };
-        let state_provider = ensure_ok!(provider_builder.build());
+        let mut state_provider = ensure_ok!(provider_builder.build());
        drop(_enter);

        // fetch parent block
@@ -399,6 +402,14 @@ where
        // use prewarming background task
        let txs = self.tx_iterator_for(&input)?;

+        // Extract the BAL, if valid and available
+        let block_access_list = ensure_ok!(input
+            .block_access_list()
+            .transpose()
+            // Eventually gets converted to a `InsertBlockErrorKind::Other`
+            .map_err(Box::<dyn std::error::Error + Send + Sync>::from))
+        .map(Arc::new);
+
        // Spawn the appropriate processor based on strategy
        let mut handle = ensure_ok!(self.spawn_payload_processor(
            env.clone(),
@@ -407,22 +418,24 @@ 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_with_caches(
+                state_provider,
+                caches,
+                cache_metrics,
+            ));
+        };

        // 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
+            let state_provider =
+                InstrumentedStateProvider::from_state_provider(&state_provider, "engine");
+            self.execute_block(&state_provider, env, &input, &mut handle)
        } else {
            self.execute_block(&state_provider, env, &input, &mut handle)
        } {
@@ -775,6 +788,7 @@ where
        parent_hash: B256,
        state: &EngineApiTreeState<N>,
        strategy: StateRootStrategy,
+        block_access_list: Option<Arc<BlockAccessList>>,
    ) -> Result<
        PayloadHandle<
            impl ExecutableTxFor<Evm> + use<N, P, Evm, V, T>,
@@ -804,12 +818,14 @@ where
                    .record(trie_input_start.elapsed().as_secs_f64());

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

                // record prewarming initialization duration
@@ -872,7 +888,7 @@ where
    /// Note: Use state root task only if prefix sets are empty, otherwise proof generation is
    /// too expensive because it requires walking all paths in every proof.
    const fn plan_state_root_computation(&self) -> StateRootStrategy {
-        if self.config.state_root_fallback() || !self.config.has_enough_parallelism() {
+        if self.config.state_root_fallback() {
            StateRootStrategy::Synchronous
        } else if self.config.use_state_root_task() {
            StateRootStrategy::StateRootTask
@@ -1243,4 +1259,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
+    }
 }
--- a/crates/engine/util/Cargo.toml
+++ b/crates/engine/util/Cargo.toml
@@ -16,7 +16,7 @@ reth-primitives-traits.workspace = true
 reth-errors.workspace = true
 reth-chainspec.workspace = true
 reth-fs-util.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-engine-tree.workspace = true
 reth-evm.workspace = true
 reth-revm.workspace = true
--- a/crates/era-utils/src/export.rs
+++ b/crates/era-utils/src/export.rs
@@ -150,6 +150,12 @@ where
        let era1_id = Era1Id::new(&config.network, start_block, block_count as u32)
            .with_hash(historical_root);

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+pub use alloy_evm::EthEvm;
+
 mod config;
 use alloy_evm::eth::spec::EthExecutorSpec;
 pub use config::{revm_spec, revm_spec_by_timestamp_and_block_number};
@@ -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)?;
--- a/crates/ethereum/node/Cargo.toml
+++ b/crates/ethereum/node/Cargo.toml
@@ -24,7 +24,7 @@ reth-provider.workspace = true
 reth-transaction-pool.workspace = true
 reth-network.workspace = true
 reth-evm.workspace = true
-reth-evm-ethereum.workspace = true
+reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-rpc.workspace = true
 reth-rpc-api.workspace = true
 reth-rpc-eth-api.workspace = true
@@ -35,7 +35,7 @@ reth-chainspec.workspace = true
 reth-revm = { workspace = true, features = ["std"] }
 reth-rpc-eth-types.workspace = true
 reth-engine-local.workspace = true
-reth-engine-primitives.workspace = true
+reth-engine-primitives = { workspace = true, features = ["std"] }
 reth-payload-primitives.workspace = true

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

 alloy-primitives.workspace = true
 alloy-provider.workspace = true
@@ -86,6 +88,9 @@ asm-keccak = [
    "reth-node-core/asm-keccak",
    "revm/asm-keccak",
 ]
+keccak-cache-global = [
+    "alloy-primitives/keccak-cache-global",
+]
 js-tracer = [
    "reth-node-builder/js-tracer",
    "reth-rpc/js-tracer",
--- a/crates/ethereum/node/src/node.rs
+++ b/crates/ethereum/node/src/node.rs
@@ -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()))
    }
 }

--- a/crates/ethereum/node/src/payload.rs
+++ b/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()),
        ))
    }
 }
--- a/crates/ethereum/node/tests/it/main.rs
+++ b/crates/ethereum/node/tests/it/main.rs
@@ -2,5 +2,6 @@

 mod builder;
 mod exex;
+mod testing;

 const fn main() {}
--- a/crates/ethereum/node/tests/it/testing.rs
+++ b/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(())
+}
--- a/crates/ethereum/payload/Cargo.toml
+++ b/crates/ethereum/payload/Cargo.toml
@@ -24,7 +24,7 @@ reth-payload-builder-primitives.workspace = true
 reth-payload-primitives.workspace = true
 reth-basic-payload-builder.workspace = true
 reth-evm.workspace = true
-reth-evm-ethereum.workspace = true
+reth-evm-ethereum = { workspace = true, features = ["std"] }
 reth-errors.workspace = true
 reth-chainspec.workspace = true
 reth-payload-validator.workspace = true
--- a/crates/ethereum/payload/src/config.rs
+++ b/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 {
--- a/crates/ethereum/payload/src/lib.rs
+++ b/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)?;
--- a/crates/ethereum/reth/Cargo.toml
+++ b/crates/ethereum/reth/Cargo.toml
@@ -79,7 +79,9 @@ arbitrary = [
    "alloy-rpc-types-engine?/arbitrary",
    "reth-codecs?/arbitrary",
 ]
-
+keccak-cache-global = [
+    "reth-node-ethereum?/keccak-cache-global",
+]
 test-utils = [
    "reth-chainspec/test-utils",
    "reth-consensus?/test-utils",
--- a/crates/evm/evm/Cargo.toml
+++ b/crates/evm/evm/Cargo.toml
@@ -32,6 +32,7 @@ auto_impl.workspace = true
 derive_more.workspace = true
 futures-util.workspace = true
 metrics = { workspace = true, optional = true }
+rayon = { workspace = true, optional = true }

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

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

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

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

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

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

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

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

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

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

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

@@ -141,7 +167,7 @@ impl From<EthVersion> for Capability {
 #[cfg(any(test, feature = "arbitrary"))]
 impl<'a> arbitrary::Arbitrary<'a> for Capability {
    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
-        let version = u.int_in_range(66..=69)?; // Valid eth protocol versions are 66-69
+        let version = u.int_in_range(66..=70)?; // Valid eth protocol versions are 66-70
                                                // Only generate valid eth protocol name for now since it's the only supported protocol
        Ok(Self::new_static("eth", version))
    }
@@ -155,6 +181,8 @@ pub struct Capabilities {
    eth_66: bool,
    eth_67: bool,
    eth_68: bool,
+    eth_69: bool,
+    eth_70: bool,
 }

 impl Capabilities {
@@ -164,6 +192,8 @@ impl Capabilities {
            eth_66: value.iter().any(Capability::is_eth_v66),
            eth_67: value.iter().any(Capability::is_eth_v67),
            eth_68: value.iter().any(Capability::is_eth_v68),
+            eth_69: value.iter().any(Capability::is_eth_v69),
+            eth_70: value.iter().any(Capability::is_eth_v70),
            inner: value,
        }
    }
@@ -182,7 +212,7 @@ impl Capabilities {
    /// Whether the peer supports `eth` sub-protocol.
    #[inline]
    pub const fn supports_eth(&self) -> bool {
-        self.eth_68 || self.eth_67 || self.eth_66
+        self.eth_70 || self.eth_69 || self.eth_68 || self.eth_67 || self.eth_66
    }

    /// Whether this peer supports eth v66 protocol.
@@ -202,6 +232,18 @@ impl Capabilities {
    pub const fn supports_eth_v68(&self) -> bool {
        self.eth_68
    }
+
+    /// Whether this peer supports eth v69 protocol.
+    #[inline]
+    pub const fn supports_eth_v69(&self) -> bool {
+        self.eth_69
+    }
+
+    /// Whether this peer supports eth v70 protocol.
+    #[inline]
+    pub const fn supports_eth_v70(&self) -> bool {
+        self.eth_70
+    }
 }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+    /// Handles partial responses for [`GetReceipts70`] queries.
+    ///
+    /// This will adhere to the soft limit but allow filling the last vec partially.
+    fn on_receipts70_request(
+        &self,
+        _peer_id: PeerId,
+        request: GetReceipts70,
+        response: oneshot::Sender<RequestResult<Receipts70<C::Receipt>>>,
+    ) {
+        self.metrics.eth_receipts_requests_received_total.increment(1);
+
+        let GetReceipts70 { first_block_receipt_index, block_hashes } = request;
+
+        let mut receipts = Vec::new();
+        let mut total_bytes = 0usize;
+        let mut last_block_incomplete = false;
+
+        for (idx, hash) in block_hashes.into_iter().enumerate() {
+            if idx >= MAX_RECEIPTS_SERVE {
+                break
+            }
+
+            let Some(mut block_receipts) =
+                self.client.receipts_by_block(BlockHashOrNumber::Hash(hash)).unwrap_or_default()
+            else {
+                break
+            };
+
+            if idx == 0 && first_block_receipt_index > 0 {
+                let skip = first_block_receipt_index as usize;
+                if skip >= block_receipts.len() {
+                    block_receipts.clear();
+                } else {
+                    block_receipts.drain(0..skip);
+                }
+            }
+
+            let block_size = block_receipts.length();
+
+            if total_bytes + block_size <= SOFT_RESPONSE_LIMIT {
+                total_bytes += block_size;
+                receipts.push(block_receipts);
+                continue;
+            }
+
+            let mut partial_block = Vec::new();
+            for receipt in block_receipts {
+                let receipt_size = receipt.length();
+                if total_bytes + receipt_size > SOFT_RESPONSE_LIMIT {
+                    break;
+                }
+                total_bytes += receipt_size;
+                partial_block.push(receipt);
+            }
+
+            receipts.push(partial_block);
+            last_block_incomplete = true;
+            break;
+        }
+
+        let _ = response.send(Ok(Receipts70 { last_block_incomplete, receipts }));
+    }
+
    #[inline]
    fn get_receipts_response<T, F>(&self, request: GetReceipts, transform_fn: F) -> Vec<Vec<T>>
    where
@@ -285,6 +349,9 @@ where
                    IncomingEthRequest::GetReceipts69 { peer_id, request, response } => {
                        this.on_receipts69_request(peer_id, request, response)
                    }
+                    IncomingEthRequest::GetReceipts70 { peer_id, request, response } => {
+                        this.on_receipts70_request(peer_id, request, response)
+                    }
                }
            },
        );
@@ -359,4 +426,15 @@ pub enum IncomingEthRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
        /// The channel sender for the response containing Receipts69.
        response: oneshot::Sender<RequestResult<Receipts69<N::Receipt>>>,
    },
+    /// Request Receipts from the peer using eth/70.
+    ///
+    /// The response should be sent through the channel.
+    GetReceipts70 {
+        /// The ID of the peer to request receipts from.
+        peer_id: PeerId,
+        /// The specific receipts requested including the `firstBlockReceiptIndex`.
+        request: GetReceipts70,
+        /// The channel sender for the response containing Receipts70.
+        response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
+    },
 }
--- a/crates/net/network/src/manager.rs
+++ b/crates/net/network/src/manager.rs
@@ -532,6 +532,13 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                    response,
                })
            }
+            PeerRequest::GetReceipts70 { request, response } => {
+                self.delegate_eth_request(IncomingEthRequest::GetReceipts70 {
+                    peer_id,
+                    request,
+                    response,
+                })
+            }
            PeerRequest::GetPooledTransactions { request, response } => {
                self.notify_tx_manager(NetworkTransactionEvent::GetPooledTransactions {
                    peer_id,
--- a/crates/net/network/src/message.rs
+++ b/crates/net/network/src/message.rs
@@ -3,7 +3,7 @@
 //! An `RLPx` stream is multiplexed via the prepended message-id of a framed message.
 //! Capabilities are exchanged via the `RLPx` `Hello` message as pairs of `(id, version)`, <https://github.com/ethereum/devp2p/blob/master/rlpx.md#capability-messaging>

-use crate::types::Receipts69;
+use crate::types::{Receipts69, Receipts70};
 use alloy_consensus::{BlockHeader, ReceiptWithBloom};
 use alloy_primitives::{Bytes, B256};
 use futures::FutureExt;
@@ -116,6 +116,11 @@ pub enum PeerResponse<N: NetworkPrimitives = EthNetworkPrimitives> {
        /// The receiver channel for the response to a receipts request.
        response: oneshot::Receiver<RequestResult<Receipts69<N::Receipt>>>,
    },
+    /// Represents a response to a request for receipts using eth/70.
+    Receipts70 {
+        /// The receiver channel for the response to a receipts request.
+        response: oneshot::Receiver<RequestResult<Receipts70<N::Receipt>>>,
+    },
 }

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

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

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

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

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

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

-        // 2. filter out transactions that are invalid or already pending import pre-size to avoid
-        //    reallocations
-        let mut new_txs = Vec::with_capacity(transactions.len());
-        for tx in transactions {
-            // recover transaction
-            let tx = match tx.try_into_recovered() {
-                Ok(tx) => tx,
+        // Remove known and invalid transactions
+        transactions.retain(|tx| {
+            if let Entry::Occupied(mut entry) = self.transactions_by_peers.entry(*tx.tx_hash()) {
+                entry.get_mut().insert(peer_id);
+                return false
+            }
+            if self.bad_imports.contains(tx.tx_hash()) {
+                trace!(target: "net::tx",
+                    peer_id=format!("{peer_id:#}"),
+                    hash=%tx.tx_hash(),
+                    client_version=%peer.client_version,
+                    "received a known bad transaction from peer"
+                );
+                has_bad_transactions = true;
+                return false;
+            }
+            true
+        });
+
+        let txs_len = transactions.len();
+
+        let new_txs = transactions
+            .into_par_iter()
+            .filter_map(|tx| match tx.try_into_recovered() {
+                Ok(tx) => Some(Pool::Transaction::from_pooled(tx)),
                Err(badtx) => {
                    trace!(target: "net::tx",
                        peer_id=format!("{peer_id:#}"),
@@ -1382,37 +1401,17 @@ where
                        client_version=%peer.client_version,
                        "failed ecrecovery for transaction"
                    );
-                    has_bad_transactions = true;
-                    continue
+                    None
                }
-            };
+            })
+            .collect::<Vec<_>>();

-            match self.transactions_by_peers.entry(*tx.tx_hash()) {
-                Entry::Occupied(mut entry) => {
-                    // transaction was already inserted
-                    entry.get_mut().insert(peer_id);
-                }
-                Entry::Vacant(entry) => {
-                    if self.bad_imports.contains(tx.tx_hash()) {
-                        trace!(target: "net::tx",
-                            peer_id=format!("{peer_id:#}"),
-                            hash=%tx.tx_hash(),
-                            client_version=%peer.client_version,
-                            "received a known bad transaction from peer"
-                        );
-                        has_bad_transactions = true;
-                    } else {
-                        // this is a new transaction that should be imported into the pool
+        has_bad_transactions |= new_txs.len() != txs_len;

-                        let pool_transaction = Pool::Transaction::from_pooled(tx);
-                        new_txs.push(pool_transaction);
-
-                        entry.insert(HashSet::from([peer_id]));
-                    }
-                }
-            }
+        // Record the transactions as seen by the peer
+        for tx in &new_txs {
+            self.transactions_by_peers.insert(*tx.hash(), HashSet::from([peer_id]));
        }
-        new_txs.shrink_to_fit();

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

--- a/crates/node/builder/src/builder/mod.rs
+++ b/crates/node/builder/src/builder/mod.rs
@@ -533,6 +533,27 @@ where
    }

    /// Modifies the addons with the given closure.
+    ///
+    /// This method provides access to methods on the addons type that don't have
+    /// direct builder methods. It's useful for advanced configuration scenarios
+    /// where you need to call addon-specific methods.
+    ///
+    /// # Examples
+    ///
+    /// ```rust,ignore
+    /// use tower::layer::util::Identity;
+    ///
+    /// let builder = NodeBuilder::new(config)
+    ///     .with_types::<EthereumNode>()
+    ///     .with_components(EthereumNode::components())
+    ///     .with_add_ons(EthereumAddOns::default())
+    ///     .map_add_ons(|addons| addons.with_rpc_middleware(Identity::default()));
+    /// ```
+    ///
+    /// # See also
+    ///
+    /// - [`NodeAddOns`] trait for available addon types
+    /// - [`crate::NodeBuilderWithComponents::extend_rpc_modules`] for RPC module configuration
    pub fn map_add_ons<F>(self, f: F) -> Self
    where
        F: FnOnce(AO) -> AO,
@@ -579,10 +600,10 @@ where
    ///     .extend_rpc_modules(|ctx| {
    ///         // Access node components, so they can used by the CustomApi
    ///         let pool = ctx.pool().clone();
-    ///         
+    ///
    ///         // Add custom RPC namespace
    ///         ctx.modules.merge_configured(CustomApi { pool }.into_rpc())?;
-    ///         
+    ///
    ///         Ok(())
    ///     })
    ///     .build()?;
@@ -838,8 +859,8 @@ impl<Node: FullNodeTypes> BuilderContext<Node> {
            .request_handler(self.provider().clone())
            .split_with_handle();

-        self.executor.spawn_critical("p2p txpool", Box::pin(txpool));
-        self.executor.spawn_critical("p2p eth request handler", Box::pin(eth));
+        self.executor.spawn_critical_blocking("p2p txpool", Box::pin(txpool));
+        self.executor.spawn_critical_blocking("p2p eth request handler", Box::pin(eth));

        let default_peers_path = self.config().datadir().known_peers();
        let known_peers_file = self.config().network.persistent_peers_file(default_peers_path);
--- a/crates/node/builder/src/builder/states.rs
+++ b/crates/node/builder/src/builder/states.rs
@@ -235,6 +235,27 @@ where
    }

    /// Modifies the addons with the given closure.
+    ///
+    /// This method provides access to methods on the addons type that don't have
+    /// direct builder methods. It's useful for advanced configuration scenarios
+    /// where you need to call addon-specific methods.
+    ///
+    /// # Examples
+    ///
+    /// ```rust,ignore
+    /// use tower::layer::util::Identity;
+    ///
+    /// let builder = NodeBuilder::new(config)
+    ///     .with_types::<EthereumNode>()
+    ///     .with_components(EthereumNode::components())
+    ///     .with_add_ons(EthereumAddOns::default())
+    ///     .map_add_ons(|addons| addons.with_rpc_middleware(Identity::default()));
+    /// ```
+    ///
+    /// # See also
+    ///
+    /// - [`NodeAddOns`] trait for available addon types
+    /// - [`crate::NodeBuilderWithComponents::extend_rpc_modules`] for RPC module configuration
    pub fn map_add_ons<F>(mut self, f: F) -> Self
    where
        F: FnOnce(AO) -> AO,
--- a/crates/node/builder/src/launch/common.rs
+++ b/crates/node/builder/src/launch/common.rs
@@ -65,7 +65,7 @@ use reth_node_metrics::{
    version::VersionInfo,
 };
 use reth_provider::{
-    providers::{NodeTypesForProvider, ProviderNodeTypes, StaticFileProvider},
+    providers::{NodeTypesForProvider, ProviderNodeTypes, RocksDBProvider, StaticFileProvider},
    BlockHashReader, BlockNumReader, ProviderError, ProviderFactory, ProviderResult,
    StageCheckpointReader, StaticFileProviderBuilder, StaticFileProviderFactory,
 };
@@ -485,9 +485,20 @@ where
                .with_blocks_per_file_for_segments(static_files_config.as_blocks_per_file_map())
                .build()?;

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

        // Check for consistency between database and static files. If it fails, it unwinds to
        // the first block that's consistent between database and static files.
--- a/crates/node/builder/src/rpc.rs
+++ b/crates/node/builder/src/rpc.rs
@@ -1010,7 +1010,7 @@ where
            .with_executor(Box::new(node.task_executor().clone()))
            .with_evm_config(node.evm_config().clone())
            .with_consensus(node.consensus().clone())
-            .build_with_auth_server(module_config, engine_api, eth_api);
+            .build_with_auth_server(module_config, engine_api, eth_api, engine_events.clone());

        // in dev mode we generate 20 random dev-signer accounts
        if config.dev.dev {
@@ -1179,6 +1179,7 @@ impl<'a, N: FullNodeComponents<Types: NodeTypes<ChainSpec: Hardforks + EthereumH
            .proof_permits(self.config.proof_permits)
            .gas_oracle_config(self.config.gas_oracle)
            .max_batch_size(self.config.max_batch_size)
+            .max_blocking_io_requests(self.config.max_blocking_io_requests)
            .pending_block_kind(self.config.pending_block_kind)
            .raw_tx_forwarder(self.config.raw_tx_forwarder)
            .evm_memory_limit(self.config.rpc_evm_memory_limit)
@@ -1188,10 +1189,7 @@ impl<'a, N: FullNodeComponents<Types: NodeTypes<ChainSpec: Hardforks + EthereumH
 /// A `EthApi` that knows how to build `eth` namespace API from [`FullNodeComponents`].
 pub trait EthApiBuilder<N: FullNodeComponents>: Default + Send + 'static {
    /// The Ethapi implementation this builder will build.
-    type EthApi: EthApiTypes
-        + FullEthApiServer<Provider = N::Provider, Pool = N::Pool>
-        + Unpin
-        + 'static;
+    type EthApi: FullEthApiServer<Provider = N::Provider, Pool = N::Pool>;

    /// Builds the [`EthApiServer`](reth_rpc_api::eth::EthApiServer) from the given context.
    fn build_eth_api(
--- a/crates/node/core/src/args/datadir_args.rs
+++ b/crates/node/core/src/args/datadir_args.rs
@@ -27,6 +27,10 @@ pub struct DatadirArgs {
        verbatim_doc_comment
    )]
    pub static_files_path: Option<PathBuf>,
+
+    /// The absolute path to store `RocksDB` database in.
+    #[arg(long = "datadir.rocksdb", value_name = "PATH", verbatim_doc_comment)]
+    pub rocksdb_path: Option<PathBuf>,
 }

 impl DatadirArgs {
--- a/crates/node/core/src/args/engine.rs
+++ b/crates/node/core/src/args/engine.rs
@@ -1,13 +1,198 @@
 //! clap [Args](clap::Args) for engine purposes

-use clap::Args;
+use clap::{builder::Resettable, Args};
 use reth_engine_primitives::{TreeConfig, DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE};
+use std::sync::OnceLock;

 use crate::node_config::{
    DEFAULT_CROSS_BLOCK_CACHE_SIZE_MB, DEFAULT_MEMORY_BLOCK_BUFFER_TARGET,
    DEFAULT_PERSISTENCE_THRESHOLD, DEFAULT_RESERVED_CPU_CORES,
 };

+/// Global static engine defaults
+static ENGINE_DEFAULTS: OnceLock<DefaultEngineValues> = OnceLock::new();
+
+/// Default values for engine that can be customized
+///
+/// Global defaults can be set via [`DefaultEngineValues::try_init`].
+#[derive(Debug, Clone)]
+pub struct DefaultEngineValues {
+    persistence_threshold: u64,
+    memory_block_buffer_target: u64,
+    legacy_state_root_task_enabled: bool,
+    state_cache_disabled: bool,
+    prewarming_disabled: bool,
+    parallel_sparse_trie_disabled: bool,
+    state_provider_metrics: bool,
+    cross_block_cache_size: u64,
+    state_root_task_compare_updates: bool,
+    accept_execution_requests_hash: bool,
+    multiproof_chunking_enabled: bool,
+    multiproof_chunk_size: usize,
+    reserved_cpu_cores: usize,
+    precompile_cache_disabled: bool,
+    state_root_fallback: bool,
+    always_process_payload_attributes_on_canonical_head: bool,
+    allow_unwind_canonical_header: bool,
+    storage_worker_count: Option<usize>,
+    account_worker_count: Option<usize>,
+}
+
+impl DefaultEngineValues {
+    /// Initialize the global engine defaults with this configuration
+    pub fn try_init(self) -> Result<(), Self> {
+        ENGINE_DEFAULTS.set(self)
+    }
+
+    /// Get a reference to the global engine defaults
+    pub fn get_global() -> &'static Self {
+        ENGINE_DEFAULTS.get_or_init(Self::default)
+    }
+
+    /// Set the default persistence threshold
+    pub const fn with_persistence_threshold(mut self, v: u64) -> Self {
+        self.persistence_threshold = v;
+        self
+    }
+
+    /// Set the default memory block buffer target
+    pub const fn with_memory_block_buffer_target(mut self, v: u64) -> Self {
+        self.memory_block_buffer_target = v;
+        self
+    }
+
+    /// Set whether to enable legacy state root task by default
+    pub const fn with_legacy_state_root_task_enabled(mut self, v: bool) -> Self {
+        self.legacy_state_root_task_enabled = v;
+        self
+    }
+
+    /// Set whether to disable state cache by default
+    pub const fn with_state_cache_disabled(mut self, v: bool) -> Self {
+        self.state_cache_disabled = v;
+        self
+    }
+
+    /// Set whether to disable prewarming by default
+    pub const fn with_prewarming_disabled(mut self, v: bool) -> Self {
+        self.prewarming_disabled = v;
+        self
+    }
+
+    /// Set whether to disable parallel sparse trie by default
+    pub const fn with_parallel_sparse_trie_disabled(mut self, v: bool) -> Self {
+        self.parallel_sparse_trie_disabled = v;
+        self
+    }
+
+    /// Set whether to enable state provider metrics by default
+    pub const fn with_state_provider_metrics(mut self, v: bool) -> Self {
+        self.state_provider_metrics = v;
+        self
+    }
+
+    /// Set the default cross-block cache size in MB
+    pub const fn with_cross_block_cache_size(mut self, v: u64) -> Self {
+        self.cross_block_cache_size = v;
+        self
+    }
+
+    /// Set whether to compare state root task updates by default
+    pub const fn with_state_root_task_compare_updates(mut self, v: bool) -> Self {
+        self.state_root_task_compare_updates = v;
+        self
+    }
+
+    /// Set whether to accept execution requests hash by default
+    pub const fn with_accept_execution_requests_hash(mut self, v: bool) -> Self {
+        self.accept_execution_requests_hash = v;
+        self
+    }
+
+    /// Set whether to enable multiproof chunking by default
+    pub const fn with_multiproof_chunking_enabled(mut self, v: bool) -> Self {
+        self.multiproof_chunking_enabled = v;
+        self
+    }
+
+    /// Set the default multiproof chunk size
+    pub const fn with_multiproof_chunk_size(mut self, v: usize) -> Self {
+        self.multiproof_chunk_size = v;
+        self
+    }
+
+    /// Set the default number of reserved CPU cores
+    pub const fn with_reserved_cpu_cores(mut self, v: usize) -> Self {
+        self.reserved_cpu_cores = v;
+        self
+    }
+
+    /// Set whether to disable precompile cache by default
+    pub const fn with_precompile_cache_disabled(mut self, v: bool) -> Self {
+        self.precompile_cache_disabled = v;
+        self
+    }
+
+    /// Set whether to enable state root fallback by default
+    pub const fn with_state_root_fallback(mut self, v: bool) -> Self {
+        self.state_root_fallback = v;
+        self
+    }
+
+    /// Set whether to always process payload attributes on canonical head by default
+    pub const fn with_always_process_payload_attributes_on_canonical_head(
+        mut self,
+        v: bool,
+    ) -> Self {
+        self.always_process_payload_attributes_on_canonical_head = v;
+        self
+    }
+
+    /// Set whether to allow unwinding canonical header by default
+    pub const fn with_allow_unwind_canonical_header(mut self, v: bool) -> Self {
+        self.allow_unwind_canonical_header = v;
+        self
+    }
+
+    /// Set the default storage worker count
+    pub const fn with_storage_worker_count(mut self, v: Option<usize>) -> Self {
+        self.storage_worker_count = v;
+        self
+    }
+
+    /// Set the default account worker count
+    pub const fn with_account_worker_count(mut self, v: Option<usize>) -> Self {
+        self.account_worker_count = v;
+        self
+    }
+}
+
+impl Default for DefaultEngineValues {
+    fn default() -> Self {
+        Self {
+            persistence_threshold: DEFAULT_PERSISTENCE_THRESHOLD,
+            memory_block_buffer_target: DEFAULT_MEMORY_BLOCK_BUFFER_TARGET,
+            legacy_state_root_task_enabled: false,
+            state_cache_disabled: false,
+            prewarming_disabled: false,
+            parallel_sparse_trie_disabled: false,
+            state_provider_metrics: false,
+            cross_block_cache_size: DEFAULT_CROSS_BLOCK_CACHE_SIZE_MB,
+            state_root_task_compare_updates: false,
+            accept_execution_requests_hash: false,
+            multiproof_chunking_enabled: true,
+            multiproof_chunk_size: DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE,
+            reserved_cpu_cores: DEFAULT_RESERVED_CPU_CORES,
+            precompile_cache_disabled: false,
+            state_root_fallback: false,
+            always_process_payload_attributes_on_canonical_head: false,
+            allow_unwind_canonical_header: false,
+            storage_worker_count: None,
+            account_worker_count: None,
+        }
+    }
+}
+
 /// Parameters for configuring the engine driver.
 #[derive(Debug, Clone, Args, PartialEq, Eq)]
 #[command(next_help_heading = "Engine")]
@@ -18,15 +203,15 @@ pub struct EngineArgs {
    ///
    /// To persist blocks as fast as the node receives them, set this value to zero. This will
    /// cause more frequent DB writes.
-    #[arg(long = "engine.persistence-threshold", default_value_t = DEFAULT_PERSISTENCE_THRESHOLD)]
+    #[arg(long = "engine.persistence-threshold", default_value_t = DefaultEngineValues::get_global().persistence_threshold)]
    pub persistence_threshold: u64,

    /// Configure the target number of blocks to keep in memory.
-    #[arg(long = "engine.memory-block-buffer-target", default_value_t = DEFAULT_MEMORY_BLOCK_BUFFER_TARGET)]
+    #[arg(long = "engine.memory-block-buffer-target", default_value_t = DefaultEngineValues::get_global().memory_block_buffer_target)]
    pub memory_block_buffer_target: u64,

    /// Enable legacy state root
-    #[arg(long = "engine.legacy-state-root", default_value = "false")]
+    #[arg(long = "engine.legacy-state-root", default_value_t = DefaultEngineValues::get_global().legacy_state_root_task_enabled)]
    pub legacy_state_root_task_enabled: bool,

    /// CAUTION: This CLI flag has no effect anymore, use --engine.disable-caching-and-prewarming
@@ -35,8 +220,12 @@ pub struct EngineArgs {
    #[deprecated]
    pub caching_and_prewarming_enabled: bool,

+    /// Disable state cache
+    #[arg(long = "engine.disable-state-cache", default_value_t = DefaultEngineValues::get_global().state_cache_disabled)]
+    pub state_cache_disabled: bool,
+
    /// Disable parallel prewarming
-    #[arg(long = "engine.disable-prewarming", alias = "engine.disable-caching-and-prewarming")]
+    #[arg(long = "engine.disable-prewarming", alias = "engine.disable-caching-and-prewarming", default_value_t = DefaultEngineValues::get_global().prewarming_disabled)]
    pub prewarming_disabled: bool,

    /// CAUTION: This CLI flag has no effect anymore, use --engine.disable-parallel-sparse-trie
@@ -46,38 +235,38 @@ pub struct EngineArgs {
    pub parallel_sparse_trie_enabled: bool,

    /// Disable the parallel sparse trie in the engine.
-    #[arg(long = "engine.disable-parallel-sparse-trie", default_value = "false")]
+    #[arg(long = "engine.disable-parallel-sparse-trie", default_value_t = DefaultEngineValues::get_global().parallel_sparse_trie_disabled)]
    pub parallel_sparse_trie_disabled: bool,

    /// Enable state provider latency metrics. This allows the engine to collect and report stats
    /// about how long state provider calls took during execution, but this does introduce slight
    /// overhead to state provider calls.
-    #[arg(long = "engine.state-provider-metrics", default_value = "false")]
+    #[arg(long = "engine.state-provider-metrics", default_value_t = DefaultEngineValues::get_global().state_provider_metrics)]
    pub state_provider_metrics: bool,

    /// Configure the size of cross-block cache in megabytes
-    #[arg(long = "engine.cross-block-cache-size", default_value_t = DEFAULT_CROSS_BLOCK_CACHE_SIZE_MB)]
+    #[arg(long = "engine.cross-block-cache-size", default_value_t = DefaultEngineValues::get_global().cross_block_cache_size)]
    pub cross_block_cache_size: u64,

    /// Enable comparing trie updates from the state root task to the trie updates from the regular
    /// state root calculation.
-    #[arg(long = "engine.state-root-task-compare-updates")]
+    #[arg(long = "engine.state-root-task-compare-updates", default_value_t = DefaultEngineValues::get_global().state_root_task_compare_updates)]
    pub state_root_task_compare_updates: bool,

    /// Enables accepting requests hash instead of an array of requests in `engine_newPayloadV4`.
-    #[arg(long = "engine.accept-execution-requests-hash")]
+    #[arg(long = "engine.accept-execution-requests-hash", default_value_t = DefaultEngineValues::get_global().accept_execution_requests_hash)]
    pub accept_execution_requests_hash: bool,

    /// Whether multiproof task should chunk proof targets.
-    #[arg(long = "engine.multiproof-chunking", default_value = "true")]
+    #[arg(long = "engine.multiproof-chunking", default_value_t = DefaultEngineValues::get_global().multiproof_chunking_enabled)]
    pub multiproof_chunking_enabled: bool,

    /// Multiproof task chunk size for proof targets.
-    #[arg(long = "engine.multiproof-chunk-size", default_value_t = DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE)]
+    #[arg(long = "engine.multiproof-chunk-size", default_value_t = DefaultEngineValues::get_global().multiproof_chunk_size)]
    pub multiproof_chunk_size: usize,

    /// Configure the number of reserved CPU cores for non-reth processes
-    #[arg(long = "engine.reserved-cpu-cores", default_value_t = DEFAULT_RESERVED_CPU_CORES)]
+    #[arg(long = "engine.reserved-cpu-cores", default_value_t = DefaultEngineValues::get_global().reserved_cpu_cores)]
    pub reserved_cpu_cores: usize,

    /// CAUTION: This CLI flag has no effect anymore, use --engine.disable-precompile-cache
@@ -87,11 +276,11 @@ pub struct EngineArgs {
    pub precompile_cache_enabled: bool,

    /// Disable precompile cache
-    #[arg(long = "engine.disable-precompile-cache", default_value = "false")]
+    #[arg(long = "engine.disable-precompile-cache", default_value_t = DefaultEngineValues::get_global().precompile_cache_disabled)]
    pub precompile_cache_disabled: bool,

    /// Enable state root fallback, useful for testing
-    #[arg(long = "engine.state-root-fallback", default_value = "false")]
+    #[arg(long = "engine.state-root-fallback", default_value_t = DefaultEngineValues::get_global().state_root_fallback)]
    pub state_root_fallback: bool,

    /// Always process payload attributes and begin a payload build process even if
@@ -101,51 +290,73 @@ pub struct EngineArgs {
    /// Note: This is a no-op on OP Stack.
    #[arg(
        long = "engine.always-process-payload-attributes-on-canonical-head",
-        default_value = "false"
+        default_value_t = DefaultEngineValues::get_global().always_process_payload_attributes_on_canonical_head
    )]
    pub always_process_payload_attributes_on_canonical_head: bool,

    /// Allow unwinding canonical header to ancestor during forkchoice updates.
    /// See `TreeConfig::unwind_canonical_header` for more details.
-    #[arg(long = "engine.allow-unwind-canonical-header", default_value = "false")]
+    #[arg(long = "engine.allow-unwind-canonical-header", default_value_t = DefaultEngineValues::get_global().allow_unwind_canonical_header)]
    pub allow_unwind_canonical_header: bool,

    /// Configure the number of storage proof workers in the Tokio blocking pool.
    /// If not specified, defaults to 2x available parallelism, clamped between 2 and 64.
-    #[arg(long = "engine.storage-worker-count")]
+    #[arg(long = "engine.storage-worker-count", default_value = Resettable::from(DefaultEngineValues::get_global().storage_worker_count.map(|v| v.to_string().into())))]
    pub storage_worker_count: Option<usize>,

    /// Configure the number of account proof workers in the Tokio blocking pool.
    /// If not specified, defaults to the same count as storage workers.
-    #[arg(long = "engine.account-worker-count")]
+    #[arg(long = "engine.account-worker-count", default_value = Resettable::from(DefaultEngineValues::get_global().account_worker_count.map(|v| v.to_string().into())))]
    pub account_worker_count: Option<usize>,
 }

 #[allow(deprecated)]
 impl Default for EngineArgs {
    fn default() -> Self {
+        let DefaultEngineValues {
+            persistence_threshold,
+            memory_block_buffer_target,
+            legacy_state_root_task_enabled,
+            state_cache_disabled,
+            prewarming_disabled,
+            parallel_sparse_trie_disabled,
+            state_provider_metrics,
+            cross_block_cache_size,
+            state_root_task_compare_updates,
+            accept_execution_requests_hash,
+            multiproof_chunking_enabled,
+            multiproof_chunk_size,
+            reserved_cpu_cores,
+            precompile_cache_disabled,
+            state_root_fallback,
+            always_process_payload_attributes_on_canonical_head,
+            allow_unwind_canonical_header,
+            storage_worker_count,
+            account_worker_count,
+        } = DefaultEngineValues::get_global().clone();
        Self {
-            persistence_threshold: DEFAULT_PERSISTENCE_THRESHOLD,
-            memory_block_buffer_target: DEFAULT_MEMORY_BLOCK_BUFFER_TARGET,
-            legacy_state_root_task_enabled: false,
-            state_root_task_compare_updates: false,
+            persistence_threshold,
+            memory_block_buffer_target,
+            legacy_state_root_task_enabled,
+            state_root_task_compare_updates,
            caching_and_prewarming_enabled: true,
-            prewarming_disabled: false,
+            state_cache_disabled,
+            prewarming_disabled,
            parallel_sparse_trie_enabled: true,
-            parallel_sparse_trie_disabled: false,
-            state_provider_metrics: false,
-            cross_block_cache_size: DEFAULT_CROSS_BLOCK_CACHE_SIZE_MB,
-            accept_execution_requests_hash: false,
-            multiproof_chunking_enabled: true,
-            multiproof_chunk_size: DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE,
-            reserved_cpu_cores: DEFAULT_RESERVED_CPU_CORES,
+            parallel_sparse_trie_disabled,
+            state_provider_metrics,
+            cross_block_cache_size,
+            accept_execution_requests_hash,
+            multiproof_chunking_enabled,
+            multiproof_chunk_size,
+            reserved_cpu_cores,
            precompile_cache_enabled: true,
-            precompile_cache_disabled: false,
-            state_root_fallback: false,
-            always_process_payload_attributes_on_canonical_head: false,
-            allow_unwind_canonical_header: false,
-            storage_worker_count: None,
-            account_worker_count: None,
+            precompile_cache_disabled,
+            state_root_fallback,
+            always_process_payload_attributes_on_canonical_head,
+            allow_unwind_canonical_header,
+            storage_worker_count,
+            account_worker_count,
        }
    }
 }
@@ -157,6 +368,7 @@ impl EngineArgs {
            .with_persistence_threshold(self.persistence_threshold)
            .with_memory_block_buffer_target(self.memory_block_buffer_target)
            .with_legacy_state_root(self.legacy_state_root_task_enabled)
+            .without_state_cache(self.state_cache_disabled)
            .without_prewarming(self.prewarming_disabled)
            .with_disable_parallel_sparse_trie(self.parallel_sparse_trie_disabled)
            .with_state_provider_metrics(self.state_provider_metrics)
@@ -202,4 +414,66 @@ mod tests {
        let args = CommandParser::<EngineArgs>::parse_from(["reth"]).args;
        assert_eq!(args, default_args);
    }
+
+    #[test]
+    #[allow(deprecated)]
+    fn engine_args() {
+        let args = EngineArgs {
+            persistence_threshold: 100,
+            memory_block_buffer_target: 50,
+            legacy_state_root_task_enabled: true,
+            caching_and_prewarming_enabled: true,
+            state_cache_disabled: true,
+            prewarming_disabled: true,
+            parallel_sparse_trie_enabled: true,
+            parallel_sparse_trie_disabled: true,
+            state_provider_metrics: true,
+            cross_block_cache_size: 256,
+            state_root_task_compare_updates: true,
+            accept_execution_requests_hash: true,
+            multiproof_chunking_enabled: true,
+            multiproof_chunk_size: 512,
+            reserved_cpu_cores: 4,
+            precompile_cache_enabled: true,
+            precompile_cache_disabled: true,
+            state_root_fallback: true,
+            always_process_payload_attributes_on_canonical_head: true,
+            allow_unwind_canonical_header: true,
+            storage_worker_count: Some(16),
+            account_worker_count: Some(8),
+        };
+
+        let parsed_args = CommandParser::<EngineArgs>::parse_from([
+            "reth",
+            "--engine.persistence-threshold",
+            "100",
+            "--engine.memory-block-buffer-target",
+            "50",
+            "--engine.legacy-state-root",
+            "--engine.disable-state-cache",
+            "--engine.disable-prewarming",
+            "--engine.disable-parallel-sparse-trie",
+            "--engine.state-provider-metrics",
+            "--engine.cross-block-cache-size",
+            "256",
+            "--engine.state-root-task-compare-updates",
+            "--engine.accept-execution-requests-hash",
+            "--engine.multiproof-chunking",
+            "--engine.multiproof-chunk-size",
+            "512",
+            "--engine.reserved-cpu-cores",
+            "4",
+            "--engine.disable-precompile-cache",
+            "--engine.state-root-fallback",
+            "--engine.always-process-payload-attributes-on-canonical-head",
+            "--engine.allow-unwind-canonical-header",
+            "--engine.storage-worker-count",
+            "16",
+            "--engine.account-worker-count",
+            "8",
+        ])
+        .args;
+
+        assert_eq!(parsed_args, args);
+    }
 }
--- a/crates/node/core/src/args/mod.rs
+++ b/crates/node/core/src/args/mod.rs
@@ -6,7 +6,7 @@ pub use network::{DiscoveryArgs, NetworkArgs};

 /// RpcServerArg struct for configuring the RPC
 mod rpc_server;
-pub use rpc_server::RpcServerArgs;
+pub use rpc_server::{DefaultRpcServerArgs, RpcServerArgs};

 /// `RpcStateCacheArgs` struct for configuring RPC state cache
 mod rpc_state_cache;
@@ -66,7 +66,7 @@ pub use benchmark_args::BenchmarkArgs;

 /// EngineArgs for configuring the engine
 mod engine;
-pub use engine::EngineArgs;
+pub use engine::{DefaultEngineValues, EngineArgs};

 /// `RessArgs` for configuring ress subprotocol.
 mod ress_args;
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Matthias Seitz	5c04d1abe1	fix: allow smaller header size	2025-12-16 17:08:53 +01:00
bigbear	4231f4b688	docs: fix incorrect API example in node-components.mdx (#20297 )	2025-12-16 15:09:29 +00:00
Léa Narzis	0b607113dc	refactor(era): make era count in era file name optional (#20292 )	2025-12-16 15:08:43 +00:00
emmmm	be4dc53b92	docs: fix --color auto option description (#20352 )	2025-12-16 15:06:04 +00:00
emmmm	4afb555d06	docs(opstack): document all rollup CLI arguments (#20374 )	2025-12-16 15:04:34 +00:00
Matthias Seitz	ab2ef99458	chore: add keccak-global (#20418 )	2025-12-16 14:59:09 +00:00
Sophia Raye	bfd4b79245	docs(trace): remove duplicate comment (#20360 )	2025-12-16 14:56:01 +00:00
Federico Gimenez	49057b1c0c	feat(storage): add with_default_tables() to register RocksDB column families at initialization (#20416 )	2025-12-16 12:59:58 +00:00
Gigi	b6772370d7	docs: fix incorrect method reference in try_recover_sealed_with_senders (#20410 )	2025-12-16 12:27:53 +00:00
Karl Yu	d72935628a	feat: add support for eth/70 eip-7975 (#20255 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-16 12:05:11 +00:00
YK	ad63b135d6	feat(storage): implement EitherWriter/EitherReader methods for RocksDB (#20408 )	2025-12-16 11:26:31 +00:00
Brian Picciano	90651ae8e8	feat(engine): Use BAL in state root validation (#20383 )	2025-12-16 11:05:51 +00:00
Matthias Seitz	bbd51862d4	chore: rm flaky bench (#20413 )	2025-12-16 09:35:38 +00:00
Arsenii Kulikov	08a16a5bde	perf: recover transactions in parallel during network import (#20385 )	2025-12-16 09:33:24 +00:00
Snezhkko	f2c39db7a2	chore(rpc): fix misleading link and comment (#20367 )	2025-12-16 09:32:25 +00:00
oooLowNeoNooo	ae9e84d6e3	fix(discv4): correct ping_interval default value in docs (#20396 )	2025-12-16 09:29:45 +00:00
theo	c51da593d1	feat(net/p2p): support fixed external addresses with DNS resolution (#20411 )	2025-12-16 09:28:31 +00:00
Matthias Seitz	0e08f9f56c	perf: remove unnecessary channels from parallel trie operations (#20406 )	2025-12-16 09:15:27 +00:00
sashass1315	7eef092110	docs(exex): sync hello-world notifications loop with code (#20403 )	2025-12-16 08:39:45 +00:00
YK	40e8241bf5	feat(storage): use RocksDBBatch in EitherWriter and related modules (#20377 )	2025-12-16 03:57:41 +00:00
dependabot[bot]	dd9ff731e4	chore(deps): bump peter-evans/create-pull-request from 7 to 8 (#20402 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-16 00:11:22 +00:00
dependabot[bot]	83f9d1837f	chore(deps): bump actions/download-artifact from 4 to 7 (#20401 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-16 00:11:00 +00:00
dependabot[bot]	68911e617b	chore(deps): bump actions/upload-artifact from 5 to 6 (#20400 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-16 00:10:40 +00:00
0xcharry	36ba6db029	chore: remove redundant .as_str() calls after to_string() (#20404 )	2025-12-16 00:10:03 +00:00
Matthias Seitz	fec4432d82	perf: defer transaction pool notifications until after lock release (#20405 )	2025-12-15 23:06:34 +00:00
Matthias Seitz	179da26305	perf: use RwLock for transaction pool listeners (#20398 ) Co-authored-by: Arsenii Kulikov <klkvrr@gmail.com>	2025-12-15 21:47:59 +00:00
Matthias Seitz	b5e7a694d2	chore: update metric once (#20371 )	2025-12-15 20:38:24 +00:00
Maxim Evtush	9489667814	fix: post-state generator to include deletions in proptest (#20276 )	2025-12-15 16:43:02 +00:00
gustavo	004877ba59	refactor(cli): cleanup repair-trie metrics (#20226 )	2025-12-15 16:41:48 +00:00
Brian Picciano	a9e36923e1	feat(trie): Proof Rewrite: Use cached branch nodes (#20075 ) Co-authored-by: YK <chiayongkang@hotmail.com> Co-authored-by: Alexey Shekhirin <5773434+shekhirin@users.noreply.github.com>	2025-12-15 15:27:04 +00:00
DaniPopes	74a3816611	ci: reduce feature powerset depth (#20379 )	2025-12-15 14:42:14 +00:00
Alexey Shekhirin	5576d4547f	revert: feat(engine): run sync state root if not enough parallelism (#20127 ) (#20378 )	2025-12-15 14:05:54 +00:00
DaniPopes	21216e2f24	perf: use indexed parallel iterators for tx recovery (#20342 )	2025-12-15 13:40:03 +00:00
YK	42c1e1afe1	feat(storage): add account history constructors to EitherWriter/EitherReader (#20366 )	2025-12-15 12:45:07 +00:00
MoNyAvA	5f7e87fa2a	docs: add blob sub-pool to tx pool docs (#20375 )	2025-12-15 12:27:54 +00:00
Matthias Seitz	1b417dacc4	chore: sanity check for u64::Max (#20373 )	2025-12-15 11:33:50 +00:00
Niven	bb952be5b5	feat(flashblocks): support eth_getBlockTransactionCount for flashblocks (#20291 ) Co-authored-by: lucas <66681646+limyeechern@users.noreply.github.com> Co-authored-by: lucas.lim <lucas.lim@okg.com>	2025-12-15 11:29:23 +00:00
Federico Magnani	f927eec880	chore: export `FlashBlockDecoder` (#20370 )	2025-12-15 11:00:46 +00:00
Tomass	9c61f5568c	fix(rpc-testing-util): use buffer_unordered in trace_block_opcode_gas_unordered (#20369 )	2025-12-15 10:38:40 +00:00
ligt	662c0486a1	feat(storage): add rocksdb provider into database provider (#20253 )	2025-12-15 10:15:57 +00:00
Matthias Seitz	997848c2a1	fix(txpool): remove stale senderinfo (#20368 )	2025-12-15 10:00:25 +00:00
Olexandr88	155bdecf3b	docs(repo): add Ethereum-specific crates section (#20363 )	2025-12-15 09:56:40 +00:00
github-actions[bot]	679234f105	chore(deps): weekly `cargo update` (#20359 ) Co-authored-by: github-merge-queue <118344674+github-merge-queue@users.noreply.github.com>	2025-12-14 20:54:42 +00:00
phrwlk	419c7b489b	fix(rpc): remove dead flashbots module config (#20364 )	2025-12-14 20:54:15 +00:00
Rej Ect	06dac07b5f	ci(hive): bump actions/cache to v5 (#20349 )	2025-12-13 09:04:07 +00:00
YK	5621132b8b	feat: add RocksDB variant to EitherReader and EitherWriter (#20288 )	2025-12-13 04:06:44 +00:00
Matthias Seitz	3380eb69c8	fix: only collect already tracked accounts (#20341 )	2025-12-12 22:09:21 +00:00
Arsenii Kulikov	0366497ada	perf: skip redundant recovery (#20343 )	2025-12-12 22:01:05 +00:00
Alexey Shekhirin	cd71f3d5a4	feat(engine): record total latencies on instrumented state provider drop (#20337 )	2025-12-12 21:14:44 +00:00
Alexey Shekhirin	64909d33e6	feat(engine): cli argument to disable state cache (#20143 )	2025-12-12 17:51:22 +00:00
Alexey Shekhirin	3c9ad31344	chore(engine): make `InstrumentedStateProvider` public (#20335 )	2025-12-12 16:41:42 +00:00
gustavo	f3e14fd061	feat(rpc): handle dedicated eth_simulate errors (#20099 )	2025-12-12 16:40:13 +00:00
Alexey Shekhirin	daf6b88dc6	feat(node): engine args defaults (#20203 )	2025-12-12 15:54:05 +00:00
emmmm	d2d58f9a0e	docs: add missing RPC namespaces to JSON-RPC intro (#20321 )	2025-12-12 15:40:38 +00:00
Matthias Seitz	ace4e515b5	chore: bump inspectors 0.33.2 (#20334 )	2025-12-12 15:39:04 +00:00
Hesham Shabanah	134164954b	feat: add `--max-peers` CLI flag (#20139 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-12 13:26:44 +00:00
Lorsmirq Benton	2775dd1f23	docs: correct comments in custom-inspector (#20304 )	2025-12-12 13:21:03 +00:00
Alexey Shekhirin	ac0f9687bd	chore(engine): move noisy multiproof debug logs to trace level (#20331 )	2025-12-12 13:01:01 +00:00
Arsenii Kulikov	a9c21a395d	perf: spawn rpc handlers as blocking (#20330 )	2025-12-12 12:15:02 +00:00
Federico Magnani	df7ad9ae45	chore(ethapi): increase visibility `tx_batch_sender` (#20315 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-12 12:14:43 +00:00
sashass1315	5903e42a98	docs: refresh repo layout crate lists (#20319 )	2025-12-12 10:59:57 +00:00
Matthias Seitz	3c41b99599	chore: lower block buffer size (#20324 )	2025-12-12 08:15:54 +00:00
pepes	d70d80fff1	fix(docs): document discv5 discovery port 9200 (#20322 )	2025-12-12 08:12:08 +00:00
gustavo	ed3a8a03d5	feat(node-core): make rpc server args customizable (#20312 )	2025-12-11 23:24:31 +00:00
YK	bfcd46d01d	feat: add `account_history_in_rocksdb` field to `StorageSettings` (#20282 )	2025-12-11 19:37:36 +00:00
Brian Picciano	194d545fae	feat(engine): Add BAL stub methods to ExecutionPayload and BlockOrPayload (#20311 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-11 19:07:43 +00:00
sashass1315	97243ec1f4	docs: fix misleading links (#20300 )	2025-12-11 18:49:18 +00:00
DaniPopes	93c1b0f52f	ci: add more sccache (#20316 )	2025-12-11 18:46:11 +00:00
Arsenii Kulikov	474c09095f	feat: bump alloy-evm (#20314 )	2025-12-11 19:46:34 +01:00
Matthias Seitz	24c298133f	feat: allow larger ws frames on client side (#20307 )	2025-12-11 16:43:10 +00:00
Block Wizard	da27336a1e	docs: add architecture diagrams to ExEx documentation (#20193 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-11 11:41:15 +00:00
Matthias Seitz	2e567d6658	feat: add semaphore for blocking IO requests (#20289 )	2025-12-11 11:35:50 +00:00
Alexey Shekhirin	28e7c8a7cb	ci: scale down depot runners (#20295 )	2025-12-11 11:33:49 +00:00
Matthias Seitz	a2a5e03cb8	perf: fetch header directly (#20294 )	2025-12-11 11:18:51 +00:00
Sophia Raye	6073aa5b4a	docs(exex): fix DebugApi comment (#20296 )	2025-12-11 10:06:31 +00:00
Karl Yu	e90cfedf3d	feat: add support for testing_ rpc namespace and testing_buildBlockV1 (#20094 ) Co-authored-by: Arsenii Kulikov <klkvrr@gmail.com> Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-11 08:56:46 +00:00
Matthias Seitz	8b27ca6fa2	chore: update engine_getBlobs metric (#20290 )	2025-12-11 08:11:54 +00:00
Tomass	1752d6fb99	chore(optimism): move predeploy constant to op-alloy (#20181 ) Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-11 07:04:01 +00:00
emmmm	ac891a780b	docs: fix stages order and add missing EraStage (#20283 )	2025-12-11 06:26:27 +00:00
Adrian	036626b8a7	docs: improve map_add_ons method documentation (#20248 )	2025-12-11 06:03:34 +00:00
josé v	68f0c9812f	feat: add `transaction_hash_numbers_in_rocksdb` field to `StorageSettings` (#20209 )	2025-12-11 01:07:12 +00:00
sashass1315	c9920c9690	docs: clarify network mode, tx gossip and NAT (#20247 )	2025-12-10 21:52:04 +00:00
Karl Yu	af82606ff4	feat: add support for debug_getBadBlock (#20177 ) Co-authored-by: Arsenii Kulikov <klkvrr@gmail.com> Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>	2025-12-10 21:03:53 +00:00
radik878	38331a362e	fix(rpc): avoid signing Optimism deposit transactions (#20254 )	2025-12-10 20:46:43 +00:00
Tomass	e8dae2ae7d	chore(deps): bump op-alloy to 0.23.0 (#20256 ) Co-authored-by: Arsenii Kulikov <klkvrr@gmail.com>	2025-12-10 20:44:54 +00:00
Sophia Raye	ce5f90175b	docs(jsonrpc): add missing debug namespace RPC methods (#20267 )	2025-12-10 17:24:29 +00:00
gustavo	8c361c87c2	feat(txpool): handle more simulated scenarios in test_utils/pool.rs (#20138 )	2025-12-10 17:13:59 +00:00
Block Wizard	4fbbb1fe54	feat: add recover_transactions_unchecked_ref to BlockBody (#20266 )	2025-12-10 17:13:08 +00:00
Brian Picciano	b7d8815104	perf(prune): use delete_current_duplicates for MerkleChangeSets tables (#20230 )	2025-12-10 13:33:11 +00:00
Alexey Shekhirin	b91cd8f451	ci: sccache (#20265 )	2025-12-10 13:05:25 +00:00
Alexey Shekhirin	09aee4e35a	ci: use 16 cores for Hive workflow (#20264 )	2025-12-10 13:02:14 +00:00
Alexey Shekhirin	505a384b10	ci: increase partitions for crate-checks to 3 (#20261 )	2025-12-10 13:02:11 +00:00
phrwlk	6e00b99b67	docs: use canonical --rollup.sequencer and note aliases (#20260 )	2025-12-10 12:18:36 +00:00
emmmm	1d389cfe7a	docs(jsonrpc): add missing debug namespace RPC methods (#20258 )	2025-12-10 12:17:50 +00:00
Matthias Seitz	2e62387469	feat: use max retries for debug consensus rpc client (#20257 )	2025-12-10 11:06:38 +00:00
Block Wizard	31133255fe	docs(reth-bench): fix incorrect authrpc.jwtsecret flag (#20249 )	2025-12-10 09:30:50 +00:00
Matthias Seitz	a6b9472d1c	fix: use generic header (#20250 )	2025-12-10 09:11:39 +00:00
forkfury	6636d2a2ad	docs: fix timestamp validation comment (#20246 )	2025-12-10 08:41:23 +00:00
YK	ab6854d159	docs(reth-bench): fix incorrect output flag in README (#20240 )	2025-12-10 07:18:34 +00:00
Charlie-Mack	5a274fc939	feat: add example for launching a node with custom rpc middleware (#20159 )	2025-12-10 07:15:46 +00:00
radik878	c9431b224b	refactor(rpc): remove dead got_notif flag from RpcService batch handler (#20171 )	2025-12-10 07:15:09 +00:00
emmmm	8cbfd91db0	docs: add missing `bodies_history` and `merkle_changesets` prune config fields (#20244 )	2025-12-10 07:10:57 +00:00
Block Wizard	43f9942ba7	docs(txpool): fix PoolSize total field comment to include blob pool (#20241 )	2025-12-10 07:05:42 +00:00
Léa Narzis	06adc3ee0c	refactor(rpc): return error instead of clamping for `get_filter_block_range` (#20218 )	2025-12-10 07:03:30 +00:00
dependabot[bot]	fbf6be4cf2	chore(deps): bump dawidd6/action-homebrew-bump-formula from 6 to 7 (#20205 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-12-10 07:01:28 +00:00
Forostovec	21d61d40d1	docs: document state and block overrides for trace_call (#20217 )	2025-12-10 07:00:59 +00:00