chore: update alloy-evm patches and fix cargo-deny

- Switch alloy-evm and alloy-op-evm patches from Rimeeeeee/evm to alloy-rs/evm staging-revm branch
- Add Soubhik-10/alloy to deny.toml allow-git for EIP-7928 BAL support
- Add TODOs for temporary git patches

.github/assets/hive/build_simulators.sh

@@ -11,14 +11,17 @@ go build .
 
 # Run each hive command in the background for each simulator and wait
 echo "Building images"
-# TODO: test code has been moved from https://github.com/ethereum/execution-spec-tests to https://github.com/ethereum/execution-specs  we need to pin eels branch with `--sim.buildarg branch=<release-branch-name>` once we have the fusaka release tagged on the new repo
-./hive -client reth --sim "ethereum/eels" --sim.buildarg fixtures=https://github.com/ethereum/execution-spec-tests/releases/download/v5.3.0/fixtures_develop.tar.gz -sim.timelimit 1s || true &
+./hive -client reth --sim "ethereum/eels" \
+  --sim.buildarg fixtures=https://github.com/ethereum/execution-spec-tests/releases/download/bal@v2.0.0/fixtures_bal.tar.gz \
+  --sim.buildarg branch=eips/amsterdam/eip-7928 \
+  --sim.timelimit 1s || true &
 ./hive -client reth --sim "ethereum/engine" -sim.timelimit 1s || true &
 ./hive -client reth --sim "devp2p" -sim.timelimit 1s || true &
 ./hive -client reth --sim "ethereum/rpc-compat" -sim.timelimit 1s || true &
 ./hive -client reth --sim "smoke/genesis" -sim.timelimit 1s || true &
 ./hive -client reth --sim "smoke/network" -sim.timelimit 1s || true &
 ./hive -client reth --sim "ethereum/sync" -sim.timelimit 1s || true &
+
 wait
 
 # Run docker save in parallel, wait and exit on error
@@ -37,7 +40,7 @@ for pid in "${saving_pids[@]}"; do
     wait "$pid" || exit
 done
 
-# Make sure we don't rebuild images on the CI jobs
+# Make sure we don't rebuild images on the CI  jobs
 git apply ../.github/assets/hive/no_sim_build.diff
 go build .
 mv ./hive ../hive_assets/

.github/assets/hive/load_images.sh

@@ -24,4 +24,4 @@ done
 
 wait
 
-docker image ls -a
+docker image ls -a

.github/workflows/bench.yml

@@ -3,9 +3,9 @@
 on:
   pull_request:
   # TODO: Disabled temporarily for https://github.com/CodSpeedHQ/runner/issues/55
-  # merge_group:
+  # merge_group :
   push:
-    branches: [main]
+    branches: ["**"]
 
 env:
   CARGO_TERM_COLOR: always

.github/workflows/book.yml

@@ -74,4 +74,4 @@ jobs:
     steps:
       - name: Deploy to GitHub Pages
         id: deployment
-        uses: actions/deploy-pages@v4
+        uses: actions/deploy-pages@v4

.github/workflows/compact.yml

@@ -9,7 +9,7 @@ on:
   pull_request:
   merge_group:
   push:
-    branches: [main]
+    branches: ["**"]
 
 env:
   CARGO_TERM_COLOR: always

.github/workflows/dependencies.yml

@@ -15,6 +15,6 @@ permissions:
 
 jobs:
   update:
-    uses: tempoxyz/ci/.github/workflows/cargo-update-pr.yml@main
+    uses: ithacaxyz/ci/.github/workflows/cargo-update-pr.yml@main
     secrets:
       token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/e2e.yml

@@ -6,7 +6,7 @@ on:
   pull_request:
   merge_group:
   push:
-    branches: [main]
+    branches: ["**"]
 
 env:
   CARGO_TERM_COLOR: always

.github/workflows/hive.yml

@@ -6,7 +6,9 @@ on:
   workflow_dispatch:
   schedule:
     - cron: "0 */6 * * *"
-
+  pull_request:
+    branches:
+      - main
 env:
   CARGO_TERM_COLOR: always
 
@@ -42,6 +44,7 @@ jobs:
         uses: actions/checkout@v6
         with:
           repository: ethereum/hive
+          ref: master
           path: hivetests
 
       - name: Get hive commit hash
@@ -127,27 +130,29 @@ jobs:
           # eth_ rpc methods
           - sim: ethereum/rpc-compat
             include:
-              - eth_blockNumber
+              # - eth_blockNumber
               - eth_call
-              - eth_chainId
-              - eth_createAccessList
-              - eth_estimateGas
-              - eth_feeHistory
-              - eth_getBalance
-              - eth_getBlockBy
-              - eth_getBlockTransactionCountBy
-              - eth_getCode
-              - eth_getProof
-              - eth_getStorage
-              - eth_getTransactionBy
-              - eth_getTransactionCount
-              - eth_getTransactionReceipt
-              - eth_sendRawTransaction
-              - eth_syncing
-              # debug_ rpc methods
-              - debug_
+              # - eth_chainId
+              # - eth_createAccessList
+              # - eth_estimateGas
+              # - eth_feeHistory
+              # - eth_getBalance
+              # - eth_getBlockBy
+              # - eth_getBlockTransactionCountBy
+              # - eth_getCode
+              # - eth_getProof
+              # - eth_getStorage
+              # - eth_getTransactionBy
+              # - eth_getTransactionCount
+              # - eth_getTransactionReceipt
+              # - eth_sendRawTransaction
+              # - eth_syncing
+              # # debug_ rpc methods
+              # - debug_
 
           # consume-engine
+          - sim: ethereum/eels/consume-engine
+            limit: .*tests/amsterdam.*
           - sim: ethereum/eels/consume-engine
             limit: .*tests/osaka.*
           - sim: ethereum/eels/consume-engine
@@ -164,10 +169,10 @@ jobs:
             limit: .*tests/homestead.*
           - sim: ethereum/eels/consume-engine
             limit: .*tests/frontier.*
-          - sim: ethereum/eels/consume-engine
-            limit: .*tests/paris.*
 
           # consume-rlp
+          - sim: ethereum/eels/consume-rlp
+            limit: .*tests/amsterdam.*
           - sim: ethereum/eels/consume-rlp
             limit: .*tests/osaka.*
           - sim: ethereum/eels/consume-rlp
@@ -184,8 +189,6 @@ jobs:
             limit: .*tests/homestead.*
           - sim: ethereum/eels/consume-rlp
             limit: .*tests/frontier.*
-          - sim: ethereum/eels/consume-rlp
-            limit: .*tests/paris.*
     needs:
       - prepare-reth-stable
       - prepare-reth-edge

.github/workflows/integration.yml

@@ -6,7 +6,7 @@ on:
   pull_request:
   merge_group:
   push:
-    branches: [main]
+    branches: ["**"]
   schedule:
     # Run once a day at 3:00 UTC
     - cron: "0 3 * * *"

.github/workflows/lint.yml

@@ -4,7 +4,7 @@ on:
   pull_request:
   merge_group:
   push:
-    branches: [main]
+    branches: ["**"]
 
 env:
   CARGO_TERM_COLOR: always
@@ -285,7 +285,7 @@ jobs:
       - run: zepter run check
 
   deny:
-    uses: tempoxyz/ci/.github/workflows/deny.yml@main
+    uses: ithacaxyz/ci/.github/workflows/deny.yml@main
 
   lint-success:
     name: lint success

.github/workflows/stage.yml

@@ -6,7 +6,7 @@ on:
   pull_request:
   merge_group:
   push:
-    branches: [main]
+    branches: ["**"]
 
 env:
   CARGO_TERM_COLOR: always

.github/workflows/unit.yml

@@ -4,9 +4,10 @@ name: unit
 
 on:
   pull_request:
+    branches: ["**"]
   merge_group:
   push:
-    branches: [main]
+    branches: ["**"]
 
 env:
   CARGO_TERM_COLOR: always

.github/workflows/windows.yml

@@ -4,9 +4,9 @@ name: windows
 
 on:
   push:
-    branches: [main]
+    branches: ["**"]
   pull_request:
-    branches: [main]
+    branches: ["**"]
   merge_group:
 
 env:

CLAUDE.md

@@ -249,7 +249,7 @@ Write comments that remain valuable after the PR is merged. Future readers won't
 unsafe impl GlobalAlloc for LimitedAllocator { ... }
 
 // Binary search requires sorted input. Panics on unsorted slices.
-fn find_index(items: &[Item], target: &Item) -> Option<usize>
+fn find_index(items: &[Item], target: &Item) -> Option
 
 // Timeout set to 5s to match EVM block processing limits
 const TRACER_TIMEOUT: Duration = Duration::from_secs(5);

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace.package]
-version = "1.10.1"
+version = "1.10.0"
 edition = "2024"
 rust-version = "1.88"
 license = "MIT OR Apache-2.0"
@@ -376,11 +376,11 @@ reth-era-utils = { path = "crates/era-utils" }
 reth-errors = { path = "crates/errors" }
 reth-eth-wire = { path = "crates/net/eth-wire" }
 reth-eth-wire-types = { path = "crates/net/eth-wire-types" }
-reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-cli = { path = "crates/ethereum/cli", default-features = false }
 reth-ethereum-consensus = { path = "crates/ethereum/consensus", default-features = false }
 reth-ethereum-engine-primitives = { path = "crates/ethereum/engine-primitives", default-features = false }
 reth-ethereum-forks = { path = "crates/ethereum/hardforks", default-features = false }
+reth-ethereum-payload-builder = { path = "crates/ethereum/payload" }
 reth-ethereum-primitives = { path = "crates/ethereum/primitives", default-features = false }
 reth-ethereum = { path = "crates/ethereum/reth" }
 reth-etl = { path = "crates/etl" }
@@ -472,7 +472,7 @@ reth-zstd-compressors = { path = "crates/storage/zstd-compressors", default-feat
 reth-ress-protocol = { path = "crates/ress/protocol" }
 reth-ress-provider = { path = "crates/ress/provider" }
 
-# revm
+# revm v103 (revm v34.0.0 with BAL EIP-7928 support)
 revm = { version = "34.0.0", default-features = false }
 revm-bytecode = { version = "8.0.0", default-features = false }
 revm-database = { version = "10.0.0", default-features = false }
@@ -480,16 +480,21 @@ revm-state = { version = "9.0.0", default-features = false }
 revm-primitives = { version = "22.0.0", default-features = false }
 revm-interpreter = { version = "32.0.0", default-features = false }
 revm-database-interface = { version = "9.0.0", default-features = false }
+revm-context = { version = "13.0.0", default-features = false }
+revm-context-interface = { version = "14.0.0", default-features = false }
+revm-inspector = { version = "15.0.0", default-features = false }
+revm-handler = { version = "15.0.0", default-features = false }
+revm-precompile = { version = "32.0.0", default-features = false }
 op-revm = { version = "15.0.0", default-features = false }
-revm-inspectors = "0.34.0"
+revm-inspectors = "0.33.2"
 
 # eth
-alloy-chains = { version = "0.2.5", default-features = false }
-alloy-dyn-abi = "1.4.3"
-alloy-eip2124 = { version = "0.2.0", default-features = false }
-alloy-eip7928 = { version = "0.3.0", default-features = false }
-alloy-evm = { version = "0.26.3", default-features = false }
 alloy-primitives = { version = "1.5.0", default-features = false, features = ["map-foldhash"] }
+alloy-chains = { version = "0.2.5", default-features = false }
+alloy-evm = { version = "0.25.2", default-features = false }
+alloy-dyn-abi = "1.4.1"
+alloy-eip7928 = { version = "0.3.0", default-features = false }
+alloy-eip2124 = { version = "0.2.0", default-features = false }
 alloy-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
 alloy-sol-macro = "1.5.0"
 alloy-sol-types = { version = "1.5.0", default-features = false }
@@ -526,7 +531,7 @@ alloy-transport-ipc = { version = "1.4.3", default-features = false }
 alloy-transport-ws = { version = "1.4.3", default-features = false }
 
 # op
-alloy-op-evm = { version = "0.26.3", default-features = false }
+alloy-op-evm = { version = "0.25.2", default-features = false }
 alloy-op-hardforks = "0.4.4"
 op-alloy-rpc-types = { version = "0.23.1", default-features = false }
 op-alloy-rpc-types-engine = { version = "0.23.1", default-features = false }
@@ -739,50 +744,54 @@ tracing-subscriber = { version = "0.3", default-features = false }
 tracing-tracy = "0.11"
 triehash = "0.8"
 typenum = "1.15.0"
-vergen = "9.1.0"
+vergen = "9.0.4"
 visibility = "0.1.1"
 walkdir = "2.3.3"
-vergen-git2 = "9.1.0"
-
+vergen-git2 = "1.0.5"
 # networking
 ipnet = "2.11"
 
 [patch.crates-io]
-# alloy-consensus = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-contract = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-eips = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-genesis = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-json-rpc = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-network = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-network-primitives = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-provider = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-pubsub = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-client = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-admin = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-anvil = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-beacon = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-debug = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-engine = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-eth = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-mev = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-trace = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-rpc-types-txpool = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-serde = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-signer = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-signer-local = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport-http = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport-ipc = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
-# alloy-transport-ws = { git = "https://github.com/alloy-rs/alloy", rev = "3049f232fbb44d1909883e154eb38ec5962f53a3" }
+alloy-consensus = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-contract = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-eips = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-genesis = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-json-rpc = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-network = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-network-primitives = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-provider = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-pubsub = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-client = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-admin = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-anvil = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-beacon = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-debug = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-engine = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-eth = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-mev = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-trace = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-rpc-types-txpool = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-serde = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-signer = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-signer-local = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-transport = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-transport-http = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-transport-ipc = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+alloy-transport-ws = { git = "https://github.com/Soubhik-10/alloy", branch = "bal-devnet-1" }
+# alloy-hardforks = { git = "https://github.com/Rimeeeeee/hardforks", branch = "amsterdam" }
 
+# alloy-op-hardforks = { git = "https://github.com/Rimeeeeee/hardforks", branch = "amsterdam" }
 # op-alloy-consensus = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }
 # op-alloy-network = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }
 # op-alloy-rpc-types = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }
 # op-alloy-rpc-types-engine = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }
 # op-alloy-rpc-jsonrpsee = { git = "https://github.com/alloy-rs/op-alloy", rev = "a79d6fc" }
 #
-# revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "1207e33" }
+# alloy-evm with BAL support for revm v34/v103
+alloy-evm = { git = "https://github.com/alloy-rs/evm", branch = "staging-revm" }
+alloy-op-evm = { git = "https://github.com/alloy-rs/evm", branch = "staging-revm" }
+revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", branch = "staging-revm" }
 #
 # jsonrpsee = { git = "https://github.com/paradigmxyz/jsonrpsee", branch = "matt/make-rpc-service-pub" }
 # jsonrpsee-core = { git = "https://github.com/paradigmxyz/jsonrpsee", branch = "matt/make-rpc-service-pub" }
@@ -792,8 +801,3 @@ ipnet = "2.11"
 
 # alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "a69f0b45a6b0286e16072cb8399e02ce6ceca353" }
 # alloy-op-evm = { git = "https://github.com/alloy-rs/evm", rev = "a69f0b45a6b0286e16072cb8399e02ce6ceca353" }
-
-# revm-inspectors = { git = "https://github.com/paradigmxyz/revm-inspectors", rev = "3020ea8" }
-
-# alloy-evm = { git = "https://github.com/alloy-rs/evm", rev = "072c248" }
-# alloy-op-evm = { git = "https://github.com/alloy-rs/evm", rev = "072c248" }

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

@@ -163,7 +163,6 @@ impl NodeManager {
             "eth,reth".to_string(),
             "--disable-discovery".to_string(),
             "--trusted-only".to_string(),
-            "--disable-tx-gossip".to_string(),
         ]);
 
         // Add tracing arguments if OTLP endpoint is configured

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

@@ -541,6 +541,7 @@ impl Command {
                 suggested_fee_recipient: alloy_primitives::Address::ZERO,
                 withdrawals: Some(vec![]),
                 parent_beacon_block_root: Some(B256::ZERO),
+                slot_number: None,
             },
             transactions: transactions.to_vec(),
             extra_data: None,

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

