feat(provider): add RocksDBProvider::clear method for table cleanup

Add clear() method to RocksDBProvider and its stub implementation to
support clearing all entries from a RocksDB table. This is needed by
the stage drop CLI command to clear RocksDB history tables.

- Adds clear<T: Table>() method that iterates and deletes all entries
- Stub implementation returns Ok(()) since stub behaves as empty DB

.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,tracy,serde-bincode-compat,serde,test-utils,arbitrary,bench,alloy-compat,min-error-logs,min-warn-logs,min-info-logs,min-debug-logs,min-trace-logs,otlp,js-tracer,portable,keccak-cache-global",
         # Do not try to add a new section to `[features]` of `A` only because `B` exposes that feature. There are edge-cases where this is still needed, but we can add them manually.
         "--left-side-feature-missing=ignore",
         # Ignore the case that `A` it outside of the workspace. Otherwise it will report errors in external dependencies that we have no influence on.

.github/actionlint.yaml

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

.github/assets/hive/run_simulator.sh

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

.github/workflows/bench.yml

@@ -11,18 +11,19 @@ env:
   CARGO_TERM_COLOR: always
   BASELINE: base
   SEED: reth
+  RUSTC_WRAPPER: "sccache"
 
 name: bench
 jobs:
   codspeed:
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     steps:
       - uses: actions/checkout@v6
         with:
           submodules: true
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
+      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/book.yml

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

.github/workflows/compact.yml

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

.github/workflows/docker-nightly.yml

@@ -28,10 +28,14 @@ jobs:
         build:
           - name: 'Build and push the nightly reth image'
             command: 'make PROFILE=maxperf docker-build-push-nightly'
+          - name: 'Build and push the nightly edge profiling reth image'
+            command: 'make PROFILE=profiling docker-build-push-nightly-edge-profiling'
           - name: 'Build and push the nightly profiling reth image'
             command: 'make PROFILE=profiling docker-build-push-nightly-profiling'
           - name: 'Build and push the nightly op-reth image'
             command: 'make IMAGE_NAME=$OP_IMAGE_NAME DOCKER_IMAGE_NAME=$OP_DOCKER_IMAGE_NAME PROFILE=maxperf op-docker-build-push-nightly'
+          - name: 'Build and push the nightly edge profiling op-reth image'
+            command: 'make IMAGE_NAME=$OP_IMAGE_NAME DOCKER_IMAGE_NAME=$OP_DOCKER_IMAGE_NAME PROFILE=profiling op-docker-build-push-nightly-edge-profiling'
           - name: 'Build and push the nightly profiling op-reth image'
             command: 'make IMAGE_NAME=$OP_IMAGE_NAME DOCKER_IMAGE_NAME=$OP_DOCKER_IMAGE_NAME PROFILE=profiling op-docker-build-push-nightly-profiling'
     steps:

.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,14 +20,14 @@ concurrency:
 jobs:
   test:
     name: e2e-testsuite
-    runs-on:
-      group: Reth
+    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:
@@ -43,4 +44,3 @@ jobs:
             --exclude 'op-reth' \
             --exclude 'reth' \
             -E 'binary(e2e_testsuite)'
-

.github/workflows/hive.yml

@@ -15,11 +15,21 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  prepare-reth:
+  prepare-reth-stable:
     uses: ./.github/workflows/prepare-reth.yml
     with:
       image_tag: ghcr.io/paradigmxyz/reth:latest
       binary_name: reth
+      cargo_features: "asm-keccak"
+      artifact_name: "reth-stable"
+
+  prepare-reth-edge:
+    uses: ./.github/workflows/prepare-reth.yml
+    with:
+      image_tag: ghcr.io/paradigmxyz/reth:latest
+      binary_name: reth
+      cargo_features: "asm-keccak edge"
+      artifact_name: "reth-edge"
 
   prepare-hive:
     if: github.repository == 'paradigmxyz/reth'
@@ -45,7 +55,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') }}
@@ -68,7 +78,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
@@ -77,6 +87,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
+        storage: [stable, edge]
         # ethereum/rpc to be deprecated:
         # https://github.com/ethereum/hive/pull/1117
         scenario:
@@ -86,7 +97,7 @@ jobs:
           - sim: devp2p
             limit: discv4
           # started failing after https://github.com/ethereum/go-ethereum/pull/31843, no
-          # action on our side, remove from here when we get unxpected passes on these tests
+          # action on our side, remove from here when we get unexpected passes on these tests
           # - sim: devp2p
           #  limit: eth
           #  include:
@@ -176,9 +187,10 @@ jobs:
           - sim: ethereum/eels/consume-rlp
             limit: .*tests/paris.*
     needs:
-      - prepare-reth
+      - prepare-reth-stable
+      - prepare-reth-edge
       - prepare-hive
-    name: run ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
+    name: ${{ matrix.storage }} / ${{ matrix.scenario.sim }}${{ matrix.scenario.limit && format(' - {0}', matrix.scenario.limit) }}
     runs-on:
       group: Reth
     permissions:
@@ -189,15 +201,15 @@ 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
+          name: reth-${{ matrix.storage }}
           path: /tmp
 
       - name: Load Docker images
@@ -247,8 +259,7 @@ jobs:
   notify-on-error:
     needs: test
     if: failure()
-    runs-on:
-      group: Reth
+    runs-on: ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.github/workflows/integration.yml

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

.github/workflows/kurtosis-op.yml

@@ -9,7 +9,7 @@ on:
 
   push:
     tags:
-      - '*'
+      - "*"
 
 env:
   CARGO_TERM_COLOR: always
@@ -32,8 +32,7 @@ jobs:
     strategy:
       fail-fast: false
     name: run kurtosis
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     needs:
       - prepare-reth
     steps:
@@ -42,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
@@ -83,12 +82,10 @@ jobs:
           kurtosis service logs -a op-devnet op-cl-2151908-2-op-node-op-reth-op-kurtosis
           exit 1
 
-
   notify-on-error:
     needs: test
     if: failure()
-    runs-on:
-      group: Reth
+    runs-on: ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.github/workflows/kurtosis.yml

@@ -9,7 +9,7 @@ on:
 
   push:
     tags:
-      - '*'
+      - "*"
 
 env:
   CARGO_TERM_COLOR: always
@@ -30,8 +30,7 @@ jobs:
     strategy:
       fail-fast: false
     name: run kurtosis
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     needs:
       - prepare-reth
     steps:
@@ -40,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
@@ -54,13 +53,12 @@ jobs:
       - name: Run kurtosis
         uses: ethpandaops/kurtosis-assertoor-github-action@v1
         with:
-          ethereum_package_args: '.github/assets/kurtosis_network_params.yaml'
+          ethereum_package_args: ".github/assets/kurtosis_network_params.yaml"
 
   notify-on-error:
     needs: test
     if: failure()
-    runs-on:
-      group: Reth
+    runs-on: ubuntu-latest
     steps:
       - name: Slack Webhook Action
         uses: rtCamp/action-slack-notify@v2

.github/workflows/lint.yml

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

.github/workflows/prepare-reth.yml

@@ -21,13 +21,17 @@ on:
         required: false
         type: string
         description: "Optional cargo package path"
+      artifact_name:
+        required: false
+        type: string
+        default: "artifacts"
+        description: "Name for the uploaded artifact"
 
 jobs:
   prepare-reth:
     if: github.repository == 'paradigmxyz/reth'
     timeout-minutes: 45
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     steps:
       - uses: actions/checkout@v6
       - run: mkdir artifacts
@@ -51,7 +55,7 @@ jobs:
 
       - name: Upload reth image
         id: upload
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
-          name: artifacts
+          name: ${{ inputs.artifact_name }}
           path: ./artifacts

.github/workflows/release-dist.yml

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

.github/workflows/release.yml

@@ -22,6 +22,7 @@ env:
   CARGO_TERM_COLOR: always
   DOCKER_IMAGE_NAME_URL: https://ghcr.io/${{ github.repository_owner }}/reth
   DOCKER_OP_IMAGE_NAME_URL: https://ghcr.io/${{ github.repository_owner }}/op-reth
+  RUSTC_WRAPPER: "sccache"
 
 jobs:
   dry-run:
@@ -51,6 +52,7 @@ jobs:
     steps:
       - uses: actions/checkout@v6
       - uses: dtolnay/rust-toolchain@stable
+      - uses: mozilla-actions/sccache-action@v0.0.9
       - name: Verify crate version matches tag
         # Check that the Cargo version starts with the tag,
         # so that Cargo version 1.4.8 can be matched against both v1.4.8 and v1.4.8-rc.1
@@ -104,6 +106,7 @@ jobs:
       - uses: dtolnay/rust-toolchain@stable
         with:
           target: ${{ matrix.configs.target }}
+      - uses: mozilla-actions/sccache-action@v0.0.9
       - name: Install cross main
         id: cross_main
         run: |
@@ -141,14 +144,14 @@ jobs:
 
       - name: Upload artifact
         if: ${{ github.event.inputs.dry_run != 'true' }}
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz
           path: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz
 
       - name: Upload signature
         if: ${{ github.event.inputs.dry_run != 'true' }}
-        uses: actions/upload-artifact@v5
+        uses: actions/upload-artifact@v6
         with:
           name: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz.asc
           path: ${{ matrix.build.binary }}-${{ needs.extract-version.outputs.VERSION }}-${{ matrix.configs.target }}.tar.gz.asc
@@ -170,7 +173,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Download artifacts
-        uses: actions/download-artifact@v6
+        uses: actions/download-artifact@v7
       - name: Generate full changelog
         id: changelog
         run: |

.github/workflows/reproducible-build.yml

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

.github/workflows/stage.yml

@@ -12,6 +12,7 @@ env:
   CARGO_TERM_COLOR: always
   FROM_BLOCK: 0
   TO_BLOCK: 50000
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -22,8 +23,7 @@ jobs:
     name: stage-run-test
     # Only run stage commands test in merge groups
     if: github.event_name == 'merge_group'
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -32,6 +32,7 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
+      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/sync-era.yml

@@ -9,6 +9,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -17,8 +18,7 @@ concurrency:
 jobs:
   sync:
     name: sync (${{ matrix.chain.bin }})
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -42,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
@@ -64,4 +65,4 @@ jobs:
           ${{ matrix.chain.bin }} stage unwind num-blocks 100 --chain ${{ matrix.chain.chain }}
       - name: Run stage unwind to block hash
         run: |
-          ${{ matrix.chain.bin }} stage unwind to-block ${{ matrix.chain.unwind-target }} --chain ${{ matrix.chain.chain }} 
+          ${{ matrix.chain.bin }} stage unwind to-block ${{ matrix.chain.unwind-target }} --chain ${{ matrix.chain.chain }}

.github/workflows/sync.yml

@@ -9,6 +9,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTC_WRAPPER: "sccache"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
@@ -17,8 +18,7 @@ concurrency:
 jobs:
   sync:
     name: sync (${{ matrix.chain.bin }})
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -42,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
@@ -63,4 +64,4 @@ jobs:
           ${{ matrix.chain.bin }} stage unwind num-blocks 100 --chain ${{ matrix.chain.chain }}
       - name: Run stage unwind to block hash
         run: |
-          ${{ matrix.chain.bin }} stage unwind to-block ${{ matrix.chain.unwind-target }} --chain ${{ matrix.chain.chain }} 
+          ${{ matrix.chain.bin }} stage unwind to-block ${{ matrix.chain.unwind-target }} --chain ${{ matrix.chain.chain }}

.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 }}
@@ -18,35 +19,28 @@ concurrency:
 
 jobs:
   test:
-    name: test / ${{ matrix.type }} (${{ matrix.partition }}/${{ matrix.total_partitions }})
-    runs-on:
-      group: Reth
+    name: test / ${{ matrix.type }} / ${{ matrix.storage }}
+    runs-on: depot-ubuntu-latest-4
     env:
       RUST_BACKTRACE: 1
+      EDGE_FEATURES: ${{ matrix.storage == 'edge' && 'edge' || '' }}
     strategy:
       matrix:
+        type: [ethereum, optimism]
+        storage: [stable, edge]
         include:
           - type: ethereum
-            args: --features "asm-keccak ethereum" --locked
-            partition: 1
-            total_partitions: 2
-          - type: ethereum
-            args: --features "asm-keccak ethereum" --locked
-            partition: 2
-            total_partitions: 2
+            features: asm-keccak ethereum
+            exclude_args: ""
           - type: optimism
-            args: --features "asm-keccak" --locked --exclude reth --exclude reth-bench --exclude "example-*" --exclude "reth-ethereum-*" --exclude "*-ethereum"
-            partition: 1
-            total_partitions: 2
-          - type: optimism
-            args: --features "asm-keccak" --locked --exclude reth --exclude reth-bench --exclude "example-*" --exclude "reth-ethereum-*" --exclude "*-ethereum"
-            partition: 2
-            total_partitions: 2
+            features: asm-keccak
+            exclude_args: --exclude reth --exclude reth-bench --exclude "example-*" --exclude "reth-ethereum-*" --exclude "*-ethereum"
     timeout-minutes: 30
     steps:
       - 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
@@ -58,15 +52,14 @@ jobs:
       - name: Run tests
         run: |
           cargo nextest run \
-            ${{ matrix.args }} --workspace \
+            --features "${{ matrix.features }} $EDGE_FEATURES" --locked \
+            ${{ matrix.exclude_args }} --workspace \
             --exclude ef-tests --no-tests=warn \
-            --partition hash:${{ matrix.partition }}/2 \
             -E "!kind(test) and not binary(e2e_testsuite)"
 
   state:
     name: Ethereum state tests
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest-4
     env:
       RUST_LOG: info,sync=error
       RUST_BACKTRACE: 1
@@ -93,6 +86,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
@@ -100,8 +94,7 @@ jobs:
 
   doc:
     name: doc tests
-    runs-on:
-      group: Reth
+    runs-on: depot-ubuntu-latest
     env:
       RUST_BACKTRACE: 1
     timeout-minutes: 30
@@ -109,6 +102,7 @@ jobs:
       - uses: actions/checkout@v6
       - uses: rui314/setup-mold@v1
       - uses: dtolnay/rust-toolchain@stable
+      - uses: mozilla-actions/sccache-action@v0.0.9
       - uses: Swatinem/rust-cache@v2
         with:
           cache-on-failure: true

.github/workflows/update-superchain.yml

@@ -27,7 +27,7 @@ jobs:
           ./fetch_superchain_config.sh
 
       - name: Create Pull Request
-        uses: peter-evans/create-pull-request@v7
+        uses: peter-evans/create-pull-request@v8
         with:
           commit-message: "chore: update superchain config"
           title: "chore: update superchain config"

.github/workflows/windows.yml

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

.gitignore

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

CLAUDE.md

@@ -18,7 +18,7 @@ Reth is a high-performance Ethereum execution client written in Rust, focusing o
 6. **Pipeline (`crates/stages/`)**: Staged sync architecture for blockchain synchronization
 7. **Trie (`crates/trie/`)**: Merkle Patricia Trie implementation with parallel state root computation
 8. **Node Builder (`crates/node/`)**: High-level node orchestration and configuration
-9  **The Consensus Engine (`crates/engine/`)**: Handles processing blocks received from the consensus layer with the Engine API (newPayload, forkchoiceUpdated)
+9. **The Consensus Engine (`crates/engine/`)**: Handles processing blocks received from the consensus layer with the Engine API (newPayload, forkchoiceUpdated)
 
 ### Key Design Principles
 

CONTRIBUTING.md

@@ -51,9 +51,7 @@ elsewhere.
 <!-- -   **Asking in the support Telegram:** The [Foundry Support Telegram][support-tg] is a fast and easy way to ask questions. -->
 <!-- -   **Opening a discussion:** This repository comes with a discussions board where you can also ask for help. Click the "Discussions" tab at the top. -->
 
-If you have reviewed existing documentation and still have questions, or you are having problems, you can get help by *
-*opening a discussion**. This repository comes with a discussions board where you can also ask for help. Click the "
-Discussions" tab at the top.
+If you have reviewed existing documentation and still have questions, or you are having problems, you can get help by **opening a discussion**. This repository comes with a discussions board where you can also ask for help. Click the "Discussions" tab at the top.
 
 As Reth is still in heavy development, the documentation can be a bit scattered. The [Reth Docs][reth-docs] is our
 current best-effort attempt at keeping up-to-date information.

Cargo.toml