@@ -69,6 +69,7 @@ pub(crate) fn prepare_payload_request(
             suggested_fee_recipient: Address::ZERO,
             withdrawals: shanghai_active.then(Vec::new),
             parent_beacon_block_root: cancun_active.then_some(B256::ZERO),
+            slot_number: None,
         },
         forkchoice_state: ForkchoiceState {
             head_block_hash: parent_hash,

bin/reth-bench/src/valid_payload.rs

@@ -181,6 +181,45 @@ pub(crate) fn payload_to_new_payload(
     target_version: Option<EngineApiMessageVersion>,
 ) -> eyre::Result<(EngineApiMessageVersion, serde_json::Value)> {
     let (version, params) = match payload {
+        ExecutionPayload::V4(payload) => {
+            let cancun = sidecar.cancun().unwrap();
+
+            if let Some(prague) = sidecar.prague() {
+                if is_optimism {
+                    (
+                        EngineApiMessageVersion::V4,
+                        serde_json::to_value((
+                            OpExecutionPayloadV4 {
+                                payload_inner: payload.payload_inner,
+                                withdrawals_root: withdrawals_root.unwrap(),
+                            },
+                            cancun.versioned_hashes.clone(),
+                            cancun.parent_beacon_block_root,
+                            Requests::default(),
+                        ))?,
+                    )
+                } else {
+                    (
+                        EngineApiMessageVersion::V4,
+                        serde_json::to_value((
+                            payload,
+                            cancun.versioned_hashes.clone(),
+                            cancun.parent_beacon_block_root,
+                            prague.requests.requests_hash(),
+                        ))?,
+                    )
+                }
+            } else {
+                (
+                    EngineApiMessageVersion::V3,
+                    serde_json::to_value((
+                        payload,
+                        cancun.versioned_hashes.clone(),
+                        cancun.parent_beacon_block_root,
+                    ))?,
+                )
+            }
+        }
         ExecutionPayload::V3(payload) => {
             let cancun = sidecar.cancun().unwrap();
 
@@ -203,7 +242,11 @@ pub(crate) fn payload_to_new_payload(
                     )
                 } else {
                     // Extract actual Requests from RequestsOrHash
-                    let requests = prague.requests.requests_hash();
+                    let requests = prague
+                        .requests
+                        .requests()
+                        .cloned()
+                        .ok_or_else(|| eyre::eyre!("Prague sidecar has hash, not requests"))?;
                     (
                         version,
                         serde_json::to_value((
@@ -285,7 +328,10 @@ pub(crate) async fn call_forkchoice_updated<N, P: EngineApiValidWaitExt<N>>(
 ) -> TransportResult<ForkchoiceUpdated> {
     // FCU V3 is used for both Cancun and Prague (there is no FCU V4)
     match message_version {
-        EngineApiMessageVersion::V3 | EngineApiMessageVersion::V4 | EngineApiMessageVersion::V5 => {
+        EngineApiMessageVersion::V3 |
+        EngineApiMessageVersion::V4 |
+        EngineApiMessageVersion::V5 |
+        EngineApiMessageVersion::V6 => {
             provider.fork_choice_updated_v3_wait(forkchoice_state, payload_attributes).await
         }
         EngineApiMessageVersion::V2 => {

crates/chain-state/src/deferred_trie.rs

@@ -192,10 +192,10 @@ impl DeferredTrieData {
                     );
                     // 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_and_sort(&sorted_hashed_state);
+                        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_and_sort(&sorted_trie_updates);
+                        Arc::make_mut(&mut overlay.nodes).extend_ref(&sorted_trie_updates);
                     }
                     overlay
                 }
@@ -242,13 +242,13 @@ impl DeferredTrieData {
 
         for ancestor in ancestors {
             let ancestor_data = ancestor.wait_cloned();
-            state_mut.extend_ref_and_sort(ancestor_data.hashed_state.as_ref());
-            nodes_mut.extend_ref_and_sort(ancestor_data.trie_updates.as_ref());
+            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_and_sort(sorted_hashed_state);
-        nodes_mut.extend_ref_and_sort(sorted_trie_updates);
+        state_mut.extend_ref(sorted_hashed_state);
+        nodes_mut.extend_ref(sorted_trie_updates);
 
         overlay
     }
@@ -287,11 +287,6 @@ impl DeferredTrieData {
                     &inputs.ancestors,
                 );
                 *state = DeferredState::Ready(computed.clone());
-
-                // Release lock before inputs (and its ancestors) drop to avoid holding it
-                // while their potential last Arc refs drop (which could trigger recursive locking)
-                drop(state);
-
                 computed
             }
         }
@@ -521,7 +516,7 @@ mod tests {
         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_and_sort(hashed_state.as_ref());
+        Arc::make_mut(&mut overlay.state).extend_ref(hashed_state.as_ref());
 
         DeferredTrieData::ready(ComputedTrieData {
             hashed_state,

crates/chain-state/src/in_memory.rs

@@ -10,15 +10,15 @@ use alloy_primitives::{map::HashMap, BlockNumber, TxHash, B256};
 use parking_lot::RwLock;
 use reth_chainspec::ChainInfo;
 use reth_ethereum_primitives::EthPrimitives;
-use reth_execution_types::{BlockExecutionOutput, BlockExecutionResult, Chain, ExecutionOutcome};
+use reth_execution_types::{Chain, ExecutionOutcome};
 use reth_metrics::{metrics::Gauge, Metrics};
 use reth_primitives_traits::{
     BlockBody as _, IndexedTx, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader,
     SignedTransaction,
 };
 use reth_storage_api::StateProviderBox;
-use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, LazyTrieData, TrieInputSorted};
-use std::{collections::BTreeMap, sync::Arc, time::Instant};
+use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, TrieInputSorted};
+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.
@@ -648,7 +648,7 @@ impl<N: NodePrimitives> BlockState<N> {
     }
 
     /// Returns the `Receipts` of executed block that determines the state.
-    pub fn receipts(&self) -> &Vec<N::Receipt> {
+    pub fn receipts(&self) -> &Vec<Vec<N::Receipt>> {
         &self.block.execution_outcome().receipts
     }
 
@@ -659,7 +659,15 @@ impl<N: NodePrimitives> BlockState<N> {
     ///
     /// This clones the vector of receipts. To avoid it, use [`Self::executed_block_receipts_ref`].
     pub fn executed_block_receipts(&self) -> Vec<N::Receipt> {
-        self.receipts().clone()
+        let receipts = self.receipts();
+
+        debug_assert!(
+            receipts.len() <= 1,
+            "Expected at most one block's worth of receipts, found {}",
+            receipts.len()
+        );
+
+        receipts.first().cloned().unwrap_or_default()
     }
 
     /// Returns a slice of `Receipt` of executed block that determines the state.
@@ -667,7 +675,15 @@ impl<N: NodePrimitives> BlockState<N> {
     /// has only one element corresponding to the executed block associated to
     /// the state.
     pub fn executed_block_receipts_ref(&self) -> &[N::Receipt] {
-        self.receipts()
+        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`.
@@ -751,7 +767,7 @@ pub struct ExecutedBlock<N: NodePrimitives = EthPrimitives> {
     /// Recovered Block
     pub recovered_block: Arc<RecoveredBlock<N::Block>>,
     /// Block's execution outcome.
-    pub execution_output: Arc<BlockExecutionOutput<N::Receipt>>,
+    pub execution_output: Arc<ExecutionOutcome<N::Receipt>>,
     /// Deferred trie data produced by execution.
     ///
     /// This allows deferring the computation of the trie data which can be expensive.
@@ -763,15 +779,7 @@ impl<N: NodePrimitives> Default for ExecutedBlock<N> {
     fn default() -> Self {
         Self {
             recovered_block: Default::default(),
-            execution_output: Arc::new(BlockExecutionOutput {
-                result: BlockExecutionResult {
-                    receipts: Default::default(),
-                    requests: Default::default(),
-                    gas_used: 0,
-                    blob_gas_used: 0,
-                },
-                state: Default::default(),
-            }),
+            execution_output: Default::default(),
             trie_data: DeferredTrieData::ready(ComputedTrieData::default()),
         }
     }
@@ -792,7 +800,7 @@ impl<N: NodePrimitives> ExecutedBlock<N> {
     /// payload builders). This is the safe default path.
     pub fn new(
         recovered_block: Arc<RecoveredBlock<N::Block>>,
-        execution_output: Arc<BlockExecutionOutput<N::Receipt>>,
+        execution_output: Arc<ExecutionOutcome<N::Receipt>>,
         trie_data: ComputedTrieData,
     ) -> Self {
         Self { recovered_block, execution_output, trie_data: DeferredTrieData::ready(trie_data) }
@@ -814,7 +822,7 @@ impl<N: NodePrimitives> ExecutedBlock<N> {
     /// Use [`Self::new()`] instead when trie data is already computed and available immediately.
     pub const fn with_deferred_trie_data(
         recovered_block: Arc<RecoveredBlock<N::Block>>,
-        execution_output: Arc<BlockExecutionOutput<N::Receipt>>,
+        execution_output: Arc<ExecutionOutcome<N::Receipt>>,
         trie_data: DeferredTrieData,
     ) -> Self {
         Self { recovered_block, execution_output, trie_data }
@@ -834,7 +842,7 @@ impl<N: NodePrimitives> ExecutedBlock<N> {
 
     /// Returns a reference to the block's execution outcome
     #[inline]
-    pub fn execution_outcome(&self) -> &BlockExecutionOutput<N::Receipt> {
+    pub fn execution_outcome(&self) -> &ExecutionOutcome<N::Receipt> {
         &self.execution_output
     }
 
@@ -934,39 +942,37 @@ impl<N: NodePrimitives<SignedTx: SignedTransaction>> NewCanonicalChain<N> {
     pub fn to_chain_notification(&self) -> CanonStateNotification<N> {
         match self {
             Self::Commit { new } => {
-                CanonStateNotification::Commit { new: Arc::new(Self::blocks_to_chain(new)) }
-            }
-            Self::Reorg { new, old } => CanonStateNotification::Reorg {
-                new: Arc::new(Self::blocks_to_chain(new)),
-                old: Arc::new(Self::blocks_to_chain(old)),
-            },
-        }
-    }
-
-    /// Converts a slice of executed blocks into a [`Chain`].
-    fn blocks_to_chain(blocks: &[ExecutedBlock<N>]) -> Chain<N> {
-        match blocks {
-            [] => Chain::default(),
-            [first, rest @ ..] => {
-                let mut chain = Chain::from_block(
-                    first.recovered_block().clone(),
-                    ExecutionOutcome::from((
-                        first.execution_outcome().clone(),
-                        first.block_number(),
-                    )),
-                    LazyTrieData::ready(first.hashed_state(), first.trie_updates()),
-                );
-                for exec in rest {
+                let new = Arc::new(new.iter().fold(Chain::default(), |mut chain, exec| {
                     chain.append_block(
                         exec.recovered_block().clone(),
-                        ExecutionOutcome::from((
-                            exec.execution_outcome().clone(),
-                            exec.block_number(),
-                        )),
-                        LazyTrieData::ready(exec.hashed_state(), exec.trie_updates()),
+                        exec.execution_outcome().clone(),
+                        exec.trie_updates(),
+                        exec.hashed_state(),
                     );
-                }
-                chain
+                    chain
+                }));
+                CanonStateNotification::Commit { new }
+            }
+            Self::Reorg { new, old } => {
+                let new = Arc::new(new.iter().fold(Chain::default(), |mut chain, exec| {
+                    chain.append_block(
+                        exec.recovered_block().clone(),
+                        exec.execution_outcome().clone(),
+                        exec.trie_updates(),
+                        exec.hashed_state(),
+                    );
+                    chain
+                }));
+                let old = Arc::new(old.iter().fold(Chain::default(), |mut chain, exec| {
+                    chain.append_block(
+                        exec.recovered_block().clone(),
+                        exec.execution_outcome().clone(),
+                        exec.trie_updates(),
+                        exec.hashed_state(),
+                    );
+                    chain
+                }));
+                CanonStateNotification::Reorg { new, old }
             }
         }
     }
@@ -1260,7 +1266,7 @@ mod tests {
 
         let state = BlockState::new(block);
 
-        assert_eq!(state.receipts(), receipts.first().unwrap());
+        assert_eq!(state.receipts(), &receipts);
     }
 
     #[test]
@@ -1537,31 +1543,33 @@ mod tests {
         let block2a =
             test_block_builder.get_executed_block_with_number(2, block1.recovered_block.hash());
 
+        let sample_execution_outcome = ExecutionOutcome {
+            receipts: vec![vec![], vec![]],
+            requests: vec![Requests::default(), Requests::default()],
+            ..Default::default()
+        };
+
         // Test commit notification
         let chain_commit = NewCanonicalChain::Commit { new: vec![block0.clone(), block1.clone()] };
 
-        // Build expected trie data map
-        let mut expected_trie_data = BTreeMap::new();
-        expected_trie_data
-            .insert(0, LazyTrieData::ready(block0.hashed_state(), block0.trie_updates()));
-        expected_trie_data
-            .insert(1, LazyTrieData::ready(block1.hashed_state(), block1.trie_updates()));
+        // 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 execution outcome (first_block matches first block number)
-        let commit_execution_outcome = ExecutionOutcome {
-            receipts: vec![vec![], vec![]],
-            requests: vec![Requests::default(), Requests::default()],
-            first_block: 0,
-            ..Default::default()
-        };
+        // 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()],
-                    commit_execution_outcome,
-                    expected_trie_data,
+                    sample_execution_outcome.clone(),
+                    expected_trie_updates,
+                    expected_hashed_state
                 ))
             }
         );
@@ -1572,39 +1580,40 @@ mod tests {
             old: vec![block1.clone(), block2.clone()],
         };
 
-        // Build expected trie data for old chain
-        let mut old_trie_data = BTreeMap::new();
-        old_trie_data.insert(1, LazyTrieData::ready(block1.hashed_state(), block1.trie_updates()));
-        old_trie_data.insert(2, LazyTrieData::ready(block2.hashed_state(), block2.trie_updates()));
+        // 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 data for new chain
-        let mut new_trie_data = BTreeMap::new();
-        new_trie_data
-            .insert(1, LazyTrieData::ready(block1a.hashed_state(), block1a.trie_updates()));
-        new_trie_data
-            .insert(2, LazyTrieData::ready(block2a.hashed_state(), block2a.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 execution outcome for reorg chains (first_block matches first block
-        // number)
-        let reorg_execution_outcome = ExecutionOutcome {
-            receipts: vec![vec![], vec![]],
-            requests: vec![Requests::default(), Requests::default()],
-            first_block: 1,
-            ..Default::default()
-        };
+        // 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()],
-                    reorg_execution_outcome.clone(),
-                    old_trie_data,
+                    sample_execution_outcome.clone(),
+                    old_trie_updates,
+                    old_hashed_state
                 )),
                 new: Arc::new(Chain::new(
                     vec![block1a.recovered_block().clone(), block2a.recovered_block().clone()],
-                    reorg_execution_outcome,
-                    new_trie_data,
+                    sample_execution_outcome,
+                    new_trie_updates,
+                    new_hashed_state
                 ))
             }
         );

crates/chain-state/src/lazy_overlay.rs

@@ -1,228 +0,0 @@
-//! Lazy overlay computation for trie input.
-//!
-//! This module provides [`LazyOverlay`], a type that computes the [`TrieInputSorted`]
-//! lazily on first access. This allows execution to start before the trie overlay
-//! is fully computed.
-
-use crate::DeferredTrieData;
-use alloy_primitives::B256;
-use reth_trie::{updates::TrieUpdatesSorted, HashedPostStateSorted, TrieInputSorted};
-use std::sync::{Arc, OnceLock};
-use tracing::{debug, trace};
-
-/// Inputs captured for lazy overlay computation.
-#[derive(Clone)]
-struct LazyOverlayInputs {
-    /// The persisted ancestor hash (anchor) this overlay should be built on.
-    anchor_hash: B256,
-    /// Deferred trie data handles for all in-memory blocks (newest to oldest).
-    blocks: Vec<DeferredTrieData>,
-}
-
-/// Lazily computed trie overlay.
-///
-/// Captures the inputs needed to compute a [`TrieInputSorted`] and defers the actual
-/// computation until first access. This is conceptually similar to [`DeferredTrieData`]
-/// but for overlay computation.
-///
-/// # Fast Path vs Slow Path
-///
-/// - **Fast path**: If the tip block's cached `anchored_trie_input` is ready and its `anchor_hash`
-///   matches our expected anchor, we can reuse it directly (O(1)).
-/// - **Slow path**: Otherwise, we merge all ancestor blocks' trie data into a new overlay.
-#[derive(Clone)]
-pub struct LazyOverlay {
-    /// Computed result, cached after first access.
-    inner: Arc<OnceLock<TrieInputSorted>>,
-    /// Inputs for lazy computation.
-    inputs: LazyOverlayInputs,
-}
-
-impl std::fmt::Debug for LazyOverlay {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("LazyOverlay")
-            .field("anchor_hash", &self.inputs.anchor_hash)
-            .field("num_blocks", &self.inputs.blocks.len())
-            .field("computed", &self.inner.get().is_some())
-            .finish()
-    }
-}
-
-impl LazyOverlay {
-    /// Create a new lazy overlay with the given anchor hash and block handles.
-    ///
-    /// # Arguments
-    ///
-    /// * `anchor_hash` - The persisted ancestor hash this overlay is built on top of
-    /// * `blocks` - Deferred trie data handles for in-memory blocks (newest to oldest)
-    pub fn new(anchor_hash: B256, blocks: Vec<DeferredTrieData>) -> Self {
-        Self { inner: Arc::new(OnceLock::new()), inputs: LazyOverlayInputs { anchor_hash, blocks } }
-    }
-
-    /// Returns the anchor hash this overlay is built on.
-    pub const fn anchor_hash(&self) -> B256 {
-        self.inputs.anchor_hash
-    }
-
-    /// Returns the number of in-memory blocks this overlay covers.
-    pub const fn num_blocks(&self) -> usize {
-        self.inputs.blocks.len()
-    }
-
-    /// Returns true if the overlay has already been computed.
-    pub fn is_computed(&self) -> bool {
-        self.inner.get().is_some()
-    }
-
-    /// Returns the computed trie input, computing it if necessary.
-    ///
-    /// The first call triggers computation (which may block waiting for deferred data).
-    /// Subsequent calls return the cached result immediately.
-    pub fn get(&self) -> &TrieInputSorted {
-        self.inner.get_or_init(|| self.compute())
-    }
-
-    /// Returns the overlay as (nodes, state) tuple for use with `OverlayStateProviderFactory`.
-    pub fn as_overlay(&self) -> (Arc<TrieUpdatesSorted>, Arc<HashedPostStateSorted>) {
-        let input = self.get();
-        (Arc::clone(&input.nodes), Arc::clone(&input.state))
-    }
-
-    /// Compute the trie input overlay.
-    fn compute(&self) -> TrieInputSorted {
-        let anchor_hash = self.inputs.anchor_hash;
-        let blocks = &self.inputs.blocks;
-
-        if blocks.is_empty() {
-            debug!(target: "chain_state::lazy_overlay", "No in-memory blocks, returning empty overlay");
-            return TrieInputSorted::default();
-        }
-
-        // Fast path: Check if tip block's overlay is ready and anchor matches.
-        // The tip block (first in list) has the cumulative overlay from all ancestors.
-        if let Some(tip) = blocks.first() {
-            let data = tip.wait_cloned();
-            if let Some(anchored) = &data.anchored_trie_input {
-                if anchored.anchor_hash == anchor_hash {
-                    trace!(target: "chain_state::lazy_overlay", %anchor_hash, "Reusing tip block's cached overlay (fast path)");
-                    return (*anchored.trie_input).clone();
-                }
-                debug!(
-                    target: "chain_state::lazy_overlay",
-                    computed_anchor = %anchored.anchor_hash,
-                    %anchor_hash,
-                    "Anchor mismatch, falling back to merge"
-                );
-            }
-        }
-
-        // Slow path: Merge all blocks' trie data into a new overlay.
-        debug!(target: "chain_state::lazy_overlay", num_blocks = blocks.len(), "Merging blocks (slow path)");
-        Self::merge_blocks(blocks)
-    }
-
-    /// Merge all blocks' trie data into a single [`TrieInputSorted`].
-    ///
-    /// Blocks are ordered newest to oldest. Uses hybrid merge algorithm that
-    /// switches between `extend_ref` (small batches) and k-way merge (large batches).
-    fn merge_blocks(blocks: &[DeferredTrieData]) -> TrieInputSorted {
-        const MERGE_BATCH_THRESHOLD: usize = 64;
-
-        if blocks.is_empty() {
-            return TrieInputSorted::default();
-        }
-
-        // Single block: use its data directly (no allocation)
-        if blocks.len() == 1 {
-            let data = blocks[0].wait_cloned();
-            return TrieInputSorted {
-                state: data.hashed_state,
-                nodes: data.trie_updates,
-                prefix_sets: Default::default(),
-            };
-        }
-
-        if blocks.len() < MERGE_BATCH_THRESHOLD {
-            // Small k: extend_ref loop with Arc::make_mut is faster.
-            // Uses copy-on-write - only clones inner data if Arc has multiple refs.
-            // Iterate oldest->newest so newer values override older ones.
-            let mut blocks_iter = blocks.iter().rev();
-            let first = blocks_iter.next().expect("blocks is non-empty");
-            let data = first.wait_cloned();
-
-            let mut state = data.hashed_state;
-            let mut nodes = data.trie_updates;
-
-            for block in blocks_iter {
-                let block_data = block.wait_cloned();
-                Arc::make_mut(&mut state).extend_ref_and_sort(block_data.hashed_state.as_ref());
-                Arc::make_mut(&mut nodes).extend_ref_and_sort(block_data.trie_updates.as_ref());
-            }
-
-            TrieInputSorted { state, nodes, prefix_sets: Default::default() }
-        } else {
-            // Large k: k-way merge is faster (O(n log k)).
-            // Collect is unavoidable here - we need all data materialized for k-way merge.
-            let trie_data: Vec<_> = blocks.iter().map(|b| b.wait_cloned()).collect();
-
-            let merged_state = HashedPostStateSorted::merge_batch(
-                trie_data.iter().map(|d| d.hashed_state.as_ref()),
-            );
-            let merged_nodes =
-                TrieUpdatesSorted::merge_batch(trie_data.iter().map(|d| d.trie_updates.as_ref()));
-
-            TrieInputSorted {
-                state: Arc::new(merged_state),
-                nodes: Arc::new(merged_nodes),
-                prefix_sets: Default::default(),
-            }
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use reth_trie::{updates::TrieUpdates, HashedPostState};
-
-    fn empty_deferred(anchor: B256) -> DeferredTrieData {
-        DeferredTrieData::pending(
-            Arc::new(HashedPostState::default()),
-            Arc::new(TrieUpdates::default()),
-            anchor,
-            Vec::new(),
-        )
-    }
-
-    #[test]
-    fn empty_blocks_returns_default() {
-        let overlay = LazyOverlay::new(B256::ZERO, vec![]);
-        let result = overlay.get();
-        assert!(result.state.is_empty());
-        assert!(result.nodes.is_empty());
-    }
-
-    #[test]
-    fn single_block_uses_data_directly() {
-        let anchor = B256::random();
-        let deferred = empty_deferred(anchor);
-        let overlay = LazyOverlay::new(anchor, vec![deferred]);
-
-        assert!(!overlay.is_computed());
-        let _ = overlay.get();
-        assert!(overlay.is_computed());
-    }
-
-    #[test]
-    fn cached_after_first_access() {
-        let overlay = LazyOverlay::new(B256::ZERO, vec![]);
-
-        // First access computes
-        let _ = overlay.get();
-        assert!(overlay.is_computed());
-
-        // Second access uses cache
-        let _ = overlay.get();
-        assert!(overlay.is_computed());
-    }
-}

crates/chain-state/src/lib.rs

@@ -14,9 +14,6 @@ pub use in_memory::*;
 mod deferred_trie;
 pub use deferred_trie::*;
 
-mod lazy_overlay;
-pub use lazy_overlay::*;
-
 mod noop;
 
 mod chain_info;

crates/chain-state/src/notifications.rs

@@ -280,6 +280,7 @@ mod tests {
             vec![block1.clone(), block2.clone()],
             ExecutionOutcome::default(),
             BTreeMap::new(),
+            BTreeMap::new(),
         ));
 
         // Create a commit notification
@@ -318,11 +319,13 @@ mod tests {
             vec![block1.clone()],
             ExecutionOutcome::default(),
             BTreeMap::new(),
+            BTreeMap::new(),
         ));
         let new_chain = Arc::new(Chain::new(
             vec![block2.clone(), block3.clone()],
             ExecutionOutcome::default(),
             BTreeMap::new(),
+            BTreeMap::new(),
         ));
 
         // Create a reorg notification
@@ -388,6 +391,7 @@ mod tests {
             vec![block1.clone(), block2.clone()],
             execution_outcome,
             BTreeMap::new(),
+            BTreeMap::new(),
         ));
 
         // Create a commit notification containing the new chain segment.
@@ -445,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, BTreeMap::new()));
+        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();
@@ -474,8 +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, BTreeMap::new()));
+        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

@@ -3,7 +3,10 @@ use crate::{
     CanonStateSubscriptions, ComputedTrieData,
 };
 use alloy_consensus::{Header, SignableTransaction, TxEip1559, TxReceipt, EMPTY_ROOT_HASH};
-use alloy_eips::eip1559::{ETHEREUM_BLOCK_GAS_LIMIT_30M, INITIAL_BASE_FEE};
+use alloy_eips::{
+    eip1559::{ETHEREUM_BLOCK_GAS_LIMIT_30M, INITIAL_BASE_FEE},
+    eip7685::Requests,
+};
 use alloy_primitives::{Address, BlockNumber, B256, U256};
 use alloy_signer::SignerSync;
 use alloy_signer_local::PrivateKeySigner;
@@ -13,7 +16,7 @@ use reth_chainspec::{ChainSpec, EthereumHardfork, MIN_TRANSACTION_GAS};
 use reth_ethereum_primitives::{
     Block, BlockBody, EthPrimitives, Receipt, Transaction, TransactionSigned,
 };
-use reth_execution_types::{BlockExecutionOutput, BlockExecutionResult, Chain, ExecutionOutcome};
+use reth_execution_types::{Chain, ExecutionOutcome};
 use reth_primitives_traits::{
     proofs::{calculate_receipt_root, calculate_transaction_root, calculate_withdrawals_root},
     Account, NodePrimitives, Recovered, RecoveredBlock, SealedBlock, SealedHeader,
@@ -170,6 +173,7 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
                 transactions: transactions.into_iter().map(|tx| tx.into_inner()).collect(),
                 ommers: Vec::new(),
                 withdrawals: Some(vec![].into()),
+                block_access_list: None,
             },
         )
     }
@@ -198,7 +202,7 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
     fn get_executed_block(
         &mut self,
         block_number: BlockNumber,
-        mut receipts: Vec<Vec<Receipt>>,
+        receipts: Vec<Vec<Receipt>>,
         parent_hash: B256,
     ) -> ExecutedBlock {
         let block = self.generate_random_block(block_number, parent_hash);
@@ -206,15 +210,12 @@ impl<N: NodePrimitives> TestBlockBuilder<N> {
         let trie_data = ComputedTrieData::default();
         ExecutedBlock::new(
             Arc::new(RecoveredBlock::new_sealed(block, senders)),
-            Arc::new(BlockExecutionOutput {
-                result: BlockExecutionResult {
-                    receipts: receipts.pop().unwrap_or_default(),
-                    requests: Default::default(),
-                    gas_used: 0,
-                    blob_gas_used: 0,
-                },
-                state: BundleState::default(),
-            }),
+            Arc::new(ExecutionOutcome::new(
+                BundleState::default(),
+                receipts,
+                block_number,
+                vec![Requests::default()],
+            )),
             trie_data,
         )
     }

crates/chainspec/src/spec.rs

@@ -45,6 +45,10 @@ use reth_network_peers::{
 };
 use reth_primitives_traits::{sync::LazyLock, BlockHeader, SealedHeader};
 
+/// The hash of an empty block access list.
+const EMPTY_BLOCK_ACCESS_LIST_HASH: B256 =
+    b256!("0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347");
+
 /// Helper method building a [`Header`] given [`Genesis`] and [`ChainHardforks`].
 pub fn make_genesis_header(genesis: &Genesis, hardforks: &ChainHardforks) -> Header {
     // If London is activated at genesis, we set the initial base fee as per EIP-1559.
@@ -79,6 +83,12 @@ pub fn make_genesis_header(genesis: &Genesis, hardforks: &ChainHardforks) -> Hea
         .active_at_timestamp(genesis.timestamp)
         .then_some(EMPTY_REQUESTS_HASH);
 
+    // If Amsterdam is activated at genesis we set block access list hash empty hash.
+    let block_access_list_hash = hardforks
+        .fork(EthereumHardfork::Amsterdam)
+        .active_at_timestamp(genesis.timestamp)
+        .then_some(EMPTY_BLOCK_ACCESS_LIST_HASH);
+
     Header {
         number: genesis.number.unwrap_or_default(),
         parent_hash: genesis.parent_hash.unwrap_or_default(),
@@ -96,6 +106,7 @@ pub fn make_genesis_header(genesis: &Genesis, hardforks: &ChainHardforks) -> Hea
         blob_gas_used,
         excess_blob_gas,
         requests_hash,
+        block_access_list_hash,
         ..Default::default()
     }
 }
@@ -300,6 +311,7 @@ pub fn create_chain_config(
         cancun_time: timestamp(EthereumHardfork::Cancun),
         prague_time: timestamp(EthereumHardfork::Prague),
         osaka_time: timestamp(EthereumHardfork::Osaka),
+        amsterdam_time: timestamp(EthereumHardfork::Amsterdam),
         bpo1_time: timestamp(EthereumHardfork::Bpo1),
         bpo2_time: timestamp(EthereumHardfork::Bpo2),
         bpo3_time: timestamp(EthereumHardfork::Bpo3),
@@ -901,6 +913,7 @@ impl From<Genesis> for ChainSpec {
             (EthereumHardfork::Bpo3.boxed(), genesis.config.bpo3_time),
             (EthereumHardfork::Bpo4.boxed(), genesis.config.bpo4_time),
             (EthereumHardfork::Bpo5.boxed(), genesis.config.bpo5_time),
+            (EthereumHardfork::Amsterdam.boxed(), genesis.config.amsterdam_time),
         ];
 
         let mut time_hardforks = time_hardfork_opts
@@ -1207,6 +1220,19 @@ impl ChainSpecBuilder {
         self
     }
 
+    /// Enable Amsterdam at genesis.
+    pub fn amsterdam_activated(mut self) -> Self {
+        self = self.osaka_activated();
+        self.hardforks.insert(EthereumHardfork::Amsterdam, ForkCondition::Timestamp(0));
+        self
+    }
+
+    /// Enable Amsterdam at the given timestamp.
+    pub fn with_amsterdam_at(mut self, timestamp: u64) -> Self {
+        self.hardforks.insert(EthereumHardfork::Amsterdam, ForkCondition::Timestamp(timestamp));
+        self
+    }
+
     /// Build the resulting [`ChainSpec`].
     ///
     /// # Panics

crates/cli/commands/src/download.rs

@@ -2,15 +2,14 @@ use crate::common::EnvironmentArgs;
 use clap::Parser;
 use eyre::Result;
 use lz4::Decoder;
-use reqwest::{blocking::Client as BlockingClient, header::RANGE, Client, StatusCode};
+use reqwest::Client;
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
 use reth_cli::chainspec::ChainSpecParser;
 use reth_fs_util as fs;
 use std::{
     borrow::Cow,
-    fs::OpenOptions,
-    io::{self, BufWriter, Read, Write},
-    path::{Path, PathBuf},
+    io::{self, Read, Write},
+    path::Path,
     sync::{Arc, OnceLock},
     time::{Duration, Instant},
 };
@@ -328,158 +327,18 @@ fn extract_from_file(path: &Path, format: CompressionFormat, target_dir: &Path)
     extract_archive(file, total_size, format, target_dir)
 }
 
-const MAX_DOWNLOAD_RETRIES: u32 = 10;
-const RETRY_BACKOFF_SECS: u64 = 5;
-
-/// Wrapper that tracks download progress while writing data.
-/// Used with [`io::copy`] to display progress during downloads.
-struct ProgressWriter<W> {
-    inner: W,
-    progress: DownloadProgress,
-}
-
-impl<W: Write> Write for ProgressWriter<W> {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let n = self.inner.write(buf)?;
-        let _ = self.progress.update(n as u64);
-        Ok(n)
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        self.inner.flush()
-    }
-}
-
-/// Downloads a file with resume support using HTTP Range requests.
-/// Automatically retries on failure, resuming from where it left off.
-/// Returns the path to the downloaded file and its total size.
-fn resumable_download(url: &str, target_dir: &Path) -> Result<(PathBuf, u64)> {
-    let file_name = Url::parse(url)
-        .ok()
-        .and_then(|u| u.path_segments()?.next_back().map(|s| s.to_string()))
-        .unwrap_or_else(|| "snapshot.tar".to_string());
-
-    let final_path = target_dir.join(&file_name);
-    let part_path = target_dir.join(format!("{file_name}.part"));
-
-    let client = BlockingClient::builder().timeout(Duration::from_secs(30)).build()?;
-
-    let mut total_size: Option<u64> = None;
-    let mut last_error: Option<eyre::Error> = None;
-
-    for attempt in 1..=MAX_DOWNLOAD_RETRIES {
-        let existing_size = fs::metadata(&part_path).map(|m| m.len()).unwrap_or(0);
-
-        if let Some(total) = total_size &&
-            existing_size >= total
-        {
-            fs::rename(&part_path, &final_path)?;
-            info!(target: "reth::cli", "Download complete: {}", final_path.display());
-            return Ok((final_path, total));
-        }
-
-        if attempt > 1 {
-            info!(target: "reth::cli",
-                "Retry attempt {}/{} - resuming from {} bytes",
-                attempt, MAX_DOWNLOAD_RETRIES, existing_size
-            );
-        }
-
-        let mut request = client.get(url);
-        if existing_size > 0 {
-            request = request.header(RANGE, format!("bytes={existing_size}-"));
-            if attempt == 1 {
-                info!(target: "reth::cli", "Resuming download from {} bytes", existing_size);
-            }
-        }
-
-        let response = match request.send().and_then(|r| r.error_for_status()) {
-            Ok(r) => r,
-            Err(e) => {
-                last_error = Some(e.into());
-                if attempt < MAX_DOWNLOAD_RETRIES {
-                    info!(target: "reth::cli",
-                        "Download failed, retrying in {} seconds...", RETRY_BACKOFF_SECS
-                    );
-                    std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS));
-                }
-                continue;
-            }
-        };
-
-        let is_partial = response.status() == StatusCode::PARTIAL_CONTENT;
-
-        let size = if is_partial {
-            response
-                .headers()
-                .get("Content-Range")
-                .and_then(|v| v.to_str().ok())
-                .and_then(|v| v.split('/').next_back())
-                .and_then(|v| v.parse().ok())
-        } else {
-            response.content_length()
-        };
-
-        if total_size.is_none() {
-            total_size = size;
-        }
-
-        let current_total = total_size.ok_or_else(|| {
-            eyre::eyre!("Server did not provide Content-Length or Content-Range header")
-        })?;
-
-        let file = if is_partial && existing_size > 0 {
-            OpenOptions::new()
-                .append(true)
-                .open(&part_path)
-                .map_err(|e| fs::FsPathError::open(e, &part_path))?
-        } else {
-            fs::create_file(&part_path)?
-        };
-
-        let start_offset = if is_partial { existing_size } else { 0 };
-        let mut progress = DownloadProgress::new(current_total);
-        progress.downloaded = start_offset;
-
-        let mut writer = ProgressWriter { inner: BufWriter::new(file), progress };
-        let mut reader = response;
-
-        let copy_result = io::copy(&mut reader, &mut writer);
-        let flush_result = writer.inner.flush();
-        println!();
-
-        if let Err(e) = copy_result.and(flush_result) {
-            last_error = Some(e.into());
-            if attempt < MAX_DOWNLOAD_RETRIES {
-                info!(target: "reth::cli",
-                    "Download interrupted, retrying in {} seconds...", RETRY_BACKOFF_SECS
-                );
-                std::thread::sleep(Duration::from_secs(RETRY_BACKOFF_SECS));
-            }
-            continue;
-        }
-
-        fs::rename(&part_path, &final_path)?;
-        info!(target: "reth::cli", "Download complete: {}", final_path.display());
-        return Ok((final_path, current_total));
-    }
-
-    Err(last_error
-        .unwrap_or_else(|| eyre::eyre!("Download failed after {} attempts", MAX_DOWNLOAD_RETRIES)))
-}
-
-/// Fetches the snapshot from a remote URL with resume support, then extracts it.
+/// Fetches the snapshot from a remote URL, uncompressing it in a streaming fashion.
 fn download_and_extract(url: &str, format: CompressionFormat, target_dir: &Path) -> Result<()> {
-    let (downloaded_path, total_size) = resumable_download(url, target_dir)?;
+    let client = reqwest::blocking::Client::builder().build()?;
+    let response = client.get(url).send()?.error_for_status()?;
 
-    info!(target: "reth::cli", "Extracting snapshot...");
-    let file = fs::open(&downloaded_path)?;
-    extract_archive(file, total_size, format, target_dir)?;
+    let total_size = response.content_length().ok_or_else(|| {
+        eyre::eyre!(
+            "Server did not provide Content-Length header. This is required for snapshot downloads"
+        )
+    })?;
 
-    fs::remove_file(&downloaded_path)?;
-    info!(target: "reth::cli", "Removed downloaded archive");
-
-    Ok(())
+    extract_archive(response, total_size, format, target_dir)
 }
 
 /// Downloads and extracts a snapshot, blocking until finished.

crates/cli/commands/src/re_execute.rs

@@ -42,9 +42,9 @@ pub struct Command<C: ChainSpecParser> {
     #[arg(long)]
     to: Option<u64>,
 
-    /// Number of tasks to run in parallel. Defaults to the number of available CPUs.
-    #[arg(long)]
-    num_tasks: Option<u64>,
+    /// Number of tasks to run in parallel
+    #[arg(long, default_value = "10")]
+    num_tasks: u64,
 
     /// Continues with execution when an invalid block is encountered and collects these blocks.
     #[arg(long)]
@@ -84,16 +84,12 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
             }
         };
 
-        let num_tasks = self.num_tasks.unwrap_or_else(|| {
-            std::thread::available_parallelism().map(|n| n.get() as u64).unwrap_or(10)
-        });
-
         let total_blocks = max_block - min_block;
         let total_gas = calculate_gas_used_from_headers(
             &provider_factory.static_file_provider(),
             min_block..=max_block,
         )?;
-        let blocks_per_task = total_blocks / num_tasks;
+        let blocks_per_task = total_blocks / self.num_tasks;
 
         let db_at = {
             let provider_factory = provider_factory.clone();
@@ -111,10 +107,10 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
         let _guard = cancellation.drop_guard();
 
         let mut tasks = JoinSet::new();
-        for i in 0..num_tasks {
+        for i in 0..self.num_tasks {
             let start_block = min_block + i * blocks_per_task;
             let end_block =
-                if i == num_tasks - 1 { max_block } else { start_block + blocks_per_task };
+                if i == self.num_tasks - 1 { max_block } else { start_block + blocks_per_task };
 
             // Spawn thread executing blocks
             let provider_factory = provider_factory.clone();
@@ -152,7 +148,7 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                     };
 
                     if let Err(err) = consensus
-                        .validate_block_post_execution(&block, &result, None)
+                        .validate_block_post_execution(&block, &result)
                         .wrap_err_with(|| {
                             format!("Failed to validate block {} {}", block.number(), block.hash())
                         })

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

@@ -15,7 +15,7 @@ use reth_db_common::{
 use reth_node_api::{HeaderTy, ReceiptTy, TxTy};
 use reth_node_core::args::StageEnum;
 use reth_provider::{
-    DBProvider, DatabaseProviderFactory, StaticFileProviderFactory, StaticFileWriter,
+    DBProvider, DatabaseProviderFactory, StaticFileProviderFactory, StaticFileWriter, TrieWriter,
 };
 use reth_prune::PruneSegment;
 use reth_stages::StageId;
@@ -113,6 +113,7 @@ impl<C: ChainSpecParser> Command<C> {
                 tx.clear::<tables::TransactionBlocks>()?;
                 tx.clear::<tables::BlockOmmers<HeaderTy<N>>>()?;
                 tx.clear::<tables::BlockWithdrawals>()?;
+                tx.clear::<tables::BlockAccessLists>()?;
                 reset_stage_checkpoint(tx, StageId::Bodies)?;
 
                 insert_genesis_header(&provider_rw, &self.env.chain)?;
@@ -167,6 +168,10 @@ impl<C: ChainSpecParser> Command<C> {
                     None,
                 )?;
             }
+            StageEnum::MerkleChangeSets => {
+                provider_rw.clear_trie_changesets()?;
+                reset_stage_checkpoint(tx, StageId::MerkleChangeSets)?;
+            }
             StageEnum::AccountHistory | StageEnum::StorageHistory => {
                 tx.clear::<tables::AccountsHistory>()?;
                 tx.clear::<tables::StoragesHistory>()?;

crates/config/src/config.rs

@@ -550,6 +550,7 @@ impl PruneConfig {
     /// - `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 {
@@ -562,6 +563,7 @@ impl PruneConfig {
                     account_history,
                     storage_history,
                     bodies_history,
+                    merkle_changesets,
                     receipts_log_filter,
                 },
         } = other;
@@ -578,6 +580,8 @@ 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; 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() {
             self.segments.receipts_log_filter = receipts_log_filter;
@@ -1087,6 +1091,7 @@ receipts = { distance = 16384 }
                 account_history: None,
                 storage_history: Some(PruneMode::Before(5000)),
                 bodies_history: None,
+                merkle_changesets: PruneMode::Before(0),
                 receipts_log_filter: ReceiptsLogPruneConfig(BTreeMap::from([(
                     Address::random(),
                     PruneMode::Full,
@@ -1103,6 +1108,7 @@ receipts = { distance = 16384 }
                 account_history: Some(PruneMode::Distance(2000)),
                 storage_history: Some(PruneMode::Distance(3000)),
                 bodies_history: None,
+                merkle_changesets: PruneMode::Distance(10000),
                 receipts_log_filter: ReceiptsLogPruneConfig(BTreeMap::from([
                     (Address::random(), PruneMode::Distance(1000)),
                     (Address::random(), PruneMode::Before(2000)),
@@ -1121,6 +1127,7 @@ receipts = { distance = 16384 }
         assert_eq!(config1.segments.receipts, Some(PruneMode::Distance(1000)));
         assert_eq!(config1.segments.account_history, Some(PruneMode::Distance(2000)));
         assert_eq!(config1.segments.storage_history, Some(PruneMode::Before(5000)));
+        assert_eq!(config1.segments.merkle_changesets, PruneMode::Distance(10000));
         assert_eq!(config1.segments.receipts_log_filter, original_filter);
     }
 

crates/consensus/common/Cargo.toml

@@ -14,11 +14,14 @@ workspace = true
 # reth
 reth-chainspec.workspace = true
 reth-consensus.workspace = true
+tracing.workspace = true
 
 # ethereum
 reth-primitives-traits.workspace = true
 alloy-consensus.workspace = true
+alloy-primitives.workspace = true
 alloy-eips.workspace = true
+alloy-rlp.workspace = true
 
 [dev-dependencies]
 alloy-primitives = { workspace = true, features = ["rand"] }
@@ -35,4 +38,6 @@ std = [
     "reth-primitives-traits/std",
     "reth-ethereum-primitives/std",
     "alloy-primitives/std",
+    "alloy-rlp/std",
+    "tracing/std",
 ]

crates/consensus/common/src/validation.rs

@@ -69,6 +69,29 @@ pub fn validate_shanghai_withdrawals<B: Block>(
     Ok(())
 }
 
+/// Validate that block access lists are present in Amsterdam
+///
+/// [EIP-7928]: https://eips.ethereum.org/EIPS/eip-7928
+#[inline]
+pub fn validate_amsterdam_block_access_lists<B: Block>(
+    block: &SealedBlock<B>,
+) -> Result<(), ConsensusError> {
+    let bal = block.body().block_access_list().ok_or(ConsensusError::BlockAccessListMissing)?;
+    let bal_hash = alloy_primitives::keccak256(alloy_rlp::encode(bal));
+    let header_bal_hash =
+        block.block_access_list_hash().ok_or(ConsensusError::BlockAccessListHashMissing)?;
+    if bal_hash != header_bal_hash {
+        tracing::error!(
+            target: "consensus",
+            ?header_bal_hash,
+            ?bal,
+            "Block access list hash mismatch in validation.rs in L81"
+        );
+        return Err(ConsensusError::InvalidBalHash);
+    }
+    Ok(())
+}
+
 /// Validate that blob gas is present in the block if Cancun is active.
 ///
 /// See [EIP-4844]: Shard Blob Transactions
@@ -131,6 +154,21 @@ where
         }
         _ => return Err(ConsensusError::WithdrawalsRootUnexpected),
     }
+    if let (Some(expected_hash), Some(body_bal)) =
+        (header.block_access_list_hash(), body.block_access_list())
+    {
+        let got_hash = alloy_primitives::keccak256(alloy_rlp::encode(body_bal));
+
+        if got_hash != expected_hash {
+            tracing::error!(
+                target: "consensus",
+                ?expected_hash,
+                ?body_bal,
+                "Block access list hash mismatch in validation.rs in L164"
+            );
+            return Err(ConsensusError::InvalidBalHash);
+        }
+    }
 
     Ok(())
 }
@@ -217,6 +255,10 @@ where
         })
     }
 
+    if chain_spec.is_amsterdam_active_at_timestamp(block.header().timestamp()) {
+        validate_amsterdam_block_access_lists(block)?;
+    }
+
     Ok(())
 }
 
@@ -492,6 +534,7 @@ mod tests {
             transactions: vec![transaction],
             ommers: vec![],
             withdrawals: Some(Withdrawals::default()),
+            block_access_list: None,
         };
 
         let block = SealedBlock::seal_slow(alloy_consensus::Block { header, body });

crates/consensus/consensus/src/lib.rs

@@ -15,12 +15,6 @@ 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;
-
-/// Pre-computed receipt root and logs bloom.
-///
-/// When provided to [`FullConsensus::validate_block_post_execution`], this allows skipping
-/// the receipt root computation and using the pre-computed values instead.
-pub type ReceiptRootBloom = (B256, Bloom);
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{
     constants::{GAS_LIMIT_BOUND_DIVISOR, MAXIMUM_GAS_LIMIT_BLOCK, MINIMUM_GAS_LIMIT},
@@ -45,15 +39,11 @@ pub trait FullConsensus<N: NodePrimitives>: Consensus<N::Block> {
     ///
     /// See the Yellow Paper sections 4.3.2 "Holistic Validity".
     ///
-    /// If `receipt_root_bloom` is provided, the implementation should use the pre-computed
-    /// receipt root and logs bloom instead of computing them from the receipts.
-    ///
     /// Note: validating blocks does not include other validations of the Consensus
     fn validate_block_post_execution(
         &self,
         block: &RecoveredBlock<N::Block>,
         result: &BlockExecutionResult<N::Receipt>,
-        receipt_root_bloom: Option<ReceiptRootBloom>,
     ) -> Result<(), ConsensusError>;
 }
 
@@ -412,9 +402,41 @@ pub enum ConsensusError {
         /// The maximum allowed RLP length.
         max_rlp_length: usize,
     },
+
+    /// Error when the hash of block access list is different from the expected hash.
+    #[error("Block header's BAL hash does not match the computed BAL hash.")]
+    InvalidBalHash,
+
+    /// Error when the block access list hash is missing.
+    #[error("block access list hash missing")]
+    BlockAccessListHashMissing,
+
+    /// Error when the block access list is different from the expected access list.
+    #[error("Block's access list is invalid.")]
+    InvalidBlockAccessList,
+
+    /// Error when the block access list is missing.
+    #[error("block access list missing")]
+    BlockAccessListMissing,
+
+    /// Error when the block access list hash is unexpected.
+    #[error("block access list hash unexpected")]
+    BlockAccessListHashUnexpected,
+
+    /// Error when the block access list contains an account change that is not present in the
+    /// computed access list.
+    #[error("Block BAL contains an account change that is not present in the computed BAL.")]
+    InvalidBalExtraAccount,
+
+    /// Error when the block access list is missing an account change that is present in the
+    /// computed access list.
+    #[error("Block BAL is missing an account change that is present in the computed BAL.")]
+    InvalidBalMissingAccount,
+
     /// EIP-7825: Transaction gas limit exceeds maximum allowed
     #[error(transparent)]
     TransactionGasLimitTooHigh(Box<TxGasLimitTooHighErr>),
+
     /// Other, likely an injected L2 error.
     #[error("{0}")]
     Other(String),

crates/consensus/consensus/src/noop.rs

@@ -18,7 +18,7 @@
 //!
 //! **Not for production use** - provides no security guarantees or consensus validation.
 
-use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
+use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
 use alloc::sync::Arc;
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{Block, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
@@ -76,7 +76,6 @@ impl<N: NodePrimitives> FullConsensus<N> for NoopConsensus {
         &self,
         _block: &RecoveredBlock<N::Block>,
         _result: &BlockExecutionResult<N::Receipt>,
-        _receipt_root_bloom: Option<ReceiptRootBloom>,
     ) -> Result<(), ConsensusError> {
         Ok(())
     }

crates/consensus/consensus/src/test_utils.rs

@@ -1,4 +1,4 @@
-use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
+use crate::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
 use core::sync::atomic::{AtomicBool, Ordering};
 use reth_execution_types::BlockExecutionResult;
 use reth_primitives_traits::{Block, NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
@@ -51,7 +51,6 @@ impl<N: NodePrimitives> FullConsensus<N> for TestConsensus {
         &self,
         _block: &RecoveredBlock<N::Block>,
         _result: &BlockExecutionResult<N::Receipt>,
-        _receipt_root_bloom: Option<ReceiptRootBloom>,
     ) -> Result<(), ConsensusError> {
         if self.fail_validation() {
             Err(ConsensusError::BaseFeeMissing)

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

@@ -56,6 +56,7 @@ pub fn generate_test_blocks(chain_spec: &ChainSpec, count: u64) -> Vec<SealedBlo
             excess_blob_gas: None,
             parent_beacon_block_root: None,
             requests_hash: None,
+            block_access_list_hash: None,
         };
 
         // Set required fields based on chain spec
@@ -106,6 +107,7 @@ pub fn generate_test_blocks(chain_spec: &ChainSpec, count: u64) -> Vec<SealedBlo
             transactions: vec![],
             ommers: vec![],
             withdrawals: header.withdrawals_root.is_some().then(Withdrawals::default),
+            block_access_list: None,
         };
 
         // Create the block

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

@@ -227,6 +227,7 @@ where
                 suggested_fee_recipient: alloy_primitives::Address::random(),
                 withdrawals: Some(vec![]),
                 parent_beacon_block_root: Some(B256::ZERO),
+                slot_number: None,
             };
 
             env.active_node_state_mut()?
@@ -299,6 +300,7 @@ where
                     suggested_fee_recipient: alloy_primitives::Address::random(),
                     withdrawals: Some(vec![]),
                     parent_beacon_block_root: Some(B256::ZERO),
+                    slot_number: None,
                 };
 
                 let fresh_fcu_result = EngineApiClient::<Engine>::fork_choice_updated_v3(

crates/e2e-test-utils/src/testsuite/setup.rs

@@ -254,6 +254,7 @@ where
                 suggested_fee_recipient: alloy_primitives::Address::ZERO,
                 withdrawals: Some(vec![]),
                 parent_beacon_block_root: Some(B256::ZERO),
+                slot_number: None,
             };
             EthPayloadBuilderAttributes::new(B256::ZERO, attributes)
         };
@@ -284,6 +285,7 @@ where
                 suggested_fee_recipient: alloy_primitives::Address::ZERO,
                 withdrawals: Some(vec![]),
                 parent_beacon_block_root: Some(B256::ZERO),
+                slot_number: None,
             };
             <<N as NodeTypes>::Payload as PayloadTypes>::PayloadBuilderAttributes::from(
                 EthPayloadBuilderAttributes::new(B256::ZERO, attributes),

crates/e2e-test-utils/tests/e2e-testsuite/main.rs

@@ -161,6 +161,7 @@ async fn test_testsuite_assert_mine_block() -> Result<()> {
                 suggested_fee_recipient: Address::random(),
                 withdrawals: None,
                 parent_beacon_block_root: None,
+                slot_number: None,
             },
         ));
 

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

@@ -839,6 +839,7 @@ mod tests {
                 receipts: vec![],
                 requests: Requests::default(),
                 gas_used: 0,
+                block_access_list: Default::default(),
                 blob_gas_used: 0,
             },
         };

crates/engine/local/src/miner.rs

@@ -204,7 +204,7 @@ where
                 EngineApiMessageVersion::default(),
             )
             .await?;
-
+        tracing::debug!(target: "engine::local", "FCU result: {res:?}");
         if !res.is_valid() {
             eyre::bail!("Invalid payload status")
         }

crates/engine/local/src/payload.rs

@@ -57,6 +57,7 @@ where
                 .chain_spec
                 .is_cancun_active_at_timestamp(timestamp)
                 .then(B256::random),
+            slot_number: None,
         }
     }
 }

crates/engine/primitives/src/lib.rs

@@ -62,7 +62,8 @@ pub trait EngineTypes:
                           + TryInto<Self::ExecutionPayloadEnvelopeV2>
                           + TryInto<Self::ExecutionPayloadEnvelopeV3>
                           + TryInto<Self::ExecutionPayloadEnvelopeV4>
-                          + TryInto<Self::ExecutionPayloadEnvelopeV5>,
+                          + TryInto<Self::ExecutionPayloadEnvelopeV5>
+                          + TryInto<Self::ExecutionPayloadEnvelopeV6>,
     > + DeserializeOwned
     + Serialize
 {
@@ -106,6 +107,14 @@ pub trait EngineTypes:
         + Send
         + Sync
         + 'static;
+    /// Execution Payload V6 envelope type.
+    type ExecutionPayloadEnvelopeV6: DeserializeOwned
+        + Serialize
+        + Clone
+        + Unpin
+        + Send
+        + Sync
+        + 'static;
 }
 
 /// Type that validates the payloads processed by the engine API.

crates/engine/service/Cargo.toml

@@ -25,7 +25,6 @@ reth-tasks.workspace = true
 reth-node-types.workspace = true
 reth-chainspec.workspace = true
 reth-engine-primitives.workspace = true
-reth-trie-db.workspace = true
 
 # async
 futures.workspace = true
@@ -41,8 +40,6 @@ reth-evm-ethereum.workspace = true
 reth-exex-types.workspace = true
 reth-primitives-traits.workspace = true
 reth-node-ethereum.workspace = true
-reth-trie-db.workspace = true
 
-alloy-eips.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tokio-stream.workspace = true

crates/engine/service/src/service.rs

@@ -26,7 +26,6 @@ use reth_provider::{
 use reth_prune::PrunerWithFactory;
 use reth_stages_api::{MetricEventsSender, Pipeline};
 use reth_tasks::TaskSpawner;
-use reth_trie_db::ChangesetCache;
 use std::{
     pin::Pin,
     sync::Arc,
@@ -85,7 +84,6 @@ where
         tree_config: TreeConfig,
         sync_metrics_tx: MetricEventsSender,
         evm_config: C,
-        changeset_cache: ChangesetCache,
     ) -> Self
     where
         V: EngineValidator<N::Payload>,
@@ -111,7 +109,6 @@ where
             tree_config,
             engine_kind,
             evm_config,
-            changeset_cache,
         );
 
         let engine_handler = EngineApiRequestHandler::new(to_tree_tx, from_tree);
@@ -159,7 +156,6 @@ mod tests {
     };
     use reth_prune::Pruner;
     use reth_tasks::TokioTaskExecutor;
-    use reth_trie_db::ChangesetCache;
     use std::sync::Arc;
     use tokio::sync::{mpsc::unbounded_channel, watch};
     use tokio_stream::wrappers::UnboundedReceiverStream;
@@ -192,8 +188,6 @@ mod tests {
         let pruner = Pruner::new_with_factory(provider_factory.clone(), vec![], 0, 0, None, rx);
         let evm_config = EthEvmConfig::new(chain_spec.clone());
 
-        let changeset_cache = ChangesetCache::new();
-
         let engine_validator = BasicEngineValidator::new(
             blockchain_db.clone(),
             consensus.clone(),
@@ -201,7 +195,6 @@ mod tests {
             engine_payload_validator,
             TreeConfig::default(),
             Box::new(NoopInvalidBlockHook::default()),
-            changeset_cache.clone(),
         );
 
         let (sync_metrics_tx, _sync_metrics_rx) = unbounded_channel();
@@ -221,7 +214,6 @@ mod tests {
             TreeConfig::default(),
             sync_metrics_tx,
             evm_config,
-            changeset_cache,
         );
     }
 }

crates/engine/tree/Cargo.toml

@@ -34,8 +34,6 @@ reth-trie-parallel.workspace = true
 reth-trie-sparse = { workspace = true, features = ["std", "metrics"] }
 reth-trie-sparse-parallel = { workspace = true, features = ["std"] }
 reth-trie.workspace = true
-reth-trie-common.workspace = true
-reth-trie-db.workspace = true
 
 # alloy
 alloy-evm.workspace = true
@@ -96,7 +94,7 @@ reth-tracing.workspace = true
 reth-node-ethereum.workspace = true
 reth-e2e-test-utils.workspace = true
 
-# revm
+# alloy
 revm-state.workspace = true
 
 assert_matches.workspace = true
@@ -135,8 +133,6 @@ test-utils = [
     "reth-static-file",
     "reth-tracing",
     "reth-trie/test-utils",
-    "reth-trie-common/test-utils",
-    "reth-trie-db/test-utils",
     "reth-trie-sparse/test-utils",
     "reth-prune-types?/test-utils",
     "reth-trie-parallel/test-utils",

crates/engine/tree/benches/channel_perf.rs

@@ -28,10 +28,10 @@ fn create_bench_state(num_accounts: usize) -> EvmState {
                 code: Default::default(),
                 account_id: None,
             },
-            original_info: Box::new(AccountInfo::default()),
             storage,
             status: AccountStatus::empty(),
             transaction_id: 0,
+            ..Default::default()
         };
 
         let address = Address::with_last_byte(i as u8);

crates/engine/tree/benches/state_root_task.rs

@@ -62,7 +62,7 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
                     storage: HashMap::default(),
                     status: AccountStatus::SelfDestructed,
                     transaction_id: 0,
-                    original_info: Box::new(AccountInfo::default()),
+                    ..Default::default()
                 }
             } else {
                 RevmAccount {
@@ -86,8 +86,8 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
                         })
                         .collect(),
                     status: AccountStatus::Touched,
-                    original_info: Box::new(AccountInfo::default()),
                     transaction_id: 0,
+                    ..Default::default()
                 }
             };
 
@@ -242,10 +242,7 @@ fn bench_state_root(c: &mut Criterion) {
                                     std::convert::identity,
                                 ),
                                 StateProviderBuilder::new(provider.clone(), genesis_hash, None),
-                                OverlayStateProviderFactory::new(
-                                    provider,
-                                    reth_trie_db::ChangesetCache::new(),
-                                ),
+                                OverlayStateProviderFactory::new(provider),
                                 &TreeConfig::default(),
                                 None,
                             );

crates/engine/tree/src/persistence.rs

@@ -159,7 +159,6 @@ where
 
         self.metrics.save_blocks_block_count.record(block_count as f64);
         self.metrics.save_blocks_duration_seconds.record(start_time.elapsed());
-
         Ok(last_block)
     }
 }

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

@@ -60,22 +60,16 @@ impl EngineApiMetrics {
     ///
     /// This method updates metrics for execution time, gas usage, and the number
     /// of accounts, storage slots and bytecodes loaded and updated.
-    ///
-    /// The optional `on_receipt` callback is invoked after each transaction with the receipt
-    /// index and a reference to all receipts collected so far. This allows callers to stream
-    /// receipts to a background task for incremental receipt root computation.
-    pub(crate) fn execute_metered<E, DB, F>(
+    pub(crate) fn execute_metered<E, DB>(
         &self,
         executor: E,
         mut transactions: impl Iterator<Item = Result<impl ExecutableTx<E>, BlockExecutionError>>,
         transaction_count: usize,
         state_hook: Box<dyn OnStateHook>,
-        mut on_receipt: F,
     ) -> Result<(BlockExecutionOutput<E::Receipt>, Vec<Address>), BlockExecutionError>
     where
         DB: alloy_evm::Database,
         E: BlockExecutor<Evm: Evm<DB: BorrowMut<State<DB>>>, Transaction: SignedTransaction>,
-        F: FnMut(&[E::Receipt]),
     {
         // clone here is cheap, all the metrics are Option<Arc<_>>. additionally
         // they are globally registered so that the data recorded in the hook will
@@ -101,21 +95,14 @@ impl EngineApiMetrics {
                 let tx = tx?;
                 senders.push(*tx.signer());
 
-                let span = debug_span!(
-                    target: "engine::tree",
-                    "execute tx",
-                    tx_hash = ?tx.tx().tx_hash(),
-                    gas_used = tracing::field::Empty,
-                );
+                let span =
+                    debug_span!(target: "engine::tree", "execute tx", tx_hash=?tx.tx().tx_hash());
                 let enter = span.entered();
                 trace!(target: "engine::tree", "Executing transaction");
                 let start = Instant::now();
                 let gas_used = executor.execute_transaction(tx)?;
                 self.executor.transaction_execution_histogram.record(start.elapsed());
 
-                // Invoke callback with the latest receipt
-                on_receipt(executor.receipts());
-
                 // record the tx gas used
                 enter.record("gas_used", gas_used);
             }
@@ -270,10 +257,7 @@ impl ForkchoiceUpdatedMetrics {
 pub(crate) struct NewPayloadStatusMetrics {
     /// Finish time of the latest new payload call.
     #[metric(skip)]
-    pub(crate) latest_finish_at: Option<Instant>,
-    /// Start time of the latest new payload call.
-    #[metric(skip)]
-    pub(crate) latest_start_at: Option<Instant>,
+    pub(crate) latest_at: Option<Instant>,
     /// The total count of new payload messages received.
     pub(crate) new_payload_messages: Counter,
     /// The total count of new payload messages that we responded to with
@@ -301,10 +285,6 @@ pub(crate) struct NewPayloadStatusMetrics {
     pub(crate) new_payload_latency: Histogram,
     /// Latency for the last new payload call.
     pub(crate) new_payload_last: Gauge,
-    /// Time from previous payload finish to current payload start (idle time).
-    pub(crate) time_between_new_payloads: Histogram,
-    /// Time from previous payload start to current payload start (total interval).
-    pub(crate) new_payload_interval: Histogram,
 }
 
 impl NewPayloadStatusMetrics {
@@ -318,14 +298,7 @@ impl NewPayloadStatusMetrics {
         let finish = Instant::now();
         let elapsed = finish - start;
 
-        if let Some(prev_finish) = self.latest_finish_at {
-            self.time_between_new_payloads.record(start - prev_finish);
-        }
-        if let Some(prev_start) = self.latest_start_at {
-            self.new_payload_interval.record(start - prev_start);
-        }
-        self.latest_finish_at = Some(finish);
-        self.latest_start_at = Some(start);
+        self.latest_at = Some(finish);
         match result {
             Ok(outcome) => match outcome.outcome.status {
                 PayloadStatusEnum::Valid => {
@@ -361,6 +334,10 @@ pub(crate) struct BlockValidationMetrics {
     pub(crate) state_root_histogram: Histogram,
     /// Histogram of deferred trie computation duration.
     pub(crate) deferred_trie_compute_duration: Histogram,
+    /// Histogram of time spent waiting for deferred trie data to become available.
+    pub(crate) deferred_trie_wait_duration: Histogram,
+    /// Trie input computation duration
+    pub(crate) trie_input_duration: Histogram,
     /// Payload conversion and validation latency
     pub(crate) payload_validation_duration: Gauge,
     /// Histogram of payload validation latency
@@ -431,13 +408,12 @@ mod tests {
     /// A simple mock executor for testing that doesn't require complex EVM setup
     struct MockExecutor {
         state: EvmState,
-        receipts: Vec<Receipt>,
         hook: Option<Box<dyn OnStateHook>>,
     }
 
     impl MockExecutor {
         fn new(state: EvmState) -> Self {
-            Self { state, receipts: vec![], hook: None }
+            Self { state, hook: None }
         }
     }
 
@@ -510,6 +486,7 @@ mod tests {
                     receipts: vec![],
                     requests: Requests::default(),
                     gas_used: 1000,
+                    block_access_list: None,
                     blob_gas_used: 0,
                 },
             ))
@@ -519,16 +496,12 @@ mod tests {
             self.hook = hook;
         }
 
-        fn evm_mut(&mut self) -> &mut Self::Evm {
-            panic!("Mock executor evm_mut() not implemented")
-        }
-
         fn evm(&self) -> &Self::Evm {
             panic!("Mock executor evm() not implemented")
         }
 
-        fn receipts(&self) -> &[Self::Receipt] {
-            &self.receipts
+        fn evm_mut(&mut self) -> &mut Self::Evm {
+            panic!("Mock executor evm_mut() not implemented")
         }
     }
 
@@ -563,12 +536,11 @@ mod tests {
         let executor = MockExecutor::new(state);
 
         // This will fail to create the EVM but should still call the hook
-        let _result = metrics.execute_metered::<_, EmptyDB, _>(
+        let _result = metrics.execute_metered::<_, EmptyDB>(
             executor,
             input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
             input.transaction_count(),
             state_hook,
-            |_| {},
         );
 
         // Check if hook was called (it might not be if finish() fails early)
@@ -611,10 +583,10 @@ mod tests {
                         code: Default::default(),
                         account_id: None,
                     },
-                    original_info: Box::new(AccountInfo::default()),
                     storage,
                     status: AccountStatus::default(),
                     transaction_id: 0,
+                    ..Default::default()
                 },
             );
             state
@@ -623,12 +595,11 @@ mod tests {
         let executor = MockExecutor::new(state);
 
         // Execute (will fail but should still update some metrics)
-        let _result = metrics.execute_metered::<_, EmptyDB, _>(
+        let _result = metrics.execute_metered::<_, EmptyDB>(
             executor,
             input.clone_transactions_recovered().map(Ok::<_, BlockExecutionError>),
             input.transaction_count(),
             state_hook,
-            |_| {},
         );
 
         let snapshot = snapshotter.snapshot().into_vec();

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

@@ -30,13 +30,11 @@ use reth_payload_primitives::{
 };
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
 use reth_provider::{
-    BlockExecutionOutput, BlockExecutionResult, BlockNumReader, BlockReader, ChangeSetReader,
-    DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StageCheckpointReader,
-    StateProviderBox, StateProviderFactory, StateReader, TransactionVariant,
+    BlockReader, DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StateProviderBox,
+    StateProviderFactory, StateReader, TransactionVariant, TrieReader,
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_stages_api::ControlFlow;
-use reth_trie_db::ChangesetCache;
 use revm::state::EvmState;
 use state::TreeState;
 use std::{fmt::Debug, ops, sync::Arc, time::Instant};
@@ -273,8 +271,6 @@ where
     engine_kind: EngineApiKind,
     /// The EVM configuration.
     evm_config: C,
-    /// Changeset cache for in-memory trie changesets
-    changeset_cache: ChangesetCache,
 }
 
 impl<N, P: Debug, T: PayloadTypes + Debug, V: Debug, C> std::fmt::Debug
@@ -299,7 +295,6 @@ where
             .field("metrics", &self.metrics)
             .field("engine_kind", &self.engine_kind)
             .field("evm_config", &self.evm_config)
-            .field("changeset_cache", &self.changeset_cache)
             .finish()
     }
 }
@@ -312,12 +307,11 @@ where
         + StateProviderFactory
         + StateReader<Receipt = N::Receipt>
         + HashedPostStateProvider
+        + TrieReader
         + Clone
         + 'static,
-    <P as DatabaseProviderFactory>::Provider: BlockReader<Block = N::Block, Header = N::BlockHeader>
-        + StageCheckpointReader
-        + ChangeSetReader
-        + BlockNumReader,
+    <P as DatabaseProviderFactory>::Provider:
+        BlockReader<Block = N::Block, Header = N::BlockHeader>,
     C: ConfigureEvm<Primitives = N> + 'static,
     T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>,
     V: EngineValidator<T>,
@@ -337,7 +331,6 @@ where
         config: TreeConfig,
         engine_kind: EngineApiKind,
         evm_config: C,
-        changeset_cache: ChangesetCache,
     ) -> Self {
         let (incoming_tx, incoming) = crossbeam_channel::unbounded();
 
@@ -358,7 +351,6 @@ where
             incoming_tx,
             engine_kind,
             evm_config,
-            changeset_cache,
         }
     }
 
@@ -378,7 +370,6 @@ where
         config: TreeConfig,
         kind: EngineApiKind,
         evm_config: C,
-        changeset_cache: ChangesetCache,
     ) -> (Sender<FromEngine<EngineApiRequest<T, N>, N::Block>>, UnboundedReceiver<EngineApiEvent<N>>)
     {
         let best_block_number = provider.best_block_number().unwrap_or(0);
@@ -410,7 +401,6 @@ where
             config,
             kind,
             evm_config,
-            changeset_cache,
         );
         let incoming = task.incoming_tx.clone();
         std::thread::Builder::new().name("Engine Task".to_string()).spawn(|| task.run()).unwrap();
@@ -592,7 +582,7 @@ where
         //      null}` if the expected and the actual arrays don't match.
         //
         // This validation **MUST** be instantly run in all cases even during active sync process.
-
+        tracing::debug!("Payload received {:?}", payload);
         let num_hash = payload.num_hash();
         let engine_event = ConsensusEngineEvent::BlockReceived(num_hash);
         self.emit_event(EngineApiEvent::BeaconConsensus(engine_event));
@@ -601,6 +591,7 @@ where
 
         // Check for invalid ancestors
         if let Some(invalid) = self.find_invalid_ancestor(&payload) {
+            tracing::debug!(target: "engine::tree", ?invalid, "found invalid ancestor for payload");
             let status = self.handle_invalid_ancestor_payload(payload, invalid)?;
             return Ok(TreeOutcome::new(status));
         }
@@ -609,6 +600,7 @@ where
         self.metrics.block_validation.record_payload_validation(start.elapsed().as_secs_f64());
 
         let status = if self.backfill_sync_state.is_idle() {
+            tracing::debug!(target: "engine::tree", "inserting payload directly");
             self.try_insert_payload(payload)?
         } else {
             self.try_buffer_payload(payload)?
@@ -647,7 +639,7 @@ where
         let parent_hash = payload.parent_hash();
         let mut latest_valid_hash = None;
 
-        match self.insert_payload(payload) {
+        match self.insert_payload(payload.clone()) {
             Ok(status) => {
                 let status = match status {
                     InsertPayloadOk::Inserted(BlockStatus::Valid) => {
@@ -669,7 +661,10 @@ where
                 Ok(PayloadStatus::new(status, latest_valid_hash))
             }
             Err(error) => match error {
-                InsertPayloadError::Block(error) => Ok(self.on_insert_block_error(error)?),
+                InsertPayloadError::Block(error) => {
+                    tracing::debug!("payload in new payload l 617 {:?}", payload);
+                    Ok(self.on_insert_block_error(error)?)
+                }
                 InsertPayloadError::Payload(error) => {
                     Ok(self.on_new_payload_error(error, num_hash, parent_hash)?)
                 }
@@ -1375,21 +1370,6 @@ where
 
         debug!(target: "engine::tree", ?last_persisted_block_hash, ?last_persisted_block_number, elapsed=?start_time.elapsed(), "Finished persisting, calling finish");
         self.persistence_state.finish(last_persisted_block_hash, last_persisted_block_number);
-
-        // Evict trie changesets for blocks below the finalized block, but keep at least 64 blocks
-        if let Some(finalized) = self.canonical_in_memory_state.get_finalized_num_hash() {
-            let min_threshold = last_persisted_block_number.saturating_sub(64);
-            let eviction_threshold = finalized.number.min(min_threshold);
-            debug!(
-                target: "engine::tree",
-                last_persisted = last_persisted_block_number,
-                finalized_number = finalized.number,
-                eviction_threshold,
-                "Evicting changesets below threshold"
-            );
-            self.changeset_cache.evict(eviction_threshold);
-        }
-
         self.on_new_persisted_block()?;
         Ok(())
     }
@@ -1478,7 +1458,7 @@ where
 
                                 self.metrics.engine.forkchoice_updated.update_response_metrics(
                                     start,
-                                    &mut self.metrics.engine.new_payload.latest_finish_at,
+                                    &mut self.metrics.engine.new_payload.latest_at,
                                     has_attrs,
                                     &output,
                                 );
@@ -1676,18 +1656,6 @@ where
                 )));
                 return Ok(());
             }
-        } else {
-            // We don't have the head block or any of its ancestors buffered. Request
-            // a download for the head block which will then trigger further sync.
-            debug!(
-                target: "engine::tree",
-                head_hash = %sync_target_state.head_block_hash,
-                "Backfill complete but head block not buffered, requesting download"
-            );
-            self.emit_event(EngineApiEvent::Download(DownloadRequest::single_block(
-                sync_target_state.head_block_hash,
-            )));
-            return Ok(());
         }
 
         // try to close the gap by executing buffered blocks that are child blocks of the new head
@@ -1855,7 +1823,6 @@ where
     /// or the database. If the required historical data (such as trie change sets) has been
     /// pruned for a given block, this operation will return an error. On archive nodes, it
     /// can retrieve any block.
-    #[instrument(level = "debug", target = "engine::tree", skip(self))]
     fn canonical_block_by_hash(&self, hash: B256) -> ProviderResult<Option<ExecutedBlock<N>>> {
         trace!(target: "engine::tree", ?hash, "Fetching executed block by hash");
         // check memory first
@@ -1868,23 +1835,12 @@ where
             .sealed_block_with_senders(hash.into(), TransactionVariant::WithHash)?
             .ok_or_else(|| ProviderError::HeaderNotFound(hash.into()))?
             .split_sealed();
-        let mut execution_output = self
+        let execution_output = self
             .provider
             .get_state(block.header().number())?
             .ok_or_else(|| ProviderError::StateForNumberNotFound(block.header().number()))?;
         let hashed_state = self.provider.hashed_post_state(execution_output.state());
-
-        debug!(
-            target: "engine::tree",
-            number = ?block.number(),
-            "computing block trie updates",
-        );
-        let db_provider = self.provider.database_provider_ro()?;
-        let trie_updates = reth_trie_db::compute_block_trie_updates(
-            &self.changeset_cache,
-            &db_provider,
-            block.number(),
-        )?;
+        let trie_updates = self.provider.get_block_trie_updates(block.number())?;
 
         let sorted_hashed_state = Arc::new(hashed_state.into_sorted());
         let sorted_trie_updates = Arc::new(trie_updates);
@@ -1892,19 +1848,9 @@ where
         let trie_data =
             ComputedTrieData::without_trie_input(sorted_hashed_state, sorted_trie_updates);
 
-        let execution_output = Arc::new(BlockExecutionOutput {
-            state: execution_output.bundle,
-            result: BlockExecutionResult {
-                receipts: execution_output.receipts.pop().unwrap_or_default(),
-                requests: execution_output.requests.pop().unwrap_or_default(),
-                gas_used: block.gas_used(),
-                blob_gas_used: block.blob_gas_used().unwrap_or_default(),
-            },
-        });
-
         Ok(Some(ExecutedBlock::new(
             Arc::new(RecoveredBlock::new_sealed(block, senders)),
-            execution_output,
+            Arc::new(execution_output),
             trie_data,
         )))
     }

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

@@ -6,6 +6,7 @@ use alloy_primitives::{keccak256, Address, StorageKey, U256};
 use reth_primitives_traits::Account;
 use reth_provider::{AccountReader, ProviderError};
 use reth_trie::{HashedPostState, HashedStorage};
+use revm_primitives::B256;
 use std::ops::Range;
 
 /// Returns the total number of storage slots (both changed and read-only) across all accounts in
@@ -101,7 +102,7 @@ impl<'a> Iterator for BALSlotIter<'a> {
                     return None;
                 }
 
-                return Some((address, StorageKey::from(slot)));
+                return Some((address, slot.into()));
             }
 
             // Move to next account
@@ -177,7 +178,7 @@ where
             let mut storage_map = HashedStorage::new(false);
 
             for slot_changes in &account_changes.storage_changes {
-                let hashed_slot = keccak256(slot_changes.slot.to_be_bytes::<32>());
+                let hashed_slot = keccak256(B256::from(slot_changes.slot));
 
                 // Get the last change for this slot
                 if let Some(last_change) = slot_changes.changes.last() {
@@ -198,7 +199,7 @@ mod tests {
     use alloy_eip7928::{
         AccountChanges, BalanceChange, CodeChange, NonceChange, SlotChanges, StorageChange,
     };
-    use alloy_primitives::{Address, Bytes, StorageKey, B256};
+    use alloy_primitives::{Address, Bytes, B256};
     use reth_revm::test_utils::StateProviderTest;
 
     #[test]
@@ -256,7 +257,7 @@ mod tests {
         assert!(result.storages.contains_key(&hashed_address));
 
         let storage = result.storages.get(&hashed_address).unwrap();
-        let hashed_slot = keccak256(slot.to_be_bytes::<32>());
+        let hashed_slot = keccak256(B256::from(slot));
 
         let stored_value = storage.storage.get(&hashed_slot).unwrap();
         assert_eq!(*stored_value, value);
@@ -416,7 +417,7 @@ mod tests {
 
         let hashed_address = keccak256(address);
         let storage = result.storages.get(&hashed_address).unwrap();
-        let hashed_slot = keccak256(slot.to_be_bytes::<32>());
+        let hashed_slot = keccak256(B256::from(slot));
 
         let stored_value = storage.storage.get(&hashed_slot).unwrap();
 

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

@@ -28,10 +28,10 @@ use reth_evm::{
     ConfigureEvm, EvmEnvFor, ExecutableTxIterator, ExecutableTxTuple, OnStateHook, SpecFor,
     TxEnvFor,
 };
+use reth_execution_types::ExecutionOutcome;
 use reth_primitives_traits::NodePrimitives;
 use reth_provider::{
-    BlockExecutionOutput, BlockReader, DatabaseProviderROFactory, StateProvider,
-    StateProviderFactory, StateReader,
+    BlockReader, DatabaseProviderROFactory, StateProvider, StateProviderFactory, StateReader,
 };
 use reth_revm::{db::BundleState, state::EvmState};
 use reth_trie::{hashed_cursor::HashedCursorFactory, trie_cursor::TrieCursorFactory};
@@ -61,7 +61,6 @@ mod configured_sparse_trie;
 pub mod executor;
 pub mod multiproof;
 pub mod prewarm;
-pub mod receipt_root_task;
 pub mod sparse_trie;
 
 use configured_sparse_trie::ConfiguredSparseTrie;
@@ -666,15 +665,13 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
 
     /// Terminates the entire caching task.
     ///
-    /// If the [`BlockExecutionOutput`] is provided it will update the shared cache using its
+    /// If the [`ExecutionOutcome`] is provided it will update the shared cache using its
     /// bundle state. Using `Arc<ExecutionOutcome>` allows sharing with the main execution
     /// path without cloning the expensive `BundleState`.
-    ///
-    /// Returns a sender for the channel that should be notified on block validation success.
     pub(super) fn terminate_caching(
         &mut self,
-        execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
-    ) -> Option<mpsc::Sender<()>> {
+        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
+    ) {
         self.prewarm_handle.terminate_caching(execution_outcome)
     }
 
@@ -710,21 +707,15 @@ impl<R: Send + Sync + 'static> CacheTaskHandle<R> {
 
     /// Terminates the entire pre-warming task.
     ///
-    /// If the [`BlockExecutionOutput`] is provided it will update the shared cache using its
+    /// If the [`ExecutionOutcome`] is provided it will update the shared cache using its
     /// bundle state. Using `Arc<ExecutionOutcome>` avoids cloning the expensive `BundleState`.
-    #[must_use = "sender must be used and notified on block validation success"]
     pub(super) fn terminate_caching(
         &mut self,
-        execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
-    ) -> Option<mpsc::Sender<()>> {
+        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
+    ) {
         if let Some(tx) = self.to_prewarm_task.take() {
-            let (valid_block_tx, valid_block_rx) = mpsc::channel();
-            let event = PrewarmTaskEvent::Terminate { execution_outcome, valid_block_rx };
+            let event = PrewarmTaskEvent::Terminate { execution_outcome };
             let _ = tx.send(event);
-
-            Some(valid_block_tx)
-        } else {
-            None
         }
     }
 }
@@ -733,10 +724,7 @@ impl<R> Drop for CacheTaskHandle<R> {
     fn drop(&mut self) {
         // Ensure we always terminate on drop - send None without needing Send + Sync bounds
         if let Some(tx) = self.to_prewarm_task.take() {
-            let _ = tx.send(PrewarmTaskEvent::Terminate {
-                execution_outcome: None,
-                valid_block_rx: mpsc::channel().1,
-            });
+            let _ = tx.send(PrewarmTaskEvent::Terminate { execution_outcome: None });
         }
     }
 }
@@ -898,7 +886,6 @@ mod tests {
     use reth_revm::db::BundleState;
     use reth_testing_utils::generators;
     use reth_trie::{test_utils::state_root, HashedPostState};
-    use reth_trie_db::ChangesetCache;
     use revm_primitives::{Address, HashMap, B256, KECCAK_EMPTY, U256};
     use revm_state::{AccountInfo, AccountStatus, EvmState, EvmStorageSlot};
     use std::sync::Arc;
@@ -1073,10 +1060,10 @@ mod tests {
                         code: Some(Default::default()),
                         account_id: None,
                     },
-                    original_info: Box::new(AccountInfo::default()),
                     storage,
                     status: AccountStatus::Touched,
                     transaction_id: 0,
+                    ..Default::default()
                 };
 
                 state_update.insert(address, account);
@@ -1156,7 +1143,7 @@ mod tests {
                 std::convert::identity,
             ),
             StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
-            OverlayStateProviderFactory::new(provider_factory, ChangesetCache::new()),
+            OverlayStateProviderFactory::new(provider_factory),
             &TreeConfig::default(),
             None, // No BAL for test
         );

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

@@ -141,27 +141,22 @@ impl ProofSequencer {
     /// Adds a proof with the corresponding state update and returns all sequential proofs and state
     /// updates if we have a continuous sequence
     fn add_proof(&mut self, sequence: u64, update: SparseTrieUpdate) -> Vec<SparseTrieUpdate> {
-        // Optimization: fast path for in-order delivery to avoid BTreeMap overhead.
-        // If this is the expected sequence, return it immediately without buffering.
-        if sequence == self.next_to_deliver {
-            let mut consecutive_proofs = Vec::with_capacity(1);
-            consecutive_proofs.push(update);
-            self.next_to_deliver += 1;
-
-            // Check if we have subsequent proofs in the pending buffer
-            while let Some(pending) = self.pending_proofs.remove(&self.next_to_deliver) {
-                consecutive_proofs.push(pending);
-                self.next_to_deliver += 1;
-            }
-
-            return consecutive_proofs;
-        }
-
-        if sequence > self.next_to_deliver {
+        if sequence >= self.next_to_deliver {
             self.pending_proofs.insert(sequence, update);
         }
 
-        Vec::new()
+        let mut consecutive_proofs = Vec::with_capacity(self.pending_proofs.len());
+        let mut current_sequence = self.next_to_deliver;
+
+        // keep collecting proofs and state updates as long as we have consecutive sequence numbers
+        while let Some(pending) = self.pending_proofs.remove(&current_sequence) {
+            consecutive_proofs.push(pending);
+            current_sequence += 1;
+        }
+
+        self.next_to_deliver += consecutive_proofs.len() as u64;
+
+        consecutive_proofs
     }
 
     /// Returns true if we still have pending proofs
@@ -1323,15 +1318,79 @@ mod tests {
     use reth_provider::{
         providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
         BlockNumReader, BlockReader, ChangeSetReader, DatabaseProviderFactory, LatestStateProvider,
-        PruneCheckpointReader, StageCheckpointReader, StateProviderBox,
+        PruneCheckpointReader, StageCheckpointReader, StateProviderBox, TrieReader,
     };
     use reth_trie::MultiProof;
-    use reth_trie_db::ChangesetCache;
     use reth_trie_parallel::proof_task::{ProofTaskCtx, ProofWorkerHandle};
     use revm_primitives::{B256, U256};
-    use std::sync::{Arc, OnceLock};
+    use std::{
+        mem,
+        sync::{Arc, OnceLock},
+    };
     use tokio::runtime::{Handle, Runtime};
 
+    /// Maximum number of targets to batch together for state update batching.
+    const STATE_UPDATE_MAX_BATCH_TARGETS: usize = 64;
+
+    /// Checks whether two `Source` values refer to the same origin.
+    fn same_source(lhs: Source, rhs: Source) -> bool {
+        match (lhs, rhs) {
+            (Source::Evm(a), Source::Evm(b)) => same_state_change_source(a, b),
+            (Source::BlockAccessList, Source::BlockAccessList) => true,
+            _ => false,
+        }
+    }
+
+    /// Checks whether two state change sources refer to the same origin.
+    fn same_state_change_source(lhs: StateChangeSource, rhs: StateChangeSource) -> bool {
+        match (lhs, rhs) {
+            (StateChangeSource::Transaction(a), StateChangeSource::Transaction(b)) => a == b,
+            (StateChangeSource::PreBlock(a), StateChangeSource::PreBlock(b)) => {
+                mem::discriminant(&a) == mem::discriminant(&b)
+            }
+            (StateChangeSource::PostBlock(a), StateChangeSource::PostBlock(b)) => {
+                mem::discriminant(&a) == mem::discriminant(&b)
+            }
+            _ => false,
+        }
+    }
+
+    /// Determines if a state update can be batched with the current batch.
+    fn can_batch_state_update(
+        batch_source: Source,
+        batch_update: &EvmState,
+        next_source: Source,
+        next_update: &EvmState,
+    ) -> bool {
+        if !same_source(batch_source, next_source) {
+            return false;
+        }
+
+        match (batch_source, next_source) {
+            (
+                Source::Evm(StateChangeSource::PreBlock(_)),
+                Source::Evm(StateChangeSource::PreBlock(_)),
+            ) |
+            (
+                Source::Evm(StateChangeSource::PostBlock(_)),
+                Source::Evm(StateChangeSource::PostBlock(_)),
+            ) => batch_update == next_update,
+            _ => true,
+        }
+    }
+
+    /// Estimates target count from `EvmState` for batching decisions.
+    fn estimate_evm_state_targets(state: &EvmState) -> usize {
+        state
+            .values()
+            .filter(|account| account.is_touched())
+            .map(|account| {
+                let changed_slots = account.storage.iter().filter(|(_, v)| v.is_changed()).count();
+                1 + changed_slots
+            })
+            .sum()
+    }
+
     /// Get a handle to the test runtime, creating it if necessary
     fn get_test_runtime_handle() -> Handle {
         static TEST_RT: OnceLock<Runtime> = OnceLock::new();
@@ -1347,6 +1406,7 @@ mod tests {
     where
         F: DatabaseProviderFactory<
                 Provider: BlockReader
+                              + TrieReader
                               + StageCheckpointReader
                               + PruneCheckpointReader
                               + ChangeSetReader
@@ -1356,8 +1416,7 @@ mod tests {
             + 'static,
     {
         let rt_handle = get_test_runtime_handle();
-        let changeset_cache = ChangesetCache::new();
-        let overlay_factory = OverlayStateProviderFactory::new(factory, changeset_cache);
+        let overlay_factory = OverlayStateProviderFactory::new(factory);
         let task_ctx = ProofTaskCtx::new(overlay_factory);
         let proof_handle = ProofWorkerHandle::new(rt_handle, task_ctx, 1, 1, false);
         let (to_sparse_trie, _receiver) = std::sync::mpsc::channel();
@@ -1369,7 +1428,7 @@ mod tests {
     fn create_cached_provider<F>(factory: F) -> CachedStateProvider<StateProviderBox>
     where
         F: DatabaseProviderFactory<
-                Provider: BlockReader + StageCheckpointReader + PruneCheckpointReader,
+                Provider: BlockReader + TrieReader + StageCheckpointReader + PruneCheckpointReader,
             > + Clone
             + Send
             + 'static,
@@ -1781,6 +1840,328 @@ mod tests {
         assert_eq!(proofs_requested, 1);
     }
 
+    /// Verifies that consecutive state update messages from the same source are batched together.
+    #[test]
+    fn test_state_update_batching() {
+        use alloy_evm::block::StateChangeSource;
+        use revm_state::Account;
+
+        let test_provider_factory = create_test_provider_factory();
+        let mut task = create_test_state_root_task(test_provider_factory);
+
+        // create multiple state updates
+        let addr1 = alloy_primitives::Address::random();
+        let addr2 = alloy_primitives::Address::random();
+
+        let mut update1 = EvmState::default();
+        update1.insert(
+            addr1,
+            Account {
+                info: revm_state::AccountInfo {
+                    balance: U256::from(100),
+                    nonce: 1,
+                    code_hash: Default::default(),
+                    code: Default::default(),
+                    account_id: Some(0),
+                },
+                original_info: Default::default(),
+                transaction_id: Default::default(),
+                storage: Default::default(),
+                status: revm_state::AccountStatus::Touched,
+            },
+        );
+
+        let mut update2 = EvmState::default();
+        update2.insert(
+            addr2,
+            Account {
+                info: revm_state::AccountInfo {
+                    balance: U256::from(200),
+                    nonce: 2,
+                    code_hash: Default::default(),
+                    code: Default::default(),
+                    account_id: Some(0),
+                },
+                original_info: Default::default(),
+                transaction_id: Default::default(),
+                storage: Default::default(),
+                status: revm_state::AccountStatus::Touched,
+            },
+        );
+
+        let source = StateChangeSource::Transaction(0);
+
+        let tx = task.tx.clone();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), update1.clone())).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), update2.clone())).unwrap();
+
+        let proofs_requested =
+            if let Ok(MultiProofMessage::StateUpdate(_src, update)) = task.rx.recv() {
+                let mut merged_update = update;
+                let mut num_batched = 1;
+
+                while let Ok(MultiProofMessage::StateUpdate(_next_source, next_update)) =
+                    task.rx.try_recv()
+                {
+                    merged_update.extend(next_update);
+                    num_batched += 1;
+                }
+
+                assert_eq!(num_batched, 2);
+                assert_eq!(merged_update.len(), 2);
+                assert!(merged_update.contains_key(&addr1));
+                assert!(merged_update.contains_key(&addr2));
+
+                task.on_state_update(source.into(), merged_update)
+            } else {
+                panic!("Expected StateUpdate message");
+            };
+        assert_eq!(proofs_requested, 1);
+    }
+
+    /// Verifies that state updates from different sources are not batched together.
+    #[test]
+    fn test_state_update_batching_separates_sources() {
+        use alloy_evm::block::StateChangeSource;
+        use revm_state::Account;
+
+        let test_provider_factory = create_test_provider_factory();
+        let task = create_test_state_root_task(test_provider_factory);
+
+        let addr_a1 = alloy_primitives::Address::random();
+        let addr_b1 = alloy_primitives::Address::random();
+        let addr_a2 = alloy_primitives::Address::random();
+
+        let create_state_update = |addr: alloy_primitives::Address, balance: u64| {
+            let mut state = EvmState::default();
+            state.insert(
+                addr,
+                Account {
+                    info: revm_state::AccountInfo {
+                        balance: U256::from(balance),
+                        nonce: 1,
+                        code_hash: Default::default(),
+                        code: Default::default(),
+                        account_id: Some(0),
+                    },
+                    original_info: Default::default(),
+                    transaction_id: Default::default(),
+                    storage: Default::default(),
+                    status: revm_state::AccountStatus::Touched,
+                },
+            );
+            state
+        };
+
+        let source_a = StateChangeSource::Transaction(1);
+        let source_b = StateChangeSource::Transaction(2);
+
+        // Queue: A1 (immediate dispatch), B1 (batched), A2 (should become pending)
+        let tx = task.tx.clone();
+        tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a1, 100)))
+            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source_b.into(), create_state_update(addr_b1, 200)))
+            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source_a.into(), create_state_update(addr_a2, 300)))
+            .unwrap();
+
+        let mut pending_msg: Option<MultiProofMessage> = None;
+
+        if let Ok(MultiProofMessage::StateUpdate(first_source, _)) = task.rx.recv() {
+            assert!(same_source(first_source, source_a.into()));
+
+            // Simulate batching loop for remaining messages
+            let mut accumulated_updates: Vec<(Source, EvmState)> = Vec::new();
+            let mut accumulated_targets = 0usize;
+
+            loop {
+                if accumulated_targets >= STATE_UPDATE_MAX_BATCH_TARGETS {
+                    break;
+                }
+                match task.rx.try_recv() {
+                    Ok(MultiProofMessage::StateUpdate(next_source, next_update)) => {
+                        if let Some((batch_source, batch_update)) = accumulated_updates.first() &&
+                            !can_batch_state_update(
+                                *batch_source,
+                                batch_update,
+                                next_source,
+                                &next_update,
+                            )
+                        {
+                            pending_msg =
+                                Some(MultiProofMessage::StateUpdate(next_source, next_update));
+                            break;
+                        }
+
+                        let next_estimate = estimate_evm_state_targets(&next_update);
+                        if next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS {
+                            pending_msg =
+                                Some(MultiProofMessage::StateUpdate(next_source, next_update));
+                            break;
+                        }
+                        if accumulated_targets + next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS &&
+                            !accumulated_updates.is_empty()
+                        {
+                            pending_msg =
+                                Some(MultiProofMessage::StateUpdate(next_source, next_update));
+                            break;
+                        }
+                        accumulated_targets += next_estimate;
+                        accumulated_updates.push((next_source, next_update));
+                    }
+                    Ok(other_msg) => {
+                        pending_msg = Some(other_msg);
+                        break;
+                    }
+                    Err(_) => break,
+                }
+            }
+
+            assert_eq!(accumulated_updates.len(), 1, "Should only batch matching sources");
+            let batch_source = accumulated_updates[0].0;
+            assert!(same_source(batch_source, source_b.into()));
+
+            let batch_source = accumulated_updates[0].0;
+            let mut merged_update = accumulated_updates.remove(0).1;
+            for (_, next_update) in accumulated_updates {
+                merged_update.extend(next_update);
+            }
+
+            assert!(same_source(batch_source, source_b.into()), "Batch should use matching source");
+            assert!(merged_update.contains_key(&addr_b1));
+            assert!(!merged_update.contains_key(&addr_a1));
+            assert!(!merged_update.contains_key(&addr_a2));
+        } else {
+            panic!("Expected first StateUpdate");
+        }
+
+        match pending_msg {
+            Some(MultiProofMessage::StateUpdate(pending_source, pending_update)) => {
+                assert!(same_source(pending_source, source_a.into()));
+                assert!(pending_update.contains_key(&addr_a2));
+            }
+            other => panic!("Expected pending StateUpdate with source_a, got {:?}", other),
+        }
+    }
+
+    /// Verifies that pre-block updates only batch when their payloads are identical.
+    #[test]
+    fn test_pre_block_updates_require_payload_match_to_batch() {
+        use alloy_evm::block::{StateChangePreBlockSource, StateChangeSource};
+        use revm_state::Account;
+
+        let test_provider_factory = create_test_provider_factory();
+        let task = create_test_state_root_task(test_provider_factory);
+
+        let addr1 = alloy_primitives::Address::random();
+        let addr2 = alloy_primitives::Address::random();
+        let addr3 = alloy_primitives::Address::random();
+
+        let create_state_update = |addr: alloy_primitives::Address, balance: u64| {
+            let mut state = EvmState::default();
+            state.insert(
+                addr,
+                Account {
+                    info: revm_state::AccountInfo {
+                        balance: U256::from(balance),
+                        nonce: 1,
+                        code_hash: Default::default(),
+                        code: Default::default(),
+                        account_id: Some(0),
+                    },
+                    original_info: Default::default(),
+                    transaction_id: Default::default(),
+                    storage: Default::default(),
+                    status: revm_state::AccountStatus::Touched,
+                },
+            );
+            state
+        };
+
+        let source = StateChangeSource::PreBlock(StateChangePreBlockSource::BeaconRootContract);
+
+        // Queue: first update dispatched immediately, next two should not merge
+        let tx = task.tx.clone();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr1, 100)))
+            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr2, 200)))
+            .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(source.into(), create_state_update(addr3, 300)))
+            .unwrap();
+
+        let mut pending_msg: Option<MultiProofMessage> = None;
+
+        if let Ok(MultiProofMessage::StateUpdate(first_source, first_update)) = task.rx.recv() {
+            assert!(same_source(first_source, source.into()));
+            assert!(first_update.contains_key(&addr1));
+
+            let mut accumulated_updates: Vec<(Source, EvmState)> = Vec::new();
+            let mut accumulated_targets = 0usize;
+
+            loop {
+                if accumulated_targets >= STATE_UPDATE_MAX_BATCH_TARGETS {
+                    break;
+                }
+                match task.rx.try_recv() {
+                    Ok(MultiProofMessage::StateUpdate(next_source, next_update)) => {
+                        if let Some((batch_source, batch_update)) = accumulated_updates.first() &&
+                            !can_batch_state_update(
+                                *batch_source,
+                                batch_update,
+                                next_source,
+                                &next_update,
+                            )
+                        {
+                            pending_msg =
+                                Some(MultiProofMessage::StateUpdate(next_source, next_update));
+                            break;
+                        }
+
+                        let next_estimate = estimate_evm_state_targets(&next_update);
+                        if next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS {
+                            pending_msg =
+                                Some(MultiProofMessage::StateUpdate(next_source, next_update));
+                            break;
+                        }
+                        if accumulated_targets + next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS &&
+                            !accumulated_updates.is_empty()
+                        {
+                            pending_msg =
+                                Some(MultiProofMessage::StateUpdate(next_source, next_update));
+                            break;
+                        }
+                        accumulated_targets += next_estimate;
+                        accumulated_updates.push((next_source, next_update));
+                    }
+                    Ok(other_msg) => {
+                        pending_msg = Some(other_msg);
+                        break;
+                    }
+                    Err(_) => break,
+                }
+            }
+
+            assert_eq!(
+                accumulated_updates.len(),
+                1,
+                "Second pre-block update should not merge with a different payload"
+            );
+            let (batched_source, batched_update) = accumulated_updates.remove(0);
+            assert!(same_source(batched_source, source.into()));
+            assert!(batched_update.contains_key(&addr2));
+            assert!(!batched_update.contains_key(&addr3));
+
+            match pending_msg {
+                Some(MultiProofMessage::StateUpdate(_, pending_update)) => {
+                    assert!(pending_update.contains_key(&addr3));
+                }
+                other => panic!("Expected pending third pre-block update, got {:?}", other),
+            }
+        } else {
+            panic!("Expected first StateUpdate");
+        }
+    }
+
     /// Verifies that different message types arriving mid-batch are not lost and preserve order.
     #[test]
     fn test_batching_preserves_ordering_with_different_message_type() {
@@ -1816,9 +2197,9 @@ mod tests {
                     nonce: 1,
                     code_hash: Default::default(),
                     code: Default::default(),
-                    account_id: None,
+                    account_id: Some(0),
                 },
-                original_info: Box::new(revm_state::AccountInfo::default()),
+                original_info: Default::default(),
                 transaction_id: Default::default(),
                 storage: Default::default(),
                 status: revm_state::AccountStatus::Touched,
@@ -1835,9 +2216,9 @@ mod tests {
                     nonce: 2,
                     code_hash: Default::default(),
                     code: Default::default(),
-                    account_id: None,
+                    account_id: Some(0),
                 },
-                original_info: Box::new(revm_state::AccountInfo::default()),
+                original_info: Default::default(),
                 transaction_id: Default::default(),
                 storage: Default::default(),
                 status: revm_state::AccountStatus::Touched,
@@ -1939,9 +2320,9 @@ mod tests {
                     nonce: 1,
                     code_hash: Default::default(),
                     code: Default::default(),
-                    account_id: None,
+                    account_id: Some(0),
                 },
-                original_info: Box::new(revm_state::AccountInfo::default()),
+                original_info: Default::default(),
                 transaction_id: Default::default(),
                 storage: Default::default(),
                 status: revm_state::AccountStatus::Touched,
@@ -1988,7 +2369,157 @@ mod tests {
                 assert_eq!(targets.len(), 1);
                 assert!(targets.contains_key(&prefetch_addr2));
             }
-            other => panic!("Expected PrefetchProofs2 in channel, got {:?}", other),
+            other => panic!("Expected remaining PrefetchProofs2 in pending_msg, got {:?}", other),
+        }
+    }
+
+    /// Verifies that pending messages from a previous batch drain get full batching treatment.
+    #[test]
+    fn test_pending_messages_get_full_batching_treatment() {
+        // Queue: [Prefetch1, State1, State2, State3, Prefetch2]
+        //
+        // Expected behavior:
+        // 1. recv() → Prefetch1
+        // 2. try_recv() → State1 is different type → pending = State1, break
+        // 3. Process Prefetch1
+        // 4. Next iteration: pending = State1 → process with batching
+        // 5. try_recv() → State2 same type → merge
+        // 6. try_recv() → State3 same type → merge
+        // 7. try_recv() → Prefetch2 different type → pending = Prefetch2, break
+        // 8. Process merged State (1+2+3)
+        //
+        // Without the state-machine fix, State1 would be processed alone (no batching).
+        use alloy_evm::block::StateChangeSource;
+        use revm_state::Account;
+
+        let test_provider_factory = create_test_provider_factory();
+        let task = create_test_state_root_task(test_provider_factory);
+
+        let prefetch_addr1 = B256::random();
+        let prefetch_addr2 = B256::random();
+        let state_addr1 = alloy_primitives::Address::random();
+        let state_addr2 = alloy_primitives::Address::random();
+        let state_addr3 = alloy_primitives::Address::random();
+
+        // Create Prefetch targets
+        let mut prefetch1 = MultiProofTargets::default();
+        prefetch1.insert(prefetch_addr1, HashSet::default());
+
+        let mut prefetch2 = MultiProofTargets::default();
+        prefetch2.insert(prefetch_addr2, HashSet::default());
+
+        // Create StateUpdates
+        let create_state_update = |addr: alloy_primitives::Address, balance: u64| {
+            let mut state = EvmState::default();
+            state.insert(
+                addr,
+                Account {
+                    info: revm_state::AccountInfo {
+                        balance: U256::from(balance),
+                        nonce: 1,
+                        code_hash: Default::default(),
+                        code: Default::default(),
+                        account_id: Some(0),
+                    },
+                    original_info: Default::default(),
+                    transaction_id: Default::default(),
+                    storage: Default::default(),
+                    status: revm_state::AccountStatus::Touched,
+                },
+            );
+            state
+        };
+
+        let source = StateChangeSource::Transaction(42);
+
+        // Queue: [Prefetch1, State1, State2, State3, Prefetch2]
+        let tx = task.tx.clone();
+        tx.send(MultiProofMessage::PrefetchProofs(prefetch1.clone())).unwrap();
+        tx.send(MultiProofMessage::StateUpdate(
+            source.into(),
+            create_state_update(state_addr1, 100),
+        ))
+        .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(
+            source.into(),
+            create_state_update(state_addr2, 200),
+        ))
+        .unwrap();
+        tx.send(MultiProofMessage::StateUpdate(
+            source.into(),
+            create_state_update(state_addr3, 300),
+        ))
+        .unwrap();
+        tx.send(MultiProofMessage::PrefetchProofs(prefetch2.clone())).unwrap();
+
+        // Simulate the state-machine loop behavior
+        let mut pending_msg: Option<MultiProofMessage> = None;
+
+        // First iteration: recv() gets Prefetch1, drains until State1
+        if let Ok(MultiProofMessage::PrefetchProofs(targets)) = task.rx.recv() {
+            let mut merged_targets = targets;
+            loop {
+                match task.rx.try_recv() {
+                    Ok(MultiProofMessage::PrefetchProofs(next_targets)) => {
+                        merged_targets.extend(next_targets);
+                    }
+                    Ok(other_msg) => {
+                        pending_msg = Some(other_msg);
+                        break;
+                    }
+                    Err(_) => break,
+                }
+            }
+            // Should have only Prefetch1 (State1 is different type)
+            assert_eq!(merged_targets.len(), 1);
+            assert!(merged_targets.contains_key(&prefetch_addr1));
+        } else {
+            panic!("Expected PrefetchProofs");
+        }
+
+        // Pending should be State1
+        assert!(matches!(pending_msg, Some(MultiProofMessage::StateUpdate(_, _))));
+
+        // Second iteration: process pending State1 WITH BATCHING
+        // This is the key test - the pending message should drain State2 and State3
+        if let Some(MultiProofMessage::StateUpdate(_src, first_update)) = pending_msg.take() {
+            let mut merged_update = first_update;
+            let mut num_batched = 1;
+
+            loop {
+                match task.rx.try_recv() {
+                    Ok(MultiProofMessage::StateUpdate(_src, next_update)) => {
+                        merged_update.extend(next_update);
+                        num_batched += 1;
+                    }
+                    Ok(other_msg) => {
+                        pending_msg = Some(other_msg);
+                        break;
+                    }
+                    Err(_) => break,
+                }
+            }
+
+            // THE KEY ASSERTION: pending State1 should have batched with State2 and State3
+            assert_eq!(
+                num_batched, 3,
+                "Pending message should get full batching treatment and merge all 3 StateUpdates"
+            );
+            assert_eq!(merged_update.len(), 3, "Should have all 3 addresses in merged update");
+            assert!(merged_update.contains_key(&state_addr1));
+            assert!(merged_update.contains_key(&state_addr2));
+            assert!(merged_update.contains_key(&state_addr3));
+        } else {
+            panic!("Expected pending StateUpdate");
+        }
+
+        // Pending should now be Prefetch2
+        match pending_msg {
+            Some(MultiProofMessage::PrefetchProofs(targets)) => {
+                assert_eq!(targets.len(), 1);
+                assert!(targets.contains_key(&prefetch_addr2));
+            }
+            _ => panic!("Prefetch2 was lost!"),
         }
     }
 

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