@@ -153,6 +153,7 @@ members = [
     "examples/custom-node-components/",
     "examples/custom-payload-builder/",
     "examples/custom-rlpx-subprotocol",
+    "examples/custom-rpc-middleware",
     "examples/custom-node",
     "examples/db-access",
     "examples/engine-api-access",
@@ -375,11 +376,11 @@ reth-era-utils = { path = "crates/era-utils" }
 reth-errors = { path = "crates/errors" }
 reth-eth-wire = { path = "crates/net/eth-wire" }
 reth-eth-wire-types = { path = "crates/net/eth-wire-types" }
+reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-cli = { path = "crates/ethereum/cli", default-features = false }
 reth-ethereum-consensus = { path = "crates/ethereum/consensus", default-features = false }
 reth-ethereum-engine-primitives = { path = "crates/ethereum/engine-primitives", default-features = false }
 reth-ethereum-forks = { path = "crates/ethereum/hardforks", default-features = false }
-reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-primitives = { path = "crates/ethereum/primitives", default-features = false }
 reth-ethereum = { path = "crates/ethereum/reth" }
 reth-etl = { path = "crates/etl" }
@@ -480,57 +481,58 @@ revm-primitives = { version = "21.0.2", default-features = false }
 revm-interpreter = { version = "31.1.0", default-features = false }
 revm-database-interface = { version = "8.0.5", default-features = false }
 op-revm = { version = "14.1.0", default-features = false }
-revm-inspectors = "0.33.1"
+revm-inspectors = "0.33.2"
 
 # eth
 alloy-chains = { version = "0.2.5", default-features = false }
 alloy-dyn-abi = "1.4.1"
 alloy-eip2124 = { version = "0.2.0", default-features = false }
-alloy-evm = { version = "0.24.1", default-features = false }
-alloy-primitives = { version = "1.4.1", default-features = false, features = ["map-foldhash"] }
+alloy-eip7928 = { version = "0.1.0", default-features = false }
+alloy-evm = { version = "0.25.1", default-features = false }
+alloy-primitives = { version = "1.5.0", default-features = false, features = ["map-foldhash"] }
 alloy-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
-alloy-sol-macro = "1.4.1"
-alloy-sol-types = { version = "1.4.1", default-features = false }
+alloy-sol-macro = "1.5.0"
+alloy-sol-types = { version = "1.5.0", default-features = false }
 alloy-trie = { version = "0.9.1", default-features = false }
 
 alloy-hardforks = "0.4.5"
 
-alloy-consensus = { version = "1.1.2", default-features = false }
-alloy-contract = { version = "1.1.2", default-features = false }
-alloy-eips = { version = "1.1.2", default-features = false }
-alloy-genesis = { version = "1.1.2", default-features = false }
-alloy-json-rpc = { version = "1.1.2", default-features = false }
-alloy-network = { version = "1.1.2", default-features = false }
-alloy-network-primitives = { version = "1.1.2", default-features = false }
-alloy-provider = { version = "1.1.2", features = ["reqwest", "debug-api"], default-features = false }
-alloy-pubsub = { version = "1.1.2", default-features = false }
-alloy-rpc-client = { version = "1.1.2", default-features = false }
-alloy-rpc-types = { version = "1.1.2", features = ["eth"], default-features = false }
-alloy-rpc-types-admin = { version = "1.1.2", default-features = false }
-alloy-rpc-types-anvil = { version = "1.1.2", default-features = false }
-alloy-rpc-types-beacon = { version = "1.1.2", default-features = false }
-alloy-rpc-types-debug = { version = "1.1.2", default-features = false }
-alloy-rpc-types-engine = { version = "1.1.2", default-features = false }
-alloy-rpc-types-eth = { version = "1.1.2", default-features = false }
-alloy-rpc-types-mev = { version = "1.1.2", default-features = false }
-alloy-rpc-types-trace = { version = "1.1.2", default-features = false }
-alloy-rpc-types-txpool = { version = "1.1.2", default-features = false }
-alloy-serde = { version = "1.1.2", default-features = false }
-alloy-signer = { version = "1.1.2", default-features = false }
-alloy-signer-local = { version = "1.1.2", default-features = false }
-alloy-transport = { version = "1.1.2" }
-alloy-transport-http = { version = "1.1.2", features = ["reqwest-rustls-tls"], default-features = false }
-alloy-transport-ipc = { version = "1.1.2", default-features = false }
-alloy-transport-ws = { version = "1.1.2", default-features = false }
+alloy-consensus = { version = "1.4.1", default-features = false }
+alloy-contract = { version = "1.4.1", default-features = false }
+alloy-eips = { version = "1.4.1", default-features = false }
+alloy-genesis = { version = "1.4.1", default-features = false }
+alloy-json-rpc = { version = "1.4.1", default-features = false }
+alloy-network = { version = "1.4.1", default-features = false }
+alloy-network-primitives = { version = "1.4.1", default-features = false }
+alloy-provider = { version = "1.4.1", features = ["reqwest", "debug-api"], default-features = false }
+alloy-pubsub = { version = "1.4.1", default-features = false }
+alloy-rpc-client = { version = "1.4.1", default-features = false }
+alloy-rpc-types = { version = "1.4.1", features = ["eth"], default-features = false }
+alloy-rpc-types-admin = { version = "1.4.1", default-features = false }
+alloy-rpc-types-anvil = { version = "1.4.1", default-features = false }
+alloy-rpc-types-beacon = { version = "1.4.1", default-features = false }
+alloy-rpc-types-debug = { version = "1.4.1", default-features = false }
+alloy-rpc-types-engine = { version = "1.4.1", default-features = false }
+alloy-rpc-types-eth = { version = "1.4.1", default-features = false }
+alloy-rpc-types-mev = { version = "1.4.1", default-features = false }
+alloy-rpc-types-trace = { version = "1.4.1", default-features = false }
+alloy-rpc-types-txpool = { version = "1.4.1", default-features = false }
+alloy-serde = { version = "1.4.1", default-features = false }
+alloy-signer = { version = "1.4.1", default-features = false }
+alloy-signer-local = { version = "1.4.1", default-features = false }
+alloy-transport = { version = "1.4.1" }
+alloy-transport-http = { version = "1.4.1", features = ["reqwest-rustls-tls"], default-features = false }
+alloy-transport-ipc = { version = "1.4.1", default-features = false }
+alloy-transport-ws = { version = "1.4.1", default-features = false }
 
 # op
-alloy-op-evm = { version = "0.24.1", default-features = false }
+alloy-op-evm = { version = "0.25.0", default-features = false }
 alloy-op-hardforks = "0.4.4"
-op-alloy-rpc-types = { version = "0.22.4", default-features = false }
-op-alloy-rpc-types-engine = { version = "0.22.4", default-features = false }
-op-alloy-network = { version = "0.22.4", default-features = false }
-op-alloy-consensus = { version = "0.22.4", default-features = false }
-op-alloy-rpc-jsonrpsee = { version = "0.22.4", default-features = false }
+op-alloy-rpc-types = { version = "0.23.1", default-features = false }
+op-alloy-rpc-types-engine = { version = "0.23.1", default-features = false }
+op-alloy-network = { version = "0.23.1", default-features = false }
+op-alloy-consensus = { version = "0.23.1", default-features = false }
+op-alloy-rpc-jsonrpsee = { version = "0.23.1", default-features = false }
 op-alloy-flz = { version = "0.13.1", default-features = false }
 
 # misc
@@ -546,12 +548,14 @@ bytes = { version = "1.5", default-features = false }
 brotli = "8"
 cfg-if = "1.0"
 clap = "4"
+color-eyre = "0.6"
 dashmap = "6.0"
 derive_more = { version = "2", default-features = false, features = ["full"] }
 dirs-next = "2.0.0"
 dyn-clone = "1.0.17"
 eyre = "0.6"
 fdlimit = "0.3.0"
+fixed-map = { version = "0.9", default-features = false }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = { version = "0.14", default-features = false }
@@ -585,6 +589,7 @@ url = { version = "2.3", default-features = false }
 zstd = "0.13"
 byteorder = "1"
 mini-moka = "0.10"
+moka = "0.12"
 tar-no-std = { version = "0.3.2", default-features = false }
 miniz_oxide = { version = "0.8.4", default-features = false }
 chrono = "0.4.41"
@@ -592,9 +597,9 @@ chrono = "0.4.41"
 # metrics
 metrics = "0.24.0"
 metrics-derive = "0.1"
-metrics-exporter-prometheus = { version = "0.16.0", default-features = false }
+metrics-exporter-prometheus = { version = "0.18.0", default-features = false }
 metrics-process = "2.1.0"
-metrics-util = { default-features = false, version = "0.19.0" }
+metrics-util = { default-features = false, version = "0.20.0" }
 
 # proc-macros
 proc-macro2 = "1.0"
@@ -680,6 +685,7 @@ ethereum_ssz = "0.9.0"
 ethereum_ssz_derive = "0.9.0"
 
 # allocators
+jemalloc_pprof = { version = "0.8", default-features = false }
 tikv-jemalloc-ctl = "0.6"
 tikv-jemallocator = "0.6"
 tracy-client = "0.18.0"
@@ -690,7 +696,7 @@ ahash = "0.8"
 anyhow = "1.0"
 bindgen = { version = "0.71", default-features = false }
 block-padding = "0.3.2"
-cc = "=1.2.15"
+cc = "1.2.15"
 cipher = "0.4.3"
 comfy-table = "7.0"
 concat-kdf = "0.1.0"
@@ -727,7 +733,9 @@ socket2 = { version = "0.5", default-features = false }
 sysinfo = { version = "0.33", default-features = false }
 tracing-journald = "0.3"
 tracing-logfmt = "0.3.3"
+tracing-samply = "0.1"
 tracing-subscriber = { version = "0.3", default-features = false }
+tracing-tracy = "0.11"
 triehash = "0.8"
 typenum = "1.15.0"
 vergen = "9.0.4"

Dockerfile

@@ -18,7 +18,7 @@ FROM chef AS builder
 COPY --from=planner /app/recipe.json recipe.json
 
 # Build profile, release by default
-ARG BUILD_PROFILE=release
+ARG BUILD_PROFILE=maxperf
 ENV BUILD_PROFILE=$BUILD_PROFILE
 
 # Extra Cargo flags

DockerfileOp

@@ -14,7 +14,7 @@ RUN cargo chef prepare --recipe-path recipe.json
 FROM chef AS builder
 COPY --from=planner /app/recipe.json recipe.json
 
-ARG BUILD_PROFILE=release
+ARG BUILD_PROFILE=maxperf
 ENV BUILD_PROFILE=$BUILD_PROFILE
 
 ARG RUSTFLAGS=""

LICENSE-APACHE

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

LICENSE-MIT

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

Makefile

@@ -276,13 +276,18 @@ docker-build-push-latest: ## Build and push a cross-arch Docker image tagged wit
 docker-build-push-nightly: ## Build and push cross-arch Docker image tagged with the latest git tag with a `-nightly` suffix, and `latest-nightly`.
 	$(call docker_build_push,nightly,nightly)
 
+.PHONY: docker-build-push-nightly-edge-profiling
+docker-build-push-nightly-edge-profiling: FEATURES := $(FEATURES) edge
+docker-build-push-nightly-edge-profiling: ## Build and push cross-arch Docker image with edge features tagged with `nightly-edge-profiling`.
+	$(call docker_build_push,nightly-edge-profiling,nightly-edge-profiling)
+
 # Create a cross-arch Docker image with the given tags and push it
 define docker_build_push
-	$(MAKE) build-x86_64-unknown-linux-gnu
+	$(MAKE) FEATURES="$(FEATURES)" build-x86_64-unknown-linux-gnu
 	mkdir -p $(BIN_DIR)/amd64
 	cp $(CARGO_TARGET_DIR)/x86_64-unknown-linux-gnu/$(PROFILE)/reth $(BIN_DIR)/amd64/reth
 
-	$(MAKE) build-aarch64-unknown-linux-gnu
+	$(MAKE) FEATURES="$(FEATURES)" build-aarch64-unknown-linux-gnu
 	mkdir -p $(BIN_DIR)/arm64
 	cp $(CARGO_TARGET_DIR)/aarch64-unknown-linux-gnu/$(PROFILE)/reth $(BIN_DIR)/arm64/reth
 
@@ -328,6 +333,11 @@ op-docker-build-push-latest: ## Build and push a cross-arch Docker image tagged
 op-docker-build-push-nightly: ## Build and push cross-arch Docker image tagged with the latest git tag with a `-nightly` suffix, and `latest-nightly`.
 	$(call op_docker_build_push,nightly,nightly)
 
+.PHONY: op-docker-build-push-nightly-edge-profiling
+op-docker-build-push-nightly-edge-profiling: FEATURES := $(FEATURES) edge
+op-docker-build-push-nightly-edge-profiling: ## Build and push cross-arch Docker image with edge features tagged with `nightly-edge-profiling`.
+	$(call op_docker_build_push,nightly-edge-profiling,nightly-edge-profiling)
+
 # Note: This requires a buildx builder with emulation support. For example:
 #
 # `docker run --privileged --rm tonistiigi/binfmt --install amd64,arm64`
@@ -347,11 +357,11 @@ op-docker-build-push-nightly-profiling: ## Build and push cross-arch Docker imag
 
 # Create a cross-arch Docker image with the given tags and push it
 define op_docker_build_push
-	$(MAKE) op-build-x86_64-unknown-linux-gnu
+	$(MAKE) FEATURES="$(FEATURES)" op-build-x86_64-unknown-linux-gnu
 	mkdir -p $(BIN_DIR)/amd64
 	cp $(CARGO_TARGET_DIR)/x86_64-unknown-linux-gnu/$(PROFILE)/op-reth $(BIN_DIR)/amd64/op-reth
 
-	$(MAKE) op-build-aarch64-unknown-linux-gnu
+	$(MAKE) FEATURES="$(FEATURES)" op-build-aarch64-unknown-linux-gnu
 	mkdir -p $(BIN_DIR)/arm64
 	cp $(CARGO_TARGET_DIR)/aarch64-unknown-linux-gnu/$(PROFILE)/op-reth $(BIN_DIR)/arm64/op-reth
 
@@ -521,10 +531,3 @@ pr:
 	make update-book-cli && \
 	cargo docs --document-private-items && \
 	make test
-
-check-features:
-	cargo hack check \
-		--package reth-codecs \
-		--package reth-primitives-traits \
-		--package reth-primitives \
-		--feature-powerset

README.md

@@ -44,7 +44,7 @@ More historical context below:
 -   We released 1.0 "production-ready" stable Reth in June 2024.
     -   Reth completed an audit with [Sigma Prime](https://sigmaprime.io/), the developers of [Lighthouse](https://github.com/sigp/lighthouse), the Rust Consensus Layer implementation. Find it [here](./audit/sigma_prime_audit_v2.pdf).
     -   Revm (the EVM used in Reth) underwent an audit with [Guido Vranken](https://x.com/guidovranken) (#1 [Ethereum Bug Bounty](https://ethereum.org/en/bug-bounty)). We will publish the results soon.
--   We released multiple iterative beta versions, up to [beta.9](https://github.com/paradigmxyz/reth/releases/tag/v0.2.0-beta.9) on Monday June 3, 2024,the last beta release.
+-   We released multiple iterative beta versions, up to [beta.9](https://github.com/paradigmxyz/reth/releases/tag/v0.2.0-beta.9) on Monday June 3, 2024, the last beta release.
 -   We released [beta](https://github.com/paradigmxyz/reth/releases/tag/v0.2.0-beta.1) on Monday March 4, 2024, our first breaking change to the database model, providing faster query speed, smaller database footprint, and allowing "history" to be mounted on separate drives.
 -   We shipped iterative improvements until the last alpha release on February 28, 2024, [0.1.0-alpha.21](https://github.com/paradigmxyz/reth/releases/tag/v0.1.0-alpha.21).
 -   We [initially announced](https://www.paradigm.xyz/2023/06/reth-alpha) [0.1.0-alpha.1](https://github.com/paradigmxyz/reth/releases/tag/v0.1.0-alpha.1) on June 20, 2023.

bin/reth-bench-compare/Cargo.toml

@@ -25,7 +25,9 @@ reth-chainspec.workspace = true
 
 # alloy
 alloy-provider = { workspace = true, features = ["reqwest-rustls-tls"], default-features = false }
+alloy-rpc-client = { workspace = true, features = ["pubsub"] }
 alloy-rpc-types-eth.workspace = true
+alloy-transport-ws.workspace = true
 alloy-primitives.workspace = true
 
 # CLI and argument parsing
@@ -69,7 +71,11 @@ jemalloc = [
     "reth-node-core/jemalloc",
 ]
 jemalloc-prof = ["reth-cli-util/jemalloc-prof"]
-tracy-allocator = ["reth-cli-util/tracy-allocator"]
+tracy-allocator = ["reth-cli-util/tracy-allocator", "tracy"]
+tracy = [
+    "reth-node-core/tracy",
+    "reth-tracing/tracy",
+]
 
 min-error-logs = [
     "tracing/release_max_level_error",

bin/reth-bench-compare/README.md

@@ -0,0 +1,50 @@
+# reth-bench-compare
+
+Compare reth performance between two git references.
+
+## Usage
+
+```bash
+reth-bench-compare \
+  --baseline-ref main \
+  --feature-ref my-feature \
+  --blocks 100 \
+  --wait-for-persistence
+```
+
+## Arguments
+
+| Argument | Description | Default | Required |
+|----------|-------------|---------|----------|
+| `--baseline-ref <REF>` | Git reference for baseline | - | Yes |
+| `--feature-ref <REF>` | Git reference to compare | - | Yes |
+| `--blocks <N>` | Number of blocks to benchmark | `100` | No |
+| `--chain <CHAIN>` | Chain to benchmark | `mainnet` | No |
+| `--datadir <PATH>` | Data directory path | OS-specific | No |
+| `--rpc-url <URL>` | RPC endpoint for block data | Chain default | No |
+| `--output-dir <PATH>` | Output directory | `./reth-bench-compare` | No |
+| `--wait-for-persistence` | Wait for block persistence | `false` | No |
+| `--persistence-threshold <N>` | Wait after every N+1 blocks | `2` | No |
+| `--wait-time <DURATION>` | Fixed delay (legacy) | - | No |
+| `--warmup-blocks <N>` | Cache warmup blocks | Same as `--blocks` | No |
+| `--draw` | Generate charts (needs Python/uv) | `false` | No |
+| `--profile` | Enable CPU profiling (needs samply) | `false` | No |
+| `-vvvv` | Debug logging | Info | No |
+| `--features <FEATURES>` | Rust features for both builds | `jemalloc,asm-keccak` | No |
+| `--rustflags <FLAGS>` | RUSTFLAGS for both builds | `-C target-cpu=native` | No |
+| `--baseline-features <FEATURES>` | Features for baseline only | Inherits `--features` | No |
+| `--feature-features <FEATURES>` | Features for feature only | Inherits `--features` | No |
+| `--baseline-rustflags <FLAGS>` | RUSTFLAGS for baseline only | Inherits `--rustflags` | No |
+| `--feature-rustflags <FLAGS>` | RUSTFLAGS for feature only | Inherits `--rustflags` | No |
+| `--baseline-args <ARGS>` | Extra args for baseline node | - | No |
+| `--feature-args <ARGS>` | Extra args for feature node | - | No |
+| `--metrics-port <PORT>` | Metrics endpoint port | `5005` | No |
+| `--sudo` | Run with elevated privileges | `false` | No |
+
+## Output
+
+Results in `./reth-bench-compare/results/<timestamp>/`:
+- `comparison_report.json` - Metrics comparison
+- `per_block_comparison.csv` - Per-block statistics
+- `baseline/` and `feature/` - Individual run results
+- `latency_comparison.png` - Chart (if `--draw` used)

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

@@ -18,6 +18,8 @@ pub(crate) struct BenchmarkRunner {
     rpc_url: String,
     jwt_secret: String,
     wait_time: Option<String>,
+    wait_for_persistence: bool,
+    persistence_threshold: Option<u64>,
     warmup_blocks: u64,
 }
 
@@ -28,6 +30,8 @@ impl BenchmarkRunner {
             rpc_url: args.get_rpc_url(),
             jwt_secret: args.jwt_secret_path().to_string_lossy().to_string(),
             wait_time: args.wait_time.clone(),
+            wait_for_persistence: args.wait_for_persistence,
+            persistence_threshold: args.persistence_threshold,
             warmup_blocks: args.get_warmup_blocks(),
         }
     }
@@ -96,13 +100,9 @@ impl BenchmarkRunner {
             &from_block.to_string(),
             "--to",
             &to_block.to_string(),
+            "--wait-time=0ms", // Warmup should avoid persistence waits.
         ]);
 
-        // Add wait-time argument if provided
-        if let Some(ref wait_time) = self.wait_time {
-            cmd.args(["--wait-time", wait_time]);
-        }
-
         cmd.env("RUST_LOG_STYLE", "never")
             .stdout(std::process::Stdio::piped())
             .stderr(std::process::Stdio::piped())
@@ -186,9 +186,16 @@ impl BenchmarkRunner {
             &output_dir.to_string_lossy(),
         ]);
 
-        // Add wait-time argument if provided
+        // Configure wait mode: wait-time takes precedence over persistence-based flow
         if let Some(ref wait_time) = self.wait_time {
             cmd.args(["--wait-time", wait_time]);
+        } else if self.wait_for_persistence {
+            cmd.arg("--wait-for-persistence");
+
+            // Add persistence threshold if specified
+            if let Some(threshold) = self.persistence_threshold {
+                cmd.args(["--persistence-threshold", &threshold.to_string()]);
+            }
         }
 
         cmd.env("RUST_LOG_STYLE", "never")

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

@@ -114,10 +114,29 @@ pub(crate) struct Args {
     #[arg(long)]
     pub profile: bool,
 
-    /// Wait time between engine API calls (passed to reth-bench)
-    #[arg(long, value_name = "DURATION")]
+    /// Optional fixed delay between engine API calls (passed to reth-bench).
+    ///
+    /// When set, reth-bench uses wait-time mode and disables persistence-based flow.
+    /// This flag remains for compatibility with older scripts.
+    #[arg(long, value_name = "DURATION", hide = true)]
     pub wait_time: Option<String>,
 
+    /// Wait for blocks to be persisted before sending the next batch (passed to reth-bench).
+    ///
+    /// When enabled, waits for every Nth block to be persisted using the
+    /// `reth_subscribePersistedBlock` subscription. This ensures the benchmark
+    /// doesn't outpace persistence.
+    #[arg(long)]
+    pub wait_for_persistence: bool,
+
+    /// Engine persistence threshold (passed to reth-bench).
+    ///
+    /// The benchmark waits after every `(threshold + 1)` blocks. By default this
+    /// matches the engine's default persistence threshold (2), so waits occur
+    /// at blocks 3, 6, 9, etc.
+    #[arg(long, value_name = "PERSISTENCE_THRESHOLD")]
+    pub persistence_threshold: Option<u64>,
+
     /// Number of blocks to run for cache warmup after clearing caches.
     /// If not specified, defaults to the same as --blocks
     #[arg(long, value_name = "N")]
@@ -128,6 +147,11 @@ pub(crate) struct Args {
     #[arg(long)]
     pub no_clear_cache: bool,
 
+    /// Skip waiting for the node to sync before starting benchmarks.
+    /// When enabled, assumes the node is already synced and skips the initial tip check.
+    #[arg(long)]
+    pub skip_wait_syncing: bool,
+
     #[command(flatten)]
     pub logs: LogArgs,
 
@@ -164,12 +188,42 @@ pub(crate) struct Args {
     #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
     pub reth_args: Vec<String>,
 
-    /// Comma-separated list of features to enable during reth compilation
+    /// Comma-separated list of features to enable during reth compilation (applied to both builds)
     ///
     /// Example: `jemalloc,asm-keccak`
     #[arg(long, value_name = "FEATURES", default_value = "jemalloc,asm-keccak")]
     pub features: String,
 
+    /// Comma-separated list of features to enable only for baseline build (overrides --features)
+    ///
+    /// Example: `--baseline-features jemalloc`
+    #[arg(long, value_name = "FEATURES")]
+    pub baseline_features: Option<String>,
+
+    /// Comma-separated list of features to enable only for feature build (overrides --features)
+    ///
+    /// Example: `--feature-features jemalloc,asm-keccak`
+    #[arg(long, value_name = "FEATURES")]
+    pub feature_features: Option<String>,
+
+    /// RUSTFLAGS to use for both baseline and feature builds
+    ///
+    /// Example: `--rustflags "-C target-cpu=native"`
+    #[arg(long, value_name = "FLAGS", default_value = "-C target-cpu=native")]
+    pub rustflags: String,
+
+    /// RUSTFLAGS to use only for baseline build (overrides --rustflags)
+    ///
+    /// Example: `--baseline-rustflags "-C target-cpu=native -C lto"`
+    #[arg(long, value_name = "FLAGS")]
+    pub baseline_rustflags: Option<String>,
+
+    /// RUSTFLAGS to use only for feature build (overrides --rustflags)
+    ///
+    /// Example: `--feature-rustflags "-C target-cpu=native -C lto"`
+    #[arg(long, value_name = "FLAGS")]
+    pub feature_rustflags: Option<String>,
+
     /// Disable automatic --debug.startup-sync-state-idle flag for specific runs.
     /// Can be "baseline", "feature", or "all".
     /// By default, the flag is passed to warmup, baseline, and feature runs.
@@ -328,7 +382,6 @@ pub(crate) async fn run_comparison(args: Args, _ctx: CliContext) -> Result<()> {
         git_manager.repo_root().to_string(),
         output_dir.clone(),
         git_manager.clone(),
-        args.features.clone(),
     )?;
     // Initialize node manager
     let mut node_manager = NodeManager::new(&args);
@@ -448,6 +501,18 @@ async fn run_compilation_phase(
         let ref_type = ref_types[i];
         let commit = &ref_commits[git_ref];
 
+        // Get per-build features and rustflags
+        let features = match ref_type {
+            "baseline" => args.baseline_features.as_ref().unwrap_or(&args.features),
+            "feature" => args.feature_features.as_ref().unwrap_or(&args.features),
+            _ => &args.features,
+        };
+        let rustflags = match ref_type {
+            "baseline" => args.baseline_rustflags.as_ref().unwrap_or(&args.rustflags),
+            "feature" => args.feature_rustflags.as_ref().unwrap_or(&args.rustflags),
+            _ => &args.rustflags,
+        };
+
         info!(
             "Compiling {} binary for reference: {} (commit: {})",
             ref_type,
@@ -459,7 +524,7 @@ async fn run_compilation_phase(
         git_manager.switch_ref(git_ref)?;
 
         // Compile reth (with caching)
-        compilation_manager.compile_reth(commit, is_optimism)?;
+        compilation_manager.compile_reth(commit, is_optimism, features, rustflags)?;
 
         info!("Completed compilation for {} reference", ref_type);
     }
@@ -471,6 +536,7 @@ async fn run_compilation_phase(
     Ok((baseline_commit, feature_commit))
 }
 
+#[allow(clippy::too_many_arguments)]
 /// Run warmup phase to warm up caches before benchmarking
 async fn run_warmup_phase(
     git_manager: &GitManager,
@@ -480,9 +546,15 @@ async fn run_warmup_phase(
     args: &Args,
     is_optimism: bool,
     baseline_commit: &str,
+    starting_tip: u64,
 ) -> Result<()> {
     info!("=== Running warmup phase ===");
 
+    // Unwind to starting block minus warmup blocks, so we end up back at starting_tip
+    let warmup_blocks = args.get_warmup_blocks();
+    let unwind_target = starting_tip.saturating_sub(warmup_blocks);
+    node_manager.unwind_to_block(unwind_target).await?;
+
     // Use baseline for warmup
     let warmup_ref = &args.baseline_ref;
 
@@ -506,17 +578,18 @@ async fn run_warmup_phase(
     // Build additional args with conditional --debug.startup-sync-state-idle flag
     let additional_args = args.build_additional_args("warmup", args.baseline_args.as_ref());
 
-    // Start reth node for warmup
-    let mut node_process =
+    // Start reth node for warmup (command is not stored for warmup phase)
+    let (mut node_process, _warmup_command) =
         node_manager.start_node(&binary_path, warmup_ref, "warmup", &additional_args).await?;
 
     // Wait for node to be ready and get its current tip
-    let current_tip = node_manager.wait_for_node_ready_and_get_tip().await?;
+    let current_tip = if args.skip_wait_syncing {
+        node_manager.wait_for_rpc_and_get_tip(&mut node_process).await?
+    } else {
+        node_manager.wait_for_node_ready_and_get_tip(&mut node_process).await?
+    };
     info!("Warmup node is ready at tip: {}", current_tip);
 
-    // Store the tip we'll unwind back to
-    let original_tip = current_tip;
-
     // Clear filesystem caches before warmup run only (unless disabled)
     if args.no_clear_cache {
         info!("Skipping filesystem cache clearing (--no-clear-cache flag set)");
@@ -527,12 +600,9 @@ async fn run_warmup_phase(
     // Run warmup to warm up caches
     benchmark_runner.run_warmup(current_tip).await?;
 
-    // Stop node before unwinding (node must be stopped to release database lock)
+    // Stop node after warmup
     node_manager.stop_node(&mut node_process).await?;
 
-    // Unwind back to starting block after warmup
-    node_manager.unwind_to_block(original_tip).await?;
-
     info!("Warmup phase completed");
     Ok(())
 }
@@ -554,6 +624,31 @@ async fn run_benchmark_workflow(
     let (baseline_commit, feature_commit) =
         run_compilation_phase(git_manager, compilation_manager, args, is_optimism).await?;
 
+    // Switch to baseline reference and get the starting tip
+    git_manager.switch_ref(&args.baseline_ref)?;
+    let binary_path =
+        compilation_manager.get_cached_binary_path_for_commit(&baseline_commit, is_optimism);
+    if !binary_path.exists() {
+        return Err(eyre!(
+            "Cached baseline binary not found at {:?}. Compilation phase should have created it.",
+            binary_path
+        ));
+    }
+
+    // Start node briefly to get the current tip, then stop it
+    info!("=== Determining initial block height ===");
+    let additional_args = args.build_additional_args("baseline", args.baseline_args.as_ref());
+    let (mut node_process, _) = node_manager
+        .start_node(&binary_path, &args.baseline_ref, "baseline", &additional_args)
+        .await?;
+    let starting_tip = if args.skip_wait_syncing {
+        node_manager.wait_for_rpc_and_get_tip(&mut node_process).await?
+    } else {
+        node_manager.wait_for_node_ready_and_get_tip(&mut node_process).await?
+    };
+    info!("Node starting tip: {}", starting_tip);
+    node_manager.stop_node(&mut node_process).await?;
+
     // Run warmup phase before benchmarking (skip if warmup_blocks is 0)
     if args.get_warmup_blocks() > 0 {
         run_warmup_phase(
@@ -564,6 +659,7 @@ async fn run_benchmark_workflow(
             args,
             is_optimism,
             &baseline_commit,
+            starting_tip,
         )
         .await?;
     } else {
@@ -579,6 +675,10 @@ async fn run_benchmark_workflow(
         let commit = commits[i];
         info!("=== Processing {} reference: {} ===", ref_type, git_ref);
 
+        // Unwind to starting block minus benchmark blocks, so we end up back at starting_tip
+        let unwind_target = starting_tip.saturating_sub(args.blocks);
+        node_manager.unwind_to_block(unwind_target).await?;
+
         // Switch to target reference
         git_manager.switch_ref(git_ref)?;
 
@@ -607,22 +707,23 @@ async fn run_benchmark_workflow(
         // Build additional args with conditional --debug.startup-sync-state-idle flag
         let additional_args = args.build_additional_args(ref_type, base_args_str);
 
-        // Start reth node
-        let mut node_process =
+        // Start reth node and capture the command for reporting
+        let (mut node_process, reth_command) =
             node_manager.start_node(&binary_path, git_ref, ref_type, &additional_args).await?;
 
         // Wait for node to be ready and get its current tip (wherever it is)
-        let current_tip = node_manager.wait_for_node_ready_and_get_tip().await?;
+        let current_tip = if args.skip_wait_syncing {
+            node_manager.wait_for_rpc_and_get_tip(&mut node_process).await?
+        } else {
+            node_manager.wait_for_node_ready_and_get_tip(&mut node_process).await?
+        };
         info!("Node is ready at tip: {}", current_tip);
 
-        // Store the tip we'll unwind back to
-        let original_tip = current_tip;
-
         // Calculate benchmark range
         // Note: reth-bench has an off-by-one error where it consumes the first block
         // of the range, so we add 1 to compensate and get exactly args.blocks blocks
-        let from_block = original_tip;
-        let to_block = original_tip + args.blocks;
+        let from_block = current_tip;
+        let to_block = current_tip + args.blocks;
 
         // Run benchmark
         let output_dir = comparison_generator.get_ref_output_dir(ref_type);
@@ -639,14 +740,12 @@ async fn run_benchmark_workflow(
         // Stop node
         node_manager.stop_node(&mut node_process).await?;
 
-        // Unwind back to original tip
-        node_manager.unwind_to_block(original_tip).await?;
-
         // Store results for comparison
         comparison_generator.add_ref_results(ref_type, &output_dir)?;
 
-        // Set the benchmark run timestamps
+        // Set the benchmark run timestamps and reth command
         comparison_generator.set_ref_timestamps(ref_type, benchmark_start, benchmark_end)?;
+        comparison_generator.set_ref_command(ref_type, reth_command)?;
 
         info!("Completed {} reference benchmark", ref_type);
     }

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

@@ -21,6 +21,8 @@ pub(crate) struct ComparisonGenerator {
     feature_ref_name: String,
     baseline_results: Option<BenchmarkResults>,
     feature_results: Option<BenchmarkResults>,
+    baseline_command: Option<String>,
+    feature_command: Option<String>,
 }
 
 /// Represents the results from a single benchmark run
@@ -37,7 +39,8 @@ pub(crate) struct BenchmarkResults {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct CombinedLatencyRow {
     pub block_number: u64,
-    pub transaction_count: u64,
+    #[serde(default)]
+    pub transaction_count: Option<u64>,
     pub gas_used: u64,
     pub new_payload_latency: u128,
 }
@@ -46,7 +49,8 @@ pub(crate) struct CombinedLatencyRow {
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub(crate) struct TotalGasRow {
     pub block_number: u64,
-    pub transaction_count: u64,
+    #[serde(default)]
+    pub transaction_count: Option<u64>,
     pub gas_used: u64,
     pub time: u128,
 }
@@ -89,11 +93,13 @@ pub(crate) struct RefInfo {
     pub summary: BenchmarkSummary,
     pub start_timestamp: Option<DateTime<Utc>>,
     pub end_timestamp: Option<DateTime<Utc>>,
+    pub reth_command: Option<String>,
 }
 
 /// Summary of the comparison between references.
 ///
 /// Percent deltas are `(feature - baseline) / baseline * 100`:
+/// - `new_payload_latency_mean_change_percent`: percent changes of the per-block means.
 /// - `new_payload_latency_p50_change_percent` / p90 / p99: percent changes of the respective
 ///   per-block percentiles.
 /// - `per_block_latency_change_mean_percent` / `per_block_latency_change_median_percent` are the
@@ -111,6 +117,7 @@ pub(crate) struct ComparisonSummary {
     pub per_block_latency_change_median_percent: f64,
     pub per_block_latency_change_std_dev_percent: f64,
     pub new_payload_total_latency_change_percent: f64,
+    pub new_payload_latency_mean_change_percent: f64,
     pub new_payload_latency_p50_change_percent: f64,
     pub new_payload_latency_p90_change_percent: f64,
     pub new_payload_latency_p99_change_percent: f64,
@@ -122,7 +129,8 @@ pub(crate) struct ComparisonSummary {
 #[derive(Debug, Serialize)]
 pub(crate) struct BlockComparison {
     pub block_number: u64,
-    pub transaction_count: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub transaction_count: Option<u64>,
     pub gas_used: u64,
     pub baseline_new_payload_latency: u128,
     pub feature_new_payload_latency: u128,
@@ -142,6 +150,8 @@ impl ComparisonGenerator {
             feature_ref_name: args.feature_ref.clone(),
             baseline_results: None,
             feature_results: None,
+            baseline_command: None,
+            feature_command: None,
         }
     }
 
@@ -206,6 +216,21 @@ impl ComparisonGenerator {
         Ok(())
     }
 
+    /// Set the reth command for a reference
+    pub(crate) fn set_ref_command(&mut self, ref_type: &str, command: String) -> Result<()> {
+        match ref_type {
+            "baseline" => {
+                self.baseline_command = Some(command);
+            }
+            "feature" => {
+                self.feature_command = Some(command);
+            }
+            _ => return Err(eyre!("Unknown reference type: {}", ref_type)),
+        }
+
+        Ok(())
+    }
+
     /// Generate the final comparison report
     pub(crate) async fn generate_comparison_report(&self) -> Result<()> {
         info!("Generating comparison report...");
@@ -230,12 +255,14 @@ impl ComparisonGenerator {
                 summary: baseline.summary.clone(),
                 start_timestamp: baseline.start_timestamp,
                 end_timestamp: baseline.end_timestamp,
+                reth_command: self.baseline_command.clone(),
             },
             feature: RefInfo {
                 ref_name: feature.ref_name.clone(),
                 summary: feature.summary.clone(),
                 start_timestamp: feature.start_timestamp,
                 end_timestamp: feature.end_timestamp,
+                reth_command: self.feature_command.clone(),
             },
             comparison_summary,
             per_block_comparisons,
@@ -420,6 +447,10 @@ impl ComparisonGenerator {
             per_block_latency_change_median_percent,
             per_block_latency_change_std_dev_percent,
             new_payload_total_latency_change_percent,
+            new_payload_latency_mean_change_percent: calc_percent_change(
+                baseline.mean_new_payload_latency_ms,
+                feature.mean_new_payload_latency_ms,
+            ),
             new_payload_latency_p50_change_percent: calc_percent_change(
                 baseline.median_new_payload_latency_ms,
                 feature.median_new_payload_latency_ms,
@@ -550,6 +581,10 @@ impl ComparisonGenerator {
             "  Total newPayload time change:                {:+.2}%",
             summary.new_payload_total_latency_change_percent
         );
+        println!(
+            "  NewPayload Latency mean:           {:+.2}%",
+            summary.new_payload_latency_mean_change_percent
+        );
         println!(
             "  NewPayload Latency p50:           {:+.2}%",
             summary.new_payload_latency_p50_change_percent
@@ -599,6 +634,9 @@ impl ComparisonGenerator {
                 end.format("%Y-%m-%d %H:%M:%S UTC")
             );
         }
+        if let Some(ref cmd) = report.baseline.reth_command {
+            println!("  Command: {}", cmd);
+        }
         println!();
 
         println!("Feature Summary:");
@@ -628,6 +666,9 @@ impl ComparisonGenerator {
                 end.format("%Y-%m-%d %H:%M:%S UTC")
             );
         }
+        if let Some(ref cmd) = report.feature.reth_command {
+            println!("  Command: {}", cmd);
+        }
         println!();
     }
 }

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

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

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

@@ -2,7 +2,9 @@
 
 use crate::cli::Args;
 use alloy_provider::{Provider, ProviderBuilder};
+use alloy_rpc_client::RpcClient;
 use alloy_rpc_types_eth::SyncStatus;
+use alloy_transport_ws::WsConnect;
 use eyre::{eyre, OptionExt, Result, WrapErr};
 #[cfg(unix)]
 use nix::sys::signal::{killpg, Signal};
@@ -18,6 +20,9 @@ use tokio::{
 };
 use tracing::{debug, info, warn};
 
+/// Default websocket RPC port used by reth
+const DEFAULT_WS_RPC_PORT: u16 = 8546;
+
 /// Manages reth node lifecycle and operations
 pub(crate) struct NodeManager {
     datadir: Option<String>,
@@ -152,7 +157,10 @@ impl NodeManager {
             metrics_arg,
             "--http".to_string(),
             "--http.api".to_string(),
-            "eth".to_string(),
+            "eth,reth".to_string(),
+            "--ws".to_string(),
+            "--ws.api".to_string(),
+            "eth,reth".to_string(),
             "--disable-discovery".to_string(),
             "--trusted-only".to_string(),
         ]);
@@ -211,6 +219,11 @@ impl NodeManager {
         cmd.arg("--");
         cmd.args(reth_args);
 
+        // Enable tracing-samply
+        if supports_samply_flags(&reth_args[0]) {
+            cmd.arg("--log.samply");
+        }
+
         // Set environment variable to disable log styling
         cmd.env("RUST_LOG_STYLE", "never");
 
@@ -240,19 +253,24 @@ impl NodeManager {
     }
 
     /// Start a reth node using the specified binary path and return the process handle
+    /// along with the formatted reth command string for reporting.
     pub(crate) async fn start_node(
         &mut self,
         binary_path: &std::path::Path,
         _git_ref: &str,
         ref_type: &str,
         additional_args: &[String],
-    ) -> Result<tokio::process::Child> {
+    ) -> Result<(tokio::process::Child, String)> {
         // Store the binary path for later use (e.g., in unwind_to_block)
         self.binary_path = Some(binary_path.to_path_buf());
 
         let binary_path_str = binary_path.to_string_lossy();
         let (reth_args, _) = self.build_reth_args(&binary_path_str, additional_args, ref_type);
 
+        // Format the reth command string for reporting
+        let reth_command = shlex::try_join(reth_args.iter().map(|s| s.as_str()))
+            .wrap_err("Failed to format reth command string")?;
+
         // Log additional arguments if any
         if !self.additional_reth_args.is_empty() {
             info!("Using common additional reth arguments: {:?}", self.additional_reth_args);
@@ -346,11 +364,16 @@ impl NodeManager {
         // Give the node a moment to start up
         sleep(Duration::from_secs(5)).await;
 
-        Ok(child)
+        Ok((child, reth_command))
     }
 
-    /// Wait for the node to be ready and return its current tip
-    pub(crate) async fn wait_for_node_ready_and_get_tip(&self) -> Result<u64> {
+    /// Wait for the node to be ready and return its current tip.
+    ///
+    /// Fails early if the node process exits before becoming ready.
+    pub(crate) async fn wait_for_node_ready_and_get_tip(
+        &self,
+        child: &mut tokio::process::Child,
+    ) -> Result<u64> {
         info!("Waiting for node to be ready and synced...");
 
         let max_wait = Duration::from_secs(120); // 2 minutes to allow for sync
@@ -361,8 +384,23 @@ impl NodeManager {
         let url = rpc_url.parse().map_err(|e| eyre!("Invalid RPC URL '{}': {}", rpc_url, e))?;
         let provider = ProviderBuilder::new().connect_http(url);
 
+        let start_time = tokio::time::Instant::now();
+        let mut iteration = 0;
+
         timeout(max_wait, async {
             loop {
+                iteration += 1;
+                debug!(
+                    "Readiness check iteration {} (elapsed: {:?})",
+                    iteration,
+                    start_time.elapsed()
+                );
+
+                // Check if the node process has exited.
+                if let Some(status) = child.try_wait()? {
+                    return Err(eyre!("Node process exited unexpectedly with {status}"));
+                }
+
                 // First check if RPC is up and node is not syncing
                 match provider.syncing().await {
                     Ok(sync_result) => {
@@ -371,24 +409,48 @@ impl NodeManager {
                                 debug!("Node is still syncing {sync_info:?}, waiting...");
                             }
                             _ => {
+                                debug!("HTTP RPC is up and node is not syncing, checking block number...");
                                 // Node is not syncing, now get the tip
                                 match provider.get_block_number().await {
                                     Ok(tip) => {
-                                        info!("Node is ready and not syncing at block: {}", tip);
-                                        return Ok(tip);
+                                        debug!("HTTP RPC ready at block: {}, checking WebSocket...", tip);
+
+                                        // Verify WebSocket RPC is ready (public endpoint, no JWT required)
+                                        let ws_url = format!("ws://localhost:{}", DEFAULT_WS_RPC_PORT);
+                                        debug!("Attempting WebSocket connection to {} (public endpoint)", ws_url);
+                                        let ws_connect = WsConnect::new(&ws_url);
+
+                                        match RpcClient::connect_pubsub(ws_connect).await
+                                        {
+                                            Ok(_) => {
+                                                info!(
+                                                    "Node is ready (HTTP and WebSocket) at block: {} (took {:?}, {} iterations)",
+                                                    tip, start_time.elapsed(), iteration
+                                                );
+                                                return Ok(tip);
+                                            }
+                                            Err(e) => {
+                                                debug!(
+                                                    "HTTP RPC ready but WebSocket not ready yet (iteration {}): {:?}",
+                                                    iteration, e
+                                                );
+                                                debug!("WebSocket error details: {}", e);
+                                            }
+                                        }
                                     }
                                     Err(e) => {
-                                        debug!("Failed to get block number: {}", e);
+                                        debug!("Failed to get block number (iteration {}): {:?}", iteration, e);
                                     }
                                 }
                             }
                         }
                     }
                     Err(e) => {
-                        debug!("Node RPC not ready yet or failed to check sync status: {}", e);
+                        debug!("Node RPC not ready yet or failed to check sync status (iteration {}): {:?}", iteration, e);
                     }
                 }
 
+                debug!("Sleeping for {:?} before next check", check_interval);
                 sleep(check_interval).await;
             }
         })
@@ -396,9 +458,79 @@ impl NodeManager {
         .wrap_err("Timed out waiting for node to be ready and synced")?
     }
 
+    /// Wait for the node RPC to be ready and return its current tip, without waiting for sync.
+    ///
+    /// This is faster than `wait_for_node_ready_and_get_tip` but may return a tip while
+    /// the node is still syncing.
+    pub(crate) async fn wait_for_rpc_and_get_tip(
+        &self,
+        child: &mut tokio::process::Child,
+    ) -> Result<u64> {
+        info!("Waiting for node RPC to be ready (skipping sync wait)...");
+
+        let max_wait = Duration::from_secs(60);
+        let check_interval = Duration::from_secs(2);
+        let rpc_url = "http://localhost:8545";
+
+        let url = rpc_url.parse().map_err(|e| eyre!("Invalid RPC URL '{}': {}", rpc_url, e))?;
+        let provider = ProviderBuilder::new().connect_http(url);
+
+        let start_time = tokio::time::Instant::now();
+        let mut iteration = 0;
+
+        timeout(max_wait, async {
+            loop {
+                iteration += 1;
+                debug!(
+                    "RPC readiness check iteration {} (elapsed: {:?})",
+                    iteration,
+                    start_time.elapsed()
+                );
+
+                if let Some(status) = child.try_wait()? {
+                    return Err(eyre!("Node process exited unexpectedly with {status}"));
+                }
+
+                match provider.get_block_number().await {
+                    Ok(tip) => {
+                        debug!("HTTP RPC ready at block: {}, checking WebSocket...", tip);
+
+                        let ws_url = format!("ws://localhost:{}", DEFAULT_WS_RPC_PORT);
+                        let ws_connect = WsConnect::new(&ws_url);
+
+                        match RpcClient::connect_pubsub(ws_connect).await {
+                            Ok(_) => {
+                                info!(
+                                    "Node RPC is ready at block: {} (took {:?}, {} iterations)",
+                                    tip,
+                                    start_time.elapsed(),
+                                    iteration
+                                );
+                                return Ok(tip);
+                            }
+                            Err(e) => {
+                                debug!(
+                                    "HTTP RPC ready but WebSocket not ready yet (iteration {}): {:?}",
+                                    iteration, e
+                                );
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        debug!("RPC not ready yet (iteration {}): {:?}", iteration, e);
+                    }
+                }
+
+                sleep(check_interval).await;
+            }
+        })
+        .await
+        .wrap_err("Timed out waiting for node RPC to be ready")?
+    }
+
     /// Stop the reth node gracefully
     pub(crate) async fn stop_node(&self, child: &mut tokio::process::Child) -> Result<()> {
-        let pid = child.id().expect("Child process ID should be available");
+        let pid = child.id().ok_or_eyre("Child process ID should be available")?;
 
         // Check if the process has already exited
         match child.try_wait() {
@@ -547,3 +679,16 @@ impl NodeManager {
         Ok(())
     }
 }
+
+fn supports_samply_flags(bin: &str) -> bool {
+    let mut cmd = std::process::Command::new(bin);
+    // NOTE: The flag to check must come before --help.
+    // We pass --help as a shortcut to not execute any command.
+    cmd.args(["--log.samply", "--help"]);
+    debug!(?cmd, "Checking samply flags support");
+    let Ok(output) = cmd.output() else {
+        return false;
+    };
+    debug!(?output, "Samply flags support check");
+    output.status.success()
+}

bin/reth-bench/Cargo.toml

@@ -16,6 +16,7 @@ workspace = true
 # reth
 reth-cli-runner.workspace = true
 reth-cli-util.workspace = true
+reth-engine-primitives.workspace = true
 reth-fs-util.workspace = true
 reth-node-api.workspace = true
 reth-node-core.workspace = true
@@ -25,10 +26,11 @@ reth-tracing.workspace = true
 # alloy
 alloy-eips.workspace = true
 alloy-json-rpc.workspace = true
+alloy-network.workspace = true
 alloy-primitives.workspace = true
-alloy-provider = { workspace = true, features = ["engine-api", "reqwest-rustls-tls"], default-features = false }
+alloy-provider = { workspace = true, features = ["engine-api", "pubsub", "reqwest-rustls-tls"], default-features = false }
 alloy-pubsub.workspace = true
-alloy-rpc-client.workspace = true
+alloy-rpc-client = { workspace = true, features = ["pubsub"] }
 alloy-rpc-types-engine.workspace = true
 alloy-transport-http.workspace = true
 alloy-transport-ipc.workspace = true
@@ -50,6 +52,9 @@ tracing.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 
+# url parsing
+url.workspace = true
+
 # async
 async-trait.workspace = true
 futures.workspace = true
@@ -58,6 +63,7 @@ tokio = { workspace = true, features = ["sync", "macros", "time", "rt-multi-thre
 # misc
 clap = { workspace = true, features = ["derive", "env"] }
 eyre.workspace = true
+color-eyre.workspace = true
 thiserror.workspace = true
 humantime.workspace = true
 
@@ -79,7 +85,11 @@ jemalloc = [
     "reth-node-core/jemalloc",
 ]
 jemalloc-prof = ["reth-cli-util/jemalloc-prof"]
-tracy-allocator = ["reth-cli-util/tracy-allocator"]
+tracy-allocator = ["reth-cli-util/tracy-allocator", "tracy"]
+tracy = [
+    "reth-node-core/tracy",
+    "reth-tracing/tracy",
+]
 
 min-error-logs = [
     "tracing/release_max_level_error",

bin/reth-bench/README.md

@@ -31,6 +31,14 @@ Otherwise, running `make maxperf` at the root of the repo should be sufficient f
 `reth-bench` contains different commands to benchmark different patterns of engine API calls.
 The `reth-bench new-payload-fcu` command is the most representative of ethereum mainnet live sync, alternating between sending `engine_newPayload` calls and `engine_forkchoiceUpdated` calls.
 
+The `new-payload-fcu` command supports two optional waiting modes that can be used together or independently:
+- `--wait-time <duration>`: Fixed sleep interval between blocks (e.g., `--wait-time 100ms`)
+- `--wait-for-persistence`: Waits for blocks to be persisted using the `reth_subscribePersistedBlock` subscription
+
+When using `--wait-for-persistence`, the benchmark waits after every `(threshold + 1)` blocks, where the threshold defaults to the engine's persistence threshold (2). This can be customized with `--persistence-threshold <N>`.
+
+By default, the WebSocket URL for persistence subscriptions is derived from `--engine-rpc-url` (converting to ws:// on port 8546). Use `--ws-rpc-url` to override this.
+
 Below is an overview of how to run a benchmark:
 
 ### Setup
@@ -80,7 +88,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 +151,5 @@ To reproduce the benchmark, first re-set the node to the block that the benchmar
 - **RPC Configuration**: The RPC endpoints should be accessible and configured correctly, specifically the RPC endpoint must support `eth_getBlockByNumber` and support fetching full transactions. The benchmark will make one RPC query per block as fast as possible, so ensure the RPC endpoint does not rate limit or block requests after a certain volume.
 - **Reproducibility**: Ensure that the node is at the same state before attempting to retry a benchmark. The `new-payload-fcu` command specifically will commit to the database, so the node must be rolled back using `reth stage unwind` to reproducibly retry benchmarks.
 - **Profiling tools**: If you are collecting CPU profiles, tools like [`samply`](https://github.com/mstange/samply) and [`perf`](https://perf.wiki.kernel.org/index.php/Main_Page) can be useful for analyzing node performance.
-- **Benchmark Data**: `reth-bench` additionally contains a `--benchmark.output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
+- **Benchmark Data**: `reth-bench` additionally contains a `--output` flag, which will output gas used benchmarks across the benchmark range in CSV format. This may be useful for further data analysis.
 - **Platform Information**: To ensure accurate and reproducible benchmarking, document the platform details, including hardware specifications, OS version, and any other relevant information before publishing any benchmarks.

bin/reth-bench/src/authenticated_transport.rs

@@ -163,7 +163,7 @@ impl AuthenticatedTransport {
 
             // shift the iat forward by one second so there is some buffer time
             let mut shifted_claims = inner_and_claims.1;
-            shifted_claims.iat -= 1;
+            shifted_claims.iat -= 30;
 
             // if the claims are out of date, reset the inner transport
             if !shifted_claims.is_within_time_window() {

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

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

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

@@ -1,5 +1,13 @@
 //! Runs the `reth bench` command, calling first newPayload for each block, then calling
 //! forkchoiceUpdated.
+//!
+//! Supports configurable waiting behavior:
+//! - **`--wait-time`**: Fixed sleep interval between blocks.
+//! - **`--wait-for-persistence`**: Waits for every Nth block to be persisted using the
+//!   `reth_subscribePersistedBlock` subscription, where N matches the engine's persistence
+//!   threshold. This ensures the benchmark doesn't outpace persistence.
+//!
+//! Both options can be used together or independently.
 
 use crate::{
     bench::{
@@ -11,16 +19,26 @@ use crate::{
     },
     valid_payload::{block_to_new_payload, call_forkchoice_updated, call_new_payload},
 };
-use alloy_provider::Provider;
+use alloy_eips::BlockNumHash;
+use alloy_network::Ethereum;
+use alloy_provider::{Provider, RootProvider};
+use alloy_pubsub::SubscriptionStream;
+use alloy_rpc_client::RpcClient;
 use alloy_rpc_types_engine::ForkchoiceState;
+use alloy_transport_ws::WsConnect;
 use clap::Parser;
 use csv::Writer;
 use eyre::{Context, OptionExt};
+use futures::StreamExt;
 use humantime::parse_duration;
 use reth_cli_runner::CliContext;
+use reth_engine_primitives::config::DEFAULT_PERSISTENCE_THRESHOLD;
 use reth_node_core::args::BenchmarkArgs;
 use std::time::{Duration, Instant};
 use tracing::{debug, info};
+use url::Url;
+
+const PERSISTENCE_CHECKPOINT_TIMEOUT: Duration = Duration::from_secs(60);
 
 /// `reth benchmark new-payload-fcu` command
 #[derive(Debug, Parser)]
@@ -30,8 +48,31 @@ pub struct Command {
     rpc_url: String,
 
     /// How long to wait after a forkchoice update before sending the next payload.
-    #[arg(long, value_name = "WAIT_TIME", value_parser = parse_duration, default_value = "250ms", verbatim_doc_comment)]
-    wait_time: Duration,
+    #[arg(long, value_name = "WAIT_TIME", value_parser = parse_duration, verbatim_doc_comment)]
+    wait_time: Option<Duration>,
+
+    /// Wait for blocks to be persisted before sending the next batch.
+    ///
+    /// When enabled, waits for every Nth block to be persisted using the
+    /// `reth_subscribePersistedBlock` subscription. This ensures the benchmark
+    /// doesn't outpace persistence.
+    ///
+    /// The subscription uses the regular RPC websocket endpoint (no JWT required).
+    #[arg(long, default_value = "false", verbatim_doc_comment)]
+    wait_for_persistence: bool,
+
+    /// Engine persistence threshold used for deciding when to wait for persistence.
+    ///
+    /// The benchmark waits after every `(threshold + 1)` blocks. By default this
+    /// matches the engine's `DEFAULT_PERSISTENCE_THRESHOLD` (2), so waits occur
+    /// at blocks 3, 6, 9, etc.
+    #[arg(
+        long = "persistence-threshold",
+        value_name = "PERSISTENCE_THRESHOLD",
+        default_value_t = DEFAULT_PERSISTENCE_THRESHOLD,
+        verbatim_doc_comment
+    )]
+    persistence_threshold: u64,
 
     /// The size of the block buffer (channel capacity) for prefetching blocks from the RPC
     /// endpoint.
@@ -50,6 +91,32 @@ pub struct Command {
 impl Command {
     /// Execute `benchmark new-payload-fcu` command
     pub async fn execute(self, _ctx: CliContext) -> eyre::Result<()> {
+        // Log mode configuration
+        if let Some(duration) = self.wait_time {
+            info!("Using wait-time mode with {}ms delay between blocks", duration.as_millis());
+        }
+        if self.wait_for_persistence {
+            info!(
+                "Persistence waiting enabled (waits after every {} blocks to match engine gap > {} behavior)",
+                self.persistence_threshold + 1,
+                self.persistence_threshold
+            );
+        }
+
+        // Set up waiter based on configured options (duration takes precedence)
+        let mut waiter = match (self.wait_time, self.wait_for_persistence) {
+            (Some(duration), _) => Some(PersistenceWaiter::with_duration(duration)),
+            (None, true) => {
+                let sub = self.setup_persistence_subscription().await?;
+                Some(PersistenceWaiter::with_subscription(
+                    sub,
+                    self.persistence_threshold,
+                    PERSISTENCE_CHECKPOINT_TIMEOUT,
+                ))
+            }
+            (None, false) => None,
+        };
+
         let BenchContext {
             benchmark_mode,
             block_provider,
@@ -110,7 +177,6 @@ impl Command {
             }
         });
 
-        // put results in a summary vec so they can be printed at the end
         let mut results = Vec::new();
         let total_benchmark_duration = Instant::now();
         let mut total_wait_time = Duration::ZERO;
@@ -121,14 +187,12 @@ impl Command {
             total_wait_time += wait_start.elapsed();
             result
         } {
-            // just put gas used here
             let gas_used = block.header.gas_used;
             let block_number = block.header.number;
             let transaction_count = block.transactions.len() as u64;
 
-            debug!(target: "reth-bench", ?block_number, "Sending payload",);
+            debug!(target: "reth-bench", ?block_number, "Sending payload");
 
-            // construct fcu to call
             let forkchoice_state = ForkchoiceState {
                 head_block_hash: head,
                 safe_block_hash: safe,
@@ -143,7 +207,6 @@ impl Command {
 
             call_forkchoice_updated(&auth_provider, version, forkchoice_state, None).await?;
 
-            // calculate the total duration and the fcu latency, record
             let total_latency = start.elapsed();
             let fcu_latency = total_latency - new_payload_result.latency;
             let combined_result = CombinedResult {
@@ -154,17 +217,15 @@ impl Command {
                 total_latency,
             };
 
-            // current duration since the start of the benchmark minus the time
-            // waiting for blocks
+            // Exclude time spent waiting on the block prefetch channel from the benchmark duration.
+            // We want to measure engine throughput, not RPC fetch latency.
             let current_duration = total_benchmark_duration.elapsed() - total_wait_time;
-
-            // convert gas used to gigagas, then compute gigagas per second
             info!(%combined_result);
 
-            // wait before sending the next payload
-            tokio::time::sleep(self.wait_time).await;
+            if let Some(w) = &mut waiter {
+                w.on_block(block_number).await?;
+            }
 
-            // record the current result
             let gas_row =
                 TotalGasRow { block_number, transaction_count, gas_used, time: current_duration };
             results.push((gas_row, combined_result));
@@ -175,24 +236,26 @@ impl Command {
             return Err(error);
         }
 
+        // Drop waiter - we don't need to wait for final blocks to persist
+        // since the benchmark goal is measuring Ggas/s of newPayload/FCU, not persistence.
+        drop(waiter);
+
         let (gas_output_results, combined_results): (_, Vec<CombinedResult>) =
             results.into_iter().unzip();
 
-        // write the csv output to files
-        if let Some(path) = self.benchmark.output {
-            // first write the combined results to a file
+        // Write CSV output files
+        if let Some(ref path) = self.benchmark.output {
             let output_path = path.join(COMBINED_OUTPUT_SUFFIX);
             info!("Writing engine api call latency output to file: {:?}", output_path);
-            let mut writer = Writer::from_path(output_path)?;
+            let mut writer = Writer::from_path(&output_path)?;
             for result in combined_results {
                 writer.serialize(result)?;
             }
             writer.flush()?;
 
-            // now write the gas output to a file
             let output_path = path.join(GAS_OUTPUT_SUFFIX);
             info!("Writing total gas output to file: {:?}", output_path);
-            let mut writer = Writer::from_path(output_path)?;
+            let mut writer = Writer::from_path(&output_path)?;
             for row in &gas_output_results {
                 writer.serialize(row)?;
             }
@@ -201,8 +264,8 @@ impl Command {
             info!("Finished writing benchmark output files to {:?}.", path);
         }
 
-        // accumulate the results and calculate the overall Ggas/s
         let gas_output = TotalGasOutput::new(gas_output_results)?;
+
         info!(
             total_duration=?gas_output.total_duration,
             total_gas_used=?gas_output.total_gas_used,
@@ -213,4 +276,278 @@ impl Command {
 
         Ok(())
     }
+
+    /// Returns the websocket RPC URL used for the persistence subscription.
+    ///
+    /// Preference:
+    /// - If `--ws-rpc-url` is provided, use it directly.
+    /// - Otherwise, derive a WS RPC URL from `--engine-rpc-url`.
+    ///
+    /// The persistence subscription endpoint (`reth_subscribePersistedBlock`) is exposed on
+    /// the regular RPC server (WS port, usually 8546), not on the engine API port (usually 8551).
+    /// Since `BenchmarkArgs` only has the engine URL by default, we convert the scheme
+    /// (http→ws, https→wss) and force the port to 8546.
+    fn derive_ws_rpc_url(&self) -> eyre::Result<Url> {
+        if let Some(ref ws_url) = self.benchmark.ws_rpc_url {
+            let parsed: Url = ws_url
+                .parse()
+                .wrap_err_with(|| format!("Failed to parse WebSocket RPC URL: {ws_url}"))?;
+            info!(target: "reth-bench", ws_url = %parsed, "Using provided WebSocket RPC URL");
+            Ok(parsed)
+        } else {
+            let derived = engine_url_to_ws_url(&self.benchmark.engine_rpc_url)?;
+            debug!(
+                target: "reth-bench",
+                engine_url = %self.benchmark.engine_rpc_url,
+                %derived,
+                "Derived WebSocket RPC URL from engine RPC URL"
+            );
+            Ok(derived)
+        }
+    }
+
+    /// Establishes a websocket connection and subscribes to `reth_subscribePersistedBlock`.
+    async fn setup_persistence_subscription(&self) -> eyre::Result<PersistenceSubscription> {
+        let ws_url = self.derive_ws_rpc_url()?;
+
+        info!("Connecting to WebSocket at {} for persistence subscription", ws_url);
+
+        let ws_connect = WsConnect::new(ws_url.to_string());
+        let client = RpcClient::connect_pubsub(ws_connect)
+            .await
+            .wrap_err("Failed to connect to WebSocket RPC endpoint")?;
+        let provider: RootProvider<Ethereum> = RootProvider::new(client);
+
+        let subscription = provider
+            .subscribe_to::<BlockNumHash>("reth_subscribePersistedBlock")
+            .await
+            .wrap_err("Failed to subscribe to persistence notifications")?;
+
+        info!("Subscribed to persistence notifications");
+
+        Ok(PersistenceSubscription::new(provider, subscription.into_stream()))
+    }
+}
+
+/// Converts an engine API URL to the default RPC websocket URL.
+///
+/// Transformations:
+/// - `http`  → `ws`
+/// - `https` → `wss`
+/// - `ws` / `wss` keep their scheme
+/// - Port is always set to `8546`, reth's default RPC websocket port.
+///
+/// This is used when we only know the engine API URL (typically `:8551`) but
+/// need to connect to the node's WS RPC endpoint for persistence events.
+fn engine_url_to_ws_url(engine_url: &str) -> eyre::Result<Url> {
+    let url: Url = engine_url
+        .parse()
+        .wrap_err_with(|| format!("Failed to parse engine RPC URL: {engine_url}"))?;
+
+    let mut ws_url = url.clone();
+
+    match ws_url.scheme() {
+        "http" => ws_url
+            .set_scheme("ws")
+            .map_err(|_| eyre::eyre!("Failed to set WS scheme for URL: {url}"))?,
+        "https" => ws_url
+            .set_scheme("wss")
+            .map_err(|_| eyre::eyre!("Failed to set WSS scheme for URL: {url}"))?,
+        "ws" | "wss" => {}
+        scheme => {
+            return Err(eyre::eyre!(
+            "Unsupported URL scheme '{scheme}' for URL: {url}. Expected http, https, ws, or wss."
+        ))
+        }
+    }
+
+    ws_url.set_port(Some(8546)).map_err(|_| eyre::eyre!("Failed to set port for URL: {url}"))?;
+
+    Ok(ws_url)
+}
+
+/// Waits until the persistence subscription reports that `target` has been persisted.
+///
+/// Consumes subscription events until `last_persisted >= target`, or returns an error if:
+/// - the subscription stream ends unexpectedly, or
+/// - `timeout` elapses before `target` is observed.
+async fn wait_for_persistence(
+    stream: &mut SubscriptionStream<BlockNumHash>,
+    target: u64,
+    last_persisted: &mut u64,
+    timeout: Duration,
+) -> eyre::Result<()> {
+    tokio::time::timeout(timeout, async {
+        while *last_persisted < target {
+            match stream.next().await {
+                Some(persisted) => {
+                    *last_persisted = persisted.number;
+                    debug!(
+                        target: "reth-bench",
+                        persisted_block = ?last_persisted,
+                        "Received persistence notification"
+                    );
+                }
+                None => {
+                    return Err(eyre::eyre!("Persistence subscription closed unexpectedly"));
+                }
+            }
+        }
+        Ok(())
+    })
+    .await
+    .map_err(|_| {
+        eyre::eyre!(
+            "Persistence timeout: target block {} not persisted within {:?}. Last persisted: {}",
+            target,
+            timeout,
+            last_persisted
+        )
+    })?
+}
+
+/// Wrapper that keeps both the subscription stream and the underlying provider alive.
+/// The provider must be kept alive for the subscription to continue receiving events.
+struct PersistenceSubscription {
+    _provider: RootProvider<Ethereum>,
+    stream: SubscriptionStream<BlockNumHash>,
+}
+
+impl PersistenceSubscription {
+    const fn new(
+        provider: RootProvider<Ethereum>,
+        stream: SubscriptionStream<BlockNumHash>,
+    ) -> Self {
+        Self { _provider: provider, stream }
+    }
+
+    const fn stream_mut(&mut self) -> &mut SubscriptionStream<BlockNumHash> {
+        &mut self.stream
+    }
+}
+
+/// Encapsulates the block waiting logic.
+///
+/// Provides a simple `on_block()` interface that handles both:
+/// - Fixed duration waits (when `wait_time` is set)
+/// - Persistence-based waits (when `subscription` is set)
+///
+/// For persistence mode, waits after every `(threshold + 1)` blocks.
+struct PersistenceWaiter {
+    wait_time: Option<Duration>,
+    subscription: Option<PersistenceSubscription>,
+    blocks_sent: u64,
+    last_persisted: u64,
+    threshold: u64,
+    timeout: Duration,
+}
+
+impl PersistenceWaiter {
+    const fn with_duration(wait_time: Duration) -> Self {
+        Self {
+            wait_time: Some(wait_time),
+            subscription: None,
+            blocks_sent: 0,
+            last_persisted: 0,
+            threshold: 0,
+            timeout: Duration::ZERO,
+        }
+    }
+
+    const fn with_subscription(
+        subscription: PersistenceSubscription,
+        threshold: u64,
+        timeout: Duration,
+    ) -> Self {
+        Self {
+            wait_time: None,
+            subscription: Some(subscription),
+            blocks_sent: 0,
+            last_persisted: 0,
+            threshold,
+            timeout,
+        }
+    }
+
+    /// Called once per block. Waits based on the configured mode.
+    #[allow(clippy::manual_is_multiple_of)]
+    async fn on_block(&mut self, block_number: u64) -> eyre::Result<()> {
+        if let Some(wait_time) = self.wait_time {
+            tokio::time::sleep(wait_time).await;
+            return Ok(());
+        }
+
+        let Some(ref mut subscription) = self.subscription else {
+            return Ok(());
+        };
+
+        self.blocks_sent += 1;
+
+        if self.blocks_sent % (self.threshold + 1) == 0 {
+            debug!(
+                target: "reth-bench",
+                target_block = ?block_number,
+                last_persisted = self.last_persisted,
+                blocks_sent = self.blocks_sent,
+                "Waiting for persistence"
+            );
+
+            wait_for_persistence(
+                subscription.stream_mut(),
+                block_number,
+                &mut self.last_persisted,
+                self.timeout,
+            )
+            .await?;
+
+            debug!(
+                target: "reth-bench",
+                persisted = self.last_persisted,
+                "Persistence caught up"
+            );
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_engine_url_to_ws_url() {
+        // http -> ws, always uses port 8546
+        let result = engine_url_to_ws_url("http://localhost:8551").unwrap();
+        assert_eq!(result.as_str(), "ws://localhost:8546/");
+
+        // https -> wss
+        let result = engine_url_to_ws_url("https://localhost:8551").unwrap();
+        assert_eq!(result.as_str(), "wss://localhost:8546/");
+
+        // Custom engine port still maps to 8546
+        let result = engine_url_to_ws_url("http://localhost:9551").unwrap();
+        assert_eq!(result.port(), Some(8546));
+
+        // Already ws passthrough
+        let result = engine_url_to_ws_url("ws://localhost:8546").unwrap();
+        assert_eq!(result.scheme(), "ws");
+
+        // Invalid inputs
+        assert!(engine_url_to_ws_url("ftp://localhost:8551").is_err());
+        assert!(engine_url_to_ws_url("not a valid url").is_err());
+    }
+
+    #[tokio::test]
+    async fn test_waiter_with_duration() {
+        let mut waiter = PersistenceWaiter::with_duration(Duration::from_millis(1));
+
+        let start = Instant::now();
+        waiter.on_block(1).await.unwrap();
+        waiter.on_block(2).await.unwrap();
+        waiter.on_block(3).await.unwrap();
+
+        // Should have waited ~3ms total
+        assert!(start.elapsed() >= Duration::from_millis(3));
+    }
 }

bin/reth-bench/src/bench_mode.rs

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

bin/reth-bench/src/main.rs

@@ -23,7 +23,7 @@ use bench::BenchmarkCommand;
 use clap::Parser;
 use reth_cli_runner::CliRunner;
 
-fn main() {
+fn main() -> eyre::Result<()> {
     // Enable backtraces unless a RUST_BACKTRACE value has already been explicitly provided.
     if std::env::var_os("RUST_BACKTRACE").is_none() {
         unsafe {
@@ -31,12 +31,11 @@ fn main() {
         }
     }
 
+    color_eyre::install()?;
+
     // Run until either exit or sigint or sigterm
-    let runner = CliRunner::try_default_runtime().unwrap();
-    runner
-        .run_command_until_exit(|ctx| {
-            let command = BenchmarkCommand::parse();
-            command.execute(ctx)
-        })
-        .unwrap();
+    let runner = CliRunner::try_default_runtime()?;
+    runner.run_command_until_exit(|ctx| BenchmarkCommand::parse().execute(ctx))?;
+
+    Ok(())
 }

bin/reth/Cargo.toml

@@ -81,7 +81,7 @@ backon.workspace = true
 tempfile.workspace = true
 
 [features]
-default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer"]
+default = ["jemalloc", "otlp", "reth-revm/portable", "js-tracer", "keccak-cache-global", "asm-keccak"]
 
 otlp = [
     "reth-ethereum-cli/otlp",
@@ -102,7 +102,10 @@ asm-keccak = [
     "reth-ethereum-cli/asm-keccak",
     "reth-node-ethereum/asm-keccak",
 ]
-
+keccak-cache-global = [
+    "reth-node-core/keccak-cache-global",
+    "reth-node-ethereum/keccak-cache-global",
+]
 jemalloc = [
     "reth-cli-util/jemalloc",
     "reth-node-core/jemalloc",
@@ -113,6 +116,11 @@ jemalloc-prof = [
     "reth-cli-util/jemalloc",
     "reth-cli-util/jemalloc-prof",
     "reth-ethereum-cli/jemalloc-prof",
+    "reth-node-metrics/jemalloc-prof",
+]
+jemalloc-symbols = [
+    "jemalloc-prof",
+    "reth-ethereum-cli/jemalloc-symbols",
 ]
 jemalloc-unprefixed = [
     "reth-cli-util/jemalloc-unprefixed",
@@ -123,6 +131,11 @@ jemalloc-unprefixed = [
 tracy-allocator = [
     "reth-cli-util/tracy-allocator",
     "reth-ethereum-cli/tracy-allocator",
+    "tracy",
+]
+tracy = [
+    "reth-ethereum-cli/tracy",
+    "reth-node-core/tracy",
 ]
 
 # Because jemalloc is default and preferred over snmalloc when both features are
@@ -163,6 +176,8 @@ min-trace-logs = [
     "reth-node-core/min-trace-logs",
 ]
 
+edge = ["reth-ethereum-cli/edge", "reth-node-core/edge"]
+
 [[bin]]
 name = "reth"
 path = "src/main.rs"

bin/reth/src/lib.rs

@@ -2,22 +2,46 @@
 //!
 //! ## Feature Flags
 //!
+//! ### Default Features
+//!
 //! - `jemalloc`: Uses [jemallocator](https://github.com/tikv/jemallocator) as the global allocator.
 //!   This is **not recommended on Windows**. See [here](https://rust-lang.github.io/rfcs/1974-global-allocators.html#jemalloc)
 //!   for more info.
+//! - `otlp`: Enables [OpenTelemetry](https://opentelemetry.io/) metrics export to a configured OTLP
+//!   collector endpoint.
+//! - `js-tracer`: Enables the `JavaScript` tracer for the `debug_trace` endpoints, allowing custom
+//!   `JavaScript`-based transaction tracing.
+//! - `keccak-cache-global`: Enables global caching for Keccak256 hashes to improve performance.
+//! - `asm-keccak`: Replaces the default, pure-Rust implementation of Keccak256 with one implemented
+//!   in assembly; see [the `keccak-asm` crate](https://github.com/DaniPopes/keccak-asm) for more
+//!   details and supported targets.
+//!
+//! ### Allocator Features
+//!
 //! - `jemalloc-prof`: Enables [jemallocator's](https://github.com/tikv/jemallocator) heap profiling
 //!   and leak detection functionality. See [jemalloc's opt.prof](https://jemalloc.net/jemalloc.3.html#opt.prof)
-//!   documentation for usage details. This is **not recommended on Windows**. See [here](https://rust-lang.github.io/rfcs/1974-global-allocators.html#jemalloc)
-//!   for more info.
-//! - `asm-keccak`: replaces the default, pure-Rust implementation of Keccak256 with one implemented
-//!   in assembly; see [the `keccak-asm` crate](https://github.com/DaniPopes/keccak-asm) for more
-//!   details and supported targets
+//!   documentation for usage details. This is **not recommended on Windows**.
+//! - `jemalloc-symbols`: Enables jemalloc symbols for profiling. Includes `jemalloc-prof`.
+//! - `jemalloc-unprefixed`: Uses unprefixed jemalloc symbols.
+//! - `tracy-allocator`: Enables [Tracy](https://github.com/wolfpld/tracy) profiler allocator
+//!   integration for memory profiling.
+//! - `snmalloc`: Uses [snmalloc](https://github.com/snmalloc/snmalloc) as the global allocator. Use
+//!   `--no-default-features` when enabling this, as jemalloc takes precedence.
+//! - `snmalloc-native`: Uses snmalloc with native CPU optimizations. Use `--no-default-features`
+//!   when enabling this.
+//!
+//! ### Log Level Features
+//!
 //! - `min-error-logs`: Disables all logs below `error` level.
 //! - `min-warn-logs`: Disables all logs below `warn` level.
 //! - `min-info-logs`: Disables all logs below `info` level. This can speed up the node, since fewer
 //!   calls to the logging component are made.
 //! - `min-debug-logs`: Disables all logs below `debug` level.
 //! - `min-trace-logs`: Disables all logs below `trace` level.
+//!
+//! ### Development Features
+//!
+//! - `dev`: Enables development mode features, including test vector generation commands.
 
 #![doc(
     html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
@@ -170,7 +194,7 @@ pub mod rpc {
         pub use reth_rpc::eth::*;
     }
 
-    /// Re-exported from `reth_rpc::rpc`.
+    /// Re-exported from `reth_rpc_server_types::result`.
     pub mod result {
         pub use reth_rpc_server_types::result::*;
     }

bin/reth/src/main.rs

@@ -3,6 +3,10 @@
 #[global_allocator]
 static ALLOC: reth_cli_util::allocator::Allocator = reth_cli_util::allocator::new_allocator();
 
+#[cfg(all(feature = "jemalloc-prof", unix))]
+#[unsafe(export_name = "_rjem_malloc_conf")]
+static MALLOC_CONF: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0";
+
 use clap::Parser;
 use reth::{args::RessArgs, cli::Cli, ress::install_ress_subprotocol};
 use reth_ethereum_cli::chainspec::EthereumChainSpecParser;

crates/chain-state/src/chain_info.rs

@@ -33,6 +33,7 @@ where
     ) -> Self {
         let (finalized_block, _) = watch::channel(finalized);
         let (safe_block, _) = watch::channel(safe);
+        let (persisted_block, _) = watch::channel(None);
 
         Self {
             inner: Arc::new(ChainInfoInner {
@@ -42,6 +43,7 @@ where
                 canonical_head: RwLock::new(head),
                 safe_block,
                 finalized_block,
+                persisted_block,
             }),
         }
     }
@@ -97,6 +99,11 @@ where
         self.inner.finalized_block.borrow().as_ref().map(SealedHeader::num_hash)
     }
 
+    /// Returns the `BlockNumHash` of the persisted block.
+    pub fn get_persisted_num_hash(&self) -> Option<BlockNumHash> {
+        *self.inner.persisted_block.borrow()
+    }
+
     /// Sets the canonical head of the chain.
     pub fn set_canonical_head(&self, header: SealedHeader<N::BlockHeader>) {
         let number = header.number();
@@ -130,6 +137,18 @@ where
         });
     }
 
+    /// Sets the persisted block of the chain.
+    pub fn set_persisted(&self, num_hash: BlockNumHash) {
+        self.inner.persisted_block.send_if_modified(|current| {
+            if current.map(|b| b.hash) != Some(num_hash.hash) {
+                let _ = current.replace(num_hash);
+                return true
+            }
+
+            false
+        });
+    }
+
     /// Subscribe to the finalized block.
     pub fn subscribe_finalized_block(
         &self,
@@ -141,6 +160,11 @@ where
     pub fn subscribe_safe_block(&self) -> watch::Receiver<Option<SealedHeader<N::BlockHeader>>> {
         self.inner.safe_block.subscribe()
     }
+
+    /// Subscribe to the persisted block.
+    pub fn subscribe_persisted_block(&self) -> watch::Receiver<Option<BlockNumHash>> {
+        self.inner.persisted_block.subscribe()
+    }
 }
 
 /// Container type for all chain info fields
@@ -159,11 +183,14 @@ struct ChainInfoInner<N: NodePrimitives = reth_ethereum_primitives::EthPrimitive
     safe_block: watch::Sender<Option<SealedHeader<N::BlockHeader>>>,
     /// The block that the beacon node considers finalized.
     finalized_block: watch::Sender<Option<SealedHeader<N::BlockHeader>>>,
+    /// The last block that was persisted to disk.
+    persisted_block: watch::Sender<Option<BlockNumHash>>,
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use alloy_primitives::B256;
     use reth_ethereum_primitives::EthPrimitives;
     use reth_testing_utils::{generators, generators::random_header};
 
@@ -338,4 +365,28 @@ mod tests {
         // Assert that the BlockNumHash returned matches the safe header
         assert_eq!(tracker.get_safe_num_hash(), Some(safe_header.num_hash()));
     }
+
+    #[test]
+    fn test_set_persisted() {
+        let mut rng = generators::rng();
+        let header = random_header(&mut rng, 10, None);
+        let tracker: ChainInfoTracker<EthPrimitives> = ChainInfoTracker::new(header, None, None);
+
+        // Initial state: persisted block should be None
+        assert!(tracker.get_persisted_num_hash().is_none());
+
+        // Set a persisted block
+        let num_hash1 = BlockNumHash::new(10, B256::random());
+        tracker.set_persisted(num_hash1);
+        assert_eq!(tracker.get_persisted_num_hash(), Some(num_hash1));
+
+        // Setting the same block again should not change anything
+        tracker.set_persisted(num_hash1);
+        assert_eq!(tracker.get_persisted_num_hash(), Some(num_hash1));
+
+        // Set a different block
+        let num_hash2 = BlockNumHash::new(20, B256::random());
+        tracker.set_persisted(num_hash2);
+        assert_eq!(tracker.get_persisted_num_hash(), Some(num_hash2));
+    }
 }

crates/chain-state/src/deferred_trie.rs

@@ -37,12 +37,19 @@ pub struct ComputedTrieData {
 
 /// Trie input bundled with its anchor hash.
 ///
-/// This is used to store the trie input and anchor hash for a block together.
+/// The `trie_input` contains the **cumulative** overlay of all in-memory ancestor blocks,
+/// not just this block's changes. Child blocks reuse the parent's overlay in O(1) by
+/// cloning the Arc-wrapped data.
+///
+/// The `anchor_hash` is metadata indicating which persisted base state this overlay
+/// sits on top of. It is CRITICAL for overlay reuse decisions: an overlay built on top
+/// of Anchor A cannot be reused for a block anchored to Anchor B, as it would result
+/// in an incorrect state.
 #[derive(Clone, Debug)]
 pub struct AnchoredTrieInput {
     /// The persisted ancestor hash this trie input is anchored to.
     pub anchor_hash: B256,
-    /// Trie input constructed from in-memory overlays.
+    /// Cumulative trie input overlay from all in-memory ancestors.
     pub trie_input: Arc<TrieInputSorted>,
 }
 
@@ -62,7 +69,8 @@ static DEFERRED_TRIE_METRICS: LazyLock<DeferredTrieMetrics> =
 /// Internal state for deferred trie data.
 enum DeferredState {
     /// Data is not yet available; raw inputs stored for fallback computation.
-    Pending(PendingInputs),
+    /// Wrapped in `Option` to allow taking ownership during computation.
+    Pending(Option<PendingInputs>),
     /// Data has been computed and is ready.
     Ready(ComputedTrieData),
 }
@@ -112,12 +120,12 @@ impl DeferredTrieData {
         ancestors: Vec<Self>,
     ) -> Self {
         Self {
-            state: Arc::new(Mutex::new(DeferredState::Pending(PendingInputs {
+            state: Arc::new(Mutex::new(DeferredState::Pending(Some(PendingInputs {
                 hashed_state,
                 trie_updates,
                 anchor_hash,
                 ancestors,
-            }))),
+            })))),
         }
     }
 
@@ -138,8 +146,9 @@ impl DeferredTrieData {
     ///
     /// # Process
     /// 1. Sort the current block's hashed state and trie updates
-    /// 2. Merge ancestor overlays (oldest -> newest, so later state takes precedence)
-    /// 3. Extend the merged overlay with this block's sorted data
+    /// 2. Reuse parent's cached overlay if available (O(1) - the common case)
+    /// 3. Otherwise, rebuild overlay from ancestors (rare fallback)
+    /// 4. Extend the overlay with this block's sorted data
     ///
     /// Used by both the async background task and the synchronous fallback path.
     ///
@@ -147,49 +156,103 @@ impl DeferredTrieData {
     /// * `hashed_state` - Unsorted hashed post-state (account/storage changes) from execution
     /// * `trie_updates` - Unsorted trie node updates from state root computation
     /// * `anchor_hash` - The persisted ancestor hash this trie input is anchored to
-    /// * `ancestors` - Deferred trie data from ancestor blocks for merging
+    /// * `ancestors` - Deferred trie data from ancestor blocks for merging (oldest -> newest)
     pub fn sort_and_build_trie_input(
-        hashed_state: &HashedPostState,
-        trie_updates: &TrieUpdates,
+        hashed_state: Arc<HashedPostState>,
+        trie_updates: Arc<TrieUpdates>,
         anchor_hash: B256,
         ancestors: &[Self],
     ) -> ComputedTrieData {
-        // Sort the current block's hashed state and trie updates
-        let sorted_hashed_state = Arc::new(hashed_state.clone_into_sorted());
-        let sorted_trie_updates = Arc::new(trie_updates.clone().into_sorted());
+        let sorted_hashed_state = match Arc::try_unwrap(hashed_state) {
+            Ok(state) => state.into_sorted(),
+            Err(arc) => arc.clone_into_sorted(),
+        };
+        let sorted_trie_updates = match Arc::try_unwrap(trie_updates) {
+            Ok(updates) => updates.into_sorted(),
+            Err(arc) => arc.clone_into_sorted(),
+        };
 
-        // Merge trie data from ancestors (oldest -> newest so later state takes precedence)
-        let mut overlay = TrieInputSorted::default();
-        for ancestor in ancestors {
-            let ancestor_data = ancestor.wait_cloned();
-            {
-                let state_mut = Arc::make_mut(&mut overlay.state);
-                state_mut.extend_ref(ancestor_data.hashed_state.as_ref());
-            }
-            {
-                let nodes_mut = Arc::make_mut(&mut overlay.nodes);
-                nodes_mut.extend_ref(ancestor_data.trie_updates.as_ref());
-            }
-        }
+        // Reuse parent's overlay if available and anchors match.
+        // We can only reuse the parent's overlay if it was built on top of the same
+        // persisted anchor. If the anchor has changed (e.g., due to persistence),
+        // the parent's overlay is relative to an old state and cannot be used.
+        let overlay = if let Some(parent) = ancestors.last() {
+            let parent_data = parent.wait_cloned();
 
-        // Extend overlay with current block's sorted data
-        {
-            let state_mut = Arc::make_mut(&mut overlay.state);
-            state_mut.extend_ref(sorted_hashed_state.as_ref());
-        }
-        {
-            let nodes_mut = Arc::make_mut(&mut overlay.nodes);
-            nodes_mut.extend_ref(sorted_trie_updates.as_ref());
-        }
+            match &parent_data.anchored_trie_input {
+                // Case 1: Parent has cached overlay AND anchors match.
+                Some(AnchoredTrieInput { anchor_hash: parent_anchor, trie_input })
+                    if *parent_anchor == anchor_hash =>
+                {
+                    // O(1): Reuse parent's overlay, extend with current block's data.
+                    let mut overlay = TrieInputSorted::new(
+                        Arc::clone(&trie_input.nodes),
+                        Arc::clone(&trie_input.state),
+                        Default::default(), // prefix_sets are per-block, not cumulative
+                    );
+                    // Only trigger COW clone if there's actually data to add.
+                    if !sorted_hashed_state.is_empty() {
+                        Arc::make_mut(&mut overlay.state).extend_ref(&sorted_hashed_state);
+                    }
+                    if !sorted_trie_updates.is_empty() {
+                        Arc::make_mut(&mut overlay.nodes).extend_ref(&sorted_trie_updates);
+                    }
+                    overlay
+                }
+                // Case 2: Parent exists but anchor mismatch or no cached overlay.
+                // We must rebuild from the ancestors list (which only contains unpersisted blocks).
+                _ => Self::merge_ancestors_into_overlay(
+                    ancestors,
+                    &sorted_hashed_state,
+                    &sorted_trie_updates,
+                ),
+            }
+        } else {
+            // Case 3: No in-memory ancestors (first block after persisted anchor).
+            // Build overlay with just this block's data.
+            Self::merge_ancestors_into_overlay(&[], &sorted_hashed_state, &sorted_trie_updates)
+        };
 
         ComputedTrieData::with_trie_input(
-            sorted_hashed_state,
-            sorted_trie_updates,
+            Arc::new(sorted_hashed_state),
+            Arc::new(sorted_trie_updates),
             anchor_hash,
             Arc::new(overlay),
         )
     }
 
+    /// Merge all ancestors and current block's data into a single overlay.
+    ///
+    /// This is a rare fallback path, only used when no ancestor has a cached
+    /// `anchored_trie_input` (e.g., blocks created via alternative constructors).
+    /// In normal operation, the parent always has a cached overlay and this
+    /// function is never called.
+    ///
+    /// Iterates ancestors oldest -> newest, then extends with current block's data,
+    /// so later state takes precedence.
+    fn merge_ancestors_into_overlay(
+        ancestors: &[Self],
+        sorted_hashed_state: &HashedPostStateSorted,
+        sorted_trie_updates: &TrieUpdatesSorted,
+    ) -> TrieInputSorted {
+        let mut overlay = TrieInputSorted::default();
+
+        let state_mut = Arc::make_mut(&mut overlay.state);
+        let nodes_mut = Arc::make_mut(&mut overlay.nodes);
+
+        for ancestor in ancestors {
+            let ancestor_data = ancestor.wait_cloned();
+            state_mut.extend_ref(ancestor_data.hashed_state.as_ref());
+            nodes_mut.extend_ref(ancestor_data.trie_updates.as_ref());
+        }
+
+        // Extend with current block's sorted data last (takes precedence)
+        state_mut.extend_ref(sorted_hashed_state);
+        nodes_mut.extend_ref(sorted_trie_updates);
+
+        overlay
+    }
+
     /// Returns trie data, computing synchronously if the async task hasn't completed.
     ///
     /// - If the async task has completed (`Ready`), returns the cached result.
@@ -204,7 +267,7 @@ impl DeferredTrieData {
     #[instrument(level = "debug", target = "engine::tree::deferred_trie", skip_all)]
     pub fn wait_cloned(&self) -> ComputedTrieData {
         let mut state = self.state.lock();
-        match &*state {
+        match &mut *state {
             // If the deferred trie data is ready, return the cached result.
             DeferredState::Ready(bundle) => {
                 DEFERRED_TRIE_METRICS.deferred_trie_async_ready.increment(1);
@@ -212,11 +275,14 @@ impl DeferredTrieData {
             }
             // If the deferred trie data is pending, compute the trie data synchronously and return
             // the result. This is the fallback path if the async task hasn't completed.
-            DeferredState::Pending(inputs) => {
+            DeferredState::Pending(maybe_inputs) => {
                 DEFERRED_TRIE_METRICS.deferred_trie_sync_fallback.increment(1);
+
+                let inputs = maybe_inputs.take().expect("inputs must be present in Pending state");
+
                 let computed = Self::sort_and_build_trie_input(
-                    &inputs.hashed_state,
-                    &inputs.trie_updates,
+                    inputs.hashed_state,
+                    inputs.trie_updates,
                     inputs.anchor_hash,
                     &inputs.ancestors,
                 );
@@ -441,4 +507,365 @@ mod tests {
         let (_, account) = &overlay_state[0];
         assert_eq!(account.unwrap().nonce, 2);
     }
+
+    /// Helper to create a ready block with anchored trie input containing specific state.
+    fn ready_block_with_state(
+        anchor_hash: B256,
+        accounts: Vec<(B256, Option<Account>)>,
+    ) -> DeferredTrieData {
+        let hashed_state = Arc::new(HashedPostStateSorted::new(accounts, B256Map::default()));
+        let trie_updates = Arc::default();
+        let mut overlay = TrieInputSorted::default();
+        Arc::make_mut(&mut overlay.state).extend_ref(hashed_state.as_ref());
+
+        DeferredTrieData::ready(ComputedTrieData {
+            hashed_state,
+            trie_updates,
+            anchored_trie_input: Some(AnchoredTrieInput {
+                anchor_hash,
+                trie_input: Arc::new(overlay),
+            }),
+        })
+    }
+
+    /// Verifies that first block after anchor (no ancestors) creates empty base overlay.
+    #[test]
+    fn first_block_after_anchor_creates_empty_base() {
+        let anchor = B256::with_last_byte(1);
+        let key = B256::with_last_byte(42);
+        let account = Account { nonce: 1, balance: U256::ZERO, bytecode_hash: None };
+
+        // First block after anchor - no ancestors
+        let first_block = DeferredTrieData::pending(
+            Arc::new(HashedPostState::default().with_accounts([(key, Some(account))])),
+            Arc::new(TrieUpdates::default()),
+            anchor,
+            vec![], // No ancestors
+        );
+
+        let result = first_block.wait_cloned();
+
+        // Should have overlay with just this block's data
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.anchor_hash, anchor);
+        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
+        let (found_key, found_account) = &overlay.trie_input.state.accounts[0];
+        assert_eq!(*found_key, key);
+        assert_eq!(found_account.unwrap().nonce, 1);
+    }
+
+    /// Verifies that parent's overlay is reused regardless of anchor.
+    #[test]
+    fn reuses_parent_overlay() {
+        let anchor = B256::with_last_byte(1);
+        let key = B256::with_last_byte(42);
+        let account = Account { nonce: 100, balance: U256::ZERO, bytecode_hash: None };
+
+        // Create parent with anchored trie input
+        let parent = ready_block_with_state(anchor, vec![(key, Some(account))]);
+
+        // Create child - should reuse parent's overlay
+        let child = DeferredTrieData::pending(
+            Arc::new(HashedPostState::default()),
+            Arc::new(TrieUpdates::default()),
+            anchor,
+            vec![parent],
+        );
+
+        let result = child.wait_cloned();
+
+        // Verify parent's account is in the overlay
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.anchor_hash, anchor);
+        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
+        let (found_key, found_account) = &overlay.trie_input.state.accounts[0];
+        assert_eq!(*found_key, key);
+        assert_eq!(found_account.unwrap().nonce, 100);
+    }
+
+    /// Verifies that parent's overlay is NOT reused when anchor changes (after persist).
+    /// The overlay data is dependent on the anchor, so it must be rebuilt from the
+    /// remaining ancestors.
+    #[test]
+    fn rebuilds_overlay_when_anchor_changes() {
+        let old_anchor = B256::with_last_byte(1);
+        let new_anchor = B256::with_last_byte(2);
+        let key = B256::with_last_byte(42);
+        let account = Account { nonce: 50, balance: U256::ZERO, bytecode_hash: None };
+
+        // Create parent with OLD anchor
+        let parent = ready_block_with_state(old_anchor, vec![(key, Some(account))]);
+
+        // Create child with NEW anchor (simulates after persist)
+        // Should NOT reuse parent's overlay because anchor changed
+        let child = DeferredTrieData::pending(
+            Arc::new(HashedPostState::default()),
+            Arc::new(TrieUpdates::default()),
+            new_anchor,
+            vec![parent],
+        );
+
+        let result = child.wait_cloned();
+
+        // Verify result uses new anchor
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.anchor_hash, new_anchor);
+
+        // Crucially, since we provided `parent` in ancestors but it has a different anchor,
+        // the code falls back to `merge_ancestors_into_overlay`.
+        // `merge_ancestors_into_overlay` reads `parent.hashed_state` (which has the account).
+        // So the account IS present, but it was obtained via REBUILD, not REUSE.
+        // We can check `DEFERRED_TRIE_METRICS` if we want to be sure, but functionally:
+        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
+        let (found_key, found_account) = &overlay.trie_input.state.accounts[0];
+        assert_eq!(*found_key, key);
+        assert_eq!(found_account.unwrap().nonce, 50);
+    }
+
+    /// Verifies that parent without `anchored_trie_input` triggers rebuild path.
+    #[test]
+    fn rebuilds_when_parent_has_no_anchored_input() {
+        let anchor = B256::with_last_byte(1);
+        let key = B256::with_last_byte(42);
+        let account = Account { nonce: 25, balance: U256::ZERO, bytecode_hash: None };
+
+        // Create parent WITHOUT anchored trie input (e.g., from without_trie_input constructor)
+        let parent_state =
+            HashedPostStateSorted::new(vec![(key, Some(account))], B256Map::default());
+        let parent = DeferredTrieData::ready(ComputedTrieData {
+            hashed_state: Arc::new(parent_state),
+            trie_updates: Arc::default(),
+            anchored_trie_input: None, // No anchored input
+        });
+
+        // Create child - should rebuild from parent's hashed_state
+        let child = DeferredTrieData::pending(
+            Arc::new(HashedPostState::default()),
+            Arc::new(TrieUpdates::default()),
+            anchor,
+            vec![parent],
+        );
+
+        let result = child.wait_cloned();
+
+        // Verify overlay is built and contains parent's data
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.anchor_hash, anchor);
+        assert_eq!(overlay.trie_input.state.accounts.len(), 1);
+    }
+
+    /// Verifies that a chain of blocks with matching anchors builds correct cumulative overlay.
+    #[test]
+    fn chain_of_blocks_builds_cumulative_overlay() {
+        let anchor = B256::with_last_byte(1);
+        let key1 = B256::with_last_byte(1);
+        let key2 = B256::with_last_byte(2);
+        let key3 = B256::with_last_byte(3);
+
+        // Block 1: sets account at key1
+        let block1 = ready_block_with_state(
+            anchor,
+            vec![(key1, Some(Account { nonce: 1, balance: U256::ZERO, bytecode_hash: None }))],
+        );
+
+        // Block 2: adds account at key2, ancestor is block1
+        let block2_hashed = HashedPostState::default().with_accounts([(
+            key2,
+            Some(Account { nonce: 2, balance: U256::ZERO, bytecode_hash: None }),
+        )]);
+        let block2 = DeferredTrieData::pending(
+            Arc::new(block2_hashed),
+            Arc::new(TrieUpdates::default()),
+            anchor,
+            vec![block1.clone()],
+        );
+        // Compute block2's trie data
+        let block2_computed = block2.wait_cloned();
+        let block2_ready = DeferredTrieData::ready(block2_computed);
+
+        // Block 3: adds account at key3, ancestor is block2 (which includes block1)
+        let block3_hashed = HashedPostState::default().with_accounts([(
+            key3,
+            Some(Account { nonce: 3, balance: U256::ZERO, bytecode_hash: None }),
+        )]);
+        let block3 = DeferredTrieData::pending(
+            Arc::new(block3_hashed),
+            Arc::new(TrieUpdates::default()),
+            anchor,
+            vec![block1, block2_ready],
+        );
+
+        let result = block3.wait_cloned();
+
+        // Verify all three accounts are in the cumulative overlay
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.trie_input.state.accounts.len(), 3);
+
+        // Accounts should be sorted by key (B256 ordering)
+        let accounts = &overlay.trie_input.state.accounts;
+        assert!(accounts.iter().any(|(k, a)| *k == key1 && a.unwrap().nonce == 1));
+        assert!(accounts.iter().any(|(k, a)| *k == key2 && a.unwrap().nonce == 2));
+        assert!(accounts.iter().any(|(k, a)| *k == key3 && a.unwrap().nonce == 3));
+    }
+
+    /// Verifies that child block's state overwrites parent's state for the same key.
+    #[test]
+    fn child_state_overwrites_parent() {
+        let anchor = B256::with_last_byte(1);
+        let key = B256::with_last_byte(42);
+
+        // Parent sets nonce to 10
+        let parent = ready_block_with_state(
+            anchor,
+            vec![(key, Some(Account { nonce: 10, balance: U256::ZERO, bytecode_hash: None }))],
+        );
+
+        // Child overwrites nonce to 99
+        let child_hashed = HashedPostState::default().with_accounts([(
+            key,
+            Some(Account { nonce: 99, balance: U256::ZERO, bytecode_hash: None }),
+        )]);
+        let child = DeferredTrieData::pending(
+            Arc::new(child_hashed),
+            Arc::new(TrieUpdates::default()),
+            anchor,
+            vec![parent],
+        );
+
+        let result = child.wait_cloned();
+
+        // Verify child's value wins (extend_ref uses later value)
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        // Note: extend_ref may result in duplicate keys; check the last occurrence
+        let accounts = &overlay.trie_input.state.accounts;
+        let last_account = accounts.iter().rfind(|(k, _)| *k == key).unwrap();
+        assert_eq!(last_account.1.unwrap().nonce, 99);
+    }
+
+    /// Stress test: verify O(N) behavior by building a chain of many blocks.
+    /// This test ensures the fix doesn't regress - previously this would be O(N²).
+    #[test]
+    fn long_chain_builds_in_linear_time() {
+        let anchor = B256::with_last_byte(1);
+        let num_blocks = 50; // Enough to notice O(N²) vs O(N) difference
+
+        let mut ancestors: Vec<DeferredTrieData> = Vec::new();
+
+        let start = Instant::now();
+
+        for i in 0..num_blocks {
+            let key = B256::with_last_byte(i as u8);
+            let account = Account { nonce: i as u64, balance: U256::ZERO, bytecode_hash: None };
+            let hashed = HashedPostState::default().with_accounts([(key, Some(account))]);
+
+            let block = DeferredTrieData::pending(
+                Arc::new(hashed),
+                Arc::new(TrieUpdates::default()),
+                anchor,
+                ancestors.clone(),
+            );
+
+            // Compute and add to ancestors for next iteration
+            let computed = block.wait_cloned();
+            ancestors.push(DeferredTrieData::ready(computed));
+        }
+
+        let elapsed = start.elapsed();
+
+        // With O(N) fix, 50 blocks should complete quickly (< 1 second)
+        // With O(N²), this would take significantly longer
+        assert!(
+            elapsed < Duration::from_secs(2),
+            "Chain of {num_blocks} blocks took {:?}, possible O(N²) regression",
+            elapsed
+        );
+
+        // Verify final overlay has all accounts
+        let final_result = ancestors.last().unwrap().wait_cloned();
+        let overlay = final_result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.trie_input.state.accounts.len(), num_blocks);
+    }
+
+    /// Verifies that a multi-ancestor overlay is rebuilt when anchor changes.
+    /// This simulates the "persist prefix then keep building" scenario where:
+    /// 1. A chain of blocks is built with anchor A
+    /// 2. Some blocks are persisted, changing anchor to B
+    /// 3. New blocks must rebuild the overlay from the remaining ancestors
+    #[test]
+    fn multi_ancestor_overlay_rebuilt_after_anchor_change() {
+        let old_anchor = B256::with_last_byte(1);
+        let new_anchor = B256::with_last_byte(2);
+        let key1 = B256::with_last_byte(1);
+        let key2 = B256::with_last_byte(2);
+        let key3 = B256::with_last_byte(3);
+        let key4 = B256::with_last_byte(4);
+
+        // Build a chain of 3 blocks with old_anchor
+        let block1 = ready_block_with_state(
+            old_anchor,
+            vec![(key1, Some(Account { nonce: 1, balance: U256::ZERO, bytecode_hash: None }))],
+        );
+
+        let block2_hashed = HashedPostState::default().with_accounts([(
+            key2,
+            Some(Account { nonce: 2, balance: U256::ZERO, bytecode_hash: None }),
+        )]);
+        let block2 = DeferredTrieData::pending(
+            Arc::new(block2_hashed),
+            Arc::new(TrieUpdates::default()),
+            old_anchor,
+            vec![block1.clone()],
+        );
+        let block2_ready = DeferredTrieData::ready(block2.wait_cloned());
+
+        let block3_hashed = HashedPostState::default().with_accounts([(
+            key3,
+            Some(Account { nonce: 3, balance: U256::ZERO, bytecode_hash: None }),
+        )]);
+        let block3 = DeferredTrieData::pending(
+            Arc::new(block3_hashed),
+            Arc::new(TrieUpdates::default()),
+            old_anchor,
+            vec![block1.clone(), block2_ready.clone()],
+        );
+        let block3_ready = DeferredTrieData::ready(block3.wait_cloned());
+
+        // Verify block3's overlay has all 3 accounts with old_anchor
+        let block3_overlay = block3_ready.wait_cloned().anchored_trie_input.unwrap();
+        assert_eq!(block3_overlay.anchor_hash, old_anchor);
+        assert_eq!(block3_overlay.trie_input.state.accounts.len(), 3);
+
+        // Now simulate persist: create block4 with NEW anchor but same ancestors.
+        // To verify correct rebuilding, we must provide ALL unpersisted ancestors.
+        // If we only provided block3, the rebuild would only see block3's state.
+        // We pass block1, block2, block3 to simulate that they are all still in memory
+        // but the anchor check forces a rebuild (e.g. artificial anchor change).
+        let block4_hashed = HashedPostState::default().with_accounts([(
+            key4,
+            Some(Account { nonce: 4, balance: U256::ZERO, bytecode_hash: None }),
+        )]);
+        let block4 = DeferredTrieData::pending(
+            Arc::new(block4_hashed),
+            Arc::new(TrieUpdates::default()),
+            new_anchor, // Different anchor - simulates post-persist
+            vec![block1, block2_ready, block3_ready],
+        );
+
+        let result = block4.wait_cloned();
+
+        // Verify:
+        // 1. New anchor is used in result
+        assert_eq!(result.anchor_hash(), Some(new_anchor));
+
+        // 2. All 4 accounts are in the overlay (rebuilt from ancestors + extended)
+        let overlay = result.anchored_trie_input.as_ref().unwrap();
+        assert_eq!(overlay.trie_input.state.accounts.len(), 4);
+
+        // 3. All accounts have correct values
+        let accounts = &overlay.trie_input.state.accounts;
+        assert!(accounts.iter().any(|(k, a)| *k == key1 && a.unwrap().nonce == 1));
+        assert!(accounts.iter().any(|(k, a)| *k == key2 && a.unwrap().nonce == 2));
+        assert!(accounts.iter().any(|(k, a)| *k == key3 && a.unwrap().nonce == 3));
+        assert!(accounts.iter().any(|(k, a)| *k == key4 && a.unwrap().nonce == 4));
+    }
 }

crates/chain-state/src/in_memory.rs

@@ -18,7 +18,7 @@ use reth_primitives_traits::{
 };
 use reth_storage_api::StateProviderBox;
 use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, TrieInputSorted};
-use std::{collections::BTreeMap, sync::Arc, time::Instant};
+use std::{collections::BTreeMap, ops::Deref, sync::Arc, time::Instant};
 use tokio::sync::{broadcast, watch};
 
 /// Size of the broadcast channel used to notify canonical state events.
@@ -86,14 +86,20 @@ impl<N: NodePrimitives> InMemoryState<N> {
     ///
     /// This tries to acquire a read lock. Drop any write locks before calling this.
     pub(crate) fn update_metrics(&self) {
-        let numbers = self.numbers.read();
-        if let Some((earliest_block_number, _)) = numbers.first_key_value() {
-            self.metrics.earliest_block.set(*earliest_block_number as f64);
+        let (count, earliest, latest) = {
+            let numbers = self.numbers.read();
+            let count = numbers.len();
+            let earliest = numbers.first_key_value().map(|(number, _)| *number);
+            let latest = numbers.last_key_value().map(|(number, _)| *number);
+            (count, earliest, latest)
+        };
+        if let Some(earliest_block_number) = earliest {
+            self.metrics.earliest_block.set(earliest_block_number as f64);
         }
-        if let Some((latest_block_number, _)) = numbers.last_key_value() {
-            self.metrics.latest_block.set(*latest_block_number as f64);
+        if let Some(latest_block_number) = latest {
+            self.metrics.latest_block.set(latest_block_number as f64);
         }
-        self.metrics.num_blocks.set(numbers.len() as f64);
+        self.metrics.num_blocks.set(count as f64);
     }
 
     /// Returns the state for a given block hash.
@@ -311,6 +317,7 @@ impl<N: NodePrimitives> CanonicalInMemoryState<N> {
     /// This will update the links between blocks and remove all blocks that are [..
     /// `persisted_height`].
     pub fn remove_persisted_blocks(&self, persisted_num_hash: BlockNumHash) {
+        self.set_persisted(persisted_num_hash);
         // if the persisted hash is not in the canonical in memory state, do nothing, because it
         // means canonical blocks were not actually persisted.
         //
@@ -438,6 +445,11 @@ impl<N: NodePrimitives> CanonicalInMemoryState<N> {
         self.inner.chain_info_tracker.set_finalized(header);
     }
 
+    /// Persisted block setter.
+    pub fn set_persisted(&self, num_hash: BlockNumHash) {
+        self.inner.chain_info_tracker.set_persisted(num_hash);
+    }
+
     /// Canonical head getter.
     pub fn get_canonical_head(&self) -> SealedHeader<N::BlockHeader> {
         self.inner.chain_info_tracker.get_canonical_head()
@@ -453,6 +465,11 @@ impl<N: NodePrimitives> CanonicalInMemoryState<N> {
         self.inner.chain_info_tracker.get_safe_header()
     }
 
+    /// Persisted block `BlockNumHash` getter.
+    pub fn get_persisted_num_hash(&self) -> Option<BlockNumHash> {
+        self.inner.chain_info_tracker.get_persisted_num_hash()
+    }
+
     /// Returns the `SealedHeader` corresponding to the pending state.
     pub fn pending_sealed_header(&self) -> Option<SealedHeader<N::BlockHeader>> {
         self.pending_state().map(|h| h.block_ref().recovered_block().clone_sealed_header())
@@ -505,6 +522,11 @@ impl<N: NodePrimitives> CanonicalInMemoryState<N> {
         self.inner.chain_info_tracker.subscribe_finalized_block()
     }
 
+    /// Subscribe to new persisted block events.
+    pub fn subscribe_persisted_block(&self) -> watch::Receiver<Option<BlockNumHash>> {
+        self.inner.chain_info_tracker.subscribe_persisted_block()
+    }
+
     /// Attempts to send a new [`CanonStateNotification`] to all active Receiver handles.
     pub fn notify_canon_state(&self, event: CanonStateNotification<N>) {
         self.inner.canon_state_notification_sender.send(event).ok();
@@ -634,6 +656,8 @@ impl<N: NodePrimitives> BlockState<N> {
     /// We assume that the `Receipts` in the executed block `ExecutionOutcome`
     /// has only one element corresponding to the executed block associated to
     /// the state.
+    ///
+    /// This clones the vector of receipts. To avoid it, use [`Self::executed_block_receipts_ref`].
     pub fn executed_block_receipts(&self) -> Vec<N::Receipt> {
         let receipts = self.receipts();
 
@@ -646,22 +670,30 @@ impl<N: NodePrimitives> BlockState<N> {
         receipts.first().cloned().unwrap_or_default()
     }
 
-    /// Returns a vector of __parent__ `BlockStates`.
+    /// Returns a slice of `Receipt` of executed block that determines the state.
+    /// We assume that the `Receipts` in the executed block `ExecutionOutcome`
+    /// has only one element corresponding to the executed block associated to
+    /// the state.
+    pub fn executed_block_receipts_ref(&self) -> &[N::Receipt] {
+        let receipts = self.receipts();
+
+        debug_assert!(
+            receipts.len() <= 1,
+            "Expected at most one block's worth of receipts, found {}",
+            receipts.len()
+        );
+
+        receipts.first().map(|receipts| receipts.deref()).unwrap_or_default()
+    }
+
+    /// Returns an iterator over __parent__ `BlockStates`.
     ///
-    /// The block state order in the output vector is newest to oldest (highest to lowest):
+    /// The block state order is newest to oldest (highest to lowest):
     /// `[5,4,3,2,1]`
     ///
     /// Note: This does not include self.
-    pub fn parent_state_chain(&self) -> Vec<&Self> {
-        let mut parents = Vec::new();
-        let mut current = self.parent.as_deref();
-
-        while let Some(parent) = current {
-            parents.push(parent);
-            current = parent.parent.as_deref();
-        }
-
-        parents
+    pub fn parent_state_chain(&self) -> impl Iterator<Item = &Self> + '_ {
+        std::iter::successors(self.parent.as_deref(), |state| state.parent.as_deref())
     }
 
     /// Returns a vector of `BlockStates` representing the entire in memory chain.
@@ -672,6 +704,11 @@ impl<N: NodePrimitives> BlockState<N> {
     }
 
     /// Appends the parent chain of this [`BlockState`] to the given vector.
+    ///
+    /// Parents are appended in order from newest to oldest (highest to lowest).
+    /// This does not include self, only the parent states.
+    ///
+    /// This is a convenience method equivalent to `chain.extend(self.parent_state_chain())`.
     pub fn append_parent_chain<'a>(&'a self, chain: &mut Vec<&'a Self>) {
         chain.extend(self.parent_state_chain());
     }
@@ -909,6 +946,8 @@ impl<N: NodePrimitives<SignedTx: SignedTransaction>> NewCanonicalChain<N> {
                     chain.append_block(
                         exec.recovered_block().clone(),
                         exec.execution_outcome().clone(),
+                        exec.trie_updates(),
+                        exec.hashed_state(),
                     );
                     chain
                 }));
@@ -919,6 +958,8 @@ impl<N: NodePrimitives<SignedTx: SignedTransaction>> NewCanonicalChain<N> {
                     chain.append_block(
                         exec.recovered_block().clone(),
                         exec.execution_outcome().clone(),
+                        exec.trie_updates(),
+                        exec.hashed_state(),
                     );
                     chain
                 }));
@@ -926,6 +967,8 @@ impl<N: NodePrimitives<SignedTx: SignedTransaction>> NewCanonicalChain<N> {
                     chain.append_block(
                         exec.recovered_block().clone(),
                         exec.execution_outcome().clone(),
+                        exec.trie_updates(),
+                        exec.hashed_state(),
                     );
                     chain
                 }));
@@ -1435,19 +1478,18 @@ mod tests {
         let mut test_block_builder: TestBlockBuilder = TestBlockBuilder::default();
         let chain = create_mock_state_chain(&mut test_block_builder, 4);
 
-        let parents = chain[3].parent_state_chain();
+        let parents: Vec<_> = chain[3].parent_state_chain().collect();
         assert_eq!(parents.len(), 3);
         assert_eq!(parents[0].block().recovered_block().number, 3);
         assert_eq!(parents[1].block().recovered_block().number, 2);
         assert_eq!(parents[2].block().recovered_block().number, 1);
 
-        let parents = chain[2].parent_state_chain();
+        let parents: Vec<_> = chain[2].parent_state_chain().collect();
         assert_eq!(parents.len(), 2);
         assert_eq!(parents[0].block().recovered_block().number, 2);
         assert_eq!(parents[1].block().recovered_block().number, 1);
 
-        let parents = chain[0].parent_state_chain();
-        assert_eq!(parents.len(), 0);
+        assert_eq!(chain[0].parent_state_chain().count(), 0);
     }
 
     #[test]
@@ -1458,8 +1500,7 @@ mod tests {
             create_mock_state(&mut test_block_builder, single_block_number, B256::random());
         let single_block_hash = single_block.block().recovered_block().hash();
 
-        let parents = single_block.parent_state_chain();
-        assert_eq!(parents.len(), 0);
+        assert_eq!(single_block.parent_state_chain().count(), 0);
 
         let block_state_chain = single_block.chain().collect::<Vec<_>>();
         assert_eq!(block_state_chain.len(), 1);
@@ -1511,13 +1552,24 @@ mod tests {
         // Test commit notification
         let chain_commit = NewCanonicalChain::Commit { new: vec![block0.clone(), block1.clone()] };
 
+        // Build expected trie updates map
+        let mut expected_trie_updates = BTreeMap::new();
+        expected_trie_updates.insert(0, block0.trie_updates());
+        expected_trie_updates.insert(1, block1.trie_updates());
+
+        // Build expected hashed state map
+        let mut expected_hashed_state = BTreeMap::new();
+        expected_hashed_state.insert(0, block0.hashed_state());
+        expected_hashed_state.insert(1, block1.hashed_state());
+
         assert_eq!(
             chain_commit.to_chain_notification(),
             CanonStateNotification::Commit {
                 new: Arc::new(Chain::new(
                     vec![block0.recovered_block().clone(), block1.recovered_block().clone()],
                     sample_execution_outcome.clone(),
-                    None
+                    expected_trie_updates,
+                    expected_hashed_state
                 ))
             }
         );
@@ -1528,18 +1580,40 @@ mod tests {
             old: vec![block1.clone(), block2.clone()],
         };
 
+        // Build expected trie updates for old chain
+        let mut old_trie_updates = BTreeMap::new();
+        old_trie_updates.insert(1, block1.trie_updates());
+        old_trie_updates.insert(2, block2.trie_updates());
+
+        // Build expected trie updates for new chain
+        let mut new_trie_updates = BTreeMap::new();
+        new_trie_updates.insert(1, block1a.trie_updates());
+        new_trie_updates.insert(2, block2a.trie_updates());
+
+        // Build expected hashed state for old chain
+        let mut old_hashed_state = BTreeMap::new();
+        old_hashed_state.insert(1, block1.hashed_state());
+        old_hashed_state.insert(2, block2.hashed_state());
+
+        // Build expected hashed state for new chain
+        let mut new_hashed_state = BTreeMap::new();
+        new_hashed_state.insert(1, block1a.hashed_state());
+        new_hashed_state.insert(2, block2a.hashed_state());
+
         assert_eq!(
             chain_reorg.to_chain_notification(),
             CanonStateNotification::Reorg {
                 old: Arc::new(Chain::new(
                     vec![block1.recovered_block().clone(), block2.recovered_block().clone()],
                     sample_execution_outcome.clone(),
-                    None
+                    old_trie_updates,
+                    old_hashed_state
                 )),
                 new: Arc::new(Chain::new(
                     vec![block1a.recovered_block().clone(), block2a.recovered_block().clone()],
                     sample_execution_outcome,
-                    None
+                    new_trie_updates,
+                    new_hashed_state
                 ))
             }
         );

crates/chain-state/src/lib.rs

@@ -23,7 +23,8 @@ mod notifications;
 pub use notifications::{
     CanonStateNotification, CanonStateNotificationSender, CanonStateNotificationStream,
     CanonStateNotifications, CanonStateSubscriptions, ForkChoiceNotifications, ForkChoiceStream,
-    ForkChoiceSubscriptions,
+    ForkChoiceSubscriptions, PersistedBlockNotifications, PersistedBlockSubscriptions,
+    WatchValueStream,
 };
 
 mod memory_overlay;

crates/chain-state/src/memory_overlay.rs

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

crates/chain-state/src/noop.rs

@@ -2,7 +2,7 @@
 
 use crate::{
     CanonStateNotifications, CanonStateSubscriptions, ForkChoiceNotifications,
-    ForkChoiceSubscriptions,
+    ForkChoiceSubscriptions, PersistedBlockNotifications, PersistedBlockSubscriptions,
 };
 use reth_primitives_traits::NodePrimitives;
 use reth_storage_api::noop::NoopProvider;
@@ -27,3 +27,10 @@ impl<C: Send + Sync, N: NodePrimitives> ForkChoiceSubscriptions for NoopProvider
         ForkChoiceNotifications(rx)
     }
 }
+
+impl<C: Send + Sync, N: NodePrimitives> PersistedBlockSubscriptions for NoopProvider<C, N> {
+    fn subscribe_persisted_block(&self) -> PersistedBlockNotifications {
+        let (_, rx) = watch::channel(None);
+        PersistedBlockNotifications(rx)
+    }
+}

crates/chain-state/src/notifications.rs

@@ -1,6 +1,6 @@
 //! Canonical chain state notification trait and types.
 
-use alloy_eips::eip2718::Encodable2718;
+use alloy_eips::{eip2718::Encodable2718, BlockNumHash};
 use derive_more::{Deref, DerefMut};
 use reth_execution_types::{BlockReceipts, Chain};
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedHeader};
@@ -205,22 +205,22 @@ pub trait ForkChoiceSubscriptions: Send + Sync {
     }
 }
 
-/// A stream for fork choice watch channels (pending, safe or finalized watchers)
+/// A stream that yields values from a `watch::Receiver<Option<T>>`, filtering out `None` values.
 #[derive(Debug)]
 #[pin_project::pin_project]
-pub struct ForkChoiceStream<T> {
+pub struct WatchValueStream<T> {
     #[pin]
     st: WatchStream<Option<T>>,
 }
 
-impl<T: Clone + Sync + Send + 'static> ForkChoiceStream<T> {
-    /// Creates a new `ForkChoiceStream`
+impl<T: Clone + Sync + Send + 'static> WatchValueStream<T> {
+    /// Creates a new [`WatchValueStream`]
     pub fn new(rx: watch::Receiver<Option<T>>) -> Self {
         Self { st: WatchStream::from_changes(rx) }
     }
 }
 
-impl<T: Clone + Sync + Send + 'static> Stream for ForkChoiceStream<T> {
+impl<T: Clone + Sync + Send + 'static> Stream for WatchValueStream<T> {
     type Item = T;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
@@ -234,6 +234,24 @@ impl<T: Clone + Sync + Send + 'static> Stream for ForkChoiceStream<T> {
     }
 }
 
+/// Alias for [`WatchValueStream`] for fork choice watch channels.
+pub type ForkChoiceStream<T> = WatchValueStream<T>;
+
+/// Wrapper around a watch receiver that receives persisted block notifications.
+#[derive(Debug, Deref, DerefMut)]
+pub struct PersistedBlockNotifications(pub watch::Receiver<Option<BlockNumHash>>);
+
+/// A trait that allows subscribing to persisted block events.
+pub trait PersistedBlockSubscriptions: Send + Sync {
+    /// Get notified when a new block is persisted to disk.
+    fn subscribe_persisted_block(&self) -> PersistedBlockNotifications;
+
+    /// Convenience method to get a stream of the persisted blocks.
+    fn persisted_block_stream(&self) -> WatchValueStream<BlockNumHash> {
+        WatchValueStream::new(self.subscribe_persisted_block().0)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -242,6 +260,7 @@ mod tests {
     use reth_ethereum_primitives::{Receipt, TransactionSigned, TxType};
     use reth_execution_types::ExecutionOutcome;
     use reth_primitives_traits::SealedBlock;
+    use std::collections::BTreeMap;
 
     #[test]
     fn test_commit_notification() {
@@ -260,7 +279,8 @@ mod tests {
         let chain: Arc<Chain> = Arc::new(Chain::new(
             vec![block1.clone(), block2.clone()],
             ExecutionOutcome::default(),
-            None,
+            BTreeMap::new(),
+            BTreeMap::new(),
         ));
 
         // Create a commit notification
@@ -295,12 +315,17 @@ mod tests {
         block3.set_block_number(3);
         block3.set_hash(block3_hash);
 
-        let old_chain: Arc<Chain> =
-            Arc::new(Chain::new(vec![block1.clone()], ExecutionOutcome::default(), None));
+        let old_chain: Arc<Chain> = Arc::new(Chain::new(
+            vec![block1.clone()],
+            ExecutionOutcome::default(),
+            BTreeMap::new(),
+            BTreeMap::new(),
+        ));
         let new_chain = Arc::new(Chain::new(
             vec![block2.clone(), block3.clone()],
             ExecutionOutcome::default(),
-            None,
+            BTreeMap::new(),
+            BTreeMap::new(),
         ));
 
         // Create a reorg notification
@@ -362,8 +387,12 @@ mod tests {
         let execution_outcome = ExecutionOutcome { receipts, ..Default::default() };
 
         // Create a new chain segment with `block1` and `block2` and the execution outcome.
-        let new_chain: Arc<Chain> =
-            Arc::new(Chain::new(vec![block1.clone(), block2.clone()], execution_outcome, None));
+        let new_chain: Arc<Chain> = Arc::new(Chain::new(
+            vec![block1.clone(), block2.clone()],
+            execution_outcome,
+            BTreeMap::new(),
+            BTreeMap::new(),
+        ));
 
         // Create a commit notification containing the new chain segment.
         let notification = CanonStateNotification::Commit { new: new_chain };
@@ -420,8 +449,12 @@ mod tests {
             ExecutionOutcome { receipts: old_receipts, ..Default::default() };
 
         // Create an old chain segment to be reverted, containing `old_block1`.
-        let old_chain: Arc<Chain> =
-            Arc::new(Chain::new(vec![old_block1.clone()], old_execution_outcome, None));
+        let old_chain: Arc<Chain> = Arc::new(Chain::new(
+            vec![old_block1.clone()],
+            old_execution_outcome,
+            BTreeMap::new(),
+            BTreeMap::new(),
+        ));
 
         // Define block2 for the new chain segment, which will be committed.
         let mut body = BlockBody::<TransactionSigned>::default();
@@ -449,7 +482,12 @@ mod tests {
             ExecutionOutcome { receipts: new_receipts, ..Default::default() };
 
         // Create a new chain segment to be committed, containing `new_block1`.
-        let new_chain = Arc::new(Chain::new(vec![new_block1.clone()], new_execution_outcome, None));
+        let new_chain = Arc::new(Chain::new(
+            vec![new_block1.clone()],
+            new_execution_outcome,
+            BTreeMap::new(),
+            BTreeMap::new(),
+        ));
 
         // Create a reorg notification with both reverted (old) and committed (new) chain segments.
         let notification = CanonStateNotification::Reorg { old: old_chain, new: new_chain };

crates/chain-state/src/test_utils.rs

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

crates/chainspec/src/spec.rs

@@ -80,6 +80,8 @@ pub fn make_genesis_header(genesis: &Genesis, hardforks: &ChainHardforks) -> Hea
         .then_some(EMPTY_REQUESTS_HASH);
 
     Header {
+        number: genesis.number.unwrap_or_default(),
+        parent_hash: genesis.parent_hash.unwrap_or_default(),
         gas_limit: genesis.gas_limit,
         difficulty: genesis.difficulty,
         nonce: genesis.nonce.into(),
@@ -968,7 +970,7 @@ impl<H: BlockHeader> EthereumHardforks for ChainSpec<H> {
 
 /// A trait for reading the current chainspec.
 #[auto_impl::auto_impl(&, Arc)]
-pub trait ChainSpecProvider: Debug + Send + Sync {
+pub trait ChainSpecProvider: Debug + Send {
     /// The chain spec type.
     type ChainSpec: EthChainSpec + 'static;
 

crates/cli/commands/Cargo.toml

@@ -83,6 +83,7 @@ backon.workspace = true
 secp256k1 = { workspace = true, features = ["global-context", "std", "recovery"] }
 tokio-stream.workspace = true
 reqwest.workspace = true
+url.workspace = true
 metrics.workspace = true
 
 # io
@@ -129,3 +130,5 @@ arbitrary = [
     "reth-primitives-traits/arbitrary",
     "reth-ethereum-primitives/arbitrary",
 ]
+
+edge = ["reth-db-common/edge", "reth-stages/rocksdb"]

crates/cli/commands/src/common.rs

@@ -23,7 +23,10 @@ use reth_node_core::{
     dirs::{ChainPath, DataDirPath},
 };
 use reth_provider::{
-    providers::{BlockchainProvider, NodeTypesForProvider, StaticFileProvider},
+    providers::{
+        BlockchainProvider, NodeTypesForProvider, RocksDBProvider, StaticFileProvider,
+        StaticFileProviderBuilder,
+    },
     ProviderFactory, StaticFileProviderFactory,
 };
 use reth_stages::{sets::DefaultStages, Pipeline, PipelineTarget};
@@ -75,10 +78,12 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
         let data_dir = self.datadir.clone().resolve_datadir(self.chain.chain());
         let db_path = data_dir.db();
         let sf_path = data_dir.static_files();
+        let rocksdb_path = data_dir.rocksdb();
 
         if access.is_read_write() {
             reth_fs_util::create_dir_all(&db_path)?;
             reth_fs_util::create_dir_all(&sf_path)?;
+            reth_fs_util::create_dir_all(&rocksdb_path)?;
         }
 
         let config_path = self.config.clone().unwrap_or_else(|| data_dir.config());
@@ -98,18 +103,32 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
         }
 
         info!(target: "reth::cli", ?db_path, ?sf_path, "Opening storage");
+        let genesis_block_number = self.chain.genesis().number.unwrap_or_default();
         let (db, sfp) = match access {
             AccessRights::RW => (
                 Arc::new(init_db(db_path, self.db.database_args())?),
-                StaticFileProvider::read_write(sf_path)?,
-            ),
-            AccessRights::RO | AccessRights::RoInconsistent => (
-                Arc::new(open_db_read_only(&db_path, self.db.database_args())?),
-                StaticFileProvider::read_only(sf_path, false)?,
+                StaticFileProviderBuilder::read_write(sf_path)
+                    .with_genesis_block_number(genesis_block_number)
+                    .build()?,
             ),
+            AccessRights::RO | AccessRights::RoInconsistent => {
+                (Arc::new(open_db_read_only(&db_path, self.db.database_args())?), {
+                    let provider = StaticFileProviderBuilder::read_only(sf_path)
+                        .with_genesis_block_number(genesis_block_number)
+                        .build()?;
+                    provider.watch_directory();
+                    provider
+                })
+            }
         };
+        // TransactionDB only support read-write mode
+        let rocksdb_provider = RocksDBProvider::builder(data_dir.rocksdb())
+            .with_default_tables()
+            .with_database_log_level(self.db.log_level)
+            .build()?;
 
-        let provider_factory = self.create_provider_factory(&config, db, sfp, access)?;
+        let provider_factory =
+            self.create_provider_factory(&config, db, sfp, rocksdb_provider, access)?;
         if access.is_read_write() {
             debug!(target: "reth::cli", chain=%self.chain.chain(), genesis=?self.chain.genesis_hash(), "Initializing genesis");
             init_genesis_with_settings(&provider_factory, self.static_files.to_settings())?;
@@ -128,6 +147,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
         config: &Config,
         db: Arc<DatabaseEnv>,
         static_file_provider: StaticFileProvider<N::Primitives>,
+        rocksdb_provider: RocksDBProvider,
         access: AccessRights,
     ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>>
     where
@@ -138,6 +158,7 @@ impl<C: ChainSpecParser> EnvironmentArgs<C> {
             db,
             self.chain.clone(),
             static_file_provider,
+            rocksdb_provider,
         )?
         .with_prune_modes(prune_modes.clone());
 

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

@@ -29,7 +29,7 @@ impl Command {
                 let static_file_provider = tool.provider_factory.static_file_provider();
                 let static_files = iter_static_files(static_file_provider.directory())?;
 
-                if let Some(segment_static_files) = static_files.get(&segment) {
+                if let Some(segment_static_files) = static_files.get(segment) {
                     for (block_range, _) in segment_static_files {
                         static_file_provider.delete_jar(segment, block_range.start())?;
                     }

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

@@ -2,19 +2,24 @@ use alloy_primitives::{hex, BlockHash};
 use clap::Parser;
 use reth_db::{
     static_file::{
-        ColumnSelectorOne, ColumnSelectorTwo, HeaderWithHashMask, ReceiptMask, TransactionMask,
-        TransactionSenderMask,
+        AccountChangesetMask, ColumnSelectorOne, ColumnSelectorTwo, HeaderWithHashMask,
+        ReceiptMask, TransactionMask, TransactionSenderMask,
     },
     RawDupSort,
 };
 use reth_db_api::{
-    table::{Decompress, DupSort, Table},
-    tables, RawKey, RawTable, Receipts, TableViewer, Transactions,
+    cursor::{DbCursorRO, DbDupCursorRO},
+    database::Database,
+    table::{Compress, Decompress, DupSort, Table},
+    tables,
+    transaction::DbTx,
+    RawKey, RawTable, Receipts, TableViewer, Transactions,
 };
 use reth_db_common::DbTool;
 use reth_node_api::{HeaderTy, ReceiptTy, TxTy};
 use reth_node_builder::NodeTypesWithDB;
-use reth_provider::{providers::ProviderNodeTypes, StaticFileProviderFactory};
+use reth_primitives_traits::ValueWithSubKey;
+use reth_provider::{providers::ProviderNodeTypes, ChangeSetReader, StaticFileProviderFactory};
 use reth_static_file_types::StaticFileSegment;
 use tracing::error;
 
@@ -39,6 +44,14 @@ enum Subcommand {
         #[arg(value_parser = maybe_json_value_parser)]
         subkey: Option<String>,
 
+        /// Optional end key for range query (exclusive upper bound)
+        #[arg(value_parser = maybe_json_value_parser)]
+        end_key: Option<String>,
+
+        /// Optional end subkey for range query (exclusive upper bound)
+        #[arg(value_parser = maybe_json_value_parser)]
+        end_subkey: Option<String>,
+
         /// Output bytes instead of human-readable decoded value
         #[arg(long)]
         raw: bool,
@@ -51,6 +64,10 @@ enum Subcommand {
         #[arg(value_parser = maybe_json_value_parser)]
         key: String,
 
+        /// The subkey to get content for, for example address in changeset
+        #[arg(value_parser = maybe_json_value_parser)]
+        subkey: Option<String>,
+
         /// Output bytes instead of human-readable decoded value
         #[arg(long)]
         raw: bool,
@@ -61,36 +78,80 @@ impl Command {
     /// Execute `db get` command
     pub fn execute<N: ProviderNodeTypes>(self, tool: &DbTool<N>) -> eyre::Result<()> {
         match self.subcommand {
-            Subcommand::Mdbx { table, key, subkey, raw } => {
-                table.view(&GetValueViewer { tool, key, subkey, raw })?
+            Subcommand::Mdbx { table, key, subkey, end_key, end_subkey, raw } => {
+                table.view(&GetValueViewer { tool, key, subkey, end_key, end_subkey, raw })?
             }
-            Subcommand::StaticFile { segment, key, raw } => {
-                let (key, mask): (u64, _) = match segment {
+            Subcommand::StaticFile { segment, key, subkey, raw } => {
+                let (key, subkey, mask): (u64, _, _) = match segment {
                     StaticFileSegment::Headers => (
                         table_key::<tables::Headers>(&key)?,
+                        None,
                         <HeaderWithHashMask<HeaderTy<N>>>::MASK,
                     ),
-                    StaticFileSegment::Transactions => {
-                        (table_key::<tables::Transactions>(&key)?, <TransactionMask<TxTy<N>>>::MASK)
-                    }
-                    StaticFileSegment::Receipts => {
-                        (table_key::<tables::Receipts>(&key)?, <ReceiptMask<ReceiptTy<N>>>::MASK)
-                    }
+                    StaticFileSegment::Transactions => (
+                        table_key::<tables::Transactions>(&key)?,
+                        None,
+                        <TransactionMask<TxTy<N>>>::MASK,
+                    ),
+                    StaticFileSegment::Receipts => (
+                        table_key::<tables::Receipts>(&key)?,
+                        None,
+                        <ReceiptMask<ReceiptTy<N>>>::MASK,
+                    ),
                     StaticFileSegment::TransactionSenders => (
                         table_key::<tables::TransactionSenders>(&key)?,
-                        <TransactionSenderMask>::MASK,
+                        None,
+                        TransactionSenderMask::MASK,
                     ),
+                    StaticFileSegment::AccountChangeSets => {
+                        let subkey =
+                            table_subkey::<tables::AccountChangeSets>(subkey.as_deref()).ok();
+                        (
+                            table_key::<tables::AccountChangeSets>(&key)?,
+                            subkey,
+                            AccountChangesetMask::MASK,
+                        )
+                    }
                 };
 
-                let content = tool
-                    .provider_factory
-                    .static_file_provider()
-                    .get_segment_provider(segment, key)?
-                    .cursor()?
-                    .get(key.into(), mask)
-                    .map(|result| {
-                        result.map(|vec| vec.iter().map(|slice| slice.to_vec()).collect::<Vec<_>>())
-                    })?;
+                // handle account changesets differently if a subkey is provided.
+                if let StaticFileSegment::AccountChangeSets = segment {
+                    let Some(subkey) = subkey else {
+                        // get all changesets for the block
+                        let changesets = tool
+                            .provider_factory
+                            .static_file_provider()
+                            .account_block_changeset(key)?;
+
+                        println!("{}", serde_json::to_string_pretty(&changesets)?);
+                        return Ok(())
+                    };
+
+                    let account = tool
+                        .provider_factory
+                        .static_file_provider()
+                        .get_account_before_block(key, subkey)?;
+
+                    if let Some(account) = account {
+                        println!("{}", serde_json::to_string_pretty(&account)?);
+                    } else {
+                        error!(target: "reth::cli", "No content for the given table key.");
+                    }
+
+                    return Ok(())
+                }
+
+                let content = tool.provider_factory.static_file_provider().find_static_file(
+                    segment,
+                    |provider| {
+                        let mut cursor = provider.cursor()?;
+                        cursor.get(key.into(), mask).map(|result| {
+                            result.map(|vec| {
+                                vec.iter().map(|slice| slice.to_vec()).collect::<Vec<_>>()
+                            })
+                        })
+                    },
+                )?;
 
                 match content {
                     Some(content) => {
@@ -126,6 +187,9 @@ impl Command {
                                         )?;
                                     println!("{}", serde_json::to_string_pretty(&sender)?);
                                 }
+                                StaticFileSegment::AccountChangeSets => {
+                                    unreachable!("account changeset static files are special cased before this match")
+                                }
                             }
                         }
                     }
@@ -154,6 +218,8 @@ struct GetValueViewer<'a, N: NodeTypesWithDB> {
     tool: &'a DbTool<N>,
     key: String,
     subkey: Option<String>,
+    end_key: Option<String>,
+    end_subkey: Option<String>,
     raw: bool,
 }
 
@@ -163,53 +229,158 @@ impl<N: ProviderNodeTypes> TableViewer<()> for GetValueViewer<'_, N> {
     fn view<T: Table>(&self) -> Result<(), Self::Error> {
         let key = table_key::<T>(&self.key)?;
 
-        let content = if self.raw {
-            self.tool
-                .get::<RawTable<T>>(RawKey::from(key))?
-                .map(|content| hex::encode_prefixed(content.raw_value()))
-        } else {
-            self.tool.get::<T>(key)?.as_ref().map(serde_json::to_string_pretty).transpose()?
-        };
+        // A non-dupsort table cannot have subkeys. The `subkey` arg becomes the `end_key`. First we
+        // check that `end_key` and `end_subkey` weren't previously given, as that wouldn't be
+        // valid.
+        if self.end_key.is_some() || self.end_subkey.is_some() {
+            return Err(eyre::eyre!("Only END_KEY can be given for non-DUPSORT tables"));
+        }
 
-        match content {
-            Some(content) => {
-                println!("{content}");
-            }
-            None => {
-                error!(target: "reth::cli", "No content for the given table key.");
-            }
-        };
+        let end_key = self.subkey.clone();
+
+        // Check if we're doing a range query
+        if let Some(ref end_key_str) = end_key {
+            let end_key = table_key::<T>(end_key_str)?;
+
+            // Use walk_range to iterate over the range
+            self.tool.provider_factory.db_ref().view(|tx| {
+                let mut cursor = tx.cursor_read::<T>()?;
+                let walker = cursor.walk_range(key..end_key)?;
+
+                for result in walker {
+                    let (k, v) = result?;
+                    let json_val = if self.raw {
+                        let raw_key = RawKey::from(k);
+                        serde_json::json!({
+                            "key": hex::encode_prefixed(raw_key.raw_key()),
+                            "val": hex::encode_prefixed(v.compress().as_ref()),
+                        })
+                    } else {
+                        serde_json::json!({
+                            "key": &k,
+                            "val": &v,
+                        })
+                    };
+
+                    println!("{}", serde_json::to_string_pretty(&json_val)?);
+                }
+
+                Ok::<_, eyre::Report>(())
+            })??;
+        } else {
+            // Single key lookup
+            let content = if self.raw {
+                self.tool
+                    .get::<RawTable<T>>(RawKey::from(key))?
+                    .map(|content| hex::encode_prefixed(content.raw_value()))
+            } else {
+                self.tool.get::<T>(key)?.as_ref().map(serde_json::to_string_pretty).transpose()?
+            };
+
+            match content {
+                Some(content) => {
+                    println!("{content}");
+                }
+                None => {
+                    error!(target: "reth::cli", "No content for the given table key.");
+                }
+            };
+        }
 
         Ok(())
     }
 
-    fn view_dupsort<T: DupSort>(&self) -> Result<(), Self::Error> {
+    fn view_dupsort<T: DupSort>(&self) -> Result<(), Self::Error>
+    where
+        T::Value: reth_primitives_traits::ValueWithSubKey<SubKey = T::SubKey>,
+    {
         // get a key for given table
         let key = table_key::<T>(&self.key)?;
 
-        // process dupsort table
-        let subkey = table_subkey::<T>(self.subkey.as_deref())?;
-
-        let content = if self.raw {
-            self.tool
-                .get_dup::<RawDupSort<T>>(RawKey::from(key), RawKey::from(subkey))?
-                .map(|content| hex::encode_prefixed(content.raw_value()))
-        } else {
-            self.tool
-                .get_dup::<T>(key, subkey)?
+        // Check if we're doing a range query
+        if let Some(ref end_key_str) = self.end_key {
+            let end_key = table_key::<T>(end_key_str)?;
+            let start_subkey = table_subkey::<T>(Some(
+                self.subkey.as_ref().expect("must have been given if end_key is given").as_str(),
+            ))?;
+            let end_subkey_parsed = self
+                .end_subkey
                 .as_ref()
-                .map(serde_json::to_string_pretty)
-                .transpose()?
-        };
+                .map(|s| table_subkey::<T>(Some(s.as_str())))
+                .transpose()?;
 
-        match content {
-            Some(content) => {
-                println!("{content}");
-            }
-            None => {
-                error!(target: "reth::cli", "No content for the given table subkey.");
-            }
-        };
+            self.tool.provider_factory.db_ref().view(|tx| {
+                let mut cursor = tx.cursor_dup_read::<T>()?;
+
+                // Seek to the starting key. If there is actually a key at the starting key then
+                // seek to the subkey within it.
+                if let Some((decoded_key, _)) = cursor.seek(key.clone())? &&
+                    decoded_key == key
+                {
+                    cursor.seek_by_key_subkey(key.clone(), start_subkey.clone())?;
+                }
+
+                // Get the current position to start iteration
+                let mut current = cursor.current()?;
+
+                while let Some((decoded_key, decoded_value)) = current {
+                    // Extract the subkey using the ValueWithSubKey trait
+                    let decoded_subkey = decoded_value.get_subkey();
+
+                    // Check if we've reached the end (exclusive)
+                    if (&decoded_key, Some(&decoded_subkey)) >=
+                        (&end_key, end_subkey_parsed.as_ref())
+                    {
+                        break;
+                    }
+
+                    // Output the entry with both key and subkey
+                    let json_val = if self.raw {
+                        let raw_key = RawKey::from(decoded_key.clone());
+                        serde_json::json!({
+                            "key": hex::encode_prefixed(raw_key.raw_key()),
+                            "val": hex::encode_prefixed(decoded_value.compress().as_ref()),
+                        })
+                    } else {
+                        serde_json::json!({
+                            "key": &decoded_key,
+                            "val": &decoded_value,
+                        })
+                    };
+
+                    println!("{}", serde_json::to_string_pretty(&json_val)?);
+
+                    // Move to next entry
+                    current = cursor.next()?;
+                }
+
+                Ok::<_, eyre::Report>(())
+            })??;
+        } else {
+            // Single key/subkey lookup
+            let subkey = table_subkey::<T>(self.subkey.as_deref())?;
+
+            let content = if self.raw {
+                self.tool
+                    .get_dup::<RawDupSort<T>>(RawKey::from(key), RawKey::from(subkey))?
+                    .map(|content| hex::encode_prefixed(content.raw_value()))
+            } else {
+                self.tool
+                    .get_dup::<T>(key, subkey)?
+                    .as_ref()
+                    .map(serde_json::to_string_pretty)
+                    .transpose()?
+            };
+
+            match content {
+                Some(content) => {
+                    println!("{content}");
+                }
+                None => {
+                    error!(target: "reth::cli", "No content for the given table subkey.");
+                }
+            };
+        }
         Ok(())
     }
 }

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

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

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

@@ -9,7 +9,10 @@ use reth_db_api::{
     transaction::{DbTx, DbTxMut},
 };
 use reth_db_common::DbTool;
-use reth_node_core::version::version_metadata;
+use reth_node_core::{
+    dirs::{ChainPath, DataDirPath},
+    version::version_metadata,
+};
 use reth_node_metrics::{
     chain::ChainSpecInfo,
     hooks::Hooks,
@@ -18,6 +21,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 +52,40 @@ 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,
+        data_dir: &ChainPath<DataDirPath>,
+    ) -> 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 pprof_dump_dir = data_dir.pprof_dumps();
 
-            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(),
+                    pprof_dump_dir,
+                );
 
-                    let handle = runtime.handle().clone();
-                    runtime.block_on(async move {
-                        let task_manager = reth_tasks::TaskManager::new(handle.clone());
-                        let task_executor = task_manager.executor();
-
-                        let config = MetricServerConfig::new(
-                            listen_addr,
-                            VersionInfo {
-                                version: version_metadata().cargo_pkg_version.as_ref(),
-                                build_timestamp: version_metadata().vergen_build_timestamp.as_ref(),
-                                cargo_features: version_metadata().vergen_cargo_features.as_ref(),
-                                git_sha: version_metadata().vergen_git_sha.as_ref(),
-                                target_triple: version_metadata()
-                                    .vergen_cargo_target_triple
-                                    .as_ref(),
-                                build_profile: version_metadata().build_profile_name.as_ref(),
-                            },
-                            ChainSpecInfo { name: chain_name },
-                            task_executor,
-                            Hooks::builder().build(),
-                        );
-
-                        // Spawn the metrics server
-                        if let Err(e) = MetricServer::new(config).serve().await {
-                            tracing::error!("Metrics server error: {}", e);
-                        }
-
-                        // Block forever to keep the runtime alive
-                        std::future::pending::<()>().await
-                    });
-                },
-            )?;
+                // Spawn the metrics server
+                if let Err(e) = MetricServer::new(config).serve().await {
+                    tracing::error!("Metrics server error: {}", e);
+                }
+            });
 
             Some(handle)
         } else {
@@ -315,8 +307,8 @@ fn verify_and_repair<N: ProviderNodeTypes>(tool: &DbTool<N>) -> eyre::Result<()>
     if inconsistent_nodes == 0 {
         info!("No inconsistencies found");
     } else {
-        info!("Repaired {} inconsistencies, committing changes", inconsistent_nodes);
         provider_rw.commit()?;
+        info!("Repaired {} inconsistencies and committed changes", inconsistent_nodes);
     }
 
     Ok(())

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

@@ -40,12 +40,32 @@ enum Subcommands {
 #[clap(rename_all = "snake_case")]
 pub enum SetCommand {
     /// Store receipts in static files instead of the database
-    ReceiptsInStaticFiles {
+    Receipts {
         #[clap(action(ArgAction::Set))]
         value: bool,
     },
     /// Store transaction senders in static files instead of the database
-    TransactionSendersInStaticFiles {
+    TransactionSenders {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store account changesets in static files instead of the database
+    AccountChangesets {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store storages history in RocksDB instead of MDBX
+    StoragesHistory {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store account history in RocksDB instead of MDBX
+    AccountHistory {
+        #[clap(action(ArgAction::Set))]
+        value: bool,
+    },
+    /// Store transaction hash numbers in RocksDB instead of MDBX
+    TxHashNumbers {
         #[clap(action(ArgAction::Set))]
         value: bool,
     },
@@ -91,11 +111,15 @@ impl Command {
         let mut settings @ StorageSettings {
             receipts_in_static_files: _,
             transaction_senders_in_static_files: _,
+            storages_history_in_rocksdb: _,
+            transaction_hash_numbers_in_rocksdb: _,
+            account_history_in_rocksdb: _,
+            account_changesets_in_static_files: _,
         } = settings.unwrap_or_else(StorageSettings::legacy);
 
         // Update the setting based on the key
         match cmd {
-            SetCommand::ReceiptsInStaticFiles { value } => {
+            SetCommand::Receipts { value } => {
                 if settings.receipts_in_static_files == value {
                     println!("receipts_in_static_files is already set to {}", value);
                     return Ok(());
@@ -103,7 +127,7 @@ impl Command {
                 settings.receipts_in_static_files = value;
                 println!("Set receipts_in_static_files = {}", value);
             }
-            SetCommand::TransactionSendersInStaticFiles { value } => {
+            SetCommand::TransactionSenders { value } => {
                 if settings.transaction_senders_in_static_files == value {
                     println!("transaction_senders_in_static_files is already set to {}", value);
                     return Ok(());
@@ -111,6 +135,38 @@ impl Command {
                 settings.transaction_senders_in_static_files = value;
                 println!("Set transaction_senders_in_static_files = {}", value);
             }
+            SetCommand::AccountChangesets { value } => {
+                if settings.account_changesets_in_static_files == value {
+                    println!("account_changesets_in_static_files is already set to {}", value);
+                    return Ok(());
+                }
+                settings.account_changesets_in_static_files = value;
+                println!("Set account_changesets_in_static_files = {}", value);
+            }
+            SetCommand::StoragesHistory { value } => {
+                if settings.storages_history_in_rocksdb == value {
+                    println!("storages_history_in_rocksdb is already set to {}", value);
+                    return Ok(());
+                }
+                settings.storages_history_in_rocksdb = value;
+                println!("Set storages_history_in_rocksdb = {}", value);
+            }
+            SetCommand::AccountHistory { value } => {
+                if settings.account_history_in_rocksdb == value {
+                    println!("account_history_in_rocksdb is already set to {}", value);
+                    return Ok(());
+                }
+                settings.account_history_in_rocksdb = value;
+                println!("Set account_history_in_rocksdb = {}", value);
+            }
+            SetCommand::TxHashNumbers { value } => {
+                if settings.transaction_hash_numbers_in_rocksdb == value {
+                    println!("transaction_hash_numbers_in_rocksdb is already set to {}", value);
+                    return Ok(());
+                }
+                settings.transaction_hash_numbers_in_rocksdb = value;
+                println!("Set transaction_hash_numbers_in_rocksdb = {}", value);
+            }
         }
 
         // Write updated settings

crates/cli/commands/src/download.rs

@@ -16,6 +16,7 @@ use std::{
 use tar::Archive;
 use tokio::task;
 use tracing::info;
+use url::Url;
 use zstd::stream::read::Decoder as ZstdDecoder;
 
 const BYTE_UNITS: [&str; 4] = ["B", "KB", "MB", "GB"];
@@ -170,12 +171,14 @@ struct DownloadProgress {
     downloaded: u64,
     total_size: u64,
     last_displayed: Instant,
+    started_at: Instant,
 }
 
 impl DownloadProgress {
     /// Creates new progress tracker with given total size
     fn new(total_size: u64) -> Self {
-        Self { downloaded: 0, total_size, last_displayed: Instant::now() }
+        let now = Instant::now();
+        Self { downloaded: 0, total_size, last_displayed: now, started_at: now }
     }
 
     /// Converts bytes to human readable format (B, KB, MB, GB)
@@ -191,6 +194,18 @@ impl DownloadProgress {
         format!("{:.2} {}", size, BYTE_UNITS[unit_index])
     }
 
+    /// Format duration as human readable string
+    fn format_duration(duration: Duration) -> String {
+        let secs = duration.as_secs();
+        if secs < 60 {
+            format!("{secs}s")
+        } else if secs < 3600 {
+            format!("{}m {}s", secs / 60, secs % 60)
+        } else {
+            format!("{}h {}m", secs / 3600, (secs % 3600) / 60)
+        }
+    }
+
     /// Updates progress bar
     fn update(&mut self, chunk_size: u64) -> Result<()> {
         self.downloaded += chunk_size;
@@ -201,8 +216,24 @@ impl DownloadProgress {
             let formatted_total = Self::format_size(self.total_size);
             let progress = (self.downloaded as f64 / self.total_size as f64) * 100.0;
 
+            // Calculate ETA based on current speed
+            let elapsed = self.started_at.elapsed();
+            let eta = if self.downloaded > 0 {
+                let remaining = self.total_size.saturating_sub(self.downloaded);
+                let speed = self.downloaded as f64 / elapsed.as_secs_f64();
+                if speed > 0.0 {
+                    Duration::from_secs_f64(remaining as f64 / speed)
+                } else {
+                    Duration::ZERO
+                }
+            } else {
+                Duration::ZERO
+            };
+            let eta_str = Self::format_duration(eta);
+
+            // Pad with spaces to clear any previous longer line
             print!(
-                "\rDownloading and extracting... {progress:.2}% ({formatted_downloaded} / {formatted_total})",
+                "\rDownloading and extracting... {progress:.2}% ({formatted_downloaded} / {formatted_total}) ETA: {eta_str}     ",
             );
             io::stdout().flush()?;
             self.last_displayed = Instant::now();
@@ -246,12 +277,18 @@ enum CompressionFormat {
 impl CompressionFormat {
     /// Detect compression format from file extension
     fn from_url(url: &str) -> Result<Self> {
-        if url.ends_with(EXTENSION_TAR_LZ4) {
+        let path =
+            Url::parse(url).map(|u| u.path().to_string()).unwrap_or_else(|_| url.to_string());
+
+        if path.ends_with(EXTENSION_TAR_LZ4) {
             Ok(Self::Lz4)
-        } else if url.ends_with(EXTENSION_TAR_ZSTD) {
+        } else if path.ends_with(EXTENSION_TAR_ZSTD) {
             Ok(Self::Zstd)
         } else {
-            Err(eyre::eyre!("Unsupported file format. Expected .tar.lz4 or .tar.zst, got: {}", url))
+            Err(eyre::eyre!(
+                "Unsupported file format. Expected .tar.lz4 or .tar.zst, got: {}",
+                path
+            ))
         }
     }
 }

crates/cli/commands/src/import_core.rs

@@ -69,9 +69,7 @@ pub async fn import_blocks_from_file<N>(
     provider_factory: ProviderFactory<N>,
     config: &Config,
     executor: impl ConfigureEvm<Primitives = N::Primitives> + 'static,
-    consensus: Arc<
-        impl FullConsensus<N::Primitives, Error = reth_consensus::ConsensusError> + 'static,
-    >,
+    consensus: Arc<impl FullConsensus<N::Primitives> + 'static>,
 ) -> eyre::Result<ImportResult>
 where
     N: ProviderNodeTypes,
@@ -198,7 +196,7 @@ pub fn build_import_pipeline_impl<N, C, E>(
 ) -> eyre::Result<(Pipeline<N>, impl futures::Stream<Item = NodeEvent<N::Primitives>> + use<N, C, E>)>
 where
     N: ProviderNodeTypes,
-    C: FullConsensus<N::Primitives, Error = reth_consensus::ConsensusError> + 'static,
+    C: FullConsensus<N::Primitives> + 'static,
     E: ConfigureEvm<Primitives = N::Primitives> + 'static,
 {
     if !file_client.has_canonical_blocks() {

crates/cli/commands/src/init_cmd.rs

@@ -1,8 +1,9 @@
 //! Command that initializes the node from a genesis file.
 
 use crate::common::{AccessRights, CliNodeTypes, Environment, EnvironmentArgs};
+use alloy_consensus::BlockHeader;
 use clap::Parser;
-use reth_chainspec::{EthChainSpec, EthereumHardforks};
+use reth_chainspec::{ChainSpecProvider, EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
 use reth_provider::BlockHashReader;
 use std::sync::Arc;
@@ -22,8 +23,9 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + EthereumHardforks>> InitComman
 
         let Environment { provider_factory, .. } = self.env.init::<N>(AccessRights::RW)?;
 
+        let genesis_block_number = provider_factory.chain_spec().genesis_header().number();
         let hash = provider_factory
-            .block_hash(0)?
+            .block_hash(genesis_block_number)?
             .ok_or_else(|| eyre::eyre!("Genesis hash not found."))?;
 
         info!(target: "reth::cli", hash = ?hash, "Genesis block written");

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

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

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

@@ -79,7 +79,7 @@ where
         + StaticFileProviderFactory<Primitives: NodePrimitives<BlockHeader: Compact>>,
 {
     provider_rw.insert_block(
-        SealedBlock::<<Provider::Primitives as NodePrimitives>::Block>::from_sealed_parts(
+        &SealedBlock::<<Provider::Primitives as NodePrimitives>::Block>::from_sealed_parts(
             header.clone(),
             Default::default(),
         )
@@ -99,6 +99,7 @@ where
 /// * Headers: It will push an empty block.
 /// * Transactions: It will not push any tx, only increments the end block range.
 /// * Receipts: It will not push any receipt, only increments the end block range.
+/// * TransactionSenders: If the segment exists, increments the end block range.
 fn append_dummy_chain<N, F>(
     sf_provider: &StaticFileProvider<N>,
     target_height: BlockNumber,
@@ -110,8 +111,15 @@ where
 {
     let (tx, rx) = std::sync::mpsc::channel();
 
-    // Spawn jobs for incrementing the block end range of transactions and receipts
-    for segment in [StaticFileSegment::Transactions, StaticFileSegment::Receipts] {
+    // Spawn jobs for incrementing the block end range of transactions, receipts, and senders.
+    for segment in [
+        StaticFileSegment::Transactions,
+        StaticFileSegment::Receipts,
+        StaticFileSegment::TransactionSenders,
+    ] {
+        if sf_provider.get_highest_static_file_block(segment).is_none() {
+            continue
+        }
         let tx_clone = tx.clone();
         let provider = sf_provider.clone();
         std::thread::spawn(move || {
@@ -151,9 +159,15 @@ where
 
     // If, for any reason, rayon crashes this verifies if all segments are at the same
     // target_height.
-    for segment in
-        [StaticFileSegment::Headers, StaticFileSegment::Receipts, StaticFileSegment::Transactions]
-    {
+    for segment in [
+        StaticFileSegment::Headers,
+        StaticFileSegment::Receipts,
+        StaticFileSegment::Transactions,
+        StaticFileSegment::TransactionSenders,
+    ] {
+        if sf_provider.get_highest_static_file_block(segment).is_none() {
+            continue
+        }
         assert_eq!(
             sf_provider.latest_writer(segment)?.user_header().block_end(),
             Some(target_height),

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

@@ -72,7 +72,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                 .split();
                 if result.len() != 1 {
                     eyre::bail!(
-                        "Invalid number of headers received. Expected: 1. Received: {}",
+                        "Invalid number of bodies received. Expected: 1. Received: {}",
                         result.len()
                     )
                 }
@@ -189,7 +189,7 @@ impl<C: ChainSpecParser> DownloadArgs<C> {
 
         let net = NetworkConfigBuilder::<N::NetworkPrimitives>::new(p2p_secret_key)
             .peer_config(config.peers_config_with_basic_nodes_from_file(None))
-            .external_ip_resolver(self.network.nat)
+            .external_ip_resolver(self.network.nat.clone())
             .network_id(self.network.network_id)
             .boot_nodes(boot_nodes.clone())
             .apply(|builder| {

crates/cli/commands/src/stage/drop.rs

@@ -14,6 +14,8 @@ use reth_db_common::{
 };
 use reth_node_api::{HeaderTy, ReceiptTy, TxTy};
 use reth_node_core::args::StageEnum;
+#[cfg(all(unix, feature = "edge"))]
+use reth_provider::RocksDBProviderFactory;
 use reth_provider::{
     DBProvider, DatabaseProviderFactory, StaticFileProviderFactory, StaticFileWriter, TrieWriter,
 };
@@ -87,6 +89,9 @@ impl<C: ChainSpecParser> Command<C> {
                             .unwrap_or_default();
                         writer.prune_transaction_senders(to_delete, 0)?;
                     }
+                    StaticFileSegment::AccountChangeSets => {
+                        writer.prune_account_changesets(highest_block)?;
+                    }
                 }
             }
         }
@@ -172,6 +177,14 @@ impl<C: ChainSpecParser> Command<C> {
                 tx.clear::<tables::AccountsHistory>()?;
                 tx.clear::<tables::StoragesHistory>()?;
 
+                // Also clear RocksDB tables if they exist
+                #[cfg(all(unix, feature = "edge"))]
+                {
+                    let rocksdb = tool.provider_factory.rocksdb_provider();
+                    rocksdb.clear::<tables::AccountsHistory>()?;
+                    rocksdb.clear::<tables::StoragesHistory>()?;
+                }
+
                 reset_stage_checkpoint(tx, StageId::IndexAccountHistory)?;
                 reset_stage_checkpoint(tx, StageId::IndexStorageHistory)?;
 
@@ -179,6 +192,14 @@ impl<C: ChainSpecParser> Command<C> {
             }
             StageEnum::TxLookup => {
                 tx.clear::<tables::TransactionHashNumbers>()?;
+
+                // Also clear RocksDB table if it exists
+                #[cfg(all(unix, feature = "edge"))]
+                {
+                    let rocksdb = tool.provider_factory.rocksdb_provider();
+                    rocksdb.clear::<tables::TransactionHashNumbers>()?;
+                }
+
                 reset_prune_checkpoint(tx, PruneSegment::TransactionLookup)?;
 
                 reset_stage_checkpoint(tx, StageId::TransactionLookup)?;

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

@@ -1,5 +1,5 @@
 use super::setup;
-use reth_consensus::{noop::NoopConsensus, ConsensusError, FullConsensus};
+use reth_consensus::{noop::NoopConsensus, FullConsensus};
 use reth_db::DatabaseEnv;
 use reth_db_api::{
     cursor::DbCursorRO, database::Database, table::TableImporter, tables, transaction::DbTx,
@@ -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};
@@ -28,7 +28,7 @@ pub(crate) async fn dump_execution_stage<N, E, C>(
 where
     N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>,
     E: ConfigureEvm<Primitives = N::Primitives> + 'static,
-    C: FullConsensus<E::Primitives, Error = ConsensusError> + 'static,
+    C: FullConsensus<E::Primitives> + 'static,
 {
     let (output_db, tip_block_number) = setup(from, to, &output_datadir.db(), db_tool)?;
 
@@ -42,6 +42,7 @@ where
                 Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).with_default_tables().build()?,
             )?,
             to,
             from,
@@ -168,7 +169,7 @@ fn dry_run<N, E, C>(
 where
     N: ProviderNodeTypes,
     E: ConfigureEvm<Primitives = N::Primitives> + 'static,
-    C: FullConsensus<E::Primitives, Error = ConsensusError> + 'static,
+    C: FullConsensus<E::Primitives> + 'static,
 {
     info!(target: "reth::cli", "Executing stage. [dry-run]");
 

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()).with_default_tables().build()?,
             )?,
             to,
             from,

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

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

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

@@ -4,7 +4,7 @@ use super::setup;
 use alloy_primitives::{Address, BlockNumber};
 use eyre::Result;
 use reth_config::config::EtlConfig;
-use reth_consensus::{ConsensusError, FullConsensus};
+use reth_consensus::FullConsensus;
 use reth_db::DatabaseEnv;
 use reth_db_api::{database::Database, models::BlockNumberAddress, table::TableImporter, tables};
 use reth_db_common::DbTool;
@@ -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::{
@@ -31,7 +31,7 @@ pub(crate) async fn dump_merkle_stage<N>(
     output_datadir: ChainPath<DataDirPath>,
     should_run: bool,
     evm_config: impl ConfigureEvm<Primitives = N::Primitives>,
-    consensus: impl FullConsensus<N::Primitives, Error = ConsensusError> + 'static,
+    consensus: impl FullConsensus<N::Primitives> + 'static,
 ) -> Result<()>
 where
     N: ProviderNodeTypes<DB = Arc<DatabaseEnv>>,
@@ -62,6 +62,7 @@ where
                 Arc::new(output_db),
                 db_tool.chain(),
                 StaticFileProvider::read_write(output_datadir.static_files())?,
+                RocksDBProvider::builder(output_datadir.rocksdb()).with_default_tables().build()?,
             )?,
             to,
             from,
@@ -78,7 +79,7 @@ fn unwind_and_copy<N: ProviderNodeTypes>(
     tip_block_number: u64,
     output_db: &DatabaseEnv,
     evm_config: impl ConfigureEvm<Primitives = N::Primitives>,
-    consensus: impl FullConsensus<N::Primitives, Error = ConsensusError> + 'static,
+    consensus: impl FullConsensus<N::Primitives> + 'static,
 ) -> eyre::Result<()> {
     let (from, to) = range;
     let provider = db_tool.provider_factory.database_provider_rw()?;

crates/cli/commands/src/stage/run.rs

@@ -11,7 +11,6 @@ use reth_cli::chainspec::ChainSpecParser;
 use reth_cli_runner::CliContext;
 use reth_cli_util::get_secret_key;
 use reth_config::config::{HashingConfig, SenderRecoveryConfig, TransactionLookupConfig};
-use reth_db_api::database_metrics::DatabaseMetrics;
 use reth_downloaders::{
     bodies::bodies::BodiesDownloaderBuilder,
     headers::reverse_headers::ReverseHeadersDownloaderBuilder,
@@ -19,19 +18,19 @@ use reth_downloaders::{
 use reth_exex::ExExManagerHandle;
 use reth_network::BlockDownloaderProvider;
 use reth_network_p2p::HeadersClient;
+use reth_node_builder::common::metrics_hooks;
 use reth_node_core::{
     args::{NetworkArgs, StageEnum},
     version::version_metadata,
 };
 use reth_node_metrics::{
     chain::ChainSpecInfo,
-    hooks::Hooks,
     server::{MetricServer, MetricServerConfig},
     version::VersionInfo,
 };
 use reth_provider::{
     ChainSpecProvider, DBProvider, DatabaseProviderFactory, StageCheckpointReader,
-    StageCheckpointWriter, StaticFileProviderFactory,
+    StageCheckpointWriter,
 };
 use reth_stages::{
     stages::{
@@ -139,20 +138,8 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                 },
                 ChainSpecInfo { name: provider_factory.chain_spec().chain().to_string() },
                 ctx.task_executor,
-                Hooks::builder()
-                    .with_hook({
-                        let db = provider_factory.db_ref().clone();
-                        move || db.report_metrics()
-                    })
-                    .with_hook({
-                        let sfp = provider_factory.static_file_provider();
-                        move || {
-                            if let Err(error) = sfp.report_metrics() {
-                                error!(%error, "Failed to report metrics from static file provider");
-                            }
-                        }
-                    })
-                    .build(),
+                metrics_hooks(&provider_factory),
+                data_dir.pprof_dumps(),
             );
 
             MetricServer::new(config).serve().await?;

crates/cli/runner/src/lib.rs

@@ -18,8 +18,8 @@ use tracing::{debug, error, trace};
 ///
 /// Provides utilities for running a cli command to completion.
 #[derive(Debug)]
-#[non_exhaustive]
 pub struct CliRunner {
+    config: CliRunnerConfig,
     tokio_runtime: tokio::runtime::Runtime,
 }
 
@@ -29,12 +29,18 @@ impl CliRunner {
     ///
     /// The default tokio runtime is multi-threaded, with both I/O and time drivers enabled.
     pub fn try_default_runtime() -> Result<Self, std::io::Error> {
-        Ok(Self { tokio_runtime: tokio_runtime()? })
+        Ok(Self { config: CliRunnerConfig::default(), tokio_runtime: tokio_runtime()? })
     }
 
     /// Create a new [`CliRunner`] from a provided tokio [`Runtime`](tokio::runtime::Runtime).
     pub const fn from_runtime(tokio_runtime: tokio::runtime::Runtime) -> Self {
-        Self { tokio_runtime }
+        Self { config: CliRunnerConfig::new(), tokio_runtime }
+    }
+
+    /// Sets the [`CliRunnerConfig`] for this runner.
+    pub const fn with_config(mut self, config: CliRunnerConfig) -> Self {
+        self.config = config;
+        self
     }
 
     /// Executes an async block on the runtime and blocks until completion.
@@ -74,7 +80,7 @@ impl CliRunner {
             // after the command has finished or exit signal was received we shutdown the task
             // manager which fires the shutdown signal to all tasks spawned via the task
             // executor and awaiting on tasks spawned with graceful shutdown
-            task_manager.graceful_shutdown_with_timeout(Duration::from_secs(5));
+            task_manager.graceful_shutdown_with_timeout(self.config.graceful_shutdown_timeout);
         }
 
         // `drop(tokio_runtime)` would block the current thread until its pools
@@ -97,6 +103,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(self.config.graceful_shutdown_timeout);
+        }
+
+        // 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
@@ -160,6 +217,38 @@ pub struct CliContext {
     pub task_executor: TaskExecutor,
 }
 
+/// Default timeout for graceful shutdown of tasks.
+const DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Configuration for [`CliRunner`].
+#[derive(Debug, Clone)]
+pub struct CliRunnerConfig {
+    /// Timeout for graceful shutdown of tasks.
+    ///
+    /// After the command completes, this is the maximum time to wait for spawned tasks
+    /// to finish before forcefully terminating them.
+    pub graceful_shutdown_timeout: Duration,
+}
+
+impl Default for CliRunnerConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl CliRunnerConfig {
+    /// Creates a new config with default values.
+    pub const fn new() -> Self {
+        Self { graceful_shutdown_timeout: DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT }
+    }
+
+    /// Sets the graceful shutdown timeout.
+    pub const fn with_graceful_shutdown_timeout(mut self, timeout: Duration) -> Self {
+        self.graceful_shutdown_timeout = timeout;
+        self
+    }
+}
+
 /// Creates a new default tokio multi-thread [Runtime](tokio::runtime::Runtime) with all features
 /// enabled
 pub fn tokio_runtime() -> Result<tokio::runtime::Runtime, std::io::Error> {

crates/cli/util/Cargo.toml

@@ -26,7 +26,8 @@ rand_08.workspace = true
 thiserror.workspace = true
 serde.workspace = true
 
-tracy-client = { workspace = true, optional = true, features = ["demangle"] }
+tracy-client = { workspace = true, optional = true }
+reth-tracing = { workspace = true, optional = true }
 
 [dev-dependencies]
 rand.workspace = true
@@ -46,7 +47,7 @@ jemalloc-prof = ["jemalloc", "tikv-jemallocator?/profiling"]
 jemalloc-unprefixed = ["jemalloc", "tikv-jemallocator?/unprefixed_malloc_on_supported_platforms"]
 
 # Wraps the selected allocator in the tracy profiling allocator
-tracy-allocator = ["dep:tracy-client"]
+tracy-allocator = ["dep:tracy-client", "dep:reth-tracing"]
 
 snmalloc = ["dep:snmalloc-rs"]
 

crates/cli/util/src/allocator.rs

@@ -25,7 +25,6 @@ cfg_if::cfg_if! {
 cfg_if::cfg_if! {
     if #[cfg(feature = "tracy-allocator")] {
         type AllocatorWrapper = tracy_client::ProfiledAllocator<AllocatorInner>;
-        tracy_client::register_demangler!();
         const fn new_allocator_wrapper() -> AllocatorWrapper {
             AllocatorWrapper::new(AllocatorInner {}, 100)
         }

crates/cli/util/src/lib.rs

@@ -8,6 +8,9 @@
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
+#[cfg(feature = "tracy-allocator")]
+use reth_tracing as _;
+
 pub mod allocator;
 pub mod cancellation;
 

crates/config/src/config.rs

@@ -2,9 +2,8 @@
 use reth_network_types::{PeersConfig, SessionsConfig};
 use reth_prune_types::PruneModes;
 use reth_stages_types::ExecutionStageThresholds;
-use reth_static_file_types::StaticFileSegment;
+use reth_static_file_types::{StaticFileMap, StaticFileSegment};
 use std::{
-    collections::HashMap,
     path::{Path, PathBuf},
     time::Duration,
 };
@@ -22,7 +21,6 @@ pub const DEFAULT_BLOCK_INTERVAL: usize = 5;
 #[cfg_attr(feature = "serde", serde(default))]
 pub struct Config {
     /// Configuration for each stage in the pipeline.
-    // TODO(onbjerg): Can we make this easier to maintain when we add/remove stages?
     pub stages: StageConfig,
     /// Configuration for pruning.
     #[cfg_attr(feature = "serde", serde(default))]
@@ -438,6 +436,8 @@ pub struct BlocksPerFileConfig {
     pub receipts: Option<u64>,
     /// Number of blocks per file for the transaction senders segment.
     pub transaction_senders: Option<u64>,
+    /// Number of blocks per file for the account changesets segment.
+    pub account_change_sets: Option<u64>,
 }
 
 impl StaticFilesConfig {
@@ -445,8 +445,13 @@ impl StaticFilesConfig {
     ///
     /// Returns an error if any blocks per file value is zero.
     pub fn validate(&self) -> eyre::Result<()> {
-        let BlocksPerFileConfig { headers, transactions, receipts, transaction_senders } =
-            self.blocks_per_file;
+        let BlocksPerFileConfig {
+            headers,
+            transactions,
+            receipts,
+            transaction_senders,
+            account_change_sets,
+        } = self.blocks_per_file;
         eyre::ensure!(headers != Some(0), "Headers segment blocks per file must be greater than 0");
         eyre::ensure!(
             transactions != Some(0),
@@ -460,15 +465,24 @@ impl StaticFilesConfig {
             transaction_senders != Some(0),
             "Transaction senders segment blocks per file must be greater than 0"
         );
+        eyre::ensure!(
+            account_change_sets != Some(0),
+            "Account changesets segment blocks per file must be greater than 0"
+        );
         Ok(())
     }
 
-    /// Converts the blocks per file configuration into a [`HashMap`] per segment.
-    pub fn as_blocks_per_file_map(&self) -> HashMap<StaticFileSegment, u64> {
-        let BlocksPerFileConfig { headers, transactions, receipts, transaction_senders } =
-            self.blocks_per_file;
+    /// Converts the blocks per file configuration into a [`StaticFileMap`].
+    pub fn as_blocks_per_file_map(&self) -> StaticFileMap<u64> {
+        let BlocksPerFileConfig {
+            headers,
+            transactions,
+            receipts,
+            transaction_senders,
+            account_change_sets,
+        } = self.blocks_per_file;
 
-        let mut map = HashMap::new();
+        let mut map = StaticFileMap::default();
         // Iterating over all possible segments allows us to do an exhaustive match here,
         // to not forget to configure new segments in the future.
         for segment in StaticFileSegment::iter() {
@@ -477,6 +491,7 @@ impl StaticFilesConfig {
                 StaticFileSegment::Transactions => transactions,
                 StaticFileSegment::Receipts => receipts,
                 StaticFileSegment::TransactionSenders => transaction_senders,
+                StaticFileSegment::AccountChangeSets => account_change_sets,
             };
 
             if let Some(blocks_per_file) = blocks_per_file {
@@ -528,11 +543,15 @@ impl PruneConfig {
 
     /// Returns whether there is any kind of receipt pruning configuration.
     pub fn has_receipts_pruning(&self) -> bool {
-        self.segments.receipts.is_some() || !self.segments.receipts_log_filter.is_empty()
+        self.segments.has_receipts_pruning()
     }
 
-    /// Merges another `PruneConfig` into this one, taking values from the other config if and only
-    /// if the corresponding value in this config is not set.
+    /// Merges values from `other` into `self`.
+    /// - `Option<PruneMode>` fields: set from `other` only if `self` is `None`.
+    /// - `block_interval`: set from `other` only if `self.block_interval ==
+    ///   DEFAULT_BLOCK_INTERVAL`.
+    /// - `merkle_changesets`: always set from `other`.
+    /// - `receipts_log_filter`: set from `other` only if `self` is empty and `other` is non-empty.
     pub fn merge(&mut self, other: Self) {
         let Self {
             block_interval,
@@ -561,7 +580,7 @@ impl PruneConfig {
         self.segments.account_history = self.segments.account_history.or(account_history);
         self.segments.storage_history = self.segments.storage_history.or(storage_history);
         self.segments.bodies_history = self.segments.bodies_history.or(bodies_history);
-        // Merkle changesets is not optional, so we just replace it if provided
+        // Merkle changesets is not optional; always take the value from `other`
         self.segments.merkle_changesets = merkle_changesets;
 
         if self.segments.receipts_log_filter.0.is_empty() && !receipts_log_filter.0.is_empty() {
@@ -1059,18 +1078,6 @@ transaction_lookup = 'full'
 receipts = { distance = 16384 }
 #";
         let _conf: Config = toml::from_str(s).unwrap();
-
-        let s = r"#
-[prune]
-block_interval = 5
-
-[prune.segments]
-sender_recovery = { distance = 16384 }
-transaction_lookup = 'full'
-receipts = 'full'
-#";
-        let err = toml::from_str::<Config>(s).unwrap_err().to_string();
-        assert!(err.contains("invalid value: string \"full\""), "{}", err);
     }
 
     #[test]

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,
@@ -492,13 +500,11 @@ mod tests {
         let expected_blob_gas_used = 10 * DATA_GAS_PER_BLOB;
 
         // validate blob, it should fail blob gas used validation
-        assert_eq!(
-            validate_block_pre_execution(&block, &chain_spec),
-            Err(ConsensusError::BlobGasUsedDiff(GotExpected {
-                got: 1,
-                expected: expected_blob_gas_used
-            }))
-        );
+        assert!(matches!(
+            validate_block_pre_execution(&block, &chain_spec).unwrap_err(),
+            ConsensusError::BlobGasUsedDiff(diff)
+                if diff.got == 1 && diff.expected == expected_blob_gas_used
+        ));
     }
 
     #[test]
@@ -509,10 +515,10 @@ mod tests {
 
         // Test exceeding default - should fail
         let header_33 = Header { extra_data: Bytes::from(vec![0; 33]), ..Default::default() };
-        assert_eq!(
-            validate_header_extra_data(&header_33, 32),
-            Err(ConsensusError::ExtraDataExceedsMax { len: 33 })
-        );
+        assert!(matches!(
+            validate_header_extra_data(&header_33, 32).unwrap_err(),
+            ConsensusError::ExtraDataExceedsMax { len } if len == 33
+        ));
 
         // Test with custom larger limit - should pass
         assert!(validate_header_extra_data(&header_33, 64).is_ok());

crates/consensus/consensus/src/lib.rs

@@ -11,9 +11,10 @@
 
 extern crate alloc;
 
-use alloc::{boxed::Box, fmt::Debug, string::String, vec::Vec};
+use alloc::{boxed::Box, fmt::Debug, string::String, sync::Arc, vec::Vec};
 use alloy_consensus::Header;
 use alloy_primitives::{BlockHash, BlockNumber, Bloom, B256};
+use core::error::Error;
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{
     constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK, MINIMUM_GAS_LIMIT},
@@ -49,15 +50,12 @@ pub trait FullConsensus<N: NodePrimitives>: Consensus<N::Block> {
 /// Consensus is a protocol that chooses canonical chain.
 #[auto_impl::auto_impl(&, Arc)]
 pub trait Consensus<B: Block>: HeaderValidator<B::Header> {
-    /// The error type related to consensus.
-    type Error;
-
     /// Ensures that body field values match the header.
     fn validate_body_against_header(
         &self,
         body: &B::Body,
         header: &SealedHeader<B::Header>,
-    ) -> Result<(), Self::Error>;
+    ) -> Result<(), ConsensusError>;
 
     /// Validate a block disregarding world state, i.e. things that can be checked before sender
     /// recovery and execution.
@@ -69,7 +67,7 @@ pub trait Consensus<B: Block>: HeaderValidator<B::Header> {
     /// **This should not be called for the genesis block**.
     ///
     /// Note: validating blocks does not include other validations of the Consensus
-    fn validate_block_pre_execution(&self, block: &SealedBlock<B>) -> Result<(), Self::Error>;
+    fn validate_block_pre_execution(&self, block: &SealedBlock<B>) -> Result<(), ConsensusError>;
 }
 
 /// `HeaderValidator` is a protocol that validates headers and their relationships.
@@ -125,7 +123,7 @@ pub trait HeaderValidator<H = Header>: Debug + Send + Sync {
 }
 
 /// Consensus Errors
-#[derive(Debug, PartialEq, Eq, Clone, thiserror::Error)]
+#[derive(Debug, Clone, thiserror::Error)]
 pub enum ConsensusError {
     /// Error when the gas used in the header exceeds the gas limit.
     #[error("block used gas ({gas_used}) is greater than gas limit ({gas_limit})")]
@@ -135,7 +133,7 @@ pub enum ConsensusError {
         /// The gas limit in the block header.
         gas_limit: u64,
     },
-    /// Error when the gas the gas limit is more than the maximum allowed.
+    /// Error when the gas limit is more than the maximum allowed.
     #[error(
         "header gas limit ({gas_limit}) exceed the maximum allowed gas limit ({MAXIMUM_GAS_LIMIT_BLOCK})"
     )]
@@ -410,6 +408,9 @@ pub enum ConsensusError {
     /// Other, likely an injected L2 error.
     #[error("{0}")]
     Other(String),
+    /// Other unspecified error.
+    #[error(transparent)]
+    Custom(#[from] Arc<dyn Error + Send + Sync>),
 }
 
 impl ConsensusError {
@@ -447,3 +448,34 @@ pub struct TxGasLimitTooHighErr {
     /// The maximum allowed gas limit
     pub max_allowed: u64,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[derive(thiserror::Error, Debug)]
+    #[error("Custom L2 consensus error")]
+    struct CustomL2Error;
+
+    #[test]
+    fn test_custom_error_conversion() {
+        // Test conversion from custom error to ConsensusError
+        let custom_err = CustomL2Error;
+        let arc_err: Arc<dyn Error + Send + Sync> = Arc::new(custom_err);
+        let consensus_err: ConsensusError = arc_err.into();
+
+        // Verify it's the Custom variant
+        assert!(matches!(consensus_err, ConsensusError::Custom(_)));
+    }
+
+    #[test]
+    fn test_custom_error_display() {
+        let custom_err = CustomL2Error;
+        let arc_err: Arc<dyn Error + Send + Sync> = Arc::new(custom_err);
+        let consensus_err: ConsensusError = arc_err.into();
+
+        // Verify the error message is preserved through transparent attribute
+        let error_message = format!("{}", consensus_err);
+        assert_eq!(error_message, "Custom L2 consensus error");
+    }
+}

crates/consensus/consensus/src/noop.rs

@@ -55,19 +55,17 @@ impl<H> HeaderValidator<H> for NoopConsensus {
 }
 
 impl<B: Block> Consensus<B> for NoopConsensus {
-    type Error = ConsensusError;
-
     /// Validates body against header (no-op implementation).
     fn validate_body_against_header(
         &self,
         _body: &B::Body,
         _header: &SealedHeader<B::Header>,
-    ) -> Result<(), Self::Error> {
+    ) -> Result<(), ConsensusError> {
         Ok(())
     }
 
     /// Validates block before execution (no-op implementation).
-    fn validate_block_pre_execution(&self, _block: &SealedBlock<B>) -> Result<(), Self::Error> {
+    fn validate_block_pre_execution(&self, _block: &SealedBlock<B>) -> Result<(), ConsensusError> {
         Ok(())
     }
 }

crates/consensus/consensus/src/test_utils.rs

@@ -61,13 +61,11 @@ impl<N: NodePrimitives> FullConsensus<N> for TestConsensus {
 }
 
 impl<B: Block> Consensus<B> for TestConsensus {
-    type Error = ConsensusError;
-
     fn validate_body_against_header(
         &self,
         _body: &B::Body,
         _header: &SealedHeader<B::Header>,
-    ) -> Result<(), Self::Error> {
+    ) -> Result<(), ConsensusError> {
         if self.fail_body_against_header() {
             Err(ConsensusError::BaseFeeMissing)
         } else {
@@ -75,7 +73,7 @@ impl<B: Block> Consensus<B> for TestConsensus {
         }
     }
 
-    fn validate_block_pre_execution(&self, _block: &SealedBlock<B>) -> Result<(), Self::Error> {
+    fn validate_block_pre_execution(&self, _block: &SealedBlock<B>) -> Result<(), ConsensusError> {
         if self.fail_validation() {
             Err(ConsensusError::BaseFeeMissing)
         } else {

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

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

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

@@ -2,11 +2,11 @@ use crate::{network::NetworkTestContext, payload::PayloadTestContext, rpc::RpcTe
 use alloy_consensus::{transaction::TxHashRef, BlockHeader};
 use alloy_eips::BlockId;
 use alloy_primitives::{BlockHash, BlockNumber, Bytes, Sealable, B256};
-use alloy_rpc_types_engine::ForkchoiceState;
+use alloy_rpc_types_engine::{ExecutionPayloadEnvelopeV5, ForkchoiceState};
 use alloy_rpc_types_eth::BlockNumberOrTag;
 use eyre::Ok;
 use futures_util::Future;
-use jsonrpsee::http_client::HttpClient;
+use jsonrpsee::{core::client::ClientT, http_client::HttpClient};
 use reth_chainspec::EthereumHardforks;
 use reth_network_api::test_utils::PeersHandleProvider;
 use reth_node_api::{
@@ -20,6 +20,7 @@ use reth_provider::{
     BlockReader, BlockReaderIdExt, CanonStateNotificationStream, CanonStateSubscriptions,
     HeaderProvider, StageCheckpointReader,
 };
+use reth_rpc_api::TestingBuildBlockRequestV1;
 use reth_rpc_builder::auth::AuthServerHandle;
 use reth_rpc_eth_api::helpers::{EthApiSpec, EthTransactions, TraceExt};
 use reth_stages_types::StageId;
@@ -319,4 +320,20 @@ where
 
         Ok(crate::testsuite::NodeClient::new_with_beacon_engine(rpc, auth, url, beacon_handle))
     }
+
+    /// Calls the `testing_buildBlockV1` RPC on this node.
+    ///
+    /// This endpoint builds a block using the provided parent, payload attributes, and
+    /// transactions. Requires the `Testing` RPC module to be enabled.
+    pub async fn testing_build_block_v1(
+        &self,
+        request: TestingBuildBlockRequestV1,
+    ) -> eyre::Result<ExecutionPayloadEnvelopeV5> {
+        let client =
+            self.rpc_client().ok_or_else(|| eyre::eyre!("HTTP RPC client not available"))?;
+
+        let res: ExecutionPayloadEnvelopeV5 =
+            client.request("testing_buildBlockV1", [request]).await?;
+        eyre::Ok(res)
+    }
 }

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

@@ -11,6 +11,7 @@ use reth_node_builder::{
     PayloadTypes,
 };
 use reth_node_core::args::{DiscoveryArgs, NetworkArgs, RpcServerArgs};
+use reth_primitives_traits::AlloyBlockHeader;
 use reth_provider::providers::BlockchainProvider;
 use reth_rpc_server_types::RpcModuleSelection;
 use reth_tasks::TaskManager;
@@ -157,8 +158,8 @@ where
                     .await?;
 
                 let node = NodeTestContext::new(node, self.attributes_generator).await?;
-
-                let genesis = node.block_hash(0);
+                let genesis_number = self.chain_spec.genesis_header().number();
+                let genesis = node.block_hash(genesis_number);
                 node.update_forkchoice(genesis, genesis).await?;
 
                 eyre::Ok(node)

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

@@ -110,9 +110,9 @@ 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
         let db_args = reth_node_core::args::DatabaseArgs::default().database_args();
         let db_env = reth_db::init_db(&db_path, db_args)?;
         let db = Arc::new(db_env);
@@ -125,6 +125,10 @@ 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)
+                .with_default_tables()
+                .build()
+                .unwrap(),
         )?;
 
         // Initialize genesis if needed
@@ -311,10 +315,12 @@ 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
         {
-            let db_env = reth_db::init_db(&db_path, DatabaseArguments::default()).unwrap();
+            let db_args = reth_node_core::args::DatabaseArgs::default().database_args();
+            let db_env = reth_db::init_db(&db_path, db_args).unwrap();
             let db = Arc::new(db_env);
 
             let provider_factory: ProviderFactory<
@@ -324,6 +330,10 @@ 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())
+                    .with_default_tables()
+                    .build()
+                    .unwrap(),
             )
             .expect("failed to create provider factory");
 
@@ -385,6 +395,10 @@ mod tests {
                 chain_spec.clone(),
                 reth_provider::providers::StaticFileProvider::read_only(static_files_path, false)
                     .unwrap(),
+                reth_provider::providers::RocksDBProvider::builder(rocksdb_dir_path)
+                    .with_default_tables()
+                    .build()
+                    .unwrap(),
             )
             .expect("failed to create provider factory");
 
@@ -466,17 +480,25 @@ mod tests {
         let datadir = temp_dir.path().join("datadir");
         std::fs::create_dir_all(&datadir).unwrap();
         let db_path = datadir.join("db");
-        let db_env = reth_db::init_db(&db_path, DatabaseArguments::default()).unwrap();
+        let db_args = reth_node_core::args::DatabaseArgs::default().database_args();
+        let db_env = reth_db::init_db(&db_path, db_args).unwrap();
         let db = Arc::new(reth_db::test_utils::TempDatabase::new(db_env, db_path));
 
         // 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)
+                .with_default_tables()
+                .build()
+                .unwrap(),
         )
         .expect("failed to create provider factory");
 

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

@@ -2,9 +2,7 @@
 
 use crate::testsuite::{Action, Environment};
 use alloy_primitives::B256;
-use alloy_rpc_types_engine::{
-    ExecutionPayloadV1, ExecutionPayloadV2, ExecutionPayloadV3, PayloadStatusEnum,
-};
+use alloy_rpc_types_engine::{ExecutionPayloadV3, PayloadStatusEnum};
 use alloy_rpc_types_eth::{Block, Header, Receipt, Transaction, TransactionRequest};
 use eyre::Result;
 use futures_util::future::BoxFuture;
@@ -131,7 +129,10 @@ where
             })?;
 
             // Convert block to ExecutionPayloadV3
-            let payload = block_to_payload_v3(block.clone());
+            let payload = ExecutionPayloadV3::from_block_unchecked(
+                block.hash(),
+                &block.map_transactions(|tx| tx.inner).into_consensus(),
+            );
 
             // Send the payload to the target node
             let target_engine = env.node_clients[self.node_idx].engine.http_client();
@@ -327,32 +328,3 @@ where
         })
     }
 }
-
-/// Helper function to convert a block to `ExecutionPayloadV3`
-fn block_to_payload_v3(block: Block) -> ExecutionPayloadV3 {
-    use alloy_primitives::U256;
-
-    ExecutionPayloadV3 {
-        payload_inner: ExecutionPayloadV2 {
-            payload_inner: ExecutionPayloadV1 {
-                parent_hash: block.header.inner.parent_hash,
-                fee_recipient: block.header.inner.beneficiary,
-                state_root: block.header.inner.state_root,
-                receipts_root: block.header.inner.receipts_root,
-                logs_bloom: block.header.inner.logs_bloom,
-                prev_randao: block.header.inner.mix_hash,
-                block_number: block.header.inner.number,
-                gas_limit: block.header.inner.gas_limit,
-                gas_used: block.header.inner.gas_used,
-                timestamp: block.header.inner.timestamp,
-                extra_data: block.header.inner.extra_data.clone(),
-                base_fee_per_gas: U256::from(block.header.inner.base_fee_per_gas.unwrap_or(0)),
-                block_hash: block.header.hash,
-                transactions: vec![], // No transactions needed for buffering tests
-            },
-            withdrawals: block.withdrawals.unwrap_or_default().to_vec(),
-        },
-        blob_gas_used: block.header.inner.blob_gas_used.unwrap_or(0),
-        excess_blob_gas: block.header.inner.excess_blob_gas.unwrap_or(0),
-    }
-}

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

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

crates/engine/primitives/src/config.rs

@@ -1,5 +1,7 @@
 //! Engine tree configuration.
 
+use alloy_eips::merge::EPOCH_SLOTS;
+
 /// Triggers persistence when the number of canonical blocks in memory exceeds this threshold.
 pub const DEFAULT_PERSISTENCE_THRESHOLD: u64 = 2;
 
@@ -40,7 +42,7 @@ pub const DEFAULT_RESERVED_CPU_CORES: usize = 1;
 /// Default maximum concurrency for prewarm task.
 pub const DEFAULT_PREWARM_MAX_CONCURRENCY: usize = 16;
 
-const DEFAULT_BLOCK_BUFFER_LIMIT: u32 = 256;
+const DEFAULT_BLOCK_BUFFER_LIMIT: u32 = EPOCH_SLOTS as u32 * 2;
 const DEFAULT_MAX_INVALID_HEADER_CACHE_LENGTH: u32 = 256;
 const DEFAULT_MAX_EXECUTE_BLOCK_BATCH_SIZE: usize = 4;
 const DEFAULT_CROSS_BLOCK_CACHE_SIZE: u64 = 4 * 1024 * 1024 * 1024;
@@ -89,6 +91,8 @@ pub struct TreeConfig {
     /// Whether to always compare trie updates from the state root task to the trie updates from
     /// the regular state root calculation.
     always_compare_trie_updates: bool,
+    /// Whether to disable state cache.
+    disable_state_cache: bool,
     /// Whether to disable parallel prewarming.
     disable_prewarming: bool,
     /// Whether to disable the parallel sparse trie state root algorithm.
@@ -97,7 +101,7 @@ pub struct TreeConfig {
     state_provider_metrics: bool,
     /// Cross-block cache size in bytes.
     cross_block_cache_size: u64,
-    /// Whether the host has enough parallelism to run state root in parallel.
+    /// Whether the host has enough parallelism to run state root task.
     has_enough_parallelism: bool,
     /// Whether multiproof task should chunk proof targets.
     multiproof_chunking_enabled: bool,
@@ -131,6 +135,8 @@ pub struct TreeConfig {
     storage_worker_count: usize,
     /// Number of account proof worker threads.
     account_worker_count: usize,
+    /// Whether to enable V2 storage proofs.
+    enable_proof_v2: bool,
 }
 
 impl Default for TreeConfig {
@@ -143,6 +149,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,
@@ -158,6 +165,7 @@ impl Default for TreeConfig {
             allow_unwind_canonical_header: false,
             storage_worker_count: default_storage_worker_count(),
             account_worker_count: default_account_worker_count(),
+            enable_proof_v2: false,
         }
     }
 }
@@ -173,6 +181,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,
@@ -188,6 +197,7 @@ impl TreeConfig {
         allow_unwind_canonical_header: bool,
         storage_worker_count: usize,
         account_worker_count: usize,
+        enable_proof_v2: bool,
     ) -> Self {
         Self {
             persistence_threshold,
@@ -197,6 +207,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,
@@ -212,6 +223,7 @@ impl TreeConfig {
             allow_unwind_canonical_header,
             storage_worker_count,
             account_worker_count,
+            enable_proof_v2,
         }
     }
 
@@ -271,7 +283,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 +380,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 +408,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;
@@ -487,4 +505,15 @@ impl TreeConfig {
         self.account_worker_count = account_worker_count.max(MIN_WORKER_COUNT);
         self
     }
+
+    /// Return whether V2 storage proofs are enabled.
+    pub const fn enable_proof_v2(&self) -> bool {
+        self.enable_proof_v2
+    }
+
+    /// Setter for whether to enable V2 storage proofs.
+    pub const fn with_enable_proof_v2(mut self, enable_proof_v2: bool) -> Self {
+        self.enable_proof_v2 = enable_proof_v2;
+        self
+    }
 }

crates/engine/primitives/src/lib.rs

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

crates/engine/service/src/service.rs

@@ -1,7 +1,7 @@
 use futures::{Stream, StreamExt};
 use pin_project::pin_project;
 use reth_chainspec::EthChainSpec;
-use reth_consensus::{ConsensusError, FullConsensus};
+use reth_consensus::FullConsensus;
 use reth_engine_primitives::{BeaconEngineMessage, ConsensusEngineEvent};
 use reth_engine_tree::{
     backfill::PipelineSync,
@@ -70,7 +70,7 @@ where
     /// Constructor for `EngineService`.
     #[expect(clippy::too_many_arguments)]
     pub fn new<V, C>(
-        consensus: Arc<dyn FullConsensus<N::Primitives, Error = ConsensusError>>,
+        consensus: Arc<dyn FullConsensus<N::Primitives>>,
         chain_spec: Arc<N::ChainSpec>,
         client: Client,
         incoming_requests: EngineMessageStream<N::Payload>,

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
@@ -51,6 +52,7 @@ futures.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt", "rt-multi-thread", "sync", "macros"] }
 mini-moka = { workspace = true, features = ["sync"] }
+moka = { workspace = true, features = ["sync"] }
 smallvec.workspace = true
 
 # metrics

crates/engine/tree/benches/state_root_task.rs

@@ -229,12 +229,19 @@ fn bench_state_root(c: &mut Criterion) {
                         black_box({
                             let mut handle = payload_processor.spawn(
                                 Default::default(),
-                                core::iter::empty::<
-                                    Result<Recovered<TransactionSigned>, core::convert::Infallible>,
-                                >(),
+                                (
+                                    Vec::<
+                                        Result<
+                                            Recovered<TransactionSigned>,
+                                            core::convert::Infallible,
+                                        >,
+                                    >::new(),
+                                    std::convert::identity,
+                                ),
                                 StateProviderBuilder::new(provider.clone(), genesis_hash, None),
                                 OverlayStateProviderFactory::new(provider),
                                 &TreeConfig::default(),
+                                None,
                             );
 
                             let mut state_hook = handle.state_hook();

crates/engine/tree/docs/root.md

@@ -128,12 +128,12 @@ we send them along with the state updates to the [Sparse Trie Task](#sparse-trie
 
 ### Finishing the calculation
 
-Once all transactions are executed, the [Engine](#engine) sends a `StateRootMessage::FinishStateUpdates` message
+Once all transactions are executed, the [Engine](#engine) sends a `StateRootMessage::FinishedStateUpdates` message
 to the State Root Task, marking the end of receiving state updates.
 
 Every time we receive a new proof from the [MultiProof Manager](#multiproof-manager), we also check
 the following conditions:
-1. Are all updates received? (`StateRootMessage::FinishStateUpdates` was sent)
+1. Are all updates received? (`StateRootMessage::FinishedStateUpdates` was sent)
 2. Is `ProofSequencer` empty? (no proofs are pending for sequencing)
 3. Are all proofs that were sent to the [`MultiProofManager::spawn_or_queue`](#multiproof-manager) finished
 calculating and were sent to the [Sparse Trie Task](#sparse-trie-task)?