@@ -30,12 +30,10 @@ use alloy_primitives::{keccak256, map::B256Set, B256};
 use crossbeam_channel::Sender as CrossbeamSender;
 use metrics::{Counter, Gauge, Histogram};
 use reth_evm::{execute::ExecutableTxFor, ConfigureEvm, Evm, EvmFor, SpecFor};
+use reth_execution_types::ExecutionOutcome;
 use reth_metrics::Metrics;
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{
-    AccountReader, BlockExecutionOutput, BlockReader, StateProvider, StateProviderFactory,
-    StateReader,
-};
+use reth_provider::{AccountReader, BlockReader, StateProvider, StateProviderFactory, StateReader};
 use reth_revm::{database::StateProviderDatabase, state::EvmState};
 use reth_trie::MultiProofTargets;
 use std::{
@@ -261,11 +259,7 @@ where
     ///
     /// This method is called from `run()` only after all execution tasks are complete.
     #[instrument(level = "debug", target = "engine::tree::payload_processor::prewarm", skip_all)]
-    fn save_cache(
-        self,
-        execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
-        valid_block_rx: mpsc::Receiver<()>,
-    ) {
+    fn save_cache(self, execution_outcome: Arc<ExecutionOutcome<N::Receipt>>) {
         let start = Instant::now();
 
         let Self { execution_cache, ctx: PrewarmContext { env, metrics, saved_cache, .. }, .. } =
@@ -283,7 +277,7 @@ where
 
                 // Insert state into cache while holding the lock
                 // Access the BundleState through the shared ExecutionOutcome
-                if new_cache.cache().insert_state(&execution_outcome.state).is_err() {
+                if new_cache.cache().insert_state(execution_outcome.state()).is_err() {
                     // Clear the cache on error to prevent having a polluted cache
                     *cached = None;
                     debug!(target: "engine::caching", "cleared execution cache on update error");
@@ -292,11 +286,9 @@ where
 
                 new_cache.update_metrics();
 
-                if valid_block_rx.recv().is_ok() {
-                    // Replace the shared cache with the new one; the previous cache (if any) is
-                    // dropped.
-                    *cached = Some(new_cache);
-                }
+                // Replace the shared cache with the new one; the previous cache (if any) is
+                // dropped.
+                *cached = Some(new_cache);
             });
 
             let elapsed = start.elapsed();
@@ -427,10 +419,9 @@ where
                     // completed executing a set of transactions
                     self.send_multi_proof_targets(proof_targets);
                 }
-                PrewarmTaskEvent::Terminate { execution_outcome, valid_block_rx } => {
+                PrewarmTaskEvent::Terminate { execution_outcome } => {
                     trace!(target: "engine::tree::payload_processor::prewarm", "Received termination signal");
-                    final_execution_outcome =
-                        Some(execution_outcome.map(|outcome| (outcome, valid_block_rx)));
+                    final_execution_outcome = Some(execution_outcome);
 
                     if finished_execution {
                         // all tasks are done, we can exit, which will save caches and exit
@@ -455,8 +446,8 @@ where
         debug!(target: "engine::tree::payload_processor::prewarm", "Completed prewarm execution");
 
         // save caches and finish using the shared ExecutionOutcome
-        if let Some(Some((execution_outcome, valid_block_rx))) = final_execution_outcome {
-            self.save_cache(execution_outcome, valid_block_rx);
+        if let Some(Some(execution_outcome)) = final_execution_outcome {
+            self.save_cache(execution_outcome);
         }
     }
 }
@@ -576,14 +567,9 @@ where
                 .entered();
             txs.recv()
         } {
-            let enter = debug_span!(
-                target: "engine::tree::payload_processor::prewarm",
-                "prewarm tx",
-                index,
-                tx_hash = %tx.tx().tx_hash(),
-                is_success = tracing::field::Empty,
-            )
-            .entered();
+            let enter =
+                debug_span!(target: "engine::tree::payload_processor::prewarm", "prewarm tx", index, tx_hash=%tx.tx().tx_hash())
+                    .entered();
 
             // create the tx env
             let start = Instant::now();
@@ -824,12 +810,7 @@ pub(super) enum PrewarmTaskEvent<R> {
     Terminate {
         /// The final execution outcome. Using `Arc` allows sharing with the main execution
         /// path without cloning the expensive `BundleState`.
-        execution_outcome: Option<Arc<BlockExecutionOutput<R>>>,
-        /// Receiver for the block validation result.
-        ///
-        /// Cache saving is racing the state root validation. We optimistically construct the
-        /// updated cache but only save it once we know the block is valid.
-        valid_block_rx: mpsc::Receiver<()>,
+        execution_outcome: Option<Arc<ExecutionOutcome<R>>>,
     },
     /// The outcome of a pre-warm task
     Outcome {

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

@@ -1,250 +0,0 @@
-//! Receipt root computation in a background task.
-//!
-//! This module provides a streaming receipt root builder that computes the receipt trie root
-//! in a background thread. Receipts are sent via a channel with their index, and for each
-//! receipt received, the builder incrementally flushes leaves to the underlying
-//! [`OrderedTrieRootEncodedBuilder`] when possible. When the channel closes, the task returns the
-//! computed root.
-
-use alloy_eips::Encodable2718;
-use alloy_primitives::{Bloom, B256};
-use crossbeam_channel::Receiver;
-use reth_primitives_traits::Receipt;
-use reth_trie_common::ordered_root::OrderedTrieRootEncodedBuilder;
-use tokio::sync::oneshot;
-
-/// Receipt with index, ready to be sent to the background task for encoding and trie building.
-#[derive(Debug, Clone)]
-pub struct IndexedReceipt<R> {
-    /// The transaction index within the block.
-    pub index: usize,
-    /// The receipt.
-    pub receipt: R,
-}
-
-impl<R> IndexedReceipt<R> {
-    /// Creates a new indexed receipt.
-    #[inline]
-    pub const fn new(index: usize, receipt: R) -> Self {
-        Self { index, receipt }
-    }
-}
-
-/// Handle for running the receipt root computation in a background task.
-///
-/// This struct holds the channels needed to receive receipts and send the result.
-/// Use [`Self::run`] to execute the computation (typically in a spawned blocking task).
-#[derive(Debug)]
-pub struct ReceiptRootTaskHandle<R> {
-    /// Receiver for indexed receipts.
-    receipt_rx: Receiver<IndexedReceipt<R>>,
-    /// Sender for the computed result.
-    result_tx: oneshot::Sender<(B256, Bloom)>,
-}
-
-impl<R: Receipt> ReceiptRootTaskHandle<R> {
-    /// Creates a new handle from the receipt receiver and result sender channels.
-    pub const fn new(
-        receipt_rx: Receiver<IndexedReceipt<R>>,
-        result_tx: oneshot::Sender<(B256, Bloom)>,
-    ) -> Self {
-        Self { receipt_rx, result_tx }
-    }
-
-    /// Runs the receipt root computation, consuming the handle.
-    ///
-    /// This method receives indexed receipts from the channel, encodes them,
-    /// and builds the trie incrementally. When all receipts have been received
-    /// (channel closed), it sends the result through the oneshot channel.
-    ///
-    /// This is designed to be called inside a blocking task (e.g., via
-    /// `executor.spawn_blocking(move || handle.run(receipts_len))`).
-    ///
-    /// # Arguments
-    ///
-    /// * `receipts_len` - The total number of receipts expected. This is needed to correctly order
-    ///   the trie keys according to RLP encoding rules.
-    ///
-    /// # Panics
-    ///
-    /// Panics if the number of receipts received doesn't match `receipts_len`.
-    pub fn run(self, receipts_len: usize) {
-        let mut builder = OrderedTrieRootEncodedBuilder::new(receipts_len);
-        let mut aggregated_bloom = Bloom::ZERO;
-        let mut encode_buf = Vec::new();
-
-        for indexed_receipt in self.receipt_rx {
-            let receipt_with_bloom = indexed_receipt.receipt.with_bloom_ref();
-
-            encode_buf.clear();
-            receipt_with_bloom.encode_2718(&mut encode_buf);
-
-            aggregated_bloom |= *receipt_with_bloom.bloom_ref();
-            builder.push_unchecked(indexed_receipt.index, &encode_buf);
-        }
-
-        let root = builder.finalize().expect("receipt root builder incomplete");
-        let _ = self.result_tx.send((root, aggregated_bloom));
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use alloy_consensus::{proofs::calculate_receipt_root, TxReceipt};
-    use alloy_primitives::{b256, hex, Address, Bytes, Log};
-    use crossbeam_channel::bounded;
-    use reth_ethereum_primitives::{Receipt, TxType};
-
-    #[tokio::test]
-    async fn test_receipt_root_task_empty() {
-        let (_tx, rx) = bounded::<IndexedReceipt<Receipt>>(1);
-        let (result_tx, result_rx) = oneshot::channel();
-        drop(_tx);
-
-        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
-        tokio::task::spawn_blocking(move || handle.run(0)).await.unwrap();
-
-        let (root, bloom) = result_rx.await.unwrap();
-
-        // Empty trie root
-        assert_eq!(root, reth_trie_common::EMPTY_ROOT_HASH);
-        assert_eq!(bloom, Bloom::ZERO);
-    }
-
-    #[tokio::test]
-    async fn test_receipt_root_task_single_receipt() {
-        let receipts: Vec<Receipt> = vec![Receipt::default()];
-
-        let (tx, rx) = bounded(1);
-        let (result_tx, result_rx) = oneshot::channel();
-        let receipts_len = receipts.len();
-
-        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
-        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
-
-        for (i, receipt) in receipts.clone().into_iter().enumerate() {
-            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
-        }
-        drop(tx);
-
-        join_handle.await.unwrap();
-        let (root, _bloom) = result_rx.await.unwrap();
-
-        // Verify against the standard calculation
-        let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
-        let expected_root = calculate_receipt_root(&receipts_with_bloom);
-
-        assert_eq!(root, expected_root);
-    }
-
-    #[tokio::test]
-    async fn test_receipt_root_task_multiple_receipts() {
-        let receipts: Vec<Receipt> = vec![Receipt::default(); 5];
-
-        let (tx, rx) = bounded(4);
-        let (result_tx, result_rx) = oneshot::channel();
-        let receipts_len = receipts.len();
-
-        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
-        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
-
-        for (i, receipt) in receipts.into_iter().enumerate() {
-            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
-        }
-        drop(tx);
-
-        join_handle.await.unwrap();
-        let (root, bloom) = result_rx.await.unwrap();
-
-        // Verify against expected values from existing test
-        assert_eq!(
-            root,
-            b256!("0x61353b4fb714dc1fccacbf7eafc4273e62f3d1eed716fe41b2a0cd2e12c63ebc")
-        );
-        assert_eq!(
-            bloom,
-            Bloom::from(hex!("00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"))
-        );
-    }
-
-    #[tokio::test]
-    async fn test_receipt_root_matches_standard_calculation() {
-        // Create some receipts with actual data
-        let receipts = vec![
-            Receipt {
-                tx_type: TxType::Legacy,
-                cumulative_gas_used: 21000,
-                success: true,
-                logs: vec![],
-            },
-            Receipt {
-                tx_type: TxType::Eip1559,
-                cumulative_gas_used: 42000,
-                success: true,
-                logs: vec![Log {
-                    address: Address::ZERO,
-                    data: alloy_primitives::LogData::new_unchecked(vec![B256::ZERO], Bytes::new()),
-                }],
-            },
-            Receipt {
-                tx_type: TxType::Eip2930,
-                cumulative_gas_used: 63000,
-                success: false,
-                logs: vec![],
-            },
-        ];
-
-        // Calculate expected values first (before we move receipts)
-        let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
-        let expected_root = calculate_receipt_root(&receipts_with_bloom);
-        let expected_bloom =
-            receipts_with_bloom.iter().fold(Bloom::ZERO, |bloom, r| bloom | r.bloom_ref());
-
-        // Calculate using the task
-        let (tx, rx) = bounded(4);
-        let (result_tx, result_rx) = oneshot::channel();
-        let receipts_len = receipts.len();
-
-        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
-        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
-
-        for (i, receipt) in receipts.into_iter().enumerate() {
-            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
-        }
-        drop(tx);
-
-        join_handle.await.unwrap();
-        let (task_root, task_bloom) = result_rx.await.unwrap();
-
-        assert_eq!(task_root, expected_root);
-        assert_eq!(task_bloom, expected_bloom);
-    }
-
-    #[tokio::test]
-    async fn test_receipt_root_task_out_of_order() {
-        let receipts: Vec<Receipt> = vec![Receipt::default(); 5];
-
-        // Calculate expected values first (before we move receipts)
-        let receipts_with_bloom: Vec<_> = receipts.iter().map(|r| r.with_bloom_ref()).collect();
-        let expected_root = calculate_receipt_root(&receipts_with_bloom);
-
-        let (tx, rx) = bounded(4);
-        let (result_tx, result_rx) = oneshot::channel();
-        let receipts_len = receipts.len();
-
-        let handle = ReceiptRootTaskHandle::new(rx, result_tx);
-        let join_handle = tokio::task::spawn_blocking(move || handle.run(receipts_len));
-
-        // Send in reverse order to test out-of-order handling
-        for (i, receipt) in receipts.into_iter().enumerate().rev() {
-            tx.send(IndexedReceipt::new(i, receipt)).unwrap();
-        }
-        drop(tx);
-
-        join_handle.await.unwrap();
-        let (root, _bloom) = result_rx.await.unwrap();
-
-        assert_eq!(root, expected_root);
-    }
-}

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

@@ -1,5 +1,11 @@
 //! Types and traits for validating blocks and payloads.
 
+/// Threshold for switching from `extend_ref` loop to `merge_batch` in `merge_overlay_trie_input`.
+///
+/// Benchmarked crossover: `extend_ref` wins up to ~64 blocks, `merge_batch` wins beyond.
+/// Using 64 as threshold since they're roughly equal there.
+const MERGE_BATCH_THRESHOLD: usize = 64;
+
 use crate::tree::{
     cached_state::CachedStateProvider,
     error::{InsertBlockError, InsertBlockErrorKind, InsertPayloadError},
@@ -15,11 +21,9 @@ use alloy_eip7928::BlockAccessList;
 use alloy_eips::{eip1898::BlockWithParent, NumHash};
 use alloy_evm::Evm;
 use alloy_primitives::B256;
-
-use crate::tree::payload_processor::receipt_root_task::{IndexedReceipt, ReceiptRootTaskHandle};
 use rayon::prelude::*;
-use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock, LazyOverlay};
-use reth_consensus::{ConsensusError, FullConsensus, ReceiptRootBloom};
+use reth_chain_state::{CanonicalInMemoryState, DeferredTrieData, ExecutedBlock};
+use reth_consensus::{ConsensusError, FullConsensus};
 use reth_engine_primitives::{
     ConfigureEngineEvm, ExecutableTxIterator, ExecutionPayload, InvalidBlockHook, PayloadValidator,
 };
@@ -37,13 +41,15 @@ use reth_primitives_traits::{
 };
 use reth_provider::{
     providers::OverlayStateProviderFactory, BlockExecutionOutput, BlockNumReader, BlockReader,
-    ChangeSetReader, DatabaseProviderFactory, DatabaseProviderROFactory, HashedPostStateProvider,
-    ProviderError, PruneCheckpointReader, StageCheckpointReader, StateProvider,
-    StateProviderFactory, StateReader,
+    ChangeSetReader, DatabaseProviderFactory, DatabaseProviderROFactory, ExecutionOutcome,
+    HashedPostStateProvider, ProviderError, PruneCheckpointReader, StageCheckpointReader,
+    StateProvider, StateProviderFactory, StateReader, TrieReader,
 };
 use reth_revm::db::State;
-use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot};
-use reth_trie_db::ChangesetCache;
+use reth_trie::{
+    updates::{TrieUpdates, TrieUpdatesSorted},
+    HashedPostState, HashedPostStateSorted, StateRoot, TrieInputSorted,
+};
 use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
 use revm_primitives::Address;
 use std::{
@@ -132,8 +138,6 @@ where
     metrics: EngineApiMetrics,
     /// Validator for the payload.
     validator: V,
-    /// Changeset cache for in-memory trie changesets
-    changeset_cache: ChangesetCache,
 }
 
 impl<N, P, Evm, V> BasicEngineValidator<P, Evm, V>
@@ -141,6 +145,7 @@ where
     N: NodePrimitives,
     P: DatabaseProviderFactory<
             Provider: BlockReader
+                          + TrieReader
                           + StageCheckpointReader
                           + PruneCheckpointReader
                           + ChangeSetReader
@@ -164,7 +169,6 @@ where
         validator: V,
         config: TreeConfig,
         invalid_block_hook: Box<dyn InvalidBlockHook<N>>,
-        changeset_cache: ChangesetCache,
     ) -> Self {
         let precompile_cache_map = PrecompileCacheMap::default();
         let payload_processor = PayloadProcessor::new(
@@ -184,7 +188,6 @@ where
             invalid_block_hook,
             metrics: EngineApiMetrics::default(),
             validator,
-            changeset_cache,
         }
     }
 
@@ -369,6 +372,7 @@ where
         }
 
         let parent_hash = input.parent_hash();
+        let block_num_hash = input.num_hash();
 
         trace!(target: "engine::tree::payload_validator", "Fetching block state provider");
         let _enter =
@@ -423,23 +427,13 @@ where
             .map_err(Box::<dyn std::error::Error + Send + Sync>::from))
         .map(Arc::new);
 
-        // Create lazy overlay from ancestors - this doesn't block, allowing execution to start
-        // before the trie data is ready. The overlay will be computed on first access.
-        let (lazy_overlay, anchor_hash) = Self::get_parent_lazy_overlay(parent_hash, ctx.state());
-
-        // Create overlay factory for payload processor (StateRootTask path needs it for
-        // multiproofs)
-        let overlay_factory =
-            OverlayStateProviderFactory::new(self.provider.clone(), self.changeset_cache.clone())
-                .with_block_hash(Some(anchor_hash))
-                .with_lazy_overlay(lazy_overlay);
-
         // Spawn the appropriate processor based on strategy
         let mut handle = ensure_ok!(self.spawn_payload_processor(
             env.clone(),
             txs,
             provider_builder,
-            overlay_factory.clone(),
+            parent_hash,
+            ctx.state(),
             strategy,
             block_access_list,
         ));
@@ -455,44 +449,19 @@ where
             state_provider = Box::new(InstrumentedStateProvider::new(state_provider, "engine"));
         }
 
-        // Execute the block and handle any execution errors.
-        // The receipt root task is spawned before execution and receives receipts incrementally
-        // as transactions complete, allowing parallel computation during execution.
-        let (output, senders, receipt_root_rx) =
-            match self.execute_block(state_provider, env, &input, &mut handle) {
-                Ok(output) => output,
-                Err(err) => return self.handle_execution_error(input, err, &parent_block),
-            };
+        // Execute the block and handle any execution errors
+        let (output, senders) = match self.execute_block(state_provider, env, &input, &mut handle) {
+            Ok(output) => output,
+            Err(err) => return self.handle_execution_error(input, err, &parent_block),
+        };
 
         // After executing the block we can stop prewarming transactions
         handle.stop_prewarming_execution();
 
-        // Create ExecutionOutcome early so we can terminate caching before validation and state
-        // root computation. Using Arc allows sharing with both the caching task and the deferred
-        // trie task without cloning the expensive BundleState.
-        let output = Arc::new(output);
-
-        // Terminate caching task early since execution is complete and caching is no longer
-        // needed. This frees up resources while state root computation continues.
-        let valid_block_tx = handle.terminate_caching(Some(output.clone()));
-
         let block = self.convert_to_block(input)?.with_senders(senders);
 
-        // Wait for the receipt root computation to complete.
-        let receipt_root_bloom = Some(
-            receipt_root_rx
-                .blocking_recv()
-                .expect("receipt root task dropped sender without result"),
-        );
-
         let hashed_state = ensure_ok_post_block!(
-            self.validate_post_execution(
-                &block,
-                &parent_block,
-                &output,
-                &mut ctx,
-                receipt_root_bloom
-            ),
+            self.validate_post_execution(&block, &parent_block, &output, &mut ctx),
             block
         );
 
@@ -525,7 +494,11 @@ where
             }
             StateRootStrategy::Parallel => {
                 debug!(target: "engine::tree::payload_validator", "Using parallel state root algorithm");
-                match self.compute_state_root_parallel(overlay_factory.clone(), &hashed_state) {
+                match self.compute_state_root_parallel(
+                    block.parent_hash(),
+                    &hashed_state,
+                    ctx.state(),
+                ) {
                     Ok(result) => {
                         let elapsed = root_time.elapsed();
                         info!(
@@ -561,7 +534,7 @@ where
             }
 
             let (root, updates) = ensure_ok_post_block!(
-                self.compute_state_root_serial(overlay_factory.clone(), &hashed_state),
+                self.compute_state_root_serial(block.parent_hash(), &hashed_state, ctx.state()),
                 block
             );
             (root, updates, root_time.elapsed())
@@ -591,18 +564,14 @@ where
             .into())
         }
 
-        if let Some(valid_block_tx) = valid_block_tx {
-            let _ = valid_block_tx.send(());
-        }
+        // Create ExecutionOutcome and wrap in Arc for sharing with both the caching task
+        // and the deferred trie task. This avoids cloning the expensive BundleState.
+        let execution_outcome = Arc::new(ExecutionOutcome::from((output, block_num_hash.number)));
 
-        Ok(self.spawn_deferred_trie_task(
-            block,
-            output,
-            &ctx,
-            hashed_state,
-            trie_output,
-            overlay_factory,
-        ))
+        // Terminate prewarming task with the shared execution outcome
+        handle.terminate_caching(Some(Arc::clone(&execution_outcome)));
+
+        Ok(self.spawn_deferred_trie_task(block, execution_outcome, &ctx, hashed_state, trie_output))
     }
 
     /// Return sealed block header from database or in-memory state by hash.
@@ -640,21 +609,13 @@ where
 
     /// Executes a block with the given state provider
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
-    #[expect(clippy::type_complexity)]
     fn execute_block<S, Err, T>(
         &mut self,
         state_provider: S,
         env: ExecutionEnv<Evm>,
         input: &BlockOrPayload<T>,
         handle: &mut PayloadHandle<impl ExecutableTxFor<Evm>, Err, N::Receipt>,
-    ) -> Result<
-        (
-            BlockExecutionOutput<N::Receipt>,
-            Vec<Address>,
-            tokio::sync::oneshot::Receiver<(B256, alloy_primitives::Bloom)>,
-        ),
-        InsertBlockErrorKind,
-    >
+    ) -> Result<(BlockExecutionOutput<N::Receipt>, Vec<Address>), InsertBlockErrorKind>
     where
         S: StateProvider + Send,
         Err: core::error::Error + Send + Sync + 'static,
@@ -663,12 +624,14 @@ where
         Evm: ConfigureEngineEvm<T::ExecutionData, Primitives = N>,
     {
         debug!(target: "engine::tree::payload_validator", "Executing block");
-
         let mut db = State::builder()
             .with_database(StateProviderDatabase::new(state_provider))
             .with_bundle_update()
+            .with_bal_builder() //TODO
             .without_state_clear()
             .build();
+        db.bal_state.bal_index = 0;
+        db.bal_state.bal_builder = Some(revm::state::bal::Bal::new());
 
         let spec_id = *env.evm_env.spec_id();
         let evm = self.evm_config.evm_with_env(&mut db, env.evm_env);
@@ -693,14 +656,6 @@ where
             });
         }
 
-        // Spawn background task to compute receipt root and logs bloom incrementally.
-        // Unbounded channel is used since tx count bounds capacity anyway (max ~30k txs per block).
-        let receipts_len = input.transaction_count();
-        let (receipt_tx, receipt_rx) = crossbeam_channel::unbounded();
-        let (result_tx, result_rx) = tokio::sync::oneshot::channel();
-        let task_handle = ReceiptRootTaskHandle::new(receipt_rx, result_tx);
-        self.payload_processor.executor().spawn_blocking(move || task_handle.run(receipts_len));
-
         let execution_start = Instant::now();
         let state_hook = Box::new(handle.state_hook());
         let (output, senders) = self.metrics.execute_metered(
@@ -708,30 +663,15 @@ where
             handle.iter_transactions().map(|res| res.map_err(BlockExecutionError::other)),
             input.transaction_count(),
             state_hook,
-            |receipts| {
-                // Send the latest receipt to the background task for incremental root computation.
-                // The receipt is cloned here; encoding happens in the background thread.
-                if let Some(receipt) = receipts.last() {
-                    // Infer tx_index from the number of receipts collected so far
-                    let tx_index = receipts.len() - 1;
-                    let _ = receipt_tx.send(IndexedReceipt::new(tx_index, receipt.clone()));
-                }
-            },
         )?;
-        drop(receipt_tx);
-
         let execution_finish = Instant::now();
         let execution_time = execution_finish.duration_since(execution_start);
         debug!(target: "engine::tree::payload_validator", elapsed = ?execution_time, "Executed block");
-        Ok((output, senders, result_rx))
+        Ok((output, senders))
     }
 
     /// Compute state root for the given hashed post state in parallel.
     ///
-    /// Uses an overlay factory which provides the state of the parent block, along with the
-    /// [`HashedPostState`] containing the changes of this block, to compute the state root and
-    /// trie updates for this block.
-    ///
     /// # Returns
     ///
     /// Returns `Ok(_)` if computed successfully.
@@ -739,39 +679,58 @@ where
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
     fn compute_state_root_parallel(
         &self,
-        overlay_factory: OverlayStateProviderFactory<P>,
+        parent_hash: B256,
         hashed_state: &HashedPostState,
+        state: &EngineApiTreeState<N>,
     ) -> Result<(B256, TrieUpdates), ParallelStateRootError> {
-        // The `hashed_state` argument will be taken into account as part of the overlay, but we
+        let (mut input, block_hash) = self.compute_trie_input(parent_hash, state)?;
+
+        // Extend state overlay with current block's sorted state.
+        input.prefix_sets.extend(hashed_state.construct_prefix_sets());
+        let sorted_hashed_state = hashed_state.clone_into_sorted();
+        Arc::make_mut(&mut input.state).extend_ref(&sorted_hashed_state);
+
+        let TrieInputSorted { nodes, state, prefix_sets: prefix_sets_mut } = input;
+
+        let factory = OverlayStateProviderFactory::new(self.provider.clone())
+            .with_block_hash(Some(block_hash))
+            .with_trie_overlay(Some(nodes))
+            .with_hashed_state_overlay(Some(state));
+
+        // The `hashed_state` argument is already taken into account as part of the overlay, but we
         // need to use the prefix sets which were generated from it to indicate to the
         // ParallelStateRoot which parts of the trie need to be recomputed.
-        let prefix_sets = hashed_state.construct_prefix_sets().freeze();
-        let overlay_factory =
-            overlay_factory.with_extended_hashed_state_overlay(hashed_state.clone_into_sorted());
-        ParallelStateRoot::new(overlay_factory, prefix_sets).incremental_root_with_updates()
+        let prefix_sets = prefix_sets_mut.freeze();
+
+        ParallelStateRoot::new(factory, prefix_sets).incremental_root_with_updates()
     }
 
     /// Compute state root for the given hashed post state in serial.
-    ///
-    /// Uses an overlay factory which provides the state of the parent block, along with the
-    /// [`HashedPostState`] containing the changes of this block, to compute the state root and
-    /// trie updates for this block.
     fn compute_state_root_serial(
         &self,
-        overlay_factory: OverlayStateProviderFactory<P>,
+        parent_hash: B256,
         hashed_state: &HashedPostState,
+        state: &EngineApiTreeState<N>,
     ) -> ProviderResult<(B256, TrieUpdates)> {
-        // The `hashed_state` argument will be taken into account as part of the overlay, but we
-        // need to use the prefix sets which were generated from it to indicate to the
-        // StateRoot which parts of the trie need to be recomputed.
-        let prefix_sets = hashed_state.construct_prefix_sets().freeze();
-        let overlay_factory =
-            overlay_factory.with_extended_hashed_state_overlay(hashed_state.clone_into_sorted());
+        let (mut input, block_hash) = self.compute_trie_input(parent_hash, state)?;
 
-        let provider = overlay_factory.database_provider_ro()?;
+        // Extend state overlay with current block's sorted state.
+        input.prefix_sets.extend(hashed_state.construct_prefix_sets());
+        let sorted_hashed_state = hashed_state.clone_into_sorted();
+        Arc::make_mut(&mut input.state).extend_ref(&sorted_hashed_state);
+
+        let TrieInputSorted { nodes, state, .. } = input;
+        let prefix_sets = hashed_state.construct_prefix_sets();
+
+        let factory = OverlayStateProviderFactory::new(self.provider.clone())
+            .with_block_hash(Some(block_hash))
+            .with_trie_overlay(Some(nodes))
+            .with_hashed_state_overlay(Some(state));
+
+        let provider = factory.database_provider_ro()?;
 
         Ok(StateRoot::new(&provider, &provider)
-            .with_prefix_sets(prefix_sets)
+            .with_prefix_sets(prefix_sets.freeze())
             .root_with_updates()?)
     }
 
@@ -781,9 +740,6 @@ where
     /// - parent header validation
     /// - post-execution consensus validation
     /// - state-root based post-execution validation
-    ///
-    /// If `receipt_root_bloom` is provided, it will be used instead of computing the receipt root
-    /// and logs bloom from the receipts.
     #[instrument(level = "debug", target = "engine::tree::payload_validator", skip_all)]
     fn validate_post_execution<T: PayloadTypes<BuiltPayload: BuiltPayload<Primitives = N>>>(
         &self,
@@ -791,7 +747,6 @@ where
         parent_block: &SealedHeader<N::BlockHeader>,
         output: &BlockExecutionOutput<N::Receipt>,
         ctx: &mut TreeCtx<'_, N>,
-        receipt_root_bloom: Option<ReceiptRootBloom>,
     ) -> Result<HashedPostState, InsertBlockErrorKind>
     where
         V: PayloadValidator<T, Block = N::Block>,
@@ -818,9 +773,7 @@ where
         let _enter =
             debug_span!(target: "engine::tree::payload_validator", "validate_block_post_execution")
                 .entered();
-        if let Err(err) =
-            self.consensus.validate_block_post_execution(block, output, receipt_root_bloom)
-        {
+        if let Err(err) = self.consensus.validate_block_post_execution(block, output) {
             // call post-block hook
             self.on_invalid_block(parent_block, block, output, None, ctx.state_mut());
             return Err(err.into())
@@ -860,11 +813,6 @@ where
     ///
     /// The method handles strategy fallbacks if the preferred approach fails, ensuring
     /// block execution always completes with a valid state root.
-    ///
-    /// # Arguments
-    ///
-    /// * `overlay_factory` - Pre-computed overlay factory for multiproof generation
-    ///   (`StateRootTask`)
     #[allow(clippy::too_many_arguments)]
     #[instrument(
         level = "debug",
@@ -877,7 +825,8 @@ where
         env: ExecutionEnv<Evm>,
         txs: T,
         provider_builder: StateProviderBuilder<N, P>,
-        overlay_factory: OverlayStateProviderFactory<P>,
+        parent_hash: B256,
+        state: &EngineApiTreeState<N>,
         strategy: StateRootStrategy,
         block_access_list: Option<Arc<BlockAccessList>>,
     ) -> Result<
@@ -890,14 +839,32 @@ where
     > {
         match strategy {
             StateRootStrategy::StateRootTask => {
+                // Compute trie input
+                let trie_input_start = Instant::now();
+                let (trie_input, block_hash) = self.compute_trie_input(parent_hash, state)?;
+
+                // Create OverlayStateProviderFactory with sorted trie data for multiproofs
+                let TrieInputSorted { nodes, state, .. } = trie_input;
+
+                let multiproof_provider_factory =
+                    OverlayStateProviderFactory::new(self.provider.clone())
+                        .with_block_hash(Some(block_hash))
+                        .with_trie_overlay(Some(nodes))
+                        .with_hashed_state_overlay(Some(state));
+
+                // Record trie input duration including OverlayStateProviderFactory setup
+                self.metrics
+                    .block_validation
+                    .trie_input_duration
+                    .record(trie_input_start.elapsed().as_secs_f64());
+
                 let spawn_start = Instant::now();
 
-                // Use the pre-computed overlay factory for multiproofs
                 let handle = self.payload_processor.spawn(
                     env,
                     txs,
                     provider_builder,
-                    overlay_factory,
+                    multiproof_provider_factory,
                     &self.config,
                     block_access_list,
                 );
@@ -991,36 +958,128 @@ where
         self.invalid_block_hook.on_invalid_block(parent_header, block, output, trie_updates);
     }
 
-    /// Creates a [`LazyOverlay`] for the parent block without blocking.
+    /// Computes [`TrieInputSorted`] for the provided parent hash by combining database state
+    /// with in-memory overlays.
     ///
-    /// Returns a lazy overlay that will compute the trie input on first access, and the anchor
-    /// block hash (the highest persisted ancestor). This allows execution to start immediately
-    /// while the trie input computation is deferred until the overlay is actually needed.
+    /// The goal of this function is to take in-memory blocks and generate a [`TrieInputSorted`]
+    /// that extends from the highest persisted ancestor up through the parent. This enables state
+    /// root computation and proof generation without requiring all blocks to be persisted
+    /// first.
     ///
-    /// If parent is on disk (no in-memory blocks), returns `None` for the lazy overlay.
-    fn get_parent_lazy_overlay(
+    /// It works as follows:
+    /// 1. Collect in-memory overlay blocks using [`crate::tree::TreeState::blocks_by_hash`]. This
+    ///    returns the highest persisted ancestor hash (`block_hash`) and the list of in-memory
+    ///    blocks building on top of it.
+    /// 2. Fast path: If the tip in-memory block's trie input is already anchored to `block_hash`
+    ///    (its `anchor_hash` matches `block_hash`), reuse it directly.
+    /// 3. Slow path: Build a new [`TrieInputSorted`] by aggregating the overlay blocks (from oldest
+    ///    to newest) on top of the database state at `block_hash`.
+    #[instrument(
+        level = "debug",
+        target = "engine::tree::payload_validator",
+        skip_all,
+        fields(parent_hash)
+    )]
+    fn compute_trie_input(
+        &self,
         parent_hash: B256,
         state: &EngineApiTreeState<N>,
-    ) -> (Option<LazyOverlay>, B256) {
-        let (anchor_hash, blocks) =
+    ) -> ProviderResult<(TrieInputSorted, B256)> {
+        let wait_start = Instant::now();
+        let (block_hash, blocks) =
             state.tree_state.blocks_by_hash(parent_hash).unwrap_or_else(|| (parent_hash, vec![]));
 
-        if blocks.is_empty() {
-            debug!(target: "engine::tree::payload_validator", "Parent found on disk, no lazy overlay needed");
-            return (None, anchor_hash);
+        // Fast path: if the tip block's anchor matches the persisted ancestor hash, reuse its
+        // TrieInput. This means the TrieInputSorted already aggregates all in-memory overlays
+        // from that ancestor, so we can avoid re-aggregation.
+        if let Some(tip_block) = blocks.first() {
+            let data = tip_block.trie_data();
+            if let (Some(anchor_hash), Some(trie_input)) =
+                (data.anchor_hash(), data.trie_input().cloned()) &&
+                anchor_hash == block_hash
+            {
+                trace!(target: "engine::tree::payload_validator", %block_hash,"Reusing trie input with matching anchor hash");
+                self.metrics
+                    .block_validation
+                    .deferred_trie_wait_duration
+                    .record(wait_start.elapsed().as_secs_f64());
+                return Ok(((*trie_input).clone(), block_hash));
+            }
         }
 
-        debug!(
-            target: "engine::tree::payload_validator",
-            %anchor_hash,
-            num_blocks = blocks.len(),
-            "Creating lazy overlay for in-memory blocks"
-        );
+        if blocks.is_empty() {
+            debug!(target: "engine::tree::payload_validator", "Parent found on disk");
+        } else {
+            debug!(target: "engine::tree::payload_validator", historical = ?block_hash, blocks = blocks.len(), "Parent found in memory");
+        }
 
-        // Extract deferred trie data handles (non-blocking)
-        let handles: Vec<DeferredTrieData> = blocks.iter().map(|b| b.trie_data_handle()).collect();
+        // Extend with contents of parent in-memory blocks directly in sorted form.
+        let input = Self::merge_overlay_trie_input(&blocks);
 
-        (Some(LazyOverlay::new(anchor_hash, handles)), anchor_hash)
+        self.metrics
+            .block_validation
+            .deferred_trie_wait_duration
+            .record(wait_start.elapsed().as_secs_f64());
+        Ok((input, block_hash))
+    }
+
+    /// Aggregates in-memory blocks into a single [`TrieInputSorted`] by combining their
+    /// state changes.
+    ///
+    /// The input `blocks` vector is ordered newest -> oldest (see `TreeState::blocks_by_hash`).
+    ///
+    /// Uses `extend_ref` loop for small k, k-way `merge_batch` for large k.
+    /// See [`MERGE_BATCH_THRESHOLD`] for crossover point.
+    fn merge_overlay_trie_input(blocks: &[ExecutedBlock<N>]) -> TrieInputSorted {
+        if blocks.is_empty() {
+            return TrieInputSorted::default();
+        }
+
+        // Single block: return Arc directly without cloning
+        if blocks.len() == 1 {
+            let data = blocks[0].trie_data();
+            return TrieInputSorted {
+                state: Arc::clone(&data.hashed_state),
+                nodes: Arc::clone(&data.trie_updates),
+                prefix_sets: Default::default(),
+            };
+        }
+
+        if blocks.len() < MERGE_BATCH_THRESHOLD {
+            // Small k: extend_ref loop is faster
+            // Iterate oldest->newest so newer values override older ones
+            let mut blocks_iter = blocks.iter().rev();
+            let first = blocks_iter.next().expect("blocks is non-empty");
+            let data = first.trie_data();
+
+            let mut state = Arc::clone(&data.hashed_state);
+            let mut nodes = Arc::clone(&data.trie_updates);
+            let state_mut = Arc::make_mut(&mut state);
+            let nodes_mut = Arc::make_mut(&mut nodes);
+
+            for block in blocks_iter {
+                let data = block.trie_data();
+                state_mut.extend_ref(data.hashed_state.as_ref());
+                nodes_mut.extend_ref(data.trie_updates.as_ref());
+            }
+
+            TrieInputSorted { state, nodes, prefix_sets: Default::default() }
+        } else {
+            // Large k: merge_batch is faster (O(n log k) via k-way merge)
+            let trie_data: Vec<_> = blocks.iter().map(|b| b.trie_data()).collect();
+
+            let merged_state = HashedPostStateSorted::merge_batch(
+                trie_data.iter().map(|d| d.hashed_state.as_ref()),
+            );
+            let merged_nodes =
+                TrieUpdatesSorted::merge_batch(trie_data.iter().map(|d| d.trie_updates.as_ref()));
+
+            TrieInputSorted {
+                state: Arc::new(merged_state),
+                nodes: Arc::new(merged_nodes),
+                prefix_sets: Default::default(),
+            }
+        }
     }
 
     /// Spawns a background task to compute and sort trie data for the executed block.
@@ -1042,11 +1101,10 @@ where
     fn spawn_deferred_trie_task(
         &self,
         block: RecoveredBlock<N::Block>,
-        execution_outcome: Arc<BlockExecutionOutput<N::Receipt>>,
+        execution_outcome: Arc<ExecutionOutcome<N::Receipt>>,
         ctx: &TreeCtx<'_, N>,
         hashed_state: HashedPostState,
         trie_output: TrieUpdates,
-        overlay_factory: OverlayStateProviderFactory<P>,
     ) -> ExecutedBlock<N> {
         // Capture parent hash and ancestor overlays for deferred trie input construction.
         let (anchor_hash, overlay_blocks) = ctx
@@ -1070,21 +1128,9 @@ where
         let deferred_handle_task = deferred_trie_data.clone();
         let block_validation_metrics = self.metrics.block_validation.clone();
 
-        // Capture block info and cache handle for changeset computation
-        let block_hash = block.hash();
-        let block_number = block.number();
-        let changeset_cache = self.changeset_cache.clone();
-
         // Spawn background task to compute trie data. Calling `wait_cloned` will compute from
         // the stored inputs and cache the result, so subsequent calls return immediately.
         let compute_trie_input_task = move || {
-            let _span = debug_span!(
-                target: "engine::tree::payload_validator",
-                "compute_trie_input_task",
-                block_number
-            )
-            .entered();
-
             let result = panic::catch_unwind(AssertUnwindSafe(|| {
                 let compute_start = Instant::now();
                 let computed = deferred_handle_task.wait_cloned();
@@ -1107,40 +1153,6 @@ where
                         .anchored_overlay_hashed_state_size
                         .record(anchored.trie_input.state.total_len() as f64);
                 }
-
-                // Compute and cache changesets using the computed trie_updates
-                let changeset_start = Instant::now();
-
-                // Get a provider from the overlay factory for trie cursor access
-                let changeset_result =
-                    overlay_factory.database_provider_ro().and_then(|provider| {
-                        reth_trie::changesets::compute_trie_changesets(
-                            &provider,
-                            &computed.trie_updates,
-                        )
-                        .map_err(ProviderError::Database)
-                    });
-
-                match changeset_result {
-                    Ok(changesets) => {
-                        debug!(
-                            target: "engine::tree::changeset",
-                            ?block_number,
-                            elapsed = ?changeset_start.elapsed(),
-                            "Computed and caching changesets"
-                        );
-
-                        changeset_cache.insert(block_hash, block_number, Arc::new(changesets));
-                    }
-                    Err(e) => {
-                        warn!(
-                            target: "engine::tree::changeset",
-                            ?block_number,
-                            ?e,
-                            "Failed to compute changesets in deferred trie task"
-                        );
-                    }
-                }
             }));
 
             if result.is_err() {
@@ -1237,6 +1249,7 @@ impl<N, Types, P, Evm, V> EngineValidator<Types> for BasicEngineValidator<P, Evm
 where
     P: DatabaseProviderFactory<
             Provider: BlockReader
+                          + TrieReader
                           + StageCheckpointReader
                           + PruneCheckpointReader
                           + ChangeSetReader
@@ -1289,7 +1302,7 @@ where
     fn on_inserted_executed_block(&self, block: ExecutedBlock<N>) {
         self.payload_processor.on_inserted_executed_block(
             block.recovered_block.block_with_parent(),
-            &block.execution_output.state,
+            block.execution_output.state(),
         );
     }
 }

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

@@ -233,9 +233,9 @@ mod tests {
         let dyn_precompile: DynPrecompile = (|_input: PrecompileInput<'_>| -> PrecompileResult {
             Ok(PrecompileOutput {
                 gas_used: 0,
-                gas_refunded: 0,
                 bytes: Bytes::default(),
                 reverted: false,
+                gas_refunded: 0,
             })
         })
         .into();
@@ -245,9 +245,9 @@ mod tests {
 
         let output = PrecompileOutput {
             gas_used: 50,
-            gas_refunded: 0,
             bytes: alloy_primitives::Bytes::copy_from_slice(b"cached_result"),
             reverted: false,
+            gas_refunded: 0,
         };
 
         let input = b"test_input";
@@ -277,9 +277,9 @@ mod tests {
 
                 Ok(PrecompileOutput {
                     gas_used: 5000,
-                    gas_refunded: 0,
                     bytes: alloy_primitives::Bytes::copy_from_slice(b"output_from_precompile_1"),
                     reverted: false,
+                    gas_refunded: 0,
                 })
             }
         })
@@ -292,9 +292,9 @@ mod tests {
 
                 Ok(PrecompileOutput {
                     gas_used: 7000,
-                    gas_refunded: 0,
                     bytes: alloy_primitives::Bytes::copy_from_slice(b"output_from_precompile_2"),
                     reverted: false,
+                    gas_refunded: 0,
                 })
             }
         })

crates/engine/tree/src/tree/tests.rs

@@ -7,7 +7,6 @@ use crate::{
         PersistTarget, TreeConfig,
     },
 };
-use reth_trie_db::ChangesetCache;
 
 use alloy_eips::eip1898::BlockWithParent;
 use alloy_primitives::{
@@ -27,7 +26,7 @@ use reth_ethereum_engine_primitives::EthEngineTypes;
 use reth_ethereum_primitives::{Block, EthPrimitives};
 use reth_evm_ethereum::MockEvmConfig;
 use reth_primitives_traits::Block as _;
-use reth_provider::test_utils::MockEthProvider;
+use reth_provider::{test_utils::MockEthProvider, ExecutionOutcome};
 use std::{
     collections::BTreeMap,
     str::FromStr,
@@ -193,7 +192,6 @@ impl TestHarness {
         let payload_builder = PayloadBuilderHandle::new(to_payload_service);
 
         let evm_config = MockEvmConfig::default();
-        let changeset_cache = ChangesetCache::new();
         let engine_validator = BasicEngineValidator::new(
             provider.clone(),
             consensus.clone(),
@@ -201,7 +199,6 @@ impl TestHarness {
             payload_validator,
             TreeConfig::default(),
             Box::new(NoopInvalidBlockHook::default()),
-            changeset_cache.clone(),
         );
 
         let tree = EngineApiTreeHandler::new(
@@ -218,7 +215,6 @@ impl TestHarness {
             TreeConfig::default().with_legacy_state_root(false).with_has_enough_parallelism(true),
             EngineApiKind::Ethereum,
             evm_config,
-            changeset_cache,
         );
 
         let block_builder = TestBlockBuilder::default().with_chain_spec((*chain_spec).clone());
@@ -392,7 +388,6 @@ impl ValidatorTestHarness {
         let provider = harness.provider.clone();
         let payload_validator = MockEngineValidator;
         let evm_config = MockEvmConfig::default();
-        let changeset_cache = ChangesetCache::new();
 
         let validator = BasicEngineValidator::new(
             provider,
@@ -401,7 +396,6 @@ impl ValidatorTestHarness {
             payload_validator,
             TreeConfig::default(),
             Box::new(NoopInvalidBlockHook::default()),
-            changeset_cache,
         );
 
         Self { harness, validator, metrics: TestMetrics::default() }
@@ -838,7 +832,7 @@ fn test_tree_state_on_new_head_deep_fork() {
     for block in &chain_a {
         test_harness.tree.state.tree_state.insert_executed(ExecutedBlock::new(
             Arc::new(block.clone()),
-            Arc::new(BlockExecutionOutput::default()),
+            Arc::new(ExecutionOutcome::default()),
             empty_trie_data(),
         ));
     }
@@ -847,7 +841,7 @@ fn test_tree_state_on_new_head_deep_fork() {
     for block in &chain_b {
         test_harness.tree.state.tree_state.insert_executed(ExecutedBlock::new(
             Arc::new(block.clone()),
-            Arc::new(BlockExecutionOutput::default()),
+            Arc::new(ExecutionOutcome::default()),
             empty_trie_data(),
         ));
     }
@@ -1008,15 +1002,6 @@ async fn test_engine_tree_live_sync_transition_required_blocks_requested() {
         _ => panic!("Unexpected event: {event:#?}"),
     }
 
-    // After backfill completes with head not buffered, we also request head download
-    let event = test_harness.from_tree_rx.recv().await.unwrap();
-    match event {
-        EngineApiEvent::Download(DownloadRequest::BlockSet(hash_set)) => {
-            assert_eq!(hash_set, HashSet::from_iter([main_chain_last_hash]));
-        }
-        _ => panic!("Unexpected event: {event:#?}"),
-    }
-
     let _ = test_harness
         .tree
         .on_engine_message(FromEngine::DownloadedBlocks(vec![main_chain

crates/engine/util/Cargo.toml

@@ -27,6 +27,9 @@ reth-payload-primitives.workspace = true
 alloy-rpc-types-engine.workspace = true
 alloy-consensus.workspace = true
 
+# revm
+revm.workspace = true
+
 # async
 tokio = { workspace = true, default-features = false }
 tokio-util.workspace = true

crates/engine/util/src/reorg.rs

@@ -283,8 +283,12 @@ where
     let mut state = State::builder()
         .with_database_ref(StateProviderDatabase::new(&state_provider))
         .with_bundle_update()
+        .with_bal_builder()
         .build();
 
+    state.bal_state.bal_index = 0;
+    state.bal_state.bal_builder = Some(revm::state::bal::Bal::new());
+
     let ctx = evm_config.context_for_block(&reorg_target).map_err(RethError::other)?;
     let evm = evm_config.evm_for_block(&mut state, &reorg_target).map_err(RethError::other)?;
     let mut builder = evm_config.create_block_builder(evm, &reorg_target_parent, ctx);

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

@@ -39,6 +39,7 @@
 //!     transactions: vec![Bytes::from(vec![1, 2, 3])],
 //!     ommers: vec![],
 //!     withdrawals: None,
+//!     block_access_list: None,
 //! };
 //! // Compress the body: rlp encoding and snappy compression
 //! let compressed_body = CompressedBody::from_body(&body)?;
@@ -581,8 +582,12 @@ mod tests {
 
     #[test]
     fn test_block_body_conversion() {
-        let block_body: BlockBody<Bytes> =
-            BlockBody { transactions: vec![], ommers: vec![], withdrawals: None };
+        let block_body: BlockBody<Bytes> = BlockBody {
+            transactions: vec![],
+            ommers: vec![],
+            withdrawals: None,
+            block_access_list: None,
+        };
 
         let compressed_body = CompressedBody::from_body(&block_body).unwrap();
 
@@ -637,7 +642,8 @@ mod tests {
 
         let withdrawals = Some(Withdrawals(vec![]));
 
-        let block_body = BlockBody { transactions, ommers: vec![], withdrawals };
+        let block_body =
+            BlockBody { transactions, ommers: vec![], withdrawals, block_access_list: None };
 
         let block = Block::new(header, block_body);
 

crates/era/src/test_utils.rs

@@ -34,6 +34,7 @@ pub(crate) fn create_header() -> Header {
         excess_blob_gas: None,
         parent_beacon_block_root: None,
         requests_hash: None,
+        block_access_list_hash: None,
     }
 }
 
@@ -138,6 +139,7 @@ pub(crate) fn create_test_block_with_compressed_data(number: BlockNumber) -> Blo
         excess_blob_gas: None,
         parent_beacon_block_root: None,
         requests_hash: None,
+        block_access_list_hash: None,
     };
 
     // Create test body
@@ -145,6 +147,7 @@ pub(crate) fn create_test_block_with_compressed_data(number: BlockNumber) -> Blo
         transactions: vec![Bytes::from(vec![(number % 256) as u8; 10])],
         ommers: vec![],
         withdrawals: Some(Withdrawals(vec![])),
+        block_access_list: None,
     };
 
     // Create test receipt list with bloom

crates/ethereum/consensus/Cargo.toml

@@ -22,6 +22,7 @@ reth-consensus.workspace = true
 alloy-eips.workspace = true
 alloy-primitives.workspace = true
 alloy-consensus.workspace = true
+alloy-rlp.workspace = true
 
 tracing.workspace = true
 
@@ -38,6 +39,7 @@ std = [
     "reth-execution-types/std",
     "reth-primitives-traits/std",
     "tracing/std",
+    "alloy-rlp/std",
 ]
 
 [dev-dependencies]

crates/ethereum/consensus/src/lib.rs

@@ -15,7 +15,7 @@ use alloc::{fmt::Debug, sync::Arc};
 use alloy_consensus::{constants::MAXIMUM_EXTRA_DATA_SIZE, EMPTY_OMMER_ROOT_HASH};
 use alloy_eips::eip7840::BlobParams;
 use reth_chainspec::{EthChainSpec, EthereumHardforks};
-use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
+use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
 use reth_consensus_common::validation::{
     validate_4844_header_standalone, validate_against_parent_4844,
     validate_against_parent_eip1559_base_fee, validate_against_parent_gas_limit,
@@ -74,14 +74,13 @@ where
         &self,
         block: &RecoveredBlock<N::Block>,
         result: &BlockExecutionResult<N::Receipt>,
-        receipt_root_bloom: Option<ReceiptRootBloom>,
     ) -> Result<(), ConsensusError> {
         validate_block_post_execution(
             block,
             &self.chain_spec,
             &result.receipts,
             &result.requests,
-            receipt_root_bloom,
+            &result.block_access_list,
         )
     }
 }
@@ -181,6 +180,15 @@ where
         } else if header.requests_hash().is_some() {
             return Err(ConsensusError::RequestsHashUnexpected)
         }
+        // if self.chain_spec.is_amsterdam_active_at_timestamp(header.timestamp()) &&
+        //     header.block_access_list_hash().is_none()
+        // {
+        //     return Err(ConsensusError::BlockAccessListHashMissing)
+        // } else if !self.chain_spec.is_amsterdam_active_at_timestamp(header.timestamp()) &&
+        //     header.block_access_list_hash().is_some()
+        // {
+        //     return Err(ConsensusError::BlockAccessListHashUnexpected)
+        // }
 
         Ok(())
     }

crates/ethereum/consensus/src/validation.rs

@@ -1,26 +1,23 @@
 use alloc::vec::Vec;
 use alloy_consensus::{proofs::calculate_receipt_root, BlockHeader, TxReceipt};
-use alloy_eips::{eip7685::Requests, Encodable2718};
+use alloy_eips::{eip7685::Requests, eip7928::BlockAccessList, Encodable2718};
 use alloy_primitives::{Bloom, Bytes, B256};
 use reth_chainspec::EthereumHardforks;
 use reth_consensus::ConsensusError;
 use reth_primitives_traits::{
-    receipt::gas_spent_by_transactions, Block, GotExpected, Receipt, RecoveredBlock,
+    receipt::gas_spent_by_transactions, Block, BlockBody, GotExpected, Receipt, RecoveredBlock,
 };
 
 /// Validate a block with regard to execution results:
 ///
 /// - Compares the receipts root in the block header to the block body
 /// - Compares the gas used in the block header to the actual gas usage after execution
-///
-/// If `receipt_root_bloom` is provided, the pre-computed receipt root and logs bloom are used
-/// instead of computing them from the receipts.
 pub fn validate_block_post_execution<B, R, ChainSpec>(
     block: &RecoveredBlock<B>,
     chain_spec: &ChainSpec,
     receipts: &[R],
     requests: &Requests,
-    receipt_root_bloom: Option<(B256, Bloom)>,
+    block_access_list: &Option<BlockAccessList>,
 ) -> Result<(), ConsensusError>
 where
     B: Block,
@@ -41,26 +38,19 @@ where
     // operation as hashing that is required for state root got calculated in every
     // transaction This was replaced with is_success flag.
     // See more about EIP here: https://eips.ethereum.org/EIPS/eip-658
-    if chain_spec.is_byzantium_active_at_block(block.header().number()) {
-        let result = if let Some((receipts_root, logs_bloom)) = receipt_root_bloom {
-            compare_receipts_root_and_logs_bloom(
-                receipts_root,
-                logs_bloom,
-                block.header().receipts_root(),
-                block.header().logs_bloom(),
-            )
-        } else {
-            verify_receipts(block.header().receipts_root(), block.header().logs_bloom(), receipts)
-        };
-
-        if let Err(error) = result {
-            let receipts = receipts
-                .iter()
-                .map(|r| Bytes::from(r.with_bloom_ref().encoded_2718()))
-                .collect::<Vec<_>>();
-            tracing::debug!(%error, ?receipts, "receipts verification failed");
-            return Err(error)
-        }
+    if chain_spec.is_byzantium_active_at_block(block.header().number()) &&
+        let Err(error) = verify_receipts(
+            block.header().receipts_root(),
+            block.header().logs_bloom(),
+            receipts,
+        )
+    {
+        let receipts = receipts
+            .iter()
+            .map(|r| Bytes::from(r.with_bloom_ref().encoded_2718()))
+            .collect::<Vec<_>>();
+        tracing::debug!(%error, ?receipts, "receipts verification failed");
+        return Err(error)
     }
 
     // Validate that the header requests hash matches the calculated requests hash
@@ -76,6 +66,33 @@ where
         }
     }
 
+    // Validate bal hash matches the calculated hash
+    if chain_spec.is_amsterdam_active_at_timestamp(block.header().timestamp()) {
+        let Some(header_block_access_list_hash) = block.header().block_access_list_hash() else {
+            return Err(ConsensusError::BlockAccessListHashMissing)
+        };
+        if let Some(bal) = block_access_list {
+            let bal_hash = alloy_primitives::keccak256(alloy_rlp::encode(bal));
+            let block_bal = block.body().block_access_list();
+            tracing::debug!("Block Bal :{:?}", block_bal);
+            if let Some(body_bal) = block_bal {
+                if body_bal.is_empty() {
+                    tracing::debug!("Hit Empty BAL : Block is {:?}", block);
+                }
+                verify_bal(body_bal, bal)?;
+            }
+
+            if bal_hash != header_block_access_list_hash {
+                tracing::debug!(
+                    ?bal_hash,
+                    ?header_block_access_list_hash,
+                    "block access list hash mismatch"
+                );
+                return Err(ConsensusError::InvalidBalHash);
+            }
+        }
+    }
+
     Ok(())
 }
 
@@ -124,6 +141,47 @@ fn compare_receipts_root_and_logs_bloom(
     Ok(())
 }
 
+/// Validates that the block access list in the body matches the expected block access list.
+fn verify_bal(
+    body_bal: &BlockAccessList,
+    expected_bal: &BlockAccessList,
+) -> Result<(), ConsensusError> {
+    if body_bal == expected_bal {
+        return Ok(());
+    }
+
+    // Extract addresses
+    let body_addrs: Vec<_> = body_bal.iter().map(|a| a.address).collect();
+    let expected_addrs: Vec<_> = expected_bal.iter().map(|a| a.address).collect();
+
+    // Missing accounts (expected but not found in body)
+    for addr in &expected_addrs {
+        if !body_addrs.contains(addr) {
+            tracing::debug!("Missing acc : computed bal {:?},body bal{:?}", expected_bal, body_bal);
+            tracing::debug!("Missing Address: {:?}", addr);
+            return Err(ConsensusError::InvalidBalMissingAccount);
+        }
+    }
+
+    // Extra accounts (body has accounts not in expected)
+    for addr in &body_addrs {
+        if !expected_addrs.contains(addr) {
+            tracing::debug!("Extra acc : computed bal {:?},body bal{:?}", expected_bal, body_bal);
+            tracing::debug!("Extra Address: {:?}", addr);
+            return Err(ConsensusError::InvalidBalExtraAccount);
+        }
+    }
+
+    tracing::debug!(
+        ?expected_bal,
+        ?body_bal,
+        "block access list in body does not match the provided block access list"
+    );
+
+    // Fallback: mismatched access lists
+    Err(ConsensusError::InvalidBlockAccessList)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ethereum/engine-primitives/src/lib.rs

@@ -17,7 +17,9 @@ pub use payload::{payload_id, BlobSidecars, EthBuiltPayload, EthPayloadBuilderAt
 mod error;
 pub use error::*;
 
-use alloy_rpc_types_engine::{ExecutionData, ExecutionPayload, ExecutionPayloadEnvelopeV5};
+use alloy_rpc_types_engine::{
+    ExecutionData, ExecutionPayload, ExecutionPayloadEnvelopeV5, ExecutionPayloadEnvelopeV6,
+};
 pub use alloy_rpc_types_engine::{
     ExecutionPayloadEnvelopeV2, ExecutionPayloadEnvelopeV3, ExecutionPayloadEnvelopeV4,
     ExecutionPayloadV1, PayloadAttributes as EthPayloadAttributes,
@@ -66,13 +68,15 @@ where
         + TryInto<ExecutionPayloadEnvelopeV2>
         + TryInto<ExecutionPayloadEnvelopeV3>
         + TryInto<ExecutionPayloadEnvelopeV4>
-        + TryInto<ExecutionPayloadEnvelopeV5>,
+        + TryInto<ExecutionPayloadEnvelopeV5>
+        + TryInto<ExecutionPayloadEnvelopeV6>,
 {
     type ExecutionPayloadEnvelopeV1 = ExecutionPayloadV1;
     type ExecutionPayloadEnvelopeV2 = ExecutionPayloadEnvelopeV2;
     type ExecutionPayloadEnvelopeV3 = ExecutionPayloadEnvelopeV3;
     type ExecutionPayloadEnvelopeV4 = ExecutionPayloadEnvelopeV4;
     type ExecutionPayloadEnvelopeV5 = ExecutionPayloadEnvelopeV5;
+    type ExecutionPayloadEnvelopeV6 = ExecutionPayloadEnvelopeV6;
 }
 
 /// A default payload type for [`EthEngineTypes`]

crates/ethereum/engine-primitives/src/payload.rs

@@ -11,8 +11,9 @@ use alloy_primitives::{Address, B256, U256};
 use alloy_rlp::Encodable;
 use alloy_rpc_types_engine::{
     BlobsBundleV1, BlobsBundleV2, ExecutionPayloadEnvelopeV2, ExecutionPayloadEnvelopeV3,
-    ExecutionPayloadEnvelopeV4, ExecutionPayloadEnvelopeV5, ExecutionPayloadFieldV2,
-    ExecutionPayloadV1, ExecutionPayloadV3, PayloadAttributes, PayloadId,
+    ExecutionPayloadEnvelopeV4, ExecutionPayloadEnvelopeV5, ExecutionPayloadEnvelopeV6,
+    ExecutionPayloadFieldV2, ExecutionPayloadV1, ExecutionPayloadV3, ExecutionPayloadV4,
+    PayloadAttributes, PayloadId,
 };
 use core::convert::Infallible;
 use reth_ethereum_primitives::EthPrimitives;
@@ -160,6 +161,38 @@ impl EthBuiltPayload {
             execution_requests: requests.unwrap_or_default(),
         })
     }
+
+    /// Try converting built payload into [`ExecutionPayloadEnvelopeV6`].
+    pub fn try_into_v6(self) -> Result<ExecutionPayloadEnvelopeV6, BuiltPayloadConversionError> {
+        let Self { block, fees, sidecars, requests, .. } = self;
+
+        let blobs_bundle = match sidecars {
+            BlobSidecars::Empty => BlobsBundleV2::empty(),
+            BlobSidecars::Eip7594(sidecars) => BlobsBundleV2::from(sidecars),
+            BlobSidecars::Eip4844(_) => {
+                return Err(BuiltPayloadConversionError::UnexpectedEip4844Sidecars)
+            }
+        };
+
+        Ok(ExecutionPayloadEnvelopeV6 {
+            execution_payload: ExecutionPayloadV4::from_block_unchecked(
+                block.hash(),
+                &Arc::unwrap_or_clone(block).into_block(),
+            ),
+            block_value: fees,
+            // From the engine API spec:
+            //
+            // > Client software **MAY** use any heuristics to decide whether to set
+            // `shouldOverrideBuilder` flag or not. If client software does not implement any
+            // heuristic this flag **SHOULD** be set to `false`.
+            //
+            // Spec:
+            // <https://github.com/ethereum/execution-apis/blob/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine/cancun.md#specification-2>
+            should_override_builder: false,
+            blobs_bundle,
+            execution_requests: requests.unwrap_or_default(),
+        })
+    }
 }
 
 impl<N: NodePrimitives> BuiltPayload for EthBuiltPayload<N> {
@@ -227,6 +260,14 @@ impl TryFrom<EthBuiltPayload> for ExecutionPayloadEnvelopeV5 {
     }
 }
 
+impl TryFrom<EthBuiltPayload> for ExecutionPayloadEnvelopeV6 {
+    type Error = BuiltPayloadConversionError;
+
+    fn try_from(value: EthBuiltPayload) -> Result<Self, Self::Error> {
+        value.try_into_v6()
+    }
+}
+
 /// An enum representing blob transaction sidecars belonging to [`EthBuiltPayload`].
 #[derive(Clone, Default, Debug)]
 pub enum BlobSidecars {
@@ -462,6 +503,7 @@ mod tests {
             .unwrap(),
             withdrawals: None,
             parent_beacon_block_root: None,
+            slot_number: None,
         };
 
         // Verify that the generated payload ID matches the expected value
@@ -499,6 +541,7 @@ mod tests {
                 },
             ]),
             parent_beacon_block_root: None,
+            slot_number: None,
         };
 
         // Verify that the generated payload ID matches the expected value
@@ -531,6 +574,7 @@ mod tests {
                 )
                 .unwrap(),
             ),
+            slot_number: None,
         };
 
         // Verify that the generated payload ID matches the expected value

crates/ethereum/evm/Cargo.toml

@@ -27,6 +27,7 @@ alloy-eips.workspace = true
 alloy-evm.workspace = true
 alloy-consensus.workspace = true
 alloy-rpc-types-engine.workspace = true
+alloy-rlp.workspace = true
 
 # Misc
 parking_lot = { workspace = true, optional = true }
@@ -57,6 +58,7 @@ std = [
     "derive_more?/std",
     "alloy-rpc-types-engine/std",
     "reth-storage-errors/std",
+    "alloy-rlp/std",
 ]
 test-utils = [
     "dep:parking_lot",

crates/ethereum/evm/src/build.rs

@@ -45,7 +45,8 @@ where
             execution_ctx: ctx,
             parent,
             transactions,
-            output: BlockExecutionResult { receipts, requests, gas_used, blob_gas_used },
+            output:
+                BlockExecutionResult { receipts, requests, gas_used, blob_gas_used, block_access_list },
             state_root,
             ..
         } = input;
@@ -90,6 +91,18 @@ where
             };
         }
 
+        let (built_block_access_list, block_access_list_hash) =
+            if self.chain_spec.is_amsterdam_active_at_timestamp(timestamp) {
+                if let Some(bal) = block_access_list {
+                    let hash = alloy_primitives::keccak256(alloy_rlp::encode(bal));
+                    (Some(bal), Some(hash))
+                } else {
+                    (None, None)
+                }
+            } else {
+                (None, None)
+            };
+
         let header = Header {
             parent_hash: ctx.parent_hash,
             ommers_hash: EMPTY_OMMER_ROOT_HASH,
@@ -112,11 +125,17 @@ where
             blob_gas_used: block_blob_gas_used,
             excess_blob_gas,
             requests_hash,
+            block_access_list_hash,
         };
 
         Ok(Block {
             header,
-            body: BlockBody { transactions, ommers: Default::default(), withdrawals },
+            body: BlockBody {
+                transactions,
+                ommers: Default::default(),
+                withdrawals,
+                block_access_list: built_block_access_list.cloned(),
+            },
         })
     }
 }

crates/ethereum/evm/src/lib.rs

@@ -188,7 +188,6 @@ where
         block: &'a SealedBlock<Block>,
     ) -> Result<EthBlockExecutionCtx<'a>, Self::Error> {
         Ok(EthBlockExecutionCtx {
-            tx_count_hint: Some(block.transaction_count()),
             parent_hash: block.header().parent_hash,
             parent_beacon_block_root: block.header().parent_beacon_block_root,
             ommers: &block.body().ommers,
@@ -203,7 +202,6 @@ where
         attributes: Self::NextBlockEnvCtx,
     ) -> Result<EthBlockExecutionCtx<'_>, Self::Error> {
         Ok(EthBlockExecutionCtx {
-            tx_count_hint: None,
             parent_hash: parent.hash(),
             parent_beacon_block_root: attributes.parent_beacon_block_root,
             ommers: &[],
@@ -283,7 +281,6 @@ where
         payload: &'a ExecutionData,
     ) -> Result<ExecutionCtxFor<'a, Self>, Self::Error> {
         Ok(EthBlockExecutionCtx {
-            tx_count_hint: Some(payload.payload.transactions().len()),
             parent_hash: payload.parent_hash(),
             parent_beacon_block_root: payload.sidecar.parent_beacon_block_root(),
             ommers: &[],

crates/ethereum/evm/src/test_utils.rs

@@ -2,7 +2,7 @@ use crate::EthEvmConfig;
 use alloc::{boxed::Box, sync::Arc, vec, vec::Vec};
 use alloy_consensus::Header;
 use alloy_eips::eip7685::Requests;
-use alloy_evm::precompiles::PrecompilesMap;
+use alloy_evm::{block::StateDB, precompiles::PrecompilesMap};
 use alloy_primitives::Bytes;
 use alloy_rpc_types_engine::ExecutionData;
 use parking_lot::Mutex;
@@ -19,7 +19,6 @@ use reth_execution_types::{BlockExecutionResult, ExecutionOutcome};
 use reth_primitives_traits::{BlockTy, SealedBlock, SealedHeader};
 use revm::{
     context::result::{ExecutionResult, Output, ResultAndState, SuccessReason},
-    database::State,
     Inspector,
 };
 
@@ -58,36 +57,30 @@ impl BlockExecutorFactory for MockEvmConfig {
 
     fn create_executor<'a, DB, I>(
         &'a self,
-        evm: EthEvm<&'a mut State<DB>, I, PrecompilesMap>,
+        evm: EthEvm<DB, I, PrecompilesMap>,
         _ctx: Self::ExecutionCtx<'a>,
     ) -> impl BlockExecutorFor<'a, Self, DB, I>
     where
-        DB: Database + 'a,
-        I: Inspector<<Self::EvmFactory as EvmFactory>::Context<&'a mut State<DB>>> + 'a,
+        DB: StateDB + Database + 'a,
+        I: Inspector<<Self::EvmFactory as EvmFactory>::Context<DB>> + 'a,
     {
-        MockExecutor {
-            result: self.exec_results.lock().pop().unwrap(),
-            evm,
-            hook: None,
-            receipts: Vec::new(),
-        }
+        MockExecutor { result: self.exec_results.lock().pop().unwrap(), evm, hook: None }
     }
 }
 
 /// Mock executor that returns a fixed execution result.
 #[derive(derive_more::Debug)]
-pub struct MockExecutor<'a, DB: Database, I> {
+pub struct MockExecutor<DB: Database, I> {
     result: ExecutionOutcome,
-    evm: EthEvm<&'a mut State<DB>, I, PrecompilesMap>,
+    evm: EthEvm<DB, I, PrecompilesMap>,
     #[debug(skip)]
     hook: Option<Box<dyn reth_evm::OnStateHook>>,
-    receipts: Vec<Receipt>,
 }
 
-impl<'a, DB: Database, I: Inspector<EthEvmContext<&'a mut State<DB>>>> BlockExecutor
-    for MockExecutor<'a, DB, I>
+impl<DB: StateDB + Database, I: Inspector<EthEvmContext<DB>>> BlockExecutor
+    for MockExecutor<DB, I>
 {
-    type Evm = EthEvm<&'a mut State<DB>, I, PrecompilesMap>;
+    type Evm = EthEvm<DB, I, PrecompilesMap>;
     type Transaction = TransactionSigned;
     type Receipt = Receipt;
 
@@ -95,10 +88,6 @@ impl<'a, DB: Database, I: Inspector<EthEvmContext<&'a mut State<DB>>>> BlockExec
         Ok(())
     }
 
-    fn receipts(&self) -> &[Self::Receipt] {
-        &self.receipts
-    }
-
     fn execute_transaction_without_commit(
         &mut self,
         _tx: impl ExecutableTx<Self>,
@@ -135,10 +124,11 @@ impl<'a, DB: Database, I: Inspector<EthEvmContext<&'a mut State<DB>>>> BlockExec
                 reqs
             }),
             gas_used: 0,
+            block_access_list: None,
             blob_gas_used: 0,
         };
 
-        evm.db_mut().bundle_state = bundle;
+        *evm.db_mut().bundle_state_mut() = bundle;
 
         Ok((evm, result))
     }

crates/ethereum/evm/tests/execute.rs

@@ -88,7 +88,12 @@ fn eip_4788_non_genesis_call() {
         .execute_one(&RecoveredBlock::new_unhashed(
             Block {
                 header: header.clone(),
-                body: BlockBody { transactions: vec![], ommers: vec![], withdrawals: None },
+                body: BlockBody {
+                    transactions: vec![],
+                    ommers: vec![],
+                    withdrawals: None,
+                    block_access_list: None,
+                },
             },
             vec![],
         ))
@@ -107,7 +112,12 @@ fn eip_4788_non_genesis_call() {
         .execute_one(&RecoveredBlock::new_unhashed(
             Block {
                 header: header.clone(),
-                body: BlockBody { transactions: vec![], ommers: vec![], withdrawals: None },
+                body: BlockBody {
+                    transactions: vec![],
+                    ommers: vec![],
+                    withdrawals: None,
+                    block_access_list: None,
+                },
             },
             vec![],
         ))
@@ -167,7 +177,12 @@ fn eip_4788_no_code_cancun() {
         .execute_one(&RecoveredBlock::new_unhashed(
             Block {
                 header,
-                body: BlockBody { transactions: vec![], ommers: vec![], withdrawals: None },
+                body: BlockBody {
+                    transactions: vec![],
+                    ommers: vec![],
+                    withdrawals: None,
+                    block_access_list: None,
+                },
             },
             vec![],
         ))
@@ -209,7 +224,12 @@ fn eip_4788_empty_account_call() {
         .execute_one(&RecoveredBlock::new_unhashed(
             Block {
                 header,
-                body: BlockBody { transactions: vec![], ommers: vec![], withdrawals: None },
+                body: BlockBody {
+                    transactions: vec![],
+                    ommers: vec![],
+                    withdrawals: None,
+                    block_access_list: None,
+                },
             },
             vec![],
         ))
@@ -799,6 +819,7 @@ fn test_balance_increment_not_duplicated() {
                 transactions: vec![],
                 ommers: vec![],
                 withdrawals: Some(vec![withdrawal].into()),
+                block_access_list: None,
             },
         },
         vec![],

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

@@ -223,6 +223,7 @@ async fn test_testing_build_block_v1_osaka() -> eyre::Result<()> {
         suggested_fee_recipient: Address::ZERO,
         withdrawals: Some(vec![]),
         parent_beacon_block_root: Some(B256::ZERO),
+        slot_number: None,
     };
 
     let request = TestingBuildBlockRequestV1 {

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

@@ -153,7 +153,7 @@ async fn maintain_txpool_reorg() -> eyre::Result<()> {
         w1.address(),
     );
     let pooled_tx1 = EthPooledTransaction::new(tx1.clone(), 200);
-    let tx_hash1 = *pooled_tx1.hash();
+    let tx_hash1 = *pooled_tx1.clone().hash();
 
     // build tx2 from wallet2
     let envelop2 = TransactionTestContext::transfer_tx(1, w2.clone()).await;
@@ -162,7 +162,7 @@ async fn maintain_txpool_reorg() -> eyre::Result<()> {
         w2.address(),
     );
     let pooled_tx2 = EthPooledTransaction::new(tx2.clone(), 200);
-    let tx_hash2 = *pooled_tx2.hash();
+    let tx_hash2 = *pooled_tx2.clone().hash();
 
     let block_info = BlockInfo {
         block_gas_limit: ETHEREUM_BLOCK_GAS_LIMIT_30M,

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

@@ -25,6 +25,7 @@ pub(crate) fn eth_payload_attributes(timestamp: u64) -> EthPayloadBuilderAttribu
         suggested_fee_recipient: Address::ZERO,
         withdrawals: Some(vec![]),
         parent_beacon_block_root: Some(B256::ZERO),
+        slot_number: None,
     };
     EthPayloadBuilderAttributes::new(B256::ZERO, attributes)
 }

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

@@ -56,6 +56,7 @@ async fn testing_rpc_build_block_works() -> eyre::Result<()> {
                 suggested_fee_recipient: Address::ZERO,
                 withdrawals: None,
                 parent_beacon_block_root: None,
+                slot_number: None,
             };
 
             let request = TestingBuildBlockRequestV1 {

crates/ethereum/payload/src/lib.rs

@@ -153,10 +153,14 @@ where
     let PayloadConfig { parent_header, attributes } = config;
 
     let state_provider = client.state_by_block_hash(parent_header.hash())?;
-    let state = StateProviderDatabase::new(state_provider.as_ref());
-    let mut db =
-        State::builder().with_database(cached_reads.as_db_mut(state)).with_bundle_update().build();
-
+    let state = StateProviderDatabase::new(&state_provider);
+    let mut db = State::builder()
+        .with_database(cached_reads.as_db_mut(state))
+        .with_bundle_update()
+        .with_bal_builder()
+        .build();
+    db.bal_state.bal_index = 0;
+    db.bal_state.bal_builder = Some(revm::state::bal::Bal::new());
     let mut builder = evm_config
         .builder_for_next_block(
             &mut db,
@@ -247,7 +251,7 @@ where
                     limit: MAX_RLP_BLOCK_SIZE,
                 },
             );
-            continue
+            continue;
         }
 
         // There's only limited amount of blob space available per block, so we need to check if

crates/ethereum/payload/src/validator.rs

@@ -3,7 +3,7 @@
 use alloy_consensus::Block;
 use alloy_rpc_types_engine::{ExecutionData, PayloadError};
 use reth_chainspec::EthereumHardforks;
-use reth_payload_validator::{cancun, prague, shanghai};
+use reth_payload_validator::{amsterdam, cancun, prague, shanghai};
 use reth_primitives_traits::{Block as _, SealedBlock, SignedTransaction};
 use std::sync::Arc;
 
@@ -103,5 +103,10 @@ where
         chain_spec.is_prague_active_at_timestamp(sealed_block.timestamp),
     )?;
 
+    amsterdam::ensure_well_formed_fields(
+        sealed_block.body(),
+        chain_spec.is_amsterdam_active_at_timestamp(sealed_block.timestamp),
+    )?;
+
     Ok(sealed_block)
 }

crates/ethereum/primitives/src/transaction.rs

@@ -236,7 +236,7 @@ impl reth_codecs::Compact for Transaction {
     // # Panics
     //
     // A panic will be triggered if an identifier larger than 3 is passed from the database. For
-    // optimism an identifier with value [`DEPOSIT_TX_TYPE_ID`] is allowed.
+    // optimism a identifier with value [`DEPOSIT_TX_TYPE_ID`] is allowed.
     fn from_compact(buf: &[u8], identifier: usize) -> (Self, &[u8]) {
         let (tx_type, buf) = TxType::from_compact(buf, identifier);
 

crates/evm/evm/src/execute.rs

@@ -1,12 +1,12 @@
 //! Traits for execution.
 
 use crate::{ConfigureEvm, Database, OnStateHook, TxEnvFor};
-use alloc::{boxed::Box, sync::Arc, vec::Vec};
+use alloc::{borrow::Cow, boxed::Box, sync::Arc, vec::Vec};
 use alloy_consensus::{BlockHeader, Header};
 use alloy_eips::eip2718::WithEncoded;
 pub use alloy_evm::block::{BlockExecutor, BlockExecutorFactory};
 use alloy_evm::{
-    block::{CommitChanges, ExecutableTx},
+    block::{CommitChanges, ExecutableTx, StateDB},
     Evm, EvmEnv, EvmFactory, RecoveredTx, ToTxEnv,
 };
 use alloy_primitives::{Address, B256};
@@ -214,7 +214,7 @@ pub struct BlockAssemblerInput<'a, 'b, F: BlockExecutorFactory, H = Header> {
     /// Output of block execution.
     pub output: &'b BlockExecutionResult<F::Receipt>,
     /// [`BundleState`] after the block execution.
-    pub bundle_state: &'a BundleState,
+    pub bundle_state: Cow<'a, BundleState>,
     /// Provider with access to state.
     #[debug(skip)]
     pub state_provider: &'b dyn StateProvider,
@@ -234,7 +234,7 @@ impl<'a, 'b, F: BlockExecutorFactory, H> BlockAssemblerInput<'a, 'b, F, H> {
         parent: &'a SealedHeader<H>,
         transactions: Vec<F::Transaction>,
         output: &'b BlockExecutionResult<F::Receipt>,
-        bundle_state: &'a BundleState,
+        bundle_state: impl Into<Cow<'a, BundleState>>,
         state_provider: &'b dyn StateProvider,
         state_root: B256,
     ) -> Self {
@@ -244,7 +244,7 @@ impl<'a, 'b, F: BlockExecutorFactory, H> BlockAssemblerInput<'a, 'b, F, H> {
             parent,
             transactions,
             output,
-            bundle_state,
+            bundle_state: bundle_state.into(),
             state_provider,
             state_root,
         }
@@ -461,8 +461,7 @@ where
     }
 }
 
-impl<'a, F, DB, Executor, Builder, N> BlockBuilder
-    for BasicBlockBuilder<'a, F, Executor, Builder, N>
+impl<'a, F, Executor, Builder, N> BlockBuilder for BasicBlockBuilder<'a, F, Executor, Builder, N>
 where
     F: BlockExecutorFactory<Transaction = N::SignedTx, Receipt = N::Receipt>,
     Executor: BlockExecutor<
@@ -470,12 +469,11 @@ where
             Spec = <F::EvmFactory as EvmFactory>::Spec,
             HaltReason = <F::EvmFactory as EvmFactory>::HaltReason,
             BlockEnv = <F::EvmFactory as EvmFactory>::BlockEnv,
-            DB = &'a mut State<DB>,
+            DB: StateDB + 'a,
         >,
         Transaction = N::SignedTx,
         Receipt = N::Receipt,
     >,
-    DB: Database + 'a,
     Builder: BlockAssembler<F, Block = N::Block>,
     N: NodePrimitives,
 {
@@ -508,13 +506,13 @@ where
         state: impl StateProvider,
     ) -> Result<BlockBuilderOutcome<N>, BlockExecutionError> {
         let (evm, result) = self.executor.finish()?;
-        let (db, evm_env) = evm.finish();
+        let (mut db, evm_env) = evm.finish();
 
         // merge all transitions into bundle state
         db.merge_transitions(BundleRetention::Reverts);
 
         // calculate the state root
-        let hashed_state = state.hashed_post_state(&db.bundle_state);
+        let hashed_state = state.hashed_post_state(db.bundle_state());
         let (state_root, trie_updates) = state
             .state_root_with_updates(hashed_state.clone())
             .map_err(BlockExecutionError::other)?;
@@ -528,7 +526,7 @@ where
             parent: self.parent,
             transactions,
             output: &result,
-            bundle_state: &db.bundle_state,
+            bundle_state: Cow::Owned(db.take_bundle()),
             state_provider: &state,
             state_root,
         })?;
@@ -564,8 +562,14 @@ pub struct BasicBlockExecutor<F, DB> {
 impl<F, DB: Database> BasicBlockExecutor<F, DB> {
     /// Creates a new `BasicBlockExecutor` with the given strategy.
     pub fn new(strategy_factory: F, db: DB) -> Self {
-        let db =
-            State::builder().with_database(db).with_bundle_update().without_state_clear().build();
+        let mut db = State::builder()
+            .with_database(db)
+            .with_bundle_update()
+            .with_bal_builder()
+            .without_state_clear()
+            .build();
+        db.bal_state.bal_index = 0;
+        db.bal_state.bal_builder = Some(revm::state::bal::Bal::new());
         Self { strategy_factory, db }
     }
 }

crates/evm/evm/src/lib.rs

@@ -18,6 +18,7 @@
 extern crate alloc;
 
 use crate::execute::{BasicBlockBuilder, Executor};
+use ::revm::context::TxEnv;
 use alloc::vec::Vec;
 use alloy_eips::{
     eip2718::{EIP2930_TX_TYPE_ID, LEGACY_TX_TYPE_ID},
@@ -25,7 +26,7 @@ use alloy_eips::{
     eip4895::Withdrawals,
 };
 use alloy_evm::{
-    block::{BlockExecutorFactory, BlockExecutorFor},
+    block::{BlockExecutorFactory, BlockExecutorFor, StateDB},
     precompiles::PrecompilesMap,
 };
 use alloy_primitives::{Address, Bytes, B256};
@@ -35,7 +36,7 @@ use reth_execution_errors::BlockExecutionError;
 use reth_primitives_traits::{
     BlockTy, HeaderTy, NodePrimitives, ReceiptTy, SealedBlock, SealedHeader, TxTy,
 };
-use revm::{context::TxEnv, database::State};
+use revm::DatabaseCommit;
 
 pub mod either;
 /// EVM environment configuration.
@@ -312,20 +313,20 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
     /// Creates a strategy with given EVM and execution context.
     fn create_executor<'a, DB, I>(
         &'a self,
-        evm: EvmFor<Self, &'a mut State<DB>, I>,
+        evm: EvmFor<Self, DB, I>,
         ctx: <Self::BlockExecutorFactory as BlockExecutorFactory>::ExecutionCtx<'a>,
     ) -> impl BlockExecutorFor<'a, Self::BlockExecutorFactory, DB, I>
     where
-        DB: Database,
-        I: InspectorFor<Self, &'a mut State<DB>> + 'a,
+        DB: StateDB + DatabaseCommit + Database + 'a,
+        I: InspectorFor<Self, DB> + 'a,
     {
         self.block_executor_factory().create_executor(evm, ctx)
     }
 
     /// Creates a strategy for execution of a given block.
-    fn executor_for_block<'a, DB: Database>(
+    fn executor_for_block<'a, DB: StateDB + DatabaseCommit + Database + 'a>(
         &'a self,
-        db: &'a mut State<DB>,
+        db: DB,
         block: &'a SealedBlock<<Self::Primitives as NodePrimitives>::Block>,
     ) -> Result<impl BlockExecutorFor<'a, Self::BlockExecutorFactory, DB>, Self::Error> {
         let evm = self.evm_for_block(db, block.header())?;
@@ -350,7 +351,7 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
     /// ```
     fn create_block_builder<'a, DB, I>(
         &'a self,
-        evm: EvmFor<Self, &'a mut State<DB>, I>,
+        evm: EvmFor<Self, DB, I>,
         parent: &'a SealedHeader<HeaderTy<Self::Primitives>>,
         ctx: <Self::BlockExecutorFactory as BlockExecutorFactory>::ExecutionCtx<'a>,
     ) -> impl BlockBuilder<
@@ -358,8 +359,8 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
         Executor: BlockExecutorFor<'a, Self::BlockExecutorFactory, DB, I>,
     >
     where
-        DB: Database,
-        I: InspectorFor<Self, &'a mut State<DB>> + 'a,
+        DB: StateDB + DatabaseCommit + Database + 'a,
+        I: InspectorFor<Self, DB> + 'a,
     {
         BasicBlockBuilder {
             executor: self.create_executor(evm, ctx.clone()),
@@ -399,9 +400,9 @@ pub trait ConfigureEvm: Clone + Debug + Send + Sync + Unpin {
     /// // Complete block building
     /// let outcome = builder.finish(state_provider)?;
     /// ```
-    fn builder_for_next_block<'a, DB: Database + 'a>(
+    fn builder_for_next_block<'a, DB: StateDB + DatabaseCommit + Database + 'a>(
         &'a self,
-        db: &'a mut State<DB>,
+        db: DB,
         parent: &'a SealedHeader<<Self::Primitives as NodePrimitives>::BlockHeader>,
         attributes: Self::NextBlockEnvCtx,
     ) -> Result<

crates/evm/execution-types/src/chain.rs

@@ -1,7 +1,7 @@
 //! Contains [Chain], a chain of blocks and their final state.
 
 use crate::ExecutionOutcome;
-use alloc::{borrow::Cow, collections::BTreeMap, vec::Vec};
+use alloc::{borrow::Cow, collections::BTreeMap, sync::Arc, vec::Vec};
 use alloy_consensus::{transaction::Recovered, BlockHeader};
 use alloy_eips::{eip1898::ForkBlock, eip2718::Encodable2718, BlockNumHash};
 use alloy_primitives::{Address, BlockHash, BlockNumber, TxHash};
@@ -10,7 +10,7 @@ use reth_primitives_traits::{
     transaction::signed::SignedTransaction, Block, BlockBody, IndexedTx, NodePrimitives,
     RecoveredBlock, SealedHeader,
 };
-use reth_trie_common::LazyTrieData;
+use reth_trie_common::{updates::TrieUpdatesSorted, HashedPostStateSorted};
 
 /// A chain of blocks and their final state.
 ///
@@ -34,10 +34,10 @@ pub struct Chain<N: NodePrimitives = reth_ethereum_primitives::EthPrimitives> {
     ///
     /// Additionally, it includes the individual state changes that led to the current state.
     execution_outcome: ExecutionOutcome<N::Receipt>,
-    /// Lazy trie data for each block in the chain, keyed by block number.
-    ///
-    /// Contains handles to lazily-initialized sorted trie updates and hashed state.
-    trie_data: BTreeMap<BlockNumber, LazyTrieData>,
+    /// State trie updates for each block in the chain, keyed by block number.
+    trie_updates: BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>>,
+    /// Hashed post state for each block in the chain, keyed by block number.
+    hashed_state: BTreeMap<BlockNumber, Arc<HashedPostStateSorted>>,
 }
 
 type ChainTxReceiptMeta<'a, N> = (
@@ -52,7 +52,8 @@ impl<N: NodePrimitives> Default for Chain<N> {
         Self {
             blocks: Default::default(),
             execution_outcome: Default::default(),
-            trie_data: Default::default(),
+            trie_updates: Default::default(),
+            hashed_state: Default::default(),
         }
     }
 }
@@ -66,23 +67,27 @@ impl<N: NodePrimitives> Chain<N> {
     pub fn new(
         blocks: impl IntoIterator<Item = RecoveredBlock<N::Block>>,
         execution_outcome: ExecutionOutcome<N::Receipt>,
-        trie_data: BTreeMap<BlockNumber, LazyTrieData>,
+        trie_updates: BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>>,
+        hashed_state: BTreeMap<BlockNumber, Arc<HashedPostStateSorted>>,
     ) -> Self {
         let blocks =
             blocks.into_iter().map(|b| (b.header().number(), b)).collect::<BTreeMap<_, _>>();
         debug_assert!(!blocks.is_empty(), "Chain should have at least one block");
 
-        Self { blocks, execution_outcome, trie_data }
+        Self { blocks, execution_outcome, trie_updates, hashed_state }
     }
 
     /// Create new Chain from a single block and its state.
     pub fn from_block(
         block: RecoveredBlock<N::Block>,
         execution_outcome: ExecutionOutcome<N::Receipt>,
-        trie_data: LazyTrieData,
+        trie_updates: Arc<TrieUpdatesSorted>,
+        hashed_state: Arc<HashedPostStateSorted>,
     ) -> Self {
         let block_number = block.header().number();
-        Self::new([block], execution_outcome, BTreeMap::from([(block_number, trie_data)]))
+        let trie_updates_map = BTreeMap::from([(block_number, trie_updates)]);
+        let hashed_state_map = BTreeMap::from([(block_number, hashed_state)]);
+        Self::new([block], execution_outcome, trie_updates_map, hashed_state_map)
     }
 
     /// Get the blocks in this chain.
@@ -100,19 +105,37 @@ impl<N: NodePrimitives> Chain<N> {
         self.blocks.values().map(|block| block.clone_sealed_header())
     }
 
-    /// Get all trie data for this chain.
-    pub const fn trie_data(&self) -> &BTreeMap<BlockNumber, LazyTrieData> {
-        &self.trie_data
+    /// Get all trie updates for this chain.
+    pub const fn trie_updates(&self) -> &BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>> {
+        &self.trie_updates
     }
 
-    /// Get trie data for a specific block number.
-    pub fn trie_data_at(&self, block_number: BlockNumber) -> Option<&LazyTrieData> {
-        self.trie_data.get(&block_number)
+    /// Get trie updates for a specific block number.
+    pub fn trie_updates_at(&self, block_number: BlockNumber) -> Option<&Arc<TrieUpdatesSorted>> {
+        self.trie_updates.get(&block_number)
     }
 
-    /// Remove all trie data for this chain.
-    pub fn clear_trie_data(&mut self) {
-        self.trie_data.clear();
+    /// Remove all trie updates for this chain.
+    pub fn clear_trie_updates(&mut self) {
+        self.trie_updates.clear();
+    }
+
+    /// Get all hashed states for this chain.
+    pub const fn hashed_state(&self) -> &BTreeMap<BlockNumber, Arc<HashedPostStateSorted>> {
+        &self.hashed_state
+    }
+
+    /// Get hashed state for a specific block number.
+    pub fn hashed_state_at(
+        &self,
+        block_number: BlockNumber,
+    ) -> Option<&Arc<HashedPostStateSorted>> {
+        self.hashed_state.get(&block_number)
+    }
+
+    /// Remove all hashed states for this chain.
+    pub fn clear_hashed_state(&mut self) {
+        self.hashed_state.clear();
     }
 
     /// Get execution outcome of this chain
@@ -160,16 +183,23 @@ impl<N: NodePrimitives> Chain<N> {
     /// Destructure the chain into its inner components:
     /// 1. The blocks contained in the chain.
     /// 2. The execution outcome representing the final state.
-    /// 3. The trie data map.
+    /// 3. The trie updates map.
+    /// 4. The hashed state map.
     #[allow(clippy::type_complexity)]
     pub fn into_inner(
         self,
     ) -> (
         ChainBlocks<'static, N::Block>,
         ExecutionOutcome<N::Receipt>,
-        BTreeMap<BlockNumber, LazyTrieData>,
+        BTreeMap<BlockNumber, Arc<TrieUpdatesSorted>>,
+        BTreeMap<BlockNumber, Arc<HashedPostStateSorted>>,
     ) {
-        (ChainBlocks { blocks: Cow::Owned(self.blocks) }, self.execution_outcome, self.trie_data)
+        (
+            ChainBlocks { blocks: Cow::Owned(self.blocks) },
+            self.execution_outcome,
+            self.trie_updates,
+            self.hashed_state,
+        )
     }
 
     /// Destructure the chain into its inner components:
@@ -299,12 +329,14 @@ impl<N: NodePrimitives> Chain<N> {
         &mut self,
         block: RecoveredBlock<N::Block>,
         execution_outcome: ExecutionOutcome<N::Receipt>,
-        trie_data: LazyTrieData,
+        trie_updates: Arc<TrieUpdatesSorted>,
+        hashed_state: Arc<HashedPostStateSorted>,
     ) {
         let block_number = block.header().number();
         self.blocks.insert(block_number, block);
         self.execution_outcome.extend(execution_outcome);
-        self.trie_data.insert(block_number, trie_data);
+        self.trie_updates.insert(block_number, trie_updates);
+        self.hashed_state.insert(block_number, hashed_state);
     }
 
     /// Merge two chains by appending the given chain into the current one.
@@ -323,7 +355,8 @@ impl<N: NodePrimitives> Chain<N> {
         // Insert blocks from other chain
         self.blocks.extend(other.blocks);
         self.execution_outcome.extend(other.execution_outcome);
-        self.trie_data.extend(other.trie_data);
+        self.trie_updates.extend(other.trie_updates);
+        self.hashed_state.extend(other.hashed_state);
 
         Ok(())
     }
@@ -550,14 +583,14 @@ pub(super) mod serde_bincode_compat {
                 execution_outcome: value.execution_outcome.as_repr(),
                 _trie_updates_legacy: None,
                 trie_updates: value
-                    .trie_data
+                    .trie_updates
                     .iter()
-                    .map(|(k, v)| (*k, v.get().trie_updates.as_ref().into()))
+                    .map(|(k, v)| (*k, v.as_ref().into()))
                     .collect(),
                 hashed_state: value
-                    .trie_data
+                    .hashed_state
                     .iter()
-                    .map(|(k, v)| (*k, v.get().hashed_state.as_ref().into()))
+                    .map(|(k, v)| (*k, v.as_ref().into()))
                     .collect(),
             }
         }
@@ -570,24 +603,19 @@ pub(super) mod serde_bincode_compat {
         >,
     {
         fn from(value: Chain<'a, N>) -> Self {
-            use reth_trie_common::LazyTrieData;
-
-            let hashed_state_map: BTreeMap<_, _> =
-                value.hashed_state.into_iter().map(|(k, v)| (k, Arc::new(v.into()))).collect();
-
-            let trie_data: BTreeMap<BlockNumber, LazyTrieData> = value
-                .trie_updates
-                .into_iter()
-                .map(|(k, v)| {
-                    let hashed_state = hashed_state_map.get(&k).cloned().unwrap_or_default();
-                    (k, LazyTrieData::ready(hashed_state, Arc::new(v.into())))
-                })
-                .collect();
-
             Self {
                 blocks: value.blocks.0.into_owned(),
                 execution_outcome: ExecutionOutcome::from_repr(value.execution_outcome),
-                trie_data,
+                trie_updates: value
+                    .trie_updates
+                    .into_iter()
+                    .map(|(k, v)| (k, Arc::new(v.into())))
+                    .collect(),
+                hashed_state: value
+                    .hashed_state
+                    .into_iter()
+                    .map(|(k, v)| (k, Arc::new(v.into())))
+                    .collect(),
             }
         }
     }
@@ -648,6 +676,7 @@ pub(super) mod serde_bincode_compat {
                         .unwrap()],
                     Default::default(),
                     BTreeMap::new(),
+                    BTreeMap::new(),
                 ),
             };
 
@@ -747,8 +776,12 @@ mod tests {
         let mut block_state_extended = execution_outcome1;
         block_state_extended.extend(execution_outcome2);
 
-        let chain: Chain =
-            Chain::new(vec![block1.clone(), block2.clone()], block_state_extended, BTreeMap::new());
+        let chain: Chain = Chain::new(
+            vec![block1.clone(), block2.clone()],
+            block_state_extended,
+            BTreeMap::new(),
+            BTreeMap::new(),
+        );
 
         // return tip state
         assert_eq!(

crates/evm/execution-types/src/execute.rs

@@ -1,5 +1,3 @@
-use alloy_primitives::{Address, B256, U256};
-use reth_primitives_traits::{Account, Bytecode};
 use revm::database::BundleState;
 
 pub use alloy_evm::block::BlockExecutionResult;
@@ -25,36 +23,3 @@ pub struct BlockExecutionOutput<T> {
     /// The changed state of the block after execution.
     pub state: BundleState,
 }
-
-impl<T> BlockExecutionOutput<T> {
-    /// Return bytecode if known.
-    pub fn bytecode(&self, code_hash: &B256) -> Option<Bytecode> {
-        self.state.bytecode(code_hash).map(Bytecode)
-    }
-
-    /// Get account if account is known.
-    pub fn account(&self, address: &Address) -> Option<Option<Account>> {
-        self.state.account(address).map(|a| a.info.as_ref().map(Into::into))
-    }
-
-    /// Get storage if value is known.
-    ///
-    /// This means that depending on status we can potentially return `U256::ZERO`.
-    pub fn storage(&self, address: &Address, storage_key: U256) -> Option<U256> {
-        self.state.account(address).and_then(|a| a.storage_slot(storage_key))
-    }
-}
-
-impl<T> Default for BlockExecutionOutput<T> {
-    fn default() -> Self {
-        Self {
-            result: BlockExecutionResult {
-                receipts: Default::default(),
-                requests: Default::default(),
-                gas_used: 0,
-                blob_gas_used: 0,
-            },
-            state: Default::default(),
-        }
-    }
-}

crates/evm/execution-types/src/execution_outcome.rs

@@ -249,14 +249,6 @@ impl<T> ExecutionOutcome<T> {
         &self.receipts[index]
     }
 
-    /// Returns an iterator over receipt slices, one per block.
-    ///
-    /// This is a more ergonomic alternative to `receipts()` that yields slices
-    /// instead of requiring indexing into a nested `Vec<Vec<T>>`.
-    pub fn receipts_iter(&self) -> impl Iterator<Item = &[T]> + '_ {
-        self.receipts.iter().map(|v| v.as_slice())
-    }
-
     /// Is execution outcome empty.
     pub const fn is_empty(&self) -> bool {
         self.len() == 0

crates/exex/exex/src/backfill/job.rs

@@ -149,7 +149,7 @@ where
             executor.into_state().take_bundle(),
             results,
         );
-        let chain = Chain::new(blocks, outcome, BTreeMap::new());
+        let chain = Chain::new(blocks, outcome, BTreeMap::new(), BTreeMap::new());
         Ok(chain)
     }
 }

crates/exex/exex/src/manager.rs

@@ -503,7 +503,6 @@ where
             }
             break
         }
-        let buffer_full = this.buffer.len() >= this.max_capacity;
 
         // Update capacity
         this.update_capacity();
@@ -537,12 +536,6 @@ where
         // Update capacity
         this.update_capacity();
 
-        // If the buffer was full and we made space, we need to wake up to accept new notifications
-        if buffer_full && this.buffer.len() < this.max_capacity {
-            debug!(target: "exex::manager", "Buffer has space again, waking up senders");
-            cx.waker().wake_by_ref();
-        }
-
         // Update watch channel block number
         let finished_height = this.exex_handles.iter_mut().try_fold(u64::MAX, |curr, exex| {
             exex.finished_height.map_or(Err(()), |height| Ok(height.number.min(curr)))
@@ -694,6 +687,7 @@ mod tests {
         BlockWriter, Chain, DBProvider, DatabaseProviderFactory, TransactionVariant,
     };
     use reth_testing_utils::generators::{self, random_block, BlockParams};
+    use std::collections::BTreeMap;
 
     fn empty_finalized_header_stream() -> ForkChoiceStream<SealedHeader> {
         let (tx, rx) = watch::channel(None);
@@ -795,7 +789,12 @@ mod tests {
         block1.set_block_number(10);
 
         let notification1 = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(vec![block1.clone()], Default::default(), Default::default())),
+            new: Arc::new(Chain::new(
+                vec![block1.clone()],
+                Default::default(),
+                Default::default(),
+                Default::default(),
+            )),
         };
 
         // Push the first notification
@@ -813,7 +812,12 @@ mod tests {
         block2.set_block_number(20);
 
         let notification2 = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(vec![block2.clone()], Default::default(), Default::default())),
+            new: Arc::new(Chain::new(
+                vec![block2.clone()],
+                Default::default(),
+                Default::default(),
+                Default::default(),
+            )),
         };
 
         exex_manager.push_notification(notification2.clone());
@@ -856,7 +860,12 @@ mod tests {
         block1.set_block_number(10);
 
         let notification1 = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(vec![block1.clone()], Default::default(), Default::default())),
+            new: Arc::new(Chain::new(
+                vec![block1.clone()],
+                Default::default(),
+                Default::default(),
+                Default::default(),
+            )),
         };
 
         exex_manager.push_notification(notification1.clone());
@@ -1084,6 +1093,7 @@ mod tests {
                 vec![Default::default()],
                 Default::default(),
                 Default::default(),
+                Default::default(),
             )),
         };
 
@@ -1154,6 +1164,7 @@ mod tests {
                 vec![Default::default()],
                 Default::default(),
                 Default::default(),
+                Default::default(),
             )),
         };
 
@@ -1198,7 +1209,12 @@ mod tests {
         block1.set_block_number(10);
 
         let notification = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(vec![block1.clone()], Default::default(), Default::default())),
+            new: Arc::new(Chain::new(
+                vec![block1.clone()],
+                Default::default(),
+                Default::default(),
+                Default::default(),
+            )),
         };
 
         let mut cx = Context::from_waker(futures::task::noop_waker_ref());
@@ -1347,11 +1363,17 @@ mod tests {
             new: Arc::new(Chain::new(
                 vec![genesis_block.clone()],
                 Default::default(),
-                Default::default(),
+                BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
         let notification = ExExNotification::ChainCommitted {
-            new: Arc::new(Chain::new(vec![block.clone()], Default::default(), Default::default())),
+            new: Arc::new(Chain::new(
+                vec![block.clone()],
+                Default::default(),
+                BTreeMap::new(),
+                BTreeMap::new(),
+            )),
         };
 
         let (finalized_headers_tx, rx) = watch::channel(None);
@@ -1421,78 +1443,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[tokio::test]
-    async fn test_deadlock_manager_wakes_after_buffer_clears() {
-        // This test simulates the scenario where the buffer fills up, ingestion pauses,
-        // and then space clears. We verify the manager wakes up to process pending items.
-
-        let temp_dir = tempfile::tempdir().unwrap();
-        let wal = Wal::new(temp_dir.path()).unwrap();
-        let provider_factory = create_test_provider_factory();
-        init_genesis(&provider_factory).unwrap();
-        let provider = BlockchainProvider::new(provider_factory.clone()).unwrap();
-
-        // 1. Setup Manager with Capacity = 1
-        let (exex_handle, _, mut notifications) = ExExHandle::new(
-            "test_exex".to_string(),
-            Default::default(),
-            provider,
-            EthEvmConfig::mainnet(),
-            wal.handle(),
-        );
-
-        let max_capacity = 2;
-        let exex_manager = ExExManager::new(
-            provider_factory,
-            vec![exex_handle],
-            max_capacity,
-            wal,
-            empty_finalized_header_stream(),
-        );
-
-        let manager_handle = exex_manager.handle();
-
-        // Spawn manager in background so it runs continuously
-        tokio::spawn(async move {
-            exex_manager.await.ok();
-        });
-
-        // Helper to create notifications
-        let mut rng = generators::rng();
-        let mut make_notif = |id: u64| {
-            let block = random_block(&mut rng, id, BlockParams::default()).try_recover().unwrap();
-            ExExNotification::ChainCommitted {
-                new: Arc::new(Chain::new(vec![block], Default::default(), Default::default())),
-            }
-        };
-
-        manager_handle.send(ExExNotificationSource::Pipeline, make_notif(1)).unwrap();
-
-        // Send the "Stuck" Item (Notification #100).
-        // At this point, the Manager loop has skipped the ingestion logic because buffer is full
-        // (buffer_full=true). This item sits in the unbounded 'handle_rx' channel waiting.
-        manager_handle.send(ExExNotificationSource::Pipeline, make_notif(100)).unwrap();
-
-        // 3. Relieve Pressure
-        // We consume items from the ExEx.
-        // As we pull items out, the ExEx frees space -> Manager sends buffered item -> Manager
-        // frees space. Once Manager frees space, the FIX (wake_by_ref) should trigger,
-        // causing it to read Notif #100.
-
-        // Consume the jam
-        let _ = notifications.next().await.unwrap();
-
-        // 4. Assert No Deadlock
-        // We expect Notification #100 next.
-        // If the wake_by_ref fix is missing, this will Time Out because the manager is sleeping
-        // despite having empty buffer.
-        let result =
-            tokio::time::timeout(std::time::Duration::from_secs(1), notifications.next()).await;
-
-        assert!(
-            result.is_ok(),
-            "Deadlock detected! Manager failed to wake up and process Pending Item #100."
-        );
-    }
 }

crates/exex/exex/src/notifications.rs

@@ -501,6 +501,7 @@ mod tests {
                 .try_recover()?],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
 
@@ -569,6 +570,7 @@ mod tests {
                 .try_recover()?],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
 
@@ -636,6 +638,7 @@ mod tests {
                 vec![exex_head_block.clone().try_recover()?],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
         wal.commit(&exex_head_notification)?;
@@ -650,6 +653,7 @@ mod tests {
                 .try_recover()?],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
 
@@ -707,6 +711,7 @@ mod tests {
                 vec![exex_head_block.clone().try_recover()?],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
         wal.commit(&exex_head_notification)?;
@@ -726,6 +731,7 @@ mod tests {
                 .try_recover()?],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
 

crates/exex/exex/src/wal/mod.rs

@@ -304,24 +304,37 @@ mod tests {
                 vec![blocks[0].clone(), blocks[1].clone()],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
         let reverted_notification = ExExNotification::ChainReverted {
-            old: Arc::new(Chain::new(vec![blocks[1].clone()], Default::default(), BTreeMap::new())),
+            old: Arc::new(Chain::new(
+                vec![blocks[1].clone()],
+                Default::default(),
+                BTreeMap::new(),
+                BTreeMap::new(),
+            )),
         };
         let committed_notification_2 = ExExNotification::ChainCommitted {
             new: Arc::new(Chain::new(
                 vec![block_1_reorged.clone(), blocks[2].clone()],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
         let reorged_notification = ExExNotification::ChainReorged {
-            old: Arc::new(Chain::new(vec![blocks[2].clone()], Default::default(), BTreeMap::new())),
+            old: Arc::new(Chain::new(
+                vec![blocks[2].clone()],
+                Default::default(),
+                BTreeMap::new(),
+                BTreeMap::new(),
+            )),
             new: Arc::new(Chain::new(
                 vec![block_2_reorged.clone(), blocks[3].clone()],
                 Default::default(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             )),
         };
 

crates/exex/exex/src/wal/storage.rs

@@ -189,28 +189,28 @@ mod tests {
     use reth_testing_utils::generators::{self, random_block};
     use reth_trie_common::{
         updates::{StorageTrieUpdates, TrieUpdates},
-        BranchNodeCompact, HashedPostState, HashedStorage, LazyTrieData, Nibbles,
+        BranchNodeCompact, HashedPostState, HashedStorage, Nibbles,
     };
     use std::{collections::BTreeMap, fs::File, sync::Arc};
 
     // wal with 1 block and tx (old 3-field format)
     // <https://github.com/paradigmxyz/reth/issues/15012>
-    #[test]
-    fn decode_notification_wal() {
-        let wal = include_bytes!("../../test-data/28.wal");
-        let notification: reth_exex_types::serde_bincode_compat::ExExNotification<
-            '_,
-            reth_ethereum_primitives::EthPrimitives,
-        > = rmp_serde::decode::from_slice(wal.as_slice()).unwrap();
-        let notification: ExExNotification = notification.into();
-        match notification {
-            ExExNotification::ChainCommitted { new } => {
-                assert_eq!(new.blocks().len(), 1);
-                assert_eq!(new.tip().transaction_count(), 1);
-            }
-            _ => panic!("unexpected notification"),
-        }
-    }
+    // #[test]
+    // fn decode_notification_wal() {
+    //     let wal = include_bytes!("../../test-data/28.wal");
+    //     let notification: reth_exex_types::serde_bincode_compat::ExExNotification<
+    //         '_,
+    //         reth_ethereum_primitives::EthPrimitives,
+    //     > = rmp_serde::decode::from_slice(wal.as_slice()).unwrap();
+    //     let notification: ExExNotification = notification.into();
+    //     match notification {
+    //         ExExNotification::ChainCommitted { new } => {
+    //             assert_eq!(new.blocks().len(), 1);
+    //             assert_eq!(new.tip().transaction_count(), 1);
+    //         }
+    //         _ => panic!("unexpected notification"),
+    //     }
+    // }
 
     // wal with 1 block and tx (new 4-field format with trie updates and hashed state)
     #[test]
@@ -241,8 +241,18 @@ mod tests {
         let new_block = random_block(&mut rng, 0, Default::default()).try_recover()?;
 
         let notification = ExExNotification::ChainReorged {
-            new: Arc::new(Chain::new(vec![new_block], Default::default(), BTreeMap::new())),
-            old: Arc::new(Chain::new(vec![old_block], Default::default(), BTreeMap::new())),
+            new: Arc::new(Chain::new(
+                vec![new_block],
+                Default::default(),
+                BTreeMap::new(),
+                BTreeMap::new(),
+            )),
+            old: Arc::new(Chain::new(
+                vec![old_block],
+                Default::default(),
+                BTreeMap::new(),
+                BTreeMap::new(),
+            )),
         };
 
         // Do a round trip serialization and deserialization
@@ -336,17 +346,13 @@ mod tests {
             )]),
         };
 
-        let trie_data = LazyTrieData::ready(
-            Arc::new(hashed_state.into_sorted()),
-            Arc::new(trie_updates.into_sorted()),
-        );
-
         let notification: ExExNotification<reth_ethereum_primitives::EthPrimitives> =
             ExExNotification::ChainCommitted {
                 new: Arc::new(Chain::new(
                     vec![block],
                     Default::default(),
-                    BTreeMap::from([(block_number, trie_data)]),
+                    BTreeMap::from([(block_number, Arc::new(trie_updates.into_sorted()))]),
+                    BTreeMap::from([(block_number, Arc::new(hashed_state.into_sorted()))]),
                 )),
             };
         Ok(notification)

crates/exex/types/src/notification.rs

@@ -223,12 +223,14 @@ pub(super) mod serde_bincode_compat {
                             .unwrap()],
                         Default::default(),
                         BTreeMap::new(),
+                        BTreeMap::new(),
                     )),
                     new: Arc::new(Chain::new(
                         vec![RecoveredBlock::arbitrary(&mut arbitrary::Unstructured::new(&bytes))
                             .unwrap()],
                         Default::default(),
                         BTreeMap::new(),
+                        BTreeMap::new(),
                     )),
                 },
             };

crates/net/banlist/src/lib.rs

@@ -106,7 +106,7 @@ impl BanList {
         self.banned_ips.contains_key(ip)
     }
 
-    /// checks the ban list to see if it contains the given peer
+    /// checks the ban list to see if it contains the given ip
     #[inline]
     pub fn is_banned_peer(&self, peer_id: &PeerId) -> bool {
         self.banned_peers.contains_key(peer_id)
@@ -117,7 +117,7 @@ impl BanList {
         self.banned_ips.remove(ip);
     }
 
-    /// Unbans the peer
+    /// Unbans the ip address
     pub fn unban_peer(&mut self, peer_id: &PeerId) {
         self.banned_peers.remove(peer_id);
     }

crates/net/downloaders/src/file_client.rs

@@ -256,6 +256,10 @@ impl<B: FullBlock<Header: reth_primitives_traits::BlockHeader>> FromReader
                     Err(err) => return Err(err),
                 };
 
+                tracing::debug!(target: "downloaders::file",
+                    block=?block,
+                    "decoded block from file chunk"
+                );
                 let block = SealedBlock::seal_slow(block);
 
                 // Validate standalone header
@@ -272,6 +276,11 @@ impl<B: FullBlock<Header: reth_primitives_traits::BlockHeader>> FromReader
                 let block_hash = block.hash();
                 let block_number = block.number();
                 let (header, body) = block.split_sealed_header_body();
+                tracing::debug!(target: "downloaders::file",
+                    header=?header,
+                    body=?body,
+                    "adding block to file client buffers"
+                );
                 headers.insert(block_number, header.unseal());
                 hash_to_number.insert(block_hash, block_number);
                 bodies.insert(block_hash, body);

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

@@ -265,6 +265,7 @@ mod tests {
                     excess_blob_gas: None,
                     parent_beacon_block_root: None,
                     requests_hash: None,
+                    block_access_list_hash:None
                 },
             ]),
         }.encode(&mut data);
@@ -302,6 +303,7 @@ mod tests {
                     excess_blob_gas: None,
                     parent_beacon_block_root: None,
                     requests_hash: None,
+                    block_access_list_hash: None
                 },
             ]),
         };
@@ -408,9 +410,11 @@ mod tests {
                             excess_blob_gas: None,
                             parent_beacon_block_root: None,
                             requests_hash: None,
+                            block_access_list_hash:None
                         },
                     ],
                     withdrawals: None,
+                    block_access_list:None
                 }
             ]),
         };
@@ -485,9 +489,11 @@ mod tests {
                             excess_blob_gas: None,
                             parent_beacon_block_root: None,
                             requests_hash: None,
+                            block_access_list_hash:None
                         },
                     ],
                     withdrawals: None,
+                    block_access_list:None
                 }
             ]),
         };

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

@@ -152,6 +152,7 @@ mod tests {
             excess_blob_gas: None,
             parent_beacon_block_root: None,
             requests_hash: None,
+            block_access_list_hash:None
         };
         assert_eq!(header.hash_slow(), expected_hash);
     }
@@ -268,6 +269,7 @@ mod tests {
             excess_blob_gas: Some(0),
             parent_beacon_block_root: None,
             requests_hash: None,
+            block_access_list_hash: None,
         };
 
         let header = Header::decode(&mut data.as_slice()).unwrap();
@@ -310,6 +312,7 @@ mod tests {
             blob_gas_used: Some(0),
             excess_blob_gas: Some(0x1600000),
             requests_hash: None,
+            block_access_list_hash: None,
         };
 
         let header = Header::decode(&mut data.as_slice()).unwrap();

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

@@ -824,6 +824,7 @@ mod tests {
                     transactions: vec![],
                     ommers: vec![],
                     withdrawals: Some(Default::default()),
+                    block_access_list: None,
                 }]
                 .into(),
             }));

crates/node/builder/Cargo.toml

@@ -54,7 +54,6 @@ reth-tasks.workspace = true
 reth-tokio-util.workspace = true
 reth-tracing.workspace = true
 reth-transaction-pool.workspace = true
-reth-trie-db = { workspace = true, features = ["metrics"] }
 reth-basic-payload-builder.workspace = true
 reth-node-ethstats.workspace = true
 
@@ -116,7 +115,6 @@ test-utils = [
     "reth-db-api/test-utils",
     "reth-provider/test-utils",
     "reth-transaction-pool/test-utils",
-    "reth-trie-db/test-utils",
     "reth-evm-ethereum/test-utils",
     "reth-node-ethereum/test-utils",
     "reth-primitives-traits/test-utils",

crates/node/builder/src/launch/common.rs

@@ -84,7 +84,6 @@ use reth_tracing::{
     tracing::{debug, error, info, warn},
 };
 use reth_transaction_pool::TransactionPool;
-use reth_trie_db::ChangesetCache;
 use std::{sync::Arc, thread::available_parallelism, time::Duration};
 use tokio::sync::{
     mpsc::{unbounded_channel, UnboundedSender},
@@ -471,10 +470,7 @@ where
     /// Returns the [`ProviderFactory`] for the attached storage after executing a consistent check
     /// between the database and static files. **It may execute a pipeline unwind if it fails this
     /// check.**
-    pub async fn create_provider_factory<N, Evm>(
-        &self,
-        changeset_cache: ChangesetCache,
-    ) -> eyre::Result<ProviderFactory<N>>
+    pub async fn create_provider_factory<N, Evm>(&self) -> eyre::Result<ProviderFactory<N>>
     where
         N: ProviderNodeTypes<DB = DB, ChainSpec = ChainSpec>,
         Evm: ConfigureEvm<Primitives = N::Primitives> + 'static,
@@ -504,8 +500,7 @@ where
             static_file_provider,
             rocksdb_provider,
         )?
-        .with_prune_modes(self.prune_modes())
-        .with_changeset_cache(changeset_cache);
+        .with_prune_modes(self.prune_modes());
 
         // Keep MDBX, static files, and RocksDB aligned. If any check fails, unwind to the
         // earliest consistent block.
@@ -598,13 +593,12 @@ where
     /// Creates a new [`ProviderFactory`] and attaches it to the launch context.
     pub async fn with_provider_factory<N, Evm>(
         self,
-        changeset_cache: ChangesetCache,
     ) -> eyre::Result<LaunchContextWith<Attached<WithConfigs<ChainSpec>, ProviderFactory<N>>>>
     where
         N: ProviderNodeTypes<DB = DB, ChainSpec = ChainSpec>,
         Evm: ConfigureEvm<Primitives = N::Primitives> + 'static,
     {
-        let factory = self.create_provider_factory::<N, Evm>(changeset_cache).await?;
+        let factory = self.create_provider_factory::<N, Evm>().await?;
         let ctx = LaunchContextWith {
             inner: self.inner,
             attachment: self.attachment.map_right(|_| factory),