fix(engine): persist blocks immediately when persistence_threshold=0

Fixes #21093

The `should_persist()` function was using `>` instead of `>=` when
comparing the block count against the persistence threshold. This meant
that with `--engine.persistence-threshold=0`, blocks would not persist
until there was at least 1 block difference (i.e., 2 blocks in memory).

With the corrected `>=` comparison, setting persistence_threshold=0 now
correctly means "persist immediately" (0 blocks allowed in memory without
triggering persistence).

.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/lint.yml

@@ -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

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"
@@ -473,22 +473,22 @@ reth-ress-protocol = { path = "crates/ress/protocol" }
 reth-ress-provider = { path = "crates/ress/provider" }
 
 # revm
-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 }
-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 }
-op-revm = { version = "15.0.0", default-features = false }
-revm-inspectors = "0.34.0"
+revm = { version = "33.1.0", default-features = false }
+revm-bytecode = { version = "7.1.1", default-features = false }
+revm-database = { version = "9.0.5", default-features = false }
+revm-state = { version = "8.1.1", default-features = false }
+revm-primitives = { version = "21.0.2", default-features = false }
+revm-interpreter = { version = "31.1.0", default-features = false }
+revm-database-interface = { version = "8.0.5", default-features = false }
+op-revm = { version = "14.1.0", default-features = false }
+revm-inspectors = "0.33.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-eip7928 = { version = "0.1.0", default-features = false }
+alloy-evm = { version = "0.25.1", default-features = false }
 alloy-primitives = { version = "1.5.0", default-features = false, features = ["map-foldhash"] }
 alloy-rlp = { version = "0.3.10", default-features = false, features = ["core-net"] }
 alloy-sol-macro = "1.5.0"
@@ -526,7 +526,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.0", 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,15 +739,15 @@ 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]
+# [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" }
@@ -792,8 +792,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

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
     }
 
@@ -950,20 +958,16 @@ impl<N: NodePrimitives<SignedTx: SignedTransaction>> NewCanonicalChain<N> {
             [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()),
+                    first.execution_outcome().clone(),
+                    first.trie_updates(),
+                    first.hashed_state(),
                 );
                 for exec in rest {
                     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
@@ -1260,7 +1264,7 @@ mod tests {
 
         let state = BlockState::new(block);
 
-        assert_eq!(state.receipts(), receipts.first().unwrap());
+        assert_eq!(state.receipts(), &receipts);
     }
 
     #[test]
@@ -1540,12 +1544,15 @@ mod tests {
         // 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 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());
 
         // Build expected execution outcome (first_block matches first block number)
         let commit_execution_outcome = ExecutionOutcome {
@@ -1561,7 +1568,8 @@ mod tests {
                 new: Arc::new(Chain::new(
                     vec![block0.recovered_block().clone(), block1.recovered_block().clone()],
                     commit_execution_outcome,
-                    expected_trie_data,
+                    expected_trie_updates,
+                    expected_hashed_state
                 ))
             }
         );
@@ -1572,17 +1580,25 @@ 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 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());
 
         // Build expected execution outcome for reorg chains (first_block matches first block
         // number)
@@ -1599,12 +1615,14 @@ mod tests {
                 old: Arc::new(Chain::new(
                     vec![block1.recovered_block().clone(), block2.recovered_block().clone()],
                     reorg_execution_outcome.clone(),
-                    old_trie_data,
+                    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,
+                    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,
@@ -198,7 +201,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 +209,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/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/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>;
 }
 

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/engine/invalid-block-hooks/src/witness.rs

@@ -448,14 +448,12 @@ mod tests {
                     nonce: account.nonce,
                     code_hash: account.bytecode_hash.unwrap_or_default(),
                     code: None,
-                    account_id: None,
                 }),
                 original_info: (i == 0).then(|| AccountInfo {
                     balance: account.balance.checked_div(U256::from(2)).unwrap_or(U256::ZERO),
                     nonce: 0,
                     code_hash: account.bytecode_hash.unwrap_or_default(),
                     code: None,
-                    account_id: None,
                 }),
                 storage,
                 status: AccountStatus::default(),

crates/engine/tree/Cargo.toml

@@ -34,7 +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
@@ -96,7 +95,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,7 +134,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",

crates/engine/tree/benches/channel_perf.rs

@@ -26,9 +26,7 @@ fn create_bench_state(num_accounts: usize) -> EvmState {
                 nonce: 10,
                 code_hash: B256::from_slice(&rng.random::<[u8; 32]>()),
                 code: Default::default(),
-                account_id: None,
             },
-            original_info: Box::new(AccountInfo::default()),
             storage,
             status: AccountStatus::empty(),
             transaction_id: 0,

crates/engine/tree/benches/state_root_task.rs

@@ -62,7 +62,6 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
                     storage: HashMap::default(),
                     status: AccountStatus::SelfDestructed,
                     transaction_id: 0,
-                    original_info: Box::new(AccountInfo::default()),
                 }
             } else {
                 RevmAccount {
@@ -71,7 +70,6 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
                         nonce: rng.random::<u64>(),
                         code_hash: KECCAK_EMPTY,
                         code: Some(Default::default()),
-                        account_id: None,
                     },
                     storage: (0..rng.random_range(0..=params.storage_slots_per_account))
                         .map(|_| {
@@ -86,7 +84,6 @@ fn create_bench_state_updates(params: &BenchParams) -> Vec<EvmState> {
                         })
                         .collect(),
                     status: AccountStatus::Touched,
-                    original_info: Box::new(AccountInfo::default()),
                     transaction_id: 0,
                 }
             };

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 }
         }
     }
 
@@ -519,16 +495,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 +535,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)
@@ -609,9 +580,7 @@ mod tests {
                         nonce: 10,
                         code_hash: B256::random(),
                         code: Default::default(),
-                        account_id: None,
                     },
-                    original_info: Box::new(AccountInfo::default()),
                     storage,
                     status: AccountStatus::default(),
                     transaction_id: 0,
@@ -623,12 +592,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,9 +30,9 @@ 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,
+    BlockNumReader, BlockReader, ChangeSetReader, DatabaseProviderFactory, HashedPostStateProvider,
+    ProviderError, StageCheckpointReader, StateProviderBox, StateProviderFactory, StateReader,
+    TransactionVariant,
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_stages_api::ControlFlow;
@@ -1478,7 +1478,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 +1676,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
@@ -1773,7 +1761,7 @@ where
         }
 
         let min_block = self.persistence_state.last_persisted_block.number;
-        self.state.tree_state.canonical_block_number().saturating_sub(min_block) >
+        self.state.tree_state.canonical_block_number().saturating_sub(min_block) >=
             self.config.persistence_threshold()
     }
 
@@ -1868,7 +1856,7 @@ 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()))?;
@@ -1892,19 +1880,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

@@ -101,7 +101,7 @@ impl<'a> Iterator for BALSlotIter<'a> {
                     return None;
                 }
 
-                return Some((address, StorageKey::from(slot)));
+                return Some((address, slot));
             }
 
             // Move to next account
@@ -177,11 +177,13 @@ 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(slot_changes.slot);
 
                 // Get the last change for this slot
                 if let Some(last_change) = slot_changes.changes.last() {
-                    storage_map.storage.insert(hashed_slot, last_change.new_value);
+                    storage_map
+                        .storage
+                        .insert(hashed_slot, U256::from_be_bytes(last_change.new_value.0));
                 }
             }
 
@@ -235,8 +237,8 @@ mod tests {
         let provider = StateProviderTest::default();
 
         let address = Address::random();
-        let slot = U256::random();
-        let value = U256::random();
+        let slot = StorageKey::random();
+        let value = B256::random();
 
         let slot_changes = SlotChanges { slot, changes: vec![StorageChange::new(0, value)] };
 
@@ -256,10 +258,10 @@ 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(slot);
 
         let stored_value = storage.storage.get(&hashed_slot).unwrap();
-        assert_eq!(*stored_value, value);
+        assert_eq!(*stored_value, U256::from_be_bytes(value.0));
     }
 
     #[test]
@@ -390,15 +392,15 @@ mod tests {
         let provider = StateProviderTest::default();
 
         let address = Address::random();
-        let slot = U256::random();
+        let slot = StorageKey::random();
 
         // Multiple changes to the same slot - should take the last one
         let slot_changes = SlotChanges {
             slot,
             changes: vec![
-                StorageChange::new(0, U256::from(100)),
-                StorageChange::new(1, U256::from(200)),
-                StorageChange::new(2, U256::from(300)),
+                StorageChange::new(0, B256::from(U256::from(100).to_be_bytes::<32>())),
+                StorageChange::new(1, B256::from(U256::from(200).to_be_bytes::<32>())),
+                StorageChange::new(2, B256::from(U256::from(300).to_be_bytes::<32>())),
             ],
         };
 
@@ -416,7 +418,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(slot);
 
         let stored_value = storage.storage.get(&hashed_slot).unwrap();
 
@@ -436,15 +438,15 @@ mod tests {
             address: addr1,
             storage_changes: vec![
                 SlotChanges {
-                    slot: U256::from(100),
-                    changes: vec![StorageChange::new(0, U256::ZERO)],
+                    slot: StorageKey::from(U256::from(100)),
+                    changes: vec![StorageChange::new(0, B256::ZERO)],
                 },
                 SlotChanges {
-                    slot: U256::from(101),
-                    changes: vec![StorageChange::new(0, U256::ZERO)],
+                    slot: StorageKey::from(U256::from(101)),
+                    changes: vec![StorageChange::new(0, B256::ZERO)],
                 },
             ],
-            storage_reads: vec![U256::from(102)],
+            storage_reads: vec![StorageKey::from(U256::from(102))],
             balance_changes: vec![],
             nonce_changes: vec![],
             code_changes: vec![],
@@ -454,10 +456,10 @@ mod tests {
         let account2 = AccountChanges {
             address: addr2,
             storage_changes: vec![SlotChanges {
-                slot: U256::from(200),
-                changes: vec![StorageChange::new(0, U256::ZERO)],
+                slot: StorageKey::from(U256::from(200)),
+                changes: vec![StorageChange::new(0, B256::ZERO)],
             }],
-            storage_reads: vec![U256::from(201)],
+            storage_reads: vec![StorageKey::from(U256::from(201))],
             balance_changes: vec![],
             nonce_changes: vec![],
             code_changes: vec![],
@@ -468,15 +470,15 @@ mod tests {
             address: addr3,
             storage_changes: vec![
                 SlotChanges {
-                    slot: U256::from(300),
-                    changes: vec![StorageChange::new(0, U256::ZERO)],
+                    slot: StorageKey::from(U256::from(300)),
+                    changes: vec![StorageChange::new(0, B256::ZERO)],
                 },
                 SlotChanges {
-                    slot: U256::from(301),
-                    changes: vec![StorageChange::new(0, U256::ZERO)],
+                    slot: StorageKey::from(U256::from(301)),
+                    changes: vec![StorageChange::new(0, B256::ZERO)],
                 },
             ],
-            storage_reads: vec![U256::from(302)],
+            storage_reads: vec![StorageKey::from(U256::from(302))],
             balance_changes: vec![],
             nonce_changes: vec![],
             code_changes: vec![],

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 });
         }
     }
 }
@@ -1071,9 +1059,7 @@ mod tests {
                         nonce: rng.random::<u64>(),
                         code_hash: KECCAK_EMPTY,
                         code: Some(Default::default()),
-                        account_id: None,
                     },
-                    original_info: Box::new(AccountInfo::default()),
                     storage,
                     status: AccountStatus::Touched,
                     transaction_id: 0,

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
@@ -1816,9 +1811,7 @@ mod tests {
                     nonce: 1,
                     code_hash: Default::default(),
                     code: Default::default(),
-                    account_id: None,
                 },
-                original_info: Box::new(revm_state::AccountInfo::default()),
                 transaction_id: Default::default(),
                 storage: Default::default(),
                 status: revm_state::AccountStatus::Touched,
@@ -1835,9 +1828,7 @@ mod tests {
                     nonce: 2,
                     code_hash: Default::default(),
                     code: Default::default(),
-                    account_id: None,
                 },
-                original_info: Box::new(revm_state::AccountInfo::default()),
                 transaction_id: Default::default(),
                 storage: Default::default(),
                 status: revm_state::AccountStatus::Touched,
@@ -1939,9 +1930,7 @@ mod tests {
                     nonce: 1,
                     code_hash: Default::default(),
                     code: Default::default(),
-                    account_id: None,
                 },
-                original_info: Box::new(revm_state::AccountInfo::default()),
                 transaction_id: Default::default(),
                 storage: Default::default(),
                 status: revm_state::AccountStatus::Touched,

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,12 +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,
 };
 use reth_revm::db::State;
-use reth_trie::{updates::TrieUpdates, HashedPostState, StateRoot};
+use reth_trie::{
+    updates::{TrieUpdates, TrieUpdatesSorted},
+    HashedPostState, HashedPostStateSorted, StateRoot, TrieInputSorted,
+};
 use reth_trie_db::ChangesetCache;
 use reth_trie_parallel::root::{ParallelStateRoot, ParallelStateRootError};
 use revm_primitives::Address;
@@ -369,6 +376,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,16 +431,26 @@ 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());
+        // Compute trie input from ancestors once, before spawning payload processor.
+        // This will be extended with the current block's hashed state after execution.
+        let trie_input_start = Instant::now();
+        let (trie_input, block_hash_for_overlay) =
+            ensure_ok!(self.compute_trie_input(parent_hash, ctx.state()));
+
+        self.metrics
+            .block_validation
+            .trie_input_duration
+            .record(trie_input_start.elapsed().as_secs_f64());
 
         // Create overlay factory for payload processor (StateRootTask path needs it for
         // multiproofs)
-        let overlay_factory =
+        let overlay_factory = {
+            let TrieInputSorted { nodes, state, .. } = &trie_input;
             OverlayStateProviderFactory::new(self.provider.clone(), self.changeset_cache.clone())
-                .with_block_hash(Some(anchor_hash))
-                .with_lazy_overlay(lazy_overlay);
+                .with_block_hash(Some(block_hash_for_overlay))
+                .with_trie_overlay(Some(Arc::clone(nodes)))
+                .with_hashed_state_overlay(Some(Arc::clone(state)))
+        };
 
         // Spawn the appropriate processor based on strategy
         let mut handle = ensure_ok!(self.spawn_payload_processor(
@@ -455,44 +473,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
         );
 
@@ -591,13 +584,16 @@ 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)));
+
+        // Terminate prewarming task with the shared execution outcome
+        handle.terminate_caching(Some(Arc::clone(&execution_outcome)));
 
         Ok(self.spawn_deferred_trie_task(
             block,
-            output,
+            execution_outcome,
             &ctx,
             hashed_state,
             trie_output,
@@ -640,21 +636,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,
@@ -693,14 +681,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,22 +688,11 @@ 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.
@@ -781,9 +750,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 +757,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 +783,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())
@@ -991,36 +954,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,7 +1097,7 @@ 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,
@@ -1289,7 +1344,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/tests.rs

@@ -27,7 +27,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,
@@ -491,7 +491,7 @@ fn test_tree_persist_block_batch() {
     let chain_spec = MAINNET.clone();
     let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());
 
-    // we need more than tree_config.persistence_threshold() +1 blocks to
+    // we need at least tree_config.persistence_threshold() + 1 blocks to
     // trigger the persistence task.
     let blocks: Vec<_> = test_block_builder
         .get_executed_blocks(1..tree_config.persistence_threshold() + 2)
@@ -531,7 +531,7 @@ async fn test_tree_persist_blocks() {
     let chain_spec = MAINNET.clone();
     let mut test_block_builder = TestBlockBuilder::eth().with_chain_spec((*chain_spec).clone());
 
-    // we need more than tree_config.persistence_threshold() +1 blocks to
+    // we need at least tree_config.persistence_threshold() + 1 blocks to
     // trigger the persistence task.
     let blocks: Vec<_> = test_block_builder
         .get_executed_blocks(1..tree_config.persistence_threshold() + 2)
@@ -838,7 +838,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 +847,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 +1008,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/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,15 +74,8 @@ 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,
-        )
+        validate_block_post_execution(block, &self.chain_spec, &result.receipts, &result.requests)
     }
 }
 

crates/ethereum/consensus/src/validation.rs

@@ -12,15 +12,11 @@ use reth_primitives_traits::{
 ///
 /// - 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)>,
 ) -> Result<(), ConsensusError>
 where
     B: Block,
@@ -41,26 +37,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

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: &[],
@@ -240,9 +238,8 @@ where
             revm_spec_by_timestamp_and_block_number(self.chain_spec(), timestamp, block_number);
 
         // configure evm env based on parent block
-        let mut cfg_env = CfgEnv::new()
-            .with_chain_id(self.chain_spec().chain().id())
-            .with_spec_and_mainnet_gas_params(spec);
+        let mut cfg_env =
+            CfgEnv::new().with_chain_id(self.chain_spec().chain().id()).with_spec(spec);
 
         if let Some(blob_params) = &blob_params {
             cfg_env.set_max_blobs_per_tx(blob_params.max_blobs_per_tx);
@@ -283,7 +280,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: &[],
@@ -411,7 +407,7 @@ mod tests {
         let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
 
         let evm_env = EvmEnv {
-            cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(SpecId::CONSTANTINOPLE),
+            cfg_env: CfgEnv::new().with_spec(SpecId::CONSTANTINOPLE),
             ..Default::default()
         };
 
@@ -478,7 +474,7 @@ mod tests {
         let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
 
         let evm_env = EvmEnv {
-            cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(SpecId::CONSTANTINOPLE),
+            cfg_env: CfgEnv::new().with_spec(SpecId::CONSTANTINOPLE),
             ..Default::default()
         };
 

crates/ethereum/evm/src/test_utils.rs

@@ -65,12 +65,7 @@ impl BlockExecutorFactory for MockEvmConfig {
         DB: Database + 'a,
         I: Inspector<<Self::EvmFactory as EvmFactory>::Context<&'a mut State<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 }
     }
 }
 
@@ -81,7 +76,6 @@ pub struct MockExecutor<'a, DB: Database, I> {
     evm: EthEvm<&'a mut State<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
@@ -95,10 +89,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>,

crates/ethereum/evm/tests/execute.rs

@@ -38,7 +38,6 @@ fn create_database_with_beacon_root_contract() -> CacheDB<EmptyDB> {
         code_hash: keccak256(BEACON_ROOTS_CODE.clone()),
         nonce: 1,
         code: Some(Bytecode::new_raw(BEACON_ROOTS_CODE.clone())),
-        account_id: None,
     };
 
     db.insert_account_info(BEACON_ROOTS_ADDRESS, beacon_root_contract_account);
@@ -54,7 +53,6 @@ fn create_database_with_withdrawal_requests_contract() -> CacheDB<EmptyDB> {
         balance: U256::ZERO,
         code_hash: keccak256(WITHDRAWAL_REQUEST_PREDEPLOY_CODE.clone()),
         code: Some(Bytecode::new_raw(WITHDRAWAL_REQUEST_PREDEPLOY_CODE.clone())),
-        account_id: None,
     };
 
     db.insert_account_info(
@@ -341,7 +339,6 @@ fn create_database_with_block_hashes(latest_block: u64) -> CacheDB<EmptyDB> {
         code_hash: keccak256(HISTORY_STORAGE_CODE.clone()),
         code: Some(Bytecode::new_raw(HISTORY_STORAGE_CODE.clone())),
         nonce: 1,
-        account_id: None,
     };
 
     db.insert_account_info(HISTORY_STORAGE_ADDRESS, blockhashes_contract_account);

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/payload/src/lib.rs

@@ -155,7 +155,7 @@ where
     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();
+        State::builder().with_database_ref(cached_reads.as_db(state)).with_bundle_update().build();
 
     let mut builder = evm_config
         .builder_for_next_block(
@@ -247,7 +247,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/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

@@ -741,7 +741,6 @@ mod tests {
             nonce,
             code_hash: KECCAK_EMPTY,
             code: None,
-            account_id: None,
         };
         state.insert_account(addr, account_info);
         state
@@ -778,13 +777,8 @@ mod tests {
 
         let mut state = setup_state_with_account(addr1, 100, 1);
 
-        let account2 = AccountInfo {
-            balance: U256::from(200),
-            nonce: 1,
-            code_hash: KECCAK_EMPTY,
-            code: None,
-            account_id: None,
-        };
+        let account2 =
+            AccountInfo { balance: U256::from(200), nonce: 1, code_hash: KECCAK_EMPTY, code: None };
         state.insert_account(addr2, account2);
 
         let mut increments = HashMap::default();
@@ -805,13 +799,8 @@ mod tests {
 
         let mut state = setup_state_with_account(addr1, 100, 1);
 
-        let account2 = AccountInfo {
-            balance: U256::from(200),
-            nonce: 1,
-            code_hash: KECCAK_EMPTY,
-            code: None,
-            account_id: None,
-        };
+        let account2 =
+            AccountInfo { balance: U256::from(200), nonce: 1, code_hash: KECCAK_EMPTY, code: None };
         state.insert_account(addr2, account2);
 
         let mut increments = HashMap::default();

crates/evm/evm/src/lib.rs

@@ -399,7 +399,7 @@ 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: Database>(
         &'a self,
         db: &'a mut State<DB>,
         parent: &'a SealedHeader<<Self::Primitives as NodePrimitives>::BlockHeader>,

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
@@ -942,20 +934,10 @@ mod tests {
         let address3 = Address::random();
 
         // Set up account info with some changes
-        let account_info1 = AccountInfo {
-            nonce: 1,
-            balance: U256::from(100),
-            code_hash: B256::ZERO,
-            code: None,
-            account_id: None,
-        };
-        let account_info2 = AccountInfo {
-            nonce: 2,
-            balance: U256::from(200),
-            code_hash: B256::ZERO,
-            code: None,
-            account_id: None,
-        };
+        let account_info1 =
+            AccountInfo { nonce: 1, balance: U256::from(100), code_hash: B256::ZERO, code: None };
+        let account_info2 =
+            AccountInfo { nonce: 2, balance: U256::from(200), code_hash: B256::ZERO, code: None };
 
         // Set up the bundle state with these accounts
         let mut bundle_state = BundleState::default();

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,7 +189,7 @@ 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};
 
@@ -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/optimism/consensus/src/lib.rs

@@ -18,7 +18,7 @@ use alloy_consensus::{
 use alloy_primitives::B64;
 use core::fmt::Debug;
 use reth_chainspec::EthChainSpec;
-use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator, ReceiptRootBloom};
+use reth_consensus::{Consensus, ConsensusError, FullConsensus, HeaderValidator};
 use reth_consensus_common::validation::{
     validate_against_parent_eip1559_base_fee, validate_against_parent_hash_number,
     validate_against_parent_timestamp, validate_cancun_gas, validate_header_base_fee,
@@ -79,9 +79,8 @@ where
         &self,
         block: &RecoveredBlock<N::Block>,
         result: &BlockExecutionResult<N::Receipt>,
-        receipt_root_bloom: Option<ReceiptRootBloom>,
     ) -> Result<(), ConsensusError> {
-        validate_block_post_execution(block.header(), &self.chain_spec, result, receipt_root_bloom)
+        validate_block_post_execution(block.header(), &self.chain_spec, result)
     }
 }
 
@@ -411,8 +410,7 @@ mod tests {
         let post_execution = <OpBeaconConsensus<OpChainSpec> as FullConsensus<OpPrimitives>>::validate_block_post_execution(
             &beacon_consensus,
             &block,
-            &result,
-            None,
+            &result
         );
 
         // validate blob, it should pass blob gas used validation
@@ -481,8 +479,7 @@ mod tests {
         let post_execution = <OpBeaconConsensus<OpChainSpec> as FullConsensus<OpPrimitives>>::validate_block_post_execution(
             &beacon_consensus,
             &block,
-            &result,
-            None,
+            &result
         );
 
         // validate blob, it should fail blob gas used validation post execution.

crates/optimism/consensus/src/validation/mod.rs

@@ -85,14 +85,10 @@ where
 ///
 /// - 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<R: DepositReceipt>(
     header: impl BlockHeader,
     chain_spec: impl OpHardforks,
     result: &BlockExecutionResult<R>,
-    receipt_root_bloom: Option<(B256, Bloom)>,
 ) -> Result<(), ConsensusError> {
     // Validate that the blob gas used is present and correctly computed if Jovian is active.
     if chain_spec.is_jovian_active_at_timestamp(header.timestamp()) {
@@ -114,32 +110,21 @@ pub fn validate_block_post_execution<R: DepositReceipt>(
     // 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(header.number()) {
-        let result = if let Some((receipts_root, logs_bloom)) = receipt_root_bloom {
-            compare_receipts_root_and_logs_bloom(
-                receipts_root,
-                logs_bloom,
-                header.receipts_root(),
-                header.logs_bloom(),
-            )
-        } else {
-            verify_receipts_optimism(
-                header.receipts_root(),
-                header.logs_bloom(),
-                receipts,
-                chain_spec,
-                header.timestamp(),
-            )
-        };
-
-        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(header.number()) &&
+        let Err(error) = verify_receipts_optimism(
+            header.receipts_root(),
+            header.logs_bloom(),
+            receipts,
+            chain_spec,
+            header.timestamp(),
+        )
+    {
+        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)
     }
 
     // Check if gas used matches the value set in header.
@@ -558,7 +543,7 @@ mod tests {
             requests: Requests::default(),
             gas_used: GAS_USED,
         };
-        validate_block_post_execution(&header, &chainspec, &result, None).unwrap();
+        validate_block_post_execution(&header, &chainspec, &result).unwrap();
     }
 
     #[test]
@@ -580,7 +565,7 @@ mod tests {
             gas_used: GAS_USED,
         };
         assert!(matches!(
-            validate_block_post_execution(&header, &chainspec, &result, None).unwrap_err(),
+            validate_block_post_execution(&header, &chainspec, &result).unwrap_err(),
             ConsensusError::BlobGasUsedDiff(diff)
                 if diff.got == BLOB_GAS_USED && diff.expected == BLOB_GAS_USED + 1
         ));

crates/optimism/evm/src/lib.rs

@@ -230,9 +230,7 @@ where
 
         let spec = revm_spec_by_timestamp_after_bedrock(self.chain_spec(), timestamp);
 
-        let cfg_env = CfgEnv::new()
-            .with_chain_id(self.chain_spec().chain().id())
-            .with_spec_and_mainnet_gas_params(spec);
+        let cfg_env = CfgEnv::new().with_chain_id(self.chain_spec().chain().id()).with_spec(spec);
 
         let blob_excess_gas_and_price = spec
             .into_eth_spec()
@@ -364,8 +362,7 @@ mod tests {
         let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
 
         // Create a custom configuration environment with a chain ID of 111
-        let cfg =
-            CfgEnv::new().with_chain_id(111).with_spec_and_mainnet_gas_params(OpSpecId::default());
+        let cfg = CfgEnv::new().with_chain_id(111).with_spec(OpSpecId::default());
 
         let evm_env = EvmEnv { cfg_env: cfg.clone(), ..Default::default() };
 
@@ -403,10 +400,8 @@ mod tests {
 
         let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
 
-        let evm_env = EvmEnv {
-            cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(OpSpecId::ECOTONE),
-            ..Default::default()
-        };
+        let evm_env =
+            EvmEnv { cfg_env: CfgEnv::new().with_spec(OpSpecId::ECOTONE), ..Default::default() };
 
         let evm = evm_config.evm_with_env(db, evm_env.clone());
 
@@ -432,8 +427,7 @@ mod tests {
         let evm_config = test_evm_config();
         let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
 
-        let cfg =
-            CfgEnv::new().with_chain_id(111).with_spec_and_mainnet_gas_params(OpSpecId::default());
+        let cfg = CfgEnv::new().with_chain_id(111).with_spec(OpSpecId::default());
         let block = BlockEnv::default();
         let evm_env = EvmEnv { block_env: block, cfg_env: cfg.clone() };
 
@@ -469,10 +463,8 @@ mod tests {
         let evm_config = test_evm_config();
         let db = CacheDB::<EmptyDBTyped<ProviderError>>::default();
 
-        let evm_env = EvmEnv {
-            cfg_env: CfgEnv::new().with_spec_and_mainnet_gas_params(OpSpecId::ECOTONE),
-            ..Default::default()
-        };
+        let evm_env =
+            EvmEnv { cfg_env: CfgEnv::new().with_spec(OpSpecId::ECOTONE), ..Default::default() };
 
         let evm = evm_config.evm_with_env_and_inspector(db, evm_env.clone(), NoOpInspector {});
 
@@ -529,8 +521,12 @@ mod tests {
 
         // Create a Chain object with a BTreeMap of blocks mapped to their block numbers,
         // including block1_hash and block2_hash, and the execution_outcome
-        let chain: Chain<OpPrimitives> =
-            Chain::new([block1, block2], execution_outcome.clone(), BTreeMap::new());
+        let chain: Chain<OpPrimitives> = Chain::new(
+            [block1, block2],
+            execution_outcome.clone(),
+            BTreeMap::new(),
+            BTreeMap::new(),
+        );
 
         // Assert that the proper receipt vector is returned for block1_hash
         assert_eq!(chain.receipts_by_block_hash(block1_hash), Some(vec![&receipt1]));

crates/optimism/flashblocks/src/service.rs

@@ -12,18 +12,10 @@ use reth_primitives_traits::{AlloyBlockHeader, BlockTy, HeaderTy, NodePrimitives
 use reth_revm::cached::CachedReads;
 use reth_storage_api::{BlockReaderIdExt, StateProviderFactory};
 use reth_tasks::TaskExecutor;
-use std::{
-    sync::Arc,
-    time::{Duration, Instant},
-};
-use tokio::{
-    sync::{oneshot, watch},
-    time::sleep,
-};
+use std::{sync::Arc, time::Instant};
+use tokio::sync::{oneshot, watch};
 use tracing::*;
 
-const CONNECTION_BACKOUT_PERIOD: Duration = Duration::from_secs(5);
-
 /// The `FlashBlockService` maintains an in-memory [`PendingFlashBlock`] built out of a sequence of
 /// [`FlashBlock`]s.
 #[derive(Debug)]
@@ -175,13 +167,7 @@ where
                             self.try_start_build_job();
                         }
                         Some(Err(err)) => {
-                            warn!(
-                                target: "flashblocks",
-                                %err,
-                                retry_period = CONNECTION_BACKOUT_PERIOD.as_secs(),
-                                "Error receiving flashblock"
-                            );
-                            sleep(CONNECTION_BACKOUT_PERIOD).await;
+                            warn!(target: "flashblocks", %err, "Error receiving flashblock");
                         }
                         None => {
                             warn!(target: "flashblocks", "Flashblock stream ended");

crates/optimism/flashblocks/src/worker.rs

@@ -8,8 +8,10 @@ use reth_evm::{
     execute::{BlockBuilder, BlockBuilderOutcome},
     ConfigureEvm,
 };
-use reth_execution_types::BlockExecutionOutput;
-use reth_primitives_traits::{BlockTy, HeaderTy, NodePrimitives, ReceiptTy, Recovered};
+use reth_execution_types::ExecutionOutcome;
+use reth_primitives_traits::{
+    AlloyBlockHeader, BlockTy, HeaderTy, NodePrimitives, ReceiptTy, Recovered,
+};
 use reth_revm::{cached::CachedReads, database::StateProviderDatabase, db::State};
 use reth_rpc_eth_types::{EthApiError, PendingBlock};
 use reth_storage_api::{noop::NoopProvider, BlockReaderIdExt, StateProviderFactory};
@@ -110,8 +112,12 @@ where
                 builder.finish(NoopProvider::default())?
             };
 
-        let execution_outcome =
-            BlockExecutionOutput { state: state.take_bundle(), result: execution_result };
+        let execution_outcome = ExecutionOutcome::new(
+            state.take_bundle(),
+            vec![execution_result.receipts],
+            block.number(),
+            vec![execution_result.requests],
+        );
 
         let pending_block = PendingBlock::with_executed_block(
             Instant::now() + Duration::from_secs(1),

crates/optimism/node/tests/it/builder.rs

@@ -19,7 +19,7 @@ use reth_optimism_node::{args::RollupArgs, OpEvmConfig, OpExecutorBuilder, OpNod
 use reth_optimism_primitives::OpPrimitives;
 use reth_provider::providers::BlockchainProvider;
 use revm::{
-    context::{BlockEnv, ContextTr, TxEnv},
+    context::{BlockEnv, Cfg, ContextTr, TxEnv},
     context_interface::result::EVMError,
     inspector::NoOpInspector,
     interpreter::interpreter::EthInterpreter,
@@ -103,7 +103,7 @@ fn test_setup_custom_precompiles() {
             input: EvmEnv<OpSpecId>,
         ) -> Self::Evm<DB, NoOpInspector> {
             let mut op_evm = OpEvmFactory::default().create_evm(db, input);
-            *op_evm.components_mut().2 = UniPrecompiles::precompiles(*op_evm.ctx().cfg().spec());
+            *op_evm.components_mut().2 = UniPrecompiles::precompiles(op_evm.ctx().cfg().spec());
 
             op_evm
         }
@@ -119,7 +119,7 @@ fn test_setup_custom_precompiles() {
         ) -> Self::Evm<DB, I> {
             let mut op_evm =
                 OpEvmFactory::default().create_evm_with_inspector(db, input, inspector);
-            *op_evm.components_mut().2 = UniPrecompiles::precompiles(*op_evm.ctx().cfg().spec());
+            *op_evm.components_mut().2 = UniPrecompiles::precompiles(op_evm.ctx().cfg().spec());
 
             op_evm
         }

crates/optimism/payload/src/builder.rs

@@ -18,7 +18,7 @@ use reth_evm::{
     op_revm::{constants::L1_BLOCK_CONTRACT, L1BlockInfo},
     ConfigureEvm, Database,
 };
-use reth_execution_types::BlockExecutionOutput;
+use reth_execution_types::ExecutionOutcome;
 use reth_optimism_forks::OpHardforks;
 use reth_optimism_primitives::{transaction::OpTransaction, L2_TO_L1_MESSAGE_PASSER_ADDRESS};
 use reth_optimism_txpool::{
@@ -375,8 +375,12 @@ impl<Txs> OpBuilder<'_, Txs> {
         let sealed_block = Arc::new(block.sealed_block().clone());
         debug!(target: "payload_builder", id=%ctx.attributes().payload_id(), sealed_block_header = ?sealed_block.header(), "sealed built block");
 
-        let execution_outcome =
-            BlockExecutionOutput { state: db.take_bundle(), result: execution_result };
+        let execution_outcome = ExecutionOutcome::new(
+            db.take_bundle(),
+            vec![execution_result.receipts],
+            block.number(),
+            Vec::new(),
+        );
 
         // create the executed block data
         let executed: BuiltPayloadExecutedBlock<N> = BuiltPayloadExecutedBlock {
@@ -630,7 +634,7 @@ where
             if sequencer_tx.value().is_eip4844() {
                 return Err(PayloadBuilderError::other(
                     OpPayloadBuilderError::BlobTransactionRejected,
-                ));
+                ))
             }
 
             // Convert the transaction to a [RecoveredTx]. This is

crates/payload/primitives/src/traits.rs

@@ -11,7 +11,7 @@ use alloy_rpc_types_engine::{PayloadAttributes as EthPayloadAttributes, PayloadI
 use core::fmt;
 use either::Either;
 use reth_chain_state::ComputedTrieData;
-use reth_execution_types::BlockExecutionOutput;
+use reth_execution_types::ExecutionOutcome;
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
 use reth_trie_common::{
     updates::{TrieUpdates, TrieUpdatesSorted},
@@ -27,7 +27,7 @@ pub struct BuiltPayloadExecutedBlock<N: NodePrimitives> {
     /// 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>>,
     /// Block's hashed state.
     ///
     /// Supports both unsorted and sorted variants so payload builders can avoid cloning in order

crates/primitives-traits/src/account.rs

@@ -238,15 +238,12 @@ impl From<Account> for AccountInfo {
             nonce: reth_acc.nonce,
             code_hash: reth_acc.bytecode_hash.unwrap_or(KECCAK_EMPTY),
             code: None,
-            account_id: None,
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
     use super::*;
     use alloy_primitives::{hex_literal::hex, B256, U256};
     use reth_codecs::Compact;
@@ -307,12 +304,11 @@ mod tests {
         assert_eq!(len, 17);
 
         let mut buf = vec![];
-        let bytecode =
-            Bytecode(RevmBytecode::LegacyAnalyzed(Arc::new(LegacyAnalyzedBytecode::new(
-                Bytes::from(&hex!("ff00")),
-                2,
-                JumpTable::from_slice(&[0], 2),
-            ))));
+        let bytecode = Bytecode(RevmBytecode::LegacyAnalyzed(LegacyAnalyzedBytecode::new(
+            Bytes::from(&hex!("ff00")),
+            2,
+            JumpTable::from_slice(&[0], 2),
+        )));
         let len = bytecode.to_compact(&mut buf);
         assert_eq!(len, 16);
 

crates/primitives-traits/src/transaction/recover.rs

@@ -1,137 +1,68 @@
 //! Helpers for recovering signers from a set of transactions
 
-use crate::{transaction::signed::RecoveryError, Recovered, SignedTransaction};
-use alloc::vec::Vec;
-use alloy_consensus::transaction::SignerRecoverable;
-use alloy_primitives::Address;
+#[cfg(feature = "rayon")]
+pub use rayon::*;
+
+#[cfg(not(feature = "rayon"))]
+pub use iter::*;
 
 #[cfg(feature = "rayon")]
-use rayon::prelude::{IntoParallelIterator, ParallelIterator};
+mod rayon {
+    use crate::{transaction::signed::RecoveryError, SignedTransaction};
+    use alloc::vec::Vec;
+    use alloy_primitives::Address;
+    use rayon::prelude::{IntoParallelIterator, ParallelIterator};
 
-/// Recovers a list of signers from a transaction list iterator.
-///
-/// Returns `Err(RecoveryError)`, if some transaction's signature is invalid.
-///
-/// When the `rayon` feature is enabled, recovery is performed in parallel.
-#[cfg(feature = "rayon")]
-pub fn recover_signers<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
-where
-    T: SignedTransaction,
-    I: IntoParallelIterator<Item = &'a T>,
-{
-    txes.into_par_iter().map(|tx| tx.recover_signer()).collect()
+    /// Recovers a list of signers from a transaction list iterator.
+    ///
+    /// Returns `Err(RecoveryError)`, if some transaction's signature is invalid
+    pub fn recover_signers<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
+    where
+        T: SignedTransaction,
+        I: IntoParallelIterator<Item = &'a T>,
+    {
+        txes.into_par_iter().map(|tx| tx.recover_signer()).collect()
+    }
+
+    /// Recovers a list of signers from a transaction list iterator _without ensuring that the
+    /// signature has a low `s` value_.
+    ///
+    /// Returns `Err(RecoveryError)`, if some transaction's signature is invalid.
+    pub fn recover_signers_unchecked<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
+    where
+        T: SignedTransaction,
+        I: IntoParallelIterator<Item = &'a T>,
+    {
+        txes.into_par_iter().map(|tx| tx.recover_signer_unchecked()).collect()
+    }
 }
 
-/// Recovers a list of signers from a transaction list iterator.
-///
-/// Returns `Err(RecoveryError)`, if some transaction's signature is invalid.
 #[cfg(not(feature = "rayon"))]
-pub fn recover_signers<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
-where
-    T: SignedTransaction,
-    I: IntoIterator<Item = &'a T>,
-{
-    txes.into_iter().map(|tx| tx.recover_signer()).collect()
-}
-
-/// Recovers a list of signers from a transaction list iterator _without ensuring that the
-/// signature has a low `s` value_.
-///
-/// Returns `Err(RecoveryError)`, if some transaction's signature is invalid.
-///
-/// When the `rayon` feature is enabled, recovery is performed in parallel.
-#[cfg(feature = "rayon")]
-pub fn recover_signers_unchecked<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
-where
-    T: SignedTransaction,
-    I: IntoParallelIterator<Item = &'a T>,
-{
-    txes.into_par_iter().map(|tx| tx.recover_signer_unchecked()).collect()
-}
-
-/// Recovers a list of signers from a transaction list iterator _without ensuring that the
-/// signature has a low `s` value_.
-///
-/// Returns `Err(RecoveryError)`, if some transaction's signature is invalid.
-#[cfg(not(feature = "rayon"))]
-pub fn recover_signers_unchecked<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
-where
-    T: SignedTransaction,
-    I: IntoIterator<Item = &'a T>,
-{
-    txes.into_iter().map(|tx| tx.recover_signer_unchecked()).collect()
-}
-
-/// Trait for items that can be used with [`try_recover_signers`].
-#[cfg(feature = "rayon")]
-pub trait TryRecoverItems: IntoParallelIterator {}
-
-/// Trait for items that can be used with [`try_recover_signers`].
-#[cfg(not(feature = "rayon"))]
-pub trait TryRecoverItems: IntoIterator {}
-
-#[cfg(feature = "rayon")]
-impl<I: IntoParallelIterator> TryRecoverItems for I {}
-
-#[cfg(not(feature = "rayon"))]
-impl<I: IntoIterator> TryRecoverItems for I {}
-
-/// Trait for decode functions that can be used with [`try_recover_signers`].
-#[cfg(feature = "rayon")]
-pub trait TryRecoverFn<Item, T>: Fn(Item) -> Result<T, RecoveryError> + Sync {}
-
-/// Trait for decode functions that can be used with [`try_recover_signers`].
-#[cfg(not(feature = "rayon"))]
-pub trait TryRecoverFn<Item, T>: Fn(Item) -> Result<T, RecoveryError> {}
-
-#[cfg(feature = "rayon")]
-impl<Item, T, F: Fn(Item) -> Result<T, RecoveryError> + Sync> TryRecoverFn<Item, T> for F {}
-
-#[cfg(not(feature = "rayon"))]
-impl<Item, T, F: Fn(Item) -> Result<T, RecoveryError>> TryRecoverFn<Item, T> for F {}
-
-/// Decodes and recovers a list of [`Recovered`] transactions from an iterator.
-///
-/// The `decode` closure transforms each item into a [`SignedTransaction`], which is then
-/// recovered.
-///
-/// Returns an error if decoding or signature recovery fails for any transaction.
-///
-/// When the `rayon` feature is enabled, recovery is performed in parallel.
-#[cfg(feature = "rayon")]
-pub fn try_recover_signers<I, F, T>(items: I, decode: F) -> Result<Vec<Recovered<T>>, RecoveryError>
-where
-    I: IntoParallelIterator,
-    F: Fn(I::Item) -> Result<T, RecoveryError> + Sync,
-    T: SignedTransaction,
-{
-    items
-        .into_par_iter()
-        .map(|item| {
-            let tx = decode(item)?;
-            SignerRecoverable::try_into_recovered(tx)
-        })
-        .collect()
-}
-
-/// Decodes and recovers a list of [`Recovered`] transactions from an iterator.
-///
-/// The `decode` closure transforms each item into a [`SignedTransaction`], which is then
-/// recovered.
-///
-/// Returns an error if decoding or signature recovery fails for any transaction.
-#[cfg(not(feature = "rayon"))]
-pub fn try_recover_signers<I, F, T>(items: I, decode: F) -> Result<Vec<Recovered<T>>, RecoveryError>
-where
-    I: IntoIterator,
-    F: Fn(I::Item) -> Result<T, RecoveryError>,
-    T: SignedTransaction,
-{
-    items
-        .into_iter()
-        .map(|item| {
-            let tx = decode(item)?;
-            SignerRecoverable::try_into_recovered(tx)
-        })
-        .collect()
+mod iter {
+    use crate::{transaction::signed::RecoveryError, SignedTransaction};
+    use alloc::vec::Vec;
+    use alloy_primitives::Address;
+
+    /// Recovers a list of signers from a transaction list iterator.
+    ///
+    /// Returns `Err(RecoveryError)`, if some transaction's signature is invalid
+    pub fn recover_signers<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
+    where
+        T: SignedTransaction,
+        I: IntoIterator<Item = &'a T>,
+    {
+        txes.into_iter().map(|tx| tx.recover_signer()).collect()
+    }
+
+    /// Recovers a list of signers from a transaction list iterator _without ensuring that the
+    /// signature has a low `s` value_.
+    ///
+    /// Returns `Err(RecoveryError)`, if some transaction's signature is invalid.
+    pub fn recover_signers_unchecked<'a, I, T>(txes: I) -> Result<Vec<Address>, RecoveryError>
+    where
+        T: SignedTransaction,
+        I: IntoIterator<Item = &'a T>,
+    {
+        txes.into_iter().map(|tx| tx.recover_signer_unchecked()).collect()
+    }
 }

crates/rpc/rpc-eth-api/src/helpers/call.rs

@@ -25,11 +25,7 @@ use reth_evm::{
 };
 use reth_node_api::BlockBody;
 use reth_primitives_traits::Recovered;
-use reth_revm::{
-    cancelled::CancelOnDrop,
-    database::StateProviderDatabase,
-    db::{bal::EvmDatabaseError, State},
-};
+use reth_revm::{cancelled::CancelOnDrop, database::StateProviderDatabase, db::State};
 use reth_rpc_convert::{RpcConvert, RpcTxReq};
 use reth_rpc_eth_types::{
     cache::db::StateProviderTraitObjWrapper,
@@ -512,7 +508,7 @@ pub trait Call:
         tx_env: TxEnvFor<Self::Evm>,
     ) -> Result<ResultAndState<HaltReasonFor<Self::Evm>>, Self::Error>
     where
-        DB: Database<Error = EvmDatabaseError<ProviderError>> + fmt::Debug,
+        DB: Database<Error = ProviderError> + fmt::Debug,
     {
         let mut evm = self.evm_config().evm_with_env(db, evm_env);
         let res = evm.transact(tx_env).map_err(Self::Error::from_evm_err)?;
@@ -530,7 +526,7 @@ pub trait Call:
         inspector: I,
     ) -> Result<ResultAndState<HaltReasonFor<Self::Evm>>, Self::Error>
     where
-        DB: Database<Error = EvmDatabaseError<ProviderError>> + fmt::Debug,
+        DB: Database<Error = ProviderError> + fmt::Debug,
         I: InspectorFor<Self::Evm, DB>,
     {
         let mut evm = self.evm_config().evm_with_env_and_inspector(db, evm_env, inspector);
@@ -707,7 +703,7 @@ pub trait Call:
         target_tx_hash: B256,
     ) -> Result<usize, Self::Error>
     where
-        DB: Database<Error = EvmDatabaseError<ProviderError>> + DatabaseCommit + core::fmt::Debug,
+        DB: Database<Error = ProviderError> + DatabaseCommit + core::fmt::Debug,
         I: IntoIterator<Item = Recovered<&'a ProviderTx<Self::Provider>>>,
     {
         let mut evm = self.evm_config().evm_with_env(db, evm_env);

crates/rpc/rpc-eth-api/src/helpers/estimate.rs

@@ -12,7 +12,7 @@ use reth_errors::ProviderError;
 use reth_evm::{ConfigureEvm, Database, Evm, EvmEnvFor, EvmFor, TransactionEnv, TxEnvFor};
 use reth_revm::{
     database::{EvmStateProvider, StateProviderDatabase},
-    db::{bal::EvmDatabaseError, State},
+    db::State,
 };
 use reth_rpc_convert::{RpcConvert, RpcTxReq};
 use reth_rpc_eth_types::{
@@ -165,7 +165,7 @@ pub trait EstimateCall: Call {
                 return Err(RpcInvalidTransactionError::GasRequiredExceedsAllowance {
                     gas_limit: tx_env.gas_limit(),
                 }
-                .into_eth_err());
+                .into_eth_err())
             }
             // Propagate other results (successful or other errors).
             ethres => ethres?,
@@ -186,7 +186,7 @@ pub trait EstimateCall: Call {
                 } else {
                     // the transaction did revert
                     Err(Self::Error::from_revert(output))
-                };
+                }
             }
         };
 
@@ -313,7 +313,7 @@ pub trait EstimateCall: Call {
         max_gas_limit: u64,
     ) -> Result<U256, Self::Error>
     where
-        DB: Database<Error = EvmDatabaseError<ProviderError>>,
+        DB: Database<Error = ProviderError>,
         EthApiError: From<DB::Error>,
     {
         let req_gas_limit = tx_env.gas_limit();

crates/rpc/rpc-eth-api/src/helpers/pending_block.rs

@@ -12,7 +12,7 @@ use reth_chain_state::{BlockState, ComputedTrieData, ExecutedBlock};
 use reth_chainspec::{ChainSpecProvider, EthChainSpec};
 use reth_errors::{BlockExecutionError, BlockValidationError, ProviderError, RethError};
 use reth_evm::{
-    execute::{BlockBuilder, BlockBuilderOutcome, BlockExecutionOutput},
+    execute::{BlockBuilder, BlockBuilderOutcome, ExecutionOutcome},
     ConfigureEvm, Evm, NextBlockEnvAttributes,
 };
 use reth_primitives_traits::{transaction::error::InvalidTransactionError, HeaderTy, SealedHeader};
@@ -363,8 +363,12 @@ pub trait LoadPendingBlock:
         let BlockBuilderOutcome { execution_result, block, hashed_state, trie_updates } =
             builder.finish(NoopProvider::default()).map_err(Self::Error::from_eth_err)?;
 
-        let execution_outcome =
-            BlockExecutionOutput { state: db.take_bundle(), result: execution_result };
+        let execution_outcome = ExecutionOutcome::new(
+            db.take_bundle(),
+            vec![execution_result.receipts],
+            block.number(),
+            vec![execution_result.requests],
+        );
 
         Ok(ExecutedBlock::new(
             block.into(),

crates/rpc/rpc-eth-api/src/helpers/trace.rs

@@ -13,10 +13,7 @@ use reth_evm::{
     Evm, EvmEnvFor, EvmFor, HaltReasonFor, InspectorFor, TxEnvFor,
 };
 use reth_primitives_traits::{BlockBody, Recovered, RecoveredBlock};
-use reth_revm::{
-    database::StateProviderDatabase,
-    db::{bal::EvmDatabaseError, State},
-};
+use reth_revm::{database::StateProviderDatabase, db::State};
 use reth_rpc_eth_types::{cache::db::StateCacheDb, EthApiError};
 use reth_storage_api::{ProviderBlock, ProviderTx};
 use revm::{context::Block, context_interface::result::ResultAndState, DatabaseCommit};
@@ -35,7 +32,7 @@ pub trait Trace: LoadState<Error: FromEvmError<Self::Evm>> + Call {
         inspector: I,
     ) -> Result<ResultAndState<HaltReasonFor<Self::Evm>>, Self::Error>
     where
-        DB: Database<Error = EvmDatabaseError<ProviderError>>,
+        DB: Database<Error = ProviderError>,
         I: InspectorFor<Self::Evm, DB>,
     {
         let mut evm = self.evm_config().evm_with_env_and_inspector(db, evm_env, inspector);

crates/rpc/rpc-eth-types/src/error/api.rs

@@ -5,7 +5,6 @@ use crate::{simulate::EthSimulateError, EthApiError, RevertError};
 use alloy_primitives::Bytes;
 use reth_errors::ProviderError;
 use reth_evm::{ConfigureEvm, EvmErrorFor, HaltReasonFor};
-use reth_revm::db::bal::EvmDatabaseError;
 use revm::{context::result::ExecutionResult, context_interface::result::HaltReason};
 
 use super::RpcInvalidTransactionError;
@@ -111,12 +110,10 @@ impl AsEthApiError for EthApiError {
 
 /// Helper trait to convert from revm errors.
 pub trait FromEvmError<Evm: ConfigureEvm>:
-    From<EvmErrorFor<Evm, EvmDatabaseError<ProviderError>>>
-    + FromEvmHalt<HaltReasonFor<Evm>>
-    + FromRevert
+    From<EvmErrorFor<Evm, ProviderError>> + FromEvmHalt<HaltReasonFor<Evm>> + FromRevert
 {
     /// Converts from EVM error to this type.
-    fn from_evm_err(err: EvmErrorFor<Evm, EvmDatabaseError<ProviderError>>) -> Self {
+    fn from_evm_err(err: EvmErrorFor<Evm, ProviderError>) -> Self {
         err.into()
     }
 
@@ -134,9 +131,7 @@ pub trait FromEvmError<Evm: ConfigureEvm>:
 
 impl<T, Evm> FromEvmError<Evm> for T
 where
-    T: From<EvmErrorFor<Evm, EvmDatabaseError<ProviderError>>>
-        + FromEvmHalt<HaltReasonFor<Evm>>
-        + FromRevert,
+    T: From<EvmErrorFor<Evm, ProviderError>> + FromEvmHalt<HaltReasonFor<Evm>> + FromRevert,
     Evm: ConfigureEvm,
 {
 }

crates/rpc/rpc-eth-types/src/error/mod.rs

@@ -11,7 +11,6 @@ pub use api::{AsEthApiError, FromEthApiError, FromEvmError, IntoEthApiError};
 use core::time::Duration;
 use reth_errors::{BlockExecutionError, BlockValidationError, RethError};
 use reth_primitives_traits::transaction::{error::InvalidTransactionError, signed::RecoveryError};
-use reth_revm::db::bal::EvmDatabaseError;
 use reth_rpc_convert::{CallFeesError, EthTxEnvError, TransactionConversionError};
 use reth_rpc_server_types::result::{
     block_id_to_str, internal_rpc_err, invalid_params_rpc_err, rpc_err, rpc_error_with_code,
@@ -20,11 +19,8 @@ use reth_transaction_pool::error::{
     Eip4844PoolTransactionError, Eip7702PoolTransactionError, InvalidPoolTransactionError,
     PoolError, PoolErrorKind, PoolTransactionError,
 };
-use revm::{
-    context_interface::result::{
-        EVMError, HaltReason, InvalidHeader, InvalidTransaction, OutOfGasError,
-    },
-    state::bal::BalError,
+use revm::context_interface::result::{
+    EVMError, HaltReason, InvalidHeader, InvalidTransaction, OutOfGasError,
 };
 use revm_inspectors::tracing::{DebugInspectorError, MuxError};
 use std::convert::Infallible;
@@ -408,24 +404,6 @@ impl From<EthTxEnvError> for EthApiError {
     }
 }
 
-impl<E> From<EvmDatabaseError<E>> for EthApiError
-where
-    E: Into<Self>,
-{
-    fn from(value: EvmDatabaseError<E>) -> Self {
-        match value {
-            EvmDatabaseError::Bal(err) => err.into(),
-            EvmDatabaseError::Database(err) => err.into(),
-        }
-    }
-}
-
-impl From<BalError> for EthApiError {
-    fn from(err: BalError) -> Self {
-        Self::EvmCustom(format!("bal error: {:?}", err))
-    }
-}
-
 #[cfg(feature = "js-tracer")]
 impl From<revm_inspectors::tracing::js::JsInspectorError> for EthApiError {
     fn from(error: revm_inspectors::tracing::js::JsInspectorError) -> Self {

crates/rpc/rpc-eth-types/src/pending_block.rs

@@ -99,7 +99,9 @@ impl<N: NodePrimitives> PendingBlock<N> {
     pub fn with_executed_block(expires_at: Instant, executed_block: ExecutedBlock<N>) -> Self {
         Self {
             expires_at,
-            receipts: Arc::new(executed_block.execution_output.receipts.clone()),
+            receipts: Arc::new(
+                executed_block.execution_output.receipts.iter().flatten().cloned().collect(),
+            ),
             executed_block,
         }
     }

crates/rpc/rpc/src/testing.rs

@@ -15,7 +15,6 @@
 //! on public-facing RPC endpoints without proper authentication.
 
 use alloy_consensus::{Header, Transaction};
-use alloy_eips::eip2718::Decodable2718;
 use alloy_evm::Evm;
 use alloy_primitives::{map::HashSet, Address, U256};
 use alloy_rpc_types_engine::ExecutionPayloadEnvelopeV5;
@@ -25,14 +24,11 @@ use reth_errors::RethError;
 use reth_ethereum_engine_primitives::EthBuiltPayload;
 use reth_ethereum_primitives::EthPrimitives;
 use reth_evm::{execute::BlockBuilder, ConfigureEvm, NextBlockEnvAttributes};
-use reth_primitives_traits::{
-    transaction::{recover::try_recover_signers, signed::RecoveryError},
-    AlloyBlockHeader as BlockTrait, TxTy,
-};
+use reth_primitives_traits::{AlloyBlockHeader as BlockTrait, Recovered, TxTy};
 use reth_revm::{database::StateProviderDatabase, db::State};
 use reth_rpc_api::{TestingApiServer, TestingBuildBlockRequestV1};
 use reth_rpc_eth_api::{helpers::Call, FromEthApiError};
-use reth_rpc_eth_types::EthApiError;
+use reth_rpc_eth_types::{utils::recover_raw_transaction, EthApiError};
 use reth_storage_api::{BlockReader, HeaderProvider};
 use revm::context::Block;
 use revm_primitives::map::DefaultHashBuilder;
@@ -110,16 +106,11 @@ where
 
                 let mut invalid_senders: HashSet<Address, DefaultHashBuilder> = HashSet::default();
 
-                // Decode and recover all transactions in parallel
-                let recovered_txs = try_recover_signers(&request.transactions, |tx| {
-                    TxTy::<Evm::Primitives>::decode_2718_exact(tx.as_ref())
-                        .map_err(RecoveryError::from_source)
-                })
-                .or(Err(EthApiError::InvalidTransactionSignature))?;
+                for (idx, tx) in request.transactions.iter().enumerate() {
+                    let tx: Recovered<TxTy<Evm::Primitives>> = recover_raw_transaction(tx)?;
+                    let sender = tx.signer();
 
-                for (idx, tx) in recovered_txs.into_iter().enumerate() {
-                    let signer = tx.signer();
-                    if skip_invalid_transactions && invalid_senders.contains(&signer) {
+                    if skip_invalid_transactions && invalid_senders.contains(&sender) {
                         continue;
                     }
 
@@ -131,17 +122,17 @@ where
                                 debug!(
                                     target: "rpc::testing",
                                     tx_idx = idx,
-                                    ?signer,
+                                    ?sender,
                                     error = ?err,
                                     "Skipping invalid transaction"
                                 );
-                                invalid_senders.insert(signer);
+                                invalid_senders.insert(sender);
                                 continue;
                             }
                             debug!(
                                 target: "rpc::testing",
                                 tx_idx = idx,
-                                ?signer,
+                                ?sender,
                                 error = ?err,
                                 "Transaction execution failed"
                             );

crates/rpc/rpc/src/validation.rs

@@ -201,7 +201,7 @@ where
         // update the cached reads
         self.update_cached_reads(parent_header_hash, request_cache).await;
 
-        self.consensus.validate_block_post_execution(&block, &output, None)?;
+        self.consensus.validate_block_post_execution(&block, &output)?;
 
         self.ensure_payment(&block, &output, &message)?;
 

crates/stages/stages/src/stages/execution.rs

@@ -351,7 +351,7 @@ where
                 })
             })?;
 
-            if let Err(err) = self.consensus.validate_block_post_execution(&block, &result, None) {
+            if let Err(err) = self.consensus.validate_block_post_execution(&block, &result) {
                 return Err(StageError::Block {
                     block: Box::new(block.block_with_parent()),
                     error: BlockErrorKind::Validation(err),
@@ -423,6 +423,7 @@ where
                 blocks,
                 state.clone(),
                 BTreeMap::new(),
+                BTreeMap::new(),
             ));
 
             if previous_input.is_some() {
@@ -524,6 +525,7 @@ where
                 blocks,
                 bundle_state_with_receipts,
                 BTreeMap::new(),
+                BTreeMap::new(),
             ));
 
             debug_assert!(

crates/stages/stages/src/stages/merkle_changesets.rs

@@ -217,7 +217,7 @@ impl MerkleChangeSets {
         let compute_cumulative_state_revert = |block_number: BlockNumber| -> HashedPostStateSorted {
             let mut cumulative_revert = HashedPostStateSorted::default();
             for n in (block_number..target_end).rev() {
-                cumulative_revert.extend_ref_and_sort(get_block_state_revert(n))
+                cumulative_revert.extend_ref(get_block_state_revert(n))
             }
             cumulative_revert
         };
@@ -270,7 +270,7 @@ impl MerkleChangeSets {
 
             let trie_overlay = Arc::clone(&nodes);
             let mut nodes_mut = Arc::unwrap_or_clone(nodes);
-            nodes_mut.extend_ref_and_sort(&this_trie_updates);
+            nodes_mut.extend_ref(&this_trie_updates);
             nodes = Arc::new(nodes_mut);
 
             // Write the changesets to the DB using the trie updates produced by the block, and the

crates/stages/stages/src/stages/utils.rs

@@ -57,12 +57,12 @@ where
     let mut collector = Collector::new(etl_config.file_size, etl_config.dir.clone());
     let mut cache: HashMap<P, Vec<u64>> = HashMap::default();
 
-    let mut collect = |cache: &mut HashMap<P, Vec<u64>>| {
-        for (key, indices) in cache.drain() {
-            let last = *indices.last().expect("qed");
+    let mut collect = |cache: &HashMap<P, Vec<u64>>| {
+        for (key, indices) in cache {
+            let last = indices.last().expect("qed");
             collector.insert(
-                sharded_key_factory(key, last),
-                BlockNumberList::new_pre_sorted(indices.into_iter()),
+                sharded_key_factory(*key, *last),
+                BlockNumberList::new_pre_sorted(indices.iter().copied()),
             )?;
         }
         Ok::<(), StageError>(())
@@ -87,12 +87,13 @@ where
             current_block_number = block_number;
             flush_counter += 1;
             if flush_counter > DEFAULT_CACHE_THRESHOLD {
-                collect(&mut cache)?;
+                collect(&cache)?;
+                cache.clear();
                 flush_counter = 0;
             }
         }
     }
-    collect(&mut cache)?;
+    collect(&cache)?;
 
     Ok(collector)
 }

crates/stateless/src/validation.rs

@@ -231,14 +231,8 @@ where
         .map_err(|e| StatelessValidationError::StatelessExecutionFailed(e.to_string()))?;
 
     // Post validation checks
-    validate_block_post_execution(
-        &current_block,
-        &chain_spec,
-        &output.receipts,
-        &output.requests,
-        None,
-    )
-    .map_err(StatelessValidationError::ConsensusValidationFailed)?;
+    validate_block_post_execution(&current_block, &chain_spec, &output.receipts, &output.requests)
+        .map_err(StatelessValidationError::ConsensusValidationFailed)?;
 
     // Compute and check the post state root
     let hashed_state = HashedPostState::from_bundle_state::<KeccakKeyHasher>(&output.state.state);

crates/stateless/src/witness_db.rs

@@ -76,7 +76,6 @@ where
                 nonce: account.nonce,
                 code_hash: account.code_hash,
                 code: None,
-                account_id: None,
             })
         })
     }

crates/storage/codecs/derive/src/compact/generator.rs

@@ -69,7 +69,7 @@ pub fn generate_from_to(
     }
 }
 
-/// Generates code to implement the `Compact` trait method `from_compact`.
+/// Generates code to implement the `Compact` trait method `to_compact`.
 fn generate_from_compact(
     fields: &FieldList,
     ident: &Ident,
@@ -155,7 +155,7 @@ fn generate_from_compact(
     }
 }
 
-/// Generates code to implement the `Compact` trait method `to_compact`.
+/// Generates code to implement the `Compact` trait method `from_compact`.
 fn generate_to_compact(
     fields: &FieldList,
     ident: &Ident,

crates/storage/codecs/derive/src/compact/mod.rs

@@ -175,7 +175,7 @@ fn should_use_alt_impl(ftype: &str, segment: &syn::PathSegment) -> bool {
         let syn::PathArguments::AngleBracketed(ref args) = segment.arguments &&
         let Some(syn::GenericArgument::Type(syn::Type::Path(arg_path))) = args.args.last() &&
         let (Some(path), 1) = (arg_path.path.segments.first(), arg_path.path.segments.len()) &&
-        ["B256", "Address", "Bloom", "TxHash", "BlockHash", "CompactPlaceholder"]
+        ["B256", "Address", "Address", "Bloom", "TxHash", "BlockHash", "CompactPlaceholder"]
             .iter()
             .any(|&s| path.ident == s)
     {

crates/storage/db-api/src/cursor.rs

@@ -62,15 +62,9 @@ pub trait DbCursorRO<T: Table> {
 
 /// A read-only cursor over the dup table `T`.
 pub trait DbDupCursorRO<T: DupSort> {
-    /// Positions the cursor at the prev KV pair of the table, returning it.
-    fn prev_dup(&mut self) -> PairResult<T>;
-
     /// Positions the cursor at the next KV pair of the table, returning it.
     fn next_dup(&mut self) -> PairResult<T>;
 
-    /// Positions the cursor at the last duplicate value of the current key.
-    fn last_dup(&mut self) -> ValueOnlyResult<T>;
-
     /// Positions the cursor at the next KV pair of the table, skipping duplicates.
     fn next_no_dup(&mut self) -> PairResult<T>;
 

crates/storage/db-api/src/mock.rs

@@ -296,18 +296,6 @@ impl<T: DupSort> DbDupCursorRO<T> for CursorMock {
         Ok(None)
     }
 
-    /// Moves to the previous duplicate entry.
-    /// **Mock behavior**: Always returns `None`.
-    fn prev_dup(&mut self) -> PairResult<T> {
-        Ok(None)
-    }
-
-    /// Moves to the last duplicate entry.
-    /// **Mock behavior**: Always returns `None`.
-    fn last_dup(&mut self) -> ValueOnlyResult<T> {
-        Ok(None)
-    }
-
     /// Moves to the next entry with a different key.
     /// **Mock behavior**: Always returns `None`.
     fn next_no_dup(&mut self) -> PairResult<T> {

crates/storage/db/src/implementation/mdbx/cursor.rs

@@ -158,25 +158,11 @@ impl<K: TransactionKind, T: Table> DbCursorRO<T> for Cursor<K, T> {
 }
 
 impl<K: TransactionKind, T: DupSort> DbDupCursorRO<T> for Cursor<K, T> {
-    /// Returns the previous `(key, value)` pair of a DUPSORT table.
-    fn prev_dup(&mut self) -> PairResult<T> {
-        decode::<T>(self.inner.prev_dup())
-    }
-
     /// Returns the next `(key, value)` pair of a DUPSORT table.
     fn next_dup(&mut self) -> PairResult<T> {
         decode::<T>(self.inner.next_dup())
     }
 
-    /// Returns the last `value` of the current duplicate `key`.
-    fn last_dup(&mut self) -> ValueOnlyResult<T> {
-        self.inner
-            .last_dup()
-            .map_err(|e| DatabaseError::Read(e.into()))?
-            .map(decode_one::<T>)
-            .transpose()
-    }
-
     /// Returns the next `(key, value)` pair skipping the duplicates.
     fn next_no_dup(&mut self) -> PairResult<T> {
         decode::<T>(self.inner.next_nodup())

crates/storage/errors/Cargo.toml

@@ -26,7 +26,6 @@ derive_more.workspace = true
 thiserror.workspace = true
 
 revm-database-interface.workspace = true
-revm-state.workspace = true
 
 [features]
 default = ["std"]
@@ -40,5 +39,4 @@ std = [
     "revm-database-interface/std",
     "reth-prune-types/std",
     "reth-static-file-types/std",
-    "revm-state/std",
 ]

crates/storage/errors/src/provider.rs

@@ -6,8 +6,7 @@ use derive_more::Display;
 use reth_primitives_traits::{transaction::signed::RecoveryError, GotExpected};
 use reth_prune_types::PruneSegmentError;
 use reth_static_file_types::StaticFileSegment;
-use revm_database_interface::{bal::EvmDatabaseError, DBErrorMarker};
-use revm_state::bal::BalError;
+use revm_database_interface::DBErrorMarker;
 
 /// Provider result type.
 pub type ProviderResult<Ok> = Result<Ok, ProviderError>;
@@ -18,9 +17,6 @@ pub enum ProviderError {
     /// Database error.
     #[error(transparent)]
     Database(#[from] DatabaseError),
-    /// BAL error.
-    #[error("BAL error:{_0}")]
-    Bal(BalError),
     /// Pruning error.
     #[error(transparent)]
     Pruning(#[from] PruneSegmentError),
@@ -211,12 +207,6 @@ impl From<RecoveryError> for ProviderError {
     }
 }
 
-impl From<ProviderError> for EvmDatabaseError<ProviderError> {
-    fn from(error: ProviderError) -> Self {
-        Self::Database(error)
-    }
-}
-
 /// A root mismatch error at a given block height.
 #[derive(Clone, Debug, PartialEq, Eq, Display)]
 #[display("root mismatch at #{block_number} ({block_hash}): {root}")]

crates/storage/libmdbx-rs/src/cursor.rs

@@ -312,7 +312,7 @@ where
     }
 
     /// Position at first key-value pair greater than or equal to specified, return both key and
-    /// data, and the return code depends on an exact match.
+    /// data, and the return code depends on a exact match.
     ///
     /// For non DupSort-ed collections this works the same as [`Self::set_range()`], but returns
     /// [false] if key found exactly and [true] if greater key was found.

crates/storage/nippy-jar/src/consistency.rs

@@ -37,12 +37,12 @@ impl<H: NippyJarHeader> NippyJarChecker<H> {
         Self { jar, data_file: None, offsets_file: None }
     }
 
-    /// It will throw an error if the [`NippyJar`] is in an inconsistent state.
+    /// It will throw an error if the [`NippyJar`] is in a inconsistent state.
     pub fn check_consistency(&mut self) -> Result<(), NippyJarError> {
         self.handle_consistency(ConsistencyFailStrategy::ThrowError)
     }
 
-    /// It will attempt to heal if the [`NippyJar`] is in an inconsistent state.
+    /// It will attempt to heal if the [`NippyJar`] is in a inconsistent state.
     ///
     /// **ATTENTION**: disk commit should be handled externally by consuming `Self`
     pub fn ensure_consistency(&mut self) -> Result<(), NippyJarError> {

crates/storage/provider/src/providers/blockchain_provider.rs

@@ -790,9 +790,7 @@ mod tests {
     use reth_db_api::models::{AccountBeforeTx, StoredBlockBodyIndices};
     use reth_errors::ProviderError;
     use reth_ethereum_primitives::{Block, Receipt};
-    use reth_execution_types::{
-        BlockExecutionOutput, BlockExecutionResult, Chain, ExecutionOutcome,
-    };
+    use reth_execution_types::{Chain, ExecutionOutcome};
     use reth_primitives_traits::{RecoveredBlock, SealedBlock, SignerRecoverable};
     use reth_storage_api::{
         BlockBodyIndicesProvider, BlockHashReader, BlockIdReader, BlockNumReader, BlockReader,
@@ -911,15 +909,8 @@ mod tests {
                 .map(|block| {
                     let senders = block.senders().expect("failed to recover senders");
                     let block_receipts = receipts.get(block.number as usize).unwrap().clone();
-                    let execution_outcome = BlockExecutionOutput {
-                        result: BlockExecutionResult {
-                            receipts: block_receipts,
-                            requests: Default::default(),
-                            gas_used: 0,
-                            blob_gas_used: 0,
-                        },
-                        state: BundleState::default(),
-                    };
+                    let execution_outcome =
+                        ExecutionOutcome { receipts: vec![block_receipts], ..Default::default() };
 
                     ExecutedBlock {
                         recovered_block: Arc::new(RecoveredBlock::new_sealed(
@@ -988,7 +979,8 @@ mod tests {
                     state.parent_state_chain().last().expect("qed").block();
                 let num_hash = lowest_memory_block.recovered_block().num_hash();
 
-                let execution_output = (*lowest_memory_block.execution_output).clone();
+                let mut execution_output = (*lowest_memory_block.execution_output).clone();
+                execution_output.first_block = lowest_memory_block.recovered_block().number;
                 lowest_memory_block.execution_output = Arc::new(execution_output);
 
                 // Push to disk
@@ -1348,7 +1340,12 @@ mod tests {
 
         // Send and receive commit notifications.
         let block_2 = test_block_builder.generate_random_block(1, block_hash_1).try_recover()?;
-        let chain = Chain::new(vec![block_2], ExecutionOutcome::default(), BTreeMap::new());
+        let chain = Chain::new(
+            vec![block_2],
+            ExecutionOutcome::default(),
+            BTreeMap::new(),
+            BTreeMap::new(),
+        );
         let commit = CanonStateNotification::Commit { new: Arc::new(chain.clone()) };
         in_memory_state.notify_canon_state(commit.clone());
         let (notification_1, notification_2) = tokio::join!(rx_1.recv(), rx_2.recv());
@@ -1358,8 +1355,12 @@ mod tests {
         // Send and receive re-org notifications.
         let block_3 = test_block_builder.generate_random_block(1, block_hash_1).try_recover()?;
         let block_4 = test_block_builder.generate_random_block(2, block_3.hash()).try_recover()?;
-        let new_chain =
-            Chain::new(vec![block_3, block_4], ExecutionOutcome::default(), BTreeMap::new());
+        let new_chain = Chain::new(
+            vec![block_3, block_4],
+            ExecutionOutcome::default(),
+            BTreeMap::new(),
+            BTreeMap::new(),
+        );
         let re_org =
             CanonStateNotification::Reorg { old: Arc::new(chain), new: Arc::new(new_chain) };
         in_memory_state.notify_canon_state(re_org.clone());
@@ -1707,8 +1708,8 @@ mod tests {
                             block.clone(),
                             senders,
                         )),
-                        execution_output: Arc::new(BlockExecutionOutput {
-                            state: BundleState::new(
+                        execution_output: Arc::new(ExecutionOutcome {
+                            bundle: BundleState::new(
                                 in_memory_state.into_iter().map(|(address, (account, _))| {
                                     (address, None, Some(account.into()), Default::default())
                                 }),
@@ -1717,12 +1718,8 @@ mod tests {
                                 })],
                                 [],
                             ),
-                            result: BlockExecutionResult {
-                                receipts: Default::default(),
-                                requests: Default::default(),
-                                gas_used: 0,
-                                blob_gas_used: 0,
-                            },
+                            first_block: first_in_memory_block,
+                            ..Default::default()
                         }),
                         ..Default::default()
                     }

crates/storage/provider/src/providers/consistent.rs

@@ -1307,7 +1307,7 @@ impl<N: ProviderNodeTypes> StorageChangeSetReader for ConsistentProvider<N> {
             let changesets = state
                 .block()
                 .execution_output
-                .state
+                .bundle
                 .reverts
                 .clone()
                 .to_plain_state_reverts()
@@ -1360,7 +1360,7 @@ impl<N: ProviderNodeTypes> ChangeSetReader for ConsistentProvider<N> {
             let changesets = state
                 .block_ref()
                 .execution_output
-                .state
+                .bundle
                 .reverts
                 .clone()
                 .to_plain_state_reverts()
@@ -1406,7 +1406,7 @@ impl<N: ProviderNodeTypes> ChangeSetReader for ConsistentProvider<N> {
             let changeset = state
                 .block_ref()
                 .execution_output
-                .state
+                .bundle
                 .reverts
                 .clone()
                 .to_plain_state_reverts()
@@ -1460,7 +1460,7 @@ impl<N: ProviderNodeTypes> ChangeSetReader for ConsistentProvider<N> {
                 let block_changesets = state
                     .block_ref()
                     .execution_output
-                    .state
+                    .bundle
                     .reverts
                     .clone()
                     .to_plain_state_reverts()
@@ -1508,7 +1508,7 @@ impl<N: ProviderNodeTypes> ChangeSetReader for ConsistentProvider<N> {
                 count += state
                     .block_ref()
                     .execution_output
-                    .state
+                    .bundle
                     .reverts
                     .clone()
                     .to_plain_state_reverts()
@@ -1551,7 +1551,7 @@ impl<N: ProviderNodeTypes> StateReader for ConsistentProvider<N> {
     ) -> ProviderResult<Option<ExecutionOutcome<Self::Receipt>>> {
         if let Some(state) = self.head_block.as_ref().and_then(|b| b.block_on_chain(block.into())) {
             let state = state.block_ref().execution_outcome().clone();
-            Ok(Some(ExecutionOutcome::from((state, block))))
+            Ok(Some(state))
         } else {
             Self::get_state(self, block..=block)
         }
@@ -1571,7 +1571,7 @@ mod tests {
     use reth_chain_state::{ExecutedBlock, NewCanonicalChain};
     use reth_db_api::models::AccountBeforeTx;
     use reth_ethereum_primitives::Block;
-    use reth_execution_types::{BlockExecutionOutput, BlockExecutionResult, ExecutionOutcome};
+    use reth_execution_types::ExecutionOutcome;
     use reth_primitives_traits::{RecoveredBlock, SealedBlock};
     use reth_storage_api::{BlockReader, BlockSource, ChangeSetReader};
     use reth_testing_utils::generators::{
@@ -1883,8 +1883,8 @@ mod tests {
                             block.clone(),
                             senders,
                         )),
-                        execution_output: Arc::new(BlockExecutionOutput {
-                            state: BundleState::new(
+                        execution_output: Arc::new(ExecutionOutcome {
+                            bundle: BundleState::new(
                                 in_memory_state.into_iter().map(|(address, (account, _))| {
                                     (address, None, Some(account.into()), Default::default())
                                 }),
@@ -1893,12 +1893,8 @@ mod tests {
                                 })],
                                 [],
                             ),
-                            result: BlockExecutionResult {
-                                receipts: Default::default(),
-                                requests: Default::default(),
-                                gas_used: 0,
-                                blob_gas_used: 0,
-                            },
+                            first_block: first_in_memory_block,
+                            ..Default::default()
                         }),
                         ..Default::default()
                     }

crates/storage/provider/src/providers/database/provider.rs

@@ -16,7 +16,7 @@ use crate::{
     HeaderSyncGapProvider, HistoricalStateProvider, HistoricalStateProviderRef, HistoryWriter,
     LatestStateProvider, LatestStateProviderRef, OriginalValuesKnown, ProviderError,
     PruneCheckpointReader, PruneCheckpointWriter, RawRocksDBBatch, RevertsInit, RocksBatchArg,
-    RocksDBProviderFactory, StageCheckpointReader, StateProviderBox, StateWriter,
+    RocksDBProviderFactory, RocksTxRefArg, StageCheckpointReader, StateProviderBox, StateWriter,
     StaticFileProviderFactory, StatsReader, StorageReader, StorageTrieWriter, TransactionVariant,
     TransactionsProvider, TransactionsProviderExt, TrieWriter,
 };
@@ -47,7 +47,7 @@ use reth_db_api::{
     transaction::{DbTx, DbTxMut},
     BlockNumberList, PlainAccountState, PlainStorageState,
 };
-use reth_execution_types::{BlockExecutionOutput, BlockExecutionResult, Chain, ExecutionOutcome};
+use reth_execution_types::{Chain, ExecutionOutcome};
 use reth_node_types::{BlockTy, BodyTy, HeaderTy, NodeTypes, ReceiptTy, TxTy};
 use reth_primitives_traits::{
     Account, Block as _, BlockBody as _, Bytecode, RecoveredBlock, SealedHeader, StorageEntry,
@@ -60,7 +60,7 @@ use reth_static_file_types::StaticFileSegment;
 use reth_storage_api::{
     BlockBodyIndicesProvider, BlockBodyReader, MetadataProvider, MetadataWriter,
     NodePrimitivesProvider, StateProvider, StateWriteConfig, StorageChangeSetReader,
-    StorageSettingsCache, TryIntoHistoricalStateProvider, WriteStateInput,
+    StorageSettingsCache, TryIntoHistoricalStateProvider,
 };
 use reth_storage_errors::provider::{ProviderResult, StaticFileWriterError};
 use reth_trie::{
@@ -493,9 +493,7 @@ impl<TX: DbTx + DbTxMut + 'static, N: NodeTypesForProvider> DatabaseProvider<TX,
                 self.prune_modes.transaction_lookup.is_none_or(|m| !m.is_full())
             {
                 let start = Instant::now();
-                let total_tx_count: usize =
-                    blocks.iter().map(|b| b.recovered_block().body().transaction_count()).sum();
-                let mut all_tx_hashes = Vec::with_capacity(total_tx_count);
+                let mut all_tx_hashes = Vec::new();
                 for (i, block) in blocks.iter().enumerate() {
                     let recovered_block = block.recovered_block();
                     let mut tx_num = tx_nums[i];
@@ -537,10 +535,7 @@ impl<TX: DbTx + DbTxMut + 'static, N: NodeTypesForProvider> DatabaseProvider<TX,
                     // Skip receipts/account changesets if they're being written to static files.
                     let start = Instant::now();
                     self.write_state(
-                        WriteStateInput::Single {
-                            outcome: execution_output,
-                            block: recovered_block.number(),
-                        },
+                        execution_output,
                         OriginalValuesKnown::No,
                         StateWriteConfig {
                             write_receipts: !sf_ctx.write_receipts,
@@ -555,55 +550,13 @@ impl<TX: DbTx + DbTxMut + 'static, N: NodeTypesForProvider> DatabaseProvider<TX,
                     let start = Instant::now();
                     self.write_hashed_state(&trie_data.hashed_state)?;
                     timings.write_hashed_state += start.elapsed();
+
+                    let start = Instant::now();
+                    self.write_trie_updates_sorted(&trie_data.trie_updates)?;
+                    timings.write_trie_updates += start.elapsed();
                 }
             }
 
-            // Write all trie updates in a single batch.
-            // This reduces cursor open/close overhead from N calls to 1.
-            // Uses hybrid algorithm: extend_ref for small batches, k-way merge for large.
-            if save_mode.with_state() {
-                const MERGE_BATCH_THRESHOLD: usize = 30;
-
-                let start = Instant::now();
-                let num_blocks = blocks.len();
-
-                let merged = if num_blocks == 0 {
-                    TrieUpdatesSorted::default()
-                } else if num_blocks == 1 {
-                    // Single block: use directly (Arc::try_unwrap avoids clone if refcount is 1)
-                    match Arc::try_unwrap(blocks[0].trie_updates()) {
-                        Ok(owned) => owned,
-                        Err(arc) => (*arc).clone(),
-                    }
-                } else if num_blocks < MERGE_BATCH_THRESHOLD {
-                    // Small k: extend_ref with Arc::make_mut (copy-on-write).
-                    // Blocks are oldest-to-newest, iterate forward so newest overrides.
-                    let mut blocks_iter = blocks.iter();
-                    let mut result = blocks_iter.next().expect("non-empty").trie_updates();
-
-                    for block in blocks_iter {
-                        Arc::make_mut(&mut result)
-                            .extend_ref_and_sort(block.trie_updates().as_ref());
-                    }
-
-                    match Arc::try_unwrap(result) {
-                        Ok(owned) => owned,
-                        Err(arc) => (*arc).clone(),
-                    }
-                } else {
-                    // Large k: k-way merge is faster (O(n log k)).
-                    // Collect Arcs first to extend lifetime, then pass refs.
-                    // Blocks are oldest-to-newest, merge_batch expects newest-to-oldest.
-                    let arcs: Vec<_> = blocks.iter().rev().map(|b| b.trie_updates()).collect();
-                    TrieUpdatesSorted::merge_batch(arcs.iter().map(|arc| arc.as_ref()))
-                };
-
-                if !merged.is_empty() {
-                    self.write_trie_updates_sorted(&merged)?;
-                }
-                timings.write_trie_updates += start.elapsed();
-            }
-
             // Full mode: update history indices
             if save_mode.with_state() {
                 let start = Instant::now();
@@ -931,6 +884,25 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
     pub fn chain_spec(&self) -> &N::ChainSpec {
         &self.chain_spec
     }
+
+    /// Executes a closure with a `RocksDB` transaction for reading.
+    ///
+    /// This helper encapsulates all the cfg-gated `RocksDB` transaction handling for reads.
+    fn with_rocksdb_tx<F, R>(&self, f: F) -> ProviderResult<R>
+    where
+        F: FnOnce(RocksTxRefArg<'_>) -> ProviderResult<R>,
+    {
+        #[cfg(all(unix, feature = "rocksdb"))]
+        let rocksdb = self.rocksdb_provider();
+        #[cfg(all(unix, feature = "rocksdb"))]
+        let rocksdb_tx = rocksdb.tx();
+        #[cfg(all(unix, feature = "rocksdb"))]
+        let rocksdb_tx_ref = &rocksdb_tx;
+        #[cfg(not(all(unix, feature = "rocksdb")))]
+        let rocksdb_tx_ref = ();
+
+        f(rocksdb_tx_ref)
+    }
 }
 
 impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
@@ -2063,17 +2035,16 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
     type Receipt = ReceiptTy<N>;
 
     #[instrument(level = "debug", target = "providers::db", skip_all)]
-    fn write_state<'a>(
+    fn write_state(
         &self,
-        execution_outcome: impl Into<WriteStateInput<'a, Self::Receipt>>,
+        execution_outcome: &ExecutionOutcome<Self::Receipt>,
         is_value_known: OriginalValuesKnown,
         config: StateWriteConfig,
     ) -> ProviderResult<()> {
-        let execution_outcome = execution_outcome.into();
         let first_block = execution_outcome.first_block();
 
         let (plain_state, reverts) =
-            execution_outcome.state().to_plain_state_and_reverts(is_value_known);
+            execution_outcome.bundle.to_plain_state_and_reverts(is_value_known);
 
         self.write_state_reverts(reverts, first_block, config)?;
         self.write_state_changes(plain_state)?;
@@ -2128,7 +2099,7 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> StateWriter
         }
 
         for (idx, (receipts, first_tx_index)) in
-            execution_outcome.receipts().zip(block_indices).enumerate()
+            execution_outcome.receipts.iter().zip(block_indices).enumerate()
         {
             let block_number = first_block + idx as u64;
 
@@ -3118,7 +3089,7 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> BlockExecutionWriter
         // Update pipeline progress
         self.update_pipeline_stages(block, true)?;
 
-        Ok(Chain::new(blocks, execution_state, BTreeMap::new()))
+        Ok(Chain::new(blocks, execution_state, BTreeMap::new(), BTreeMap::new()))
     }
 
     fn remove_block_and_execution_above(&self, block: BlockNumber) -> ProviderResult<()> {
@@ -3157,15 +3128,12 @@ impl<TX: DbTxMut + DbTx + 'static, N: NodeTypesForProvider> BlockWriter
         // Wrap block in ExecutedBlock with empty execution output (no receipts/state/trie)
         let executed_block = ExecutedBlock::new(
             Arc::new(block.clone()),
-            Arc::new(BlockExecutionOutput {
-                result: BlockExecutionResult {
-                    receipts: Default::default(),
-                    requests: Default::default(),
-                    gas_used: 0,
-                    blob_gas_used: 0,
-                },
-                state: Default::default(),
-            }),
+            Arc::new(ExecutionOutcome::new(
+                Default::default(),
+                Vec::<Vec<ReceiptTy<N>>>::new(),
+                block_number,
+                vec![],
+            )),
             ComputedTrieData::default(),
         );
 

crates/storage/provider/src/providers/mod.rs

@@ -16,8 +16,8 @@ pub use static_file::{
 mod state;
 pub use state::{
     historical::{
-        compute_history_rank, history_info, needs_prev_shard_check, HistoricalStateProvider,
-        HistoricalStateProviderRef, HistoryInfo, LowestAvailableBlocks,
+        history_info, needs_prev_shard_check, HistoricalStateProvider, HistoricalStateProviderRef,
+        HistoryInfo, LowestAvailableBlocks,
     },
     latest::{LatestStateProvider, LatestStateProviderRef},
     overlay::{OverlayStateProvider, OverlayStateProviderFactory},

crates/storage/provider/src/providers/rocksdb/invariants.rs

@@ -164,7 +164,16 @@ impl RocksDBProvider {
                 self.prune_transaction_hash_numbers_in_range(provider, 0..=highest_tx)?;
             }
             (None, None) => {
-                // Both MDBX and static files are empty, nothing to check.
+                // Both MDBX and static files are empty.
+                // If checkpoint says we should have data, that's an inconsistency.
+                if checkpoint > 0 {
+                    tracing::warn!(
+                        target: "reth::providers::rocksdb",
+                        checkpoint,
+                        "Checkpoint set but no transaction data exists, unwind needed"
+                    );
+                    return Ok(Some(0));
+                }
             }
         }
 
@@ -254,27 +263,16 @@ impl RocksDBProvider {
                 }
 
                 // Find the max highest_block_number (excluding u64::MAX sentinel) across all
-                // entries. Also track if we found any non-sentinel entries.
+                // entries
                 let mut max_highest_block = 0u64;
-                let mut found_non_sentinel = false;
                 for result in self.iter::<tables::StoragesHistory>()? {
                     let (key, _) = result?;
                     let highest = key.sharded_key.highest_block_number;
-                    if highest != u64::MAX {
-                        found_non_sentinel = true;
-                        if highest > max_highest_block {
-                            max_highest_block = highest;
-                        }
+                    if highest != u64::MAX && highest > max_highest_block {
+                        max_highest_block = highest;
                     }
                 }
 
-                // If all entries are sentinel entries (u64::MAX), treat as first-run scenario.
-                // This means no completed shards exist (only sentinel shards with
-                // highest_block_number=u64::MAX), so no actual history has been indexed.
-                if !found_non_sentinel {
-                    return Ok(None);
-                }
-
                 // If any entry has highest_block > checkpoint, prune excess
                 if max_highest_block > checkpoint {
                     tracing::info!(
@@ -298,7 +296,11 @@ impl RocksDBProvider {
                 Ok(None)
             }
             None => {
-                // Empty RocksDB table, nothing to check.
+                // Empty RocksDB table
+                if checkpoint > 0 {
+                    // Stage says we should have data but we don't
+                    return Ok(Some(0));
+                }
                 Ok(None)
             }
         }
@@ -375,27 +377,16 @@ impl RocksDBProvider {
                 }
 
                 // Find the max highest_block_number (excluding u64::MAX sentinel) across all
-                // entries. Also track if we found any non-sentinel entries.
+                // entries
                 let mut max_highest_block = 0u64;
-                let mut found_non_sentinel = false;
                 for result in self.iter::<tables::AccountsHistory>()? {
                     let (key, _) = result?;
                     let highest = key.highest_block_number;
-                    if highest != u64::MAX {
-                        found_non_sentinel = true;
-                        if highest > max_highest_block {
-                            max_highest_block = highest;
-                        }
+                    if highest != u64::MAX && highest > max_highest_block {
+                        max_highest_block = highest;
                     }
                 }
 
-                // If all entries are sentinel entries (u64::MAX), treat as first-run scenario.
-                // This means no completed shards exist (only sentinel shards with
-                // highest_block_number=u64::MAX), so no actual history has been indexed.
-                if !found_non_sentinel {
-                    return Ok(None);
-                }
-
                 // If any entry has highest_block > checkpoint, prune excess
                 if max_highest_block > checkpoint {
                     tracing::info!(
@@ -422,7 +413,11 @@ impl RocksDBProvider {
                 Ok(None)
             }
             None => {
-                // Empty RocksDB table, nothing to check.
+                // Empty RocksDB table
+                if checkpoint > 0 {
+                    // Stage says we should have data but we don't
+                    return Ok(Some(0));
+                }
                 Ok(None)
             }
         }
@@ -547,7 +542,7 @@ mod tests {
     }
 
     #[test]
-    fn test_check_consistency_empty_rocksdb_with_checkpoint_is_first_run() {
+    fn test_check_consistency_empty_rocksdb_with_checkpoint_needs_unwind() {
         let temp_dir = TempDir::new().unwrap();
         let rocksdb = RocksDBBuilder::new(temp_dir.path())
             .with_table::<tables::TransactionHashNumbers>()
@@ -571,10 +566,10 @@ mod tests {
 
         let provider = factory.database_provider_ro().unwrap();
 
-        // RocksDB is empty but checkpoint says block 100 was processed.
-        // This is treated as a first-run/migration scenario - no unwind needed.
+        // RocksDB is empty but checkpoint says block 100 was processed
+        // This means RocksDB is missing data and we need to unwind to rebuild
         let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(result, None, "Empty data with checkpoint is treated as first run");
+        assert_eq!(result, Some(0), "Should require unwind to block 0 to rebuild RocksDB");
     }
 
     #[test]
@@ -655,7 +650,7 @@ mod tests {
     }
 
     #[test]
-    fn test_check_consistency_storages_history_empty_with_checkpoint_is_first_run() {
+    fn test_check_consistency_storages_history_empty_with_checkpoint_needs_unwind() {
         let temp_dir = TempDir::new().unwrap();
         let rocksdb = RocksDBBuilder::new(temp_dir.path())
             .with_table::<tables::StoragesHistory>()
@@ -679,10 +674,9 @@ mod tests {
 
         let provider = factory.database_provider_ro().unwrap();
 
-        // RocksDB is empty but checkpoint says block 100 was processed.
-        // This is treated as a first-run/migration scenario - no unwind needed.
+        // RocksDB is empty but checkpoint says block 100 was processed
         let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(result, None, "Empty RocksDB with checkpoint is treated as first run");
+        assert_eq!(result, Some(0), "Should require unwind to block 0 to rebuild StoragesHistory");
     }
 
     #[test]
@@ -984,97 +978,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn test_check_consistency_storages_history_sentinel_only_with_checkpoint_is_first_run() {
-        let temp_dir = TempDir::new().unwrap();
-        let rocksdb = RocksDBBuilder::new(temp_dir.path())
-            .with_table::<tables::StoragesHistory>()
-            .build()
-            .unwrap();
-
-        // Insert ONLY sentinel entries (highest_block_number = u64::MAX)
-        // This simulates a scenario where history tracking started but no shards were completed
-        let key_sentinel_1 = StorageShardedKey::new(Address::ZERO, B256::ZERO, u64::MAX);
-        let key_sentinel_2 = StorageShardedKey::new(Address::random(), B256::random(), u64::MAX);
-        let block_list = BlockNumberList::new_pre_sorted([10, 20, 30]);
-        rocksdb.put::<tables::StoragesHistory>(key_sentinel_1, &block_list).unwrap();
-        rocksdb.put::<tables::StoragesHistory>(key_sentinel_2, &block_list).unwrap();
-
-        // Verify entries exist (not empty table)
-        assert!(rocksdb.first::<tables::StoragesHistory>().unwrap().is_some());
-
-        // Create a test provider factory for MDBX
-        let factory = create_test_provider_factory();
-        factory.set_storage_settings_cache(
-            StorageSettings::legacy().with_storages_history_in_rocksdb(true),
-        );
-
-        // Set a checkpoint indicating we should have processed up to block 100
-        {
-            let provider = factory.database_provider_rw().unwrap();
-            provider
-                .save_stage_checkpoint(StageId::IndexStorageHistory, StageCheckpoint::new(100))
-                .unwrap();
-            provider.commit().unwrap();
-        }
-
-        let provider = factory.database_provider_ro().unwrap();
-
-        // RocksDB has only sentinel entries (no completed shards) but checkpoint is set.
-        // This is treated as a first-run/migration scenario - no unwind needed.
-        let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(
-            result, None,
-            "Sentinel-only entries with checkpoint should be treated as first run"
-        );
-    }
-
-    #[test]
-    fn test_check_consistency_accounts_history_sentinel_only_with_checkpoint_is_first_run() {
-        use reth_db_api::models::ShardedKey;
-
-        let temp_dir = TempDir::new().unwrap();
-        let rocksdb = RocksDBBuilder::new(temp_dir.path())
-            .with_table::<tables::AccountsHistory>()
-            .build()
-            .unwrap();
-
-        // Insert ONLY sentinel entries (highest_block_number = u64::MAX)
-        let key_sentinel_1 = ShardedKey::new(Address::ZERO, u64::MAX);
-        let key_sentinel_2 = ShardedKey::new(Address::random(), u64::MAX);
-        let block_list = BlockNumberList::new_pre_sorted([10, 20, 30]);
-        rocksdb.put::<tables::AccountsHistory>(key_sentinel_1, &block_list).unwrap();
-        rocksdb.put::<tables::AccountsHistory>(key_sentinel_2, &block_list).unwrap();
-
-        // Verify entries exist (not empty table)
-        assert!(rocksdb.first::<tables::AccountsHistory>().unwrap().is_some());
-
-        // Create a test provider factory for MDBX
-        let factory = create_test_provider_factory();
-        factory.set_storage_settings_cache(
-            StorageSettings::legacy().with_account_history_in_rocksdb(true),
-        );
-
-        // Set a checkpoint indicating we should have processed up to block 100
-        {
-            let provider = factory.database_provider_rw().unwrap();
-            provider
-                .save_stage_checkpoint(StageId::IndexAccountHistory, StageCheckpoint::new(100))
-                .unwrap();
-            provider.commit().unwrap();
-        }
-
-        let provider = factory.database_provider_ro().unwrap();
-
-        // RocksDB has only sentinel entries (no completed shards) but checkpoint is set.
-        // This is treated as a first-run/migration scenario - no unwind needed.
-        let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(
-            result, None,
-            "Sentinel-only entries with checkpoint should be treated as first run"
-        );
-    }
-
     #[test]
     fn test_check_consistency_storages_history_behind_checkpoint_single_entry() {
         use reth_db_api::models::storage_sharded_key::StorageShardedKey;
@@ -1232,7 +1135,7 @@ mod tests {
     }
 
     #[test]
-    fn test_check_consistency_accounts_history_empty_with_checkpoint_is_first_run() {
+    fn test_check_consistency_accounts_history_empty_with_checkpoint_needs_unwind() {
         let temp_dir = TempDir::new().unwrap();
         let rocksdb = RocksDBBuilder::new(temp_dir.path())
             .with_table::<tables::AccountsHistory>()
@@ -1256,10 +1159,9 @@ mod tests {
 
         let provider = factory.database_provider_ro().unwrap();
 
-        // RocksDB is empty but checkpoint says block 100 was processed.
-        // This is treated as a first-run/migration scenario - no unwind needed.
+        // RocksDB is empty but checkpoint says block 100 was processed
         let result = rocksdb.check_consistency(&provider).unwrap();
-        assert_eq!(result, None, "Empty RocksDB with checkpoint is treated as first run");
+        assert_eq!(result, Some(0), "Should require unwind to block 0 to rebuild AccountsHistory");
     }
 
     #[test]

crates/storage/provider/src/providers/rocksdb/provider.rs

@@ -1,15 +1,11 @@
 use super::metrics::{RocksDBMetrics, RocksDBOperation};
-use crate::providers::{compute_history_rank, needs_prev_shard_check, HistoryInfo};
+use crate::providers::{needs_prev_shard_check, HistoryInfo};
 use alloy_consensus::transaction::TxHashRef;
 use alloy_primitives::{Address, BlockNumber, TxNumber, B256};
-use itertools::Itertools;
 use parking_lot::Mutex;
 use reth_chain_state::ExecutedBlock;
 use reth_db_api::{
-    models::{
-        sharded_key::NUM_OF_INDICES_IN_SHARD, storage_sharded_key::StorageShardedKey, ShardedKey,
-        StorageSettings,
-    },
+    models::{storage_sharded_key::StorageShardedKey, ShardedKey, StorageSettings},
     table::{Compress, Decode, Decompress, Encode, Table},
     tables, BlockNumberList, DatabaseError,
 };
@@ -591,15 +587,15 @@ impl RocksDBProvider {
         let mut account_history: BTreeMap<Address, Vec<u64>> = BTreeMap::new();
         for (block_idx, block) in blocks.iter().enumerate() {
             let block_number = ctx.first_block_number + block_idx as u64;
-            let bundle = &block.execution_outcome().state;
+            let bundle = &block.execution_outcome().bundle;
             for &address in bundle.state().keys() {
                 account_history.entry(address).or_default().push(block_number);
             }
         }
-
-        // Write account history using proper shard append logic
-        for (address, indices) in account_history {
-            batch.append_account_history_shard(address, indices)?;
+        for (address, blocks) in account_history {
+            let key = ShardedKey::new(address, u64::MAX);
+            let value = BlockNumberList::new_pre_sorted(blocks);
+            batch.put::<tables::AccountsHistory>(key, &value)?;
         }
         ctx.pending_batches.lock().push(batch.into_inner());
         Ok(())
@@ -616,7 +612,7 @@ impl RocksDBProvider {
         let mut storage_history: BTreeMap<(Address, B256), Vec<u64>> = BTreeMap::new();
         for (block_idx, block) in blocks.iter().enumerate() {
             let block_number = ctx.first_block_number + block_idx as u64;
-            let bundle = &block.execution_outcome().state;
+            let bundle = &block.execution_outcome().bundle;
             for (&address, account) in bundle.state() {
                 for &slot in account.storage.keys() {
                     let key = B256::new(slot.to_be_bytes());
@@ -624,10 +620,10 @@ impl RocksDBProvider {
                 }
             }
         }
-
-        // Write storage history using proper shard append logic
-        for ((address, slot), indices) in storage_history {
-            batch.append_storage_history_shard(address, slot, indices)?;
+        for ((address, slot), blocks) in storage_history {
+            let key = StorageShardedKey::new(address, slot, u64::MAX);
+            let value = BlockNumberList::new_pre_sorted(blocks);
+            batch.put::<tables::StoragesHistory>(key, &value)?;
         }
         ctx.pending_batches.lock().push(batch.into_inner());
         Ok(())
@@ -718,129 +714,6 @@ impl<'a> RocksDBBatch<'a> {
     pub fn into_inner(self) -> WriteBatchWithTransaction<true> {
         self.inner
     }
-
-    /// Appends indices to an account history shard with proper shard management.
-    ///
-    /// Loads the existing shard (if any), appends new indices, and rechunks into
-    /// multiple shards if needed (respecting `NUM_OF_INDICES_IN_SHARD` limit).
-    ///
-    /// # Requirements
-    ///
-    /// - The `indices` MUST be strictly increasing and contain no duplicates.
-    /// - This method MUST only be called once per address per batch. The batch reads existing
-    ///   shards from committed DB state, not from pending writes. Calling twice for the same
-    ///   address will cause the second call to overwrite the first.
-    pub fn append_account_history_shard(
-        &mut self,
-        address: Address,
-        indices: impl IntoIterator<Item = u64>,
-    ) -> ProviderResult<()> {
-        let indices: Vec<u64> = indices.into_iter().collect();
-
-        if indices.is_empty() {
-            return Ok(());
-        }
-
-        debug_assert!(
-            indices.windows(2).all(|w| w[0] < w[1]),
-            "indices must be strictly increasing: {:?}",
-            indices
-        );
-
-        let last_key = ShardedKey::new(address, u64::MAX);
-        let last_shard_opt = self.provider.get::<tables::AccountsHistory>(last_key.clone())?;
-        let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
-
-        last_shard.append(indices).map_err(ProviderError::other)?;
-
-        // Fast path: all indices fit in one shard
-        if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
-            self.put::<tables::AccountsHistory>(last_key, &last_shard)?;
-            return Ok(());
-        }
-
-        // Slow path: rechunk into multiple shards
-        let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
-        let mut chunks_peekable = chunks.into_iter().peekable();
-
-        while let Some(chunk) = chunks_peekable.next() {
-            let shard = BlockNumberList::new_pre_sorted(chunk);
-            let highest_block_number = if chunks_peekable.peek().is_some() {
-                shard.iter().next_back().expect("`chunks` does not return empty list")
-            } else {
-                u64::MAX
-            };
-
-            self.put::<tables::AccountsHistory>(
-                ShardedKey::new(address, highest_block_number),
-                &shard,
-            )?;
-        }
-
-        Ok(())
-    }
-
-    /// Appends indices to a storage history shard with proper shard management.
-    ///
-    /// Loads the existing shard (if any), appends new indices, and rechunks into
-    /// multiple shards if needed (respecting `NUM_OF_INDICES_IN_SHARD` limit).
-    ///
-    /// # Requirements
-    ///
-    /// - The `indices` MUST be strictly increasing and contain no duplicates.
-    /// - This method MUST only be called once per (address, `storage_key`) pair per batch. The
-    ///   batch reads existing shards from committed DB state, not from pending writes. Calling
-    ///   twice for the same key will cause the second call to overwrite the first.
-    pub fn append_storage_history_shard(
-        &mut self,
-        address: Address,
-        storage_key: B256,
-        indices: impl IntoIterator<Item = u64>,
-    ) -> ProviderResult<()> {
-        let indices: Vec<u64> = indices.into_iter().collect();
-
-        if indices.is_empty() {
-            return Ok(());
-        }
-
-        debug_assert!(
-            indices.windows(2).all(|w| w[0] < w[1]),
-            "indices must be strictly increasing: {:?}",
-            indices
-        );
-
-        let last_key = StorageShardedKey::last(address, storage_key);
-        let last_shard_opt = self.provider.get::<tables::StoragesHistory>(last_key.clone())?;
-        let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
-
-        last_shard.append(indices).map_err(ProviderError::other)?;
-
-        // Fast path: all indices fit in one shard
-        if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
-            self.put::<tables::StoragesHistory>(last_key, &last_shard)?;
-            return Ok(());
-        }
-
-        // Slow path: rechunk into multiple shards
-        let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
-        let mut chunks_peekable = chunks.into_iter().peekable();
-
-        while let Some(chunk) = chunks_peekable.next() {
-            let shard = BlockNumberList::new_pre_sorted(chunk);
-            let highest_block_number = if chunks_peekable.peek().is_some() {
-                shard.iter().next_back().expect("`chunks` does not return empty list")
-            } else {
-                u64::MAX
-            };
-
-            self.put::<tables::StoragesHistory>(
-                StorageShardedKey::new(address, storage_key, highest_block_number),
-                &shard,
-            )?;
-        }
-
-        Ok(())
-    }
 }
 
 /// `RocksDB` transaction wrapper providing MDBX-like semantics.
@@ -1028,16 +901,6 @@ impl<'db> RocksTx<'db> {
     where
         T: Table<Value = BlockNumberList>,
     {
-        // History may be pruned if a lowest available block is set.
-        let is_maybe_pruned = lowest_available_block_number.is_some();
-        let fallback = || {
-            Ok(if is_maybe_pruned {
-                HistoryInfo::MaybeInPlainState
-            } else {
-                HistoryInfo::NotYetWritten
-            })
-        };
-
         let cf = self.provider.0.db.cf_handle(T::NAME).ok_or_else(|| {
             ProviderError::Database(DatabaseError::Other(format!(
                 "column family not found: {}",
@@ -1055,28 +918,53 @@ impl<'db> RocksTx<'db> {
 
         if !iter.valid() {
             // No shard found at or after target block.
-            //
-            // (MaybeInPlainState) The key may have been written, but due to pruning we may not have
-            // changesets and history, so we need to make a plain state lookup.
-            // (HistoryInfo::NotYetWritten) The key has not been written to at all.
-            return fallback();
+            return if lowest_available_block_number.is_some() {
+                // The key may have been written, but due to pruning we may not have changesets
+                // and history, so we need to make a plain state lookup.
+                Ok(HistoryInfo::MaybeInPlainState)
+            } else {
+                // The key has not been written to at all.
+                Ok(HistoryInfo::NotYetWritten)
+            };
         }
 
         // Check if the found key matches our target entity.
         let Some(key_bytes) = iter.key() else {
-            return fallback();
+            return if lowest_available_block_number.is_some() {
+                Ok(HistoryInfo::MaybeInPlainState)
+            } else {
+                Ok(HistoryInfo::NotYetWritten)
+            };
         };
         if !key_matches(key_bytes)? {
             // The found key is for a different entity.
-            return fallback();
+            return if lowest_available_block_number.is_some() {
+                Ok(HistoryInfo::MaybeInPlainState)
+            } else {
+                Ok(HistoryInfo::NotYetWritten)
+            };
         }
 
         // Decompress the block list for this shard.
         let Some(value_bytes) = iter.value() else {
-            return fallback();
+            return if lowest_available_block_number.is_some() {
+                Ok(HistoryInfo::MaybeInPlainState)
+            } else {
+                Ok(HistoryInfo::NotYetWritten)
+            };
         };
         let chunk = BlockNumberList::decompress(value_bytes)?;
-        let (rank, found_block) = compute_history_rank(&chunk, block_number);
+
+        // Get the rank of the first entry before or equal to our block.
+        let mut rank = chunk.rank(block_number);
+
+        // Adjust the rank, so that we have the rank of the first entry strictly before our
+        // block (not equal to it).
+        if rank.checked_sub(1).and_then(|r| chunk.select(r)) == Some(block_number) {
+            rank -= 1;
+        }
+
+        let found_block = chunk.select(rank);
 
         // Lazy check for previous shard - only called when needed.
         // If we can step to a previous shard for this same key, history already exists,
@@ -1215,11 +1103,7 @@ mod tests {
     use crate::providers::HistoryInfo;
     use alloy_primitives::{Address, TxHash, B256};
     use reth_db_api::{
-        models::{
-            sharded_key::{ShardedKey, NUM_OF_INDICES_IN_SHARD},
-            storage_sharded_key::StorageShardedKey,
-            IntegerList,
-        },
+        models::{sharded_key::ShardedKey, storage_sharded_key::StorageShardedKey, IntegerList},
         table::Table,
         tables,
     };
@@ -1568,156 +1452,4 @@ mod tests {
 
         tx.rollback().unwrap();
     }
-
-    #[test]
-    fn test_account_history_shard_split_at_boundary() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
-
-        let address = Address::from([0x42; 20]);
-        let limit = NUM_OF_INDICES_IN_SHARD;
-
-        // Add exactly NUM_OF_INDICES_IN_SHARD + 1 indices to trigger a split
-        let indices: Vec<u64> = (0..=(limit as u64)).collect();
-        let mut batch = provider.batch();
-        batch.append_account_history_shard(address, indices).unwrap();
-        batch.commit().unwrap();
-
-        // Should have 2 shards: one completed shard and one sentinel shard
-        let completed_key = ShardedKey::new(address, (limit - 1) as u64);
-        let sentinel_key = ShardedKey::new(address, u64::MAX);
-
-        let completed_shard = provider.get::<tables::AccountsHistory>(completed_key).unwrap();
-        let sentinel_shard = provider.get::<tables::AccountsHistory>(sentinel_key).unwrap();
-
-        assert!(completed_shard.is_some(), "completed shard should exist");
-        assert!(sentinel_shard.is_some(), "sentinel shard should exist");
-
-        let completed_shard = completed_shard.unwrap();
-        let sentinel_shard = sentinel_shard.unwrap();
-
-        assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
-        assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
-    }
-
-    #[test]
-    fn test_account_history_multiple_shard_splits() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
-
-        let address = Address::from([0x43; 20]);
-        let limit = NUM_OF_INDICES_IN_SHARD;
-
-        // First batch: add NUM_OF_INDICES_IN_SHARD indices
-        let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
-        let mut batch = provider.batch();
-        batch.append_account_history_shard(address, first_batch_indices).unwrap();
-        batch.commit().unwrap();
-
-        // Should have just a sentinel shard (exactly at limit, not over)
-        let sentinel_key = ShardedKey::new(address, u64::MAX);
-        let shard = provider.get::<tables::AccountsHistory>(sentinel_key.clone()).unwrap();
-        assert!(shard.is_some());
-        assert_eq!(shard.unwrap().len(), limit as u64);
-
-        // Second batch: add another NUM_OF_INDICES_IN_SHARD + 1 indices (causing 2 more shards)
-        let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
-        let mut batch = provider.batch();
-        batch.append_account_history_shard(address, second_batch_indices).unwrap();
-        batch.commit().unwrap();
-
-        // Now we should have: 2 completed shards + 1 sentinel shard
-        let first_completed = ShardedKey::new(address, (limit - 1) as u64);
-        let second_completed = ShardedKey::new(address, (2 * limit - 1) as u64);
-
-        assert!(
-            provider.get::<tables::AccountsHistory>(first_completed).unwrap().is_some(),
-            "first completed shard should exist"
-        );
-        assert!(
-            provider.get::<tables::AccountsHistory>(second_completed).unwrap().is_some(),
-            "second completed shard should exist"
-        );
-        assert!(
-            provider.get::<tables::AccountsHistory>(sentinel_key).unwrap().is_some(),
-            "sentinel shard should exist"
-        );
-    }
-
-    #[test]
-    fn test_storage_history_shard_split_at_boundary() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
-
-        let address = Address::from([0x44; 20]);
-        let slot = B256::from([0x55; 32]);
-        let limit = NUM_OF_INDICES_IN_SHARD;
-
-        // Add exactly NUM_OF_INDICES_IN_SHARD + 1 indices to trigger a split
-        let indices: Vec<u64> = (0..=(limit as u64)).collect();
-        let mut batch = provider.batch();
-        batch.append_storage_history_shard(address, slot, indices).unwrap();
-        batch.commit().unwrap();
-
-        // Should have 2 shards: one completed shard and one sentinel shard
-        let completed_key = StorageShardedKey::new(address, slot, (limit - 1) as u64);
-        let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
-
-        let completed_shard = provider.get::<tables::StoragesHistory>(completed_key).unwrap();
-        let sentinel_shard = provider.get::<tables::StoragesHistory>(sentinel_key).unwrap();
-
-        assert!(completed_shard.is_some(), "completed shard should exist");
-        assert!(sentinel_shard.is_some(), "sentinel shard should exist");
-
-        let completed_shard = completed_shard.unwrap();
-        let sentinel_shard = sentinel_shard.unwrap();
-
-        assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
-        assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
-    }
-
-    #[test]
-    fn test_storage_history_multiple_shard_splits() {
-        let temp_dir = TempDir::new().unwrap();
-        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
-
-        let address = Address::from([0x46; 20]);
-        let slot = B256::from([0x57; 32]);
-        let limit = NUM_OF_INDICES_IN_SHARD;
-
-        // First batch: add NUM_OF_INDICES_IN_SHARD indices
-        let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
-        let mut batch = provider.batch();
-        batch.append_storage_history_shard(address, slot, first_batch_indices).unwrap();
-        batch.commit().unwrap();
-
-        // Should have just a sentinel shard (exactly at limit, not over)
-        let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
-        let shard = provider.get::<tables::StoragesHistory>(sentinel_key.clone()).unwrap();
-        assert!(shard.is_some());
-        assert_eq!(shard.unwrap().len(), limit as u64);
-
-        // Second batch: add another NUM_OF_INDICES_IN_SHARD + 1 indices (causing 2 more shards)
-        let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
-        let mut batch = provider.batch();
-        batch.append_storage_history_shard(address, slot, second_batch_indices).unwrap();
-        batch.commit().unwrap();
-
-        // Now we should have: 2 completed shards + 1 sentinel shard
-        let first_completed = StorageShardedKey::new(address, slot, (limit - 1) as u64);
-        let second_completed = StorageShardedKey::new(address, slot, (2 * limit - 1) as u64);
-
-        assert!(
-            provider.get::<tables::StoragesHistory>(first_completed).unwrap().is_some(),
-            "first completed shard should exist"
-        );
-        assert!(
-            provider.get::<tables::StoragesHistory>(second_completed).unwrap().is_some(),
-            "second completed shard should exist"
-        );
-        assert!(
-            provider.get::<tables::StoragesHistory>(sentinel_key).unwrap().is_some(),
-            "sentinel shard should exist"
-        );
-    }
 }

crates/storage/provider/src/providers/state/historical.rs

@@ -1,11 +1,12 @@
 use crate::{
-    AccountReader, BlockHashReader, ChangeSetReader, EitherReader, HashedPostStateProvider,
-    ProviderError, RocksDBProviderFactory, StateProvider, StateRootProvider,
+    AccountReader, BlockHashReader, ChangeSetReader, HashedPostStateProvider, ProviderError,
+    StateProvider, StateRootProvider,
 };
 use alloy_eips::merge::EPOCH_SLOTS;
 use alloy_primitives::{Address, BlockNumber, Bytes, StorageKey, StorageValue, B256};
 use reth_db_api::{
     cursor::{DbCursorRO, DbDupCursorRO},
+    models::{storage_sharded_key::StorageShardedKey, ShardedKey},
     table::Table,
     tables,
     transaction::DbTx,
@@ -13,8 +14,7 @@ use reth_db_api::{
 };
 use reth_primitives_traits::{Account, Bytecode};
 use reth_storage_api::{
-    BlockNumReader, BytecodeReader, DBProvider, NodePrimitivesProvider, StateProofProvider,
-    StorageRootProvider, StorageSettingsCache,
+    BlockNumReader, BytecodeReader, DBProvider, StateProofProvider, StorageRootProvider,
 };
 use reth_storage_errors::provider::ProviderResult;
 use reth_trie::{
@@ -127,47 +127,38 @@ impl<'b, Provider: DBProvider + ChangeSetReader + BlockNumReader>
         Self { provider, block_number, lowest_available_blocks }
     }
 
-    /// Lookup an account in the `AccountsHistory` table using `EitherReader`.
-    pub fn account_history_lookup(&self, address: Address) -> ProviderResult<HistoryInfo>
-    where
-        Provider: StorageSettingsCache + RocksDBProviderFactory + NodePrimitivesProvider,
-    {
+    /// Lookup an account in the `AccountsHistory` table
+    pub fn account_history_lookup(&self, address: Address) -> ProviderResult<HistoryInfo> {
         if !self.lowest_available_blocks.is_account_history_available(self.block_number) {
             return Err(ProviderError::StateAtBlockPruned(self.block_number))
         }
 
-        self.provider.with_rocksdb_tx(|rocks_tx_ref| {
-            let mut reader = EitherReader::new_accounts_history(self.provider, rocks_tx_ref)?;
-            reader.account_history_info(
-                address,
-                self.block_number,
-                self.lowest_available_blocks.account_history_block_number,
-            )
-        })
+        // history key to search IntegerList of block number changesets.
+        let history_key = ShardedKey::new(address, self.block_number);
+        self.history_info_lookup::<tables::AccountsHistory, _>(
+            history_key,
+            |key| key.key == address,
+            self.lowest_available_blocks.account_history_block_number,
+        )
     }
 
-    /// Lookup a storage key in the `StoragesHistory` table using `EitherReader`.
+    /// Lookup a storage key in the `StoragesHistory` table
     pub fn storage_history_lookup(
         &self,
         address: Address,
         storage_key: StorageKey,
-    ) -> ProviderResult<HistoryInfo>
-    where
-        Provider: StorageSettingsCache + RocksDBProviderFactory + NodePrimitivesProvider,
-    {
+    ) -> ProviderResult<HistoryInfo> {
         if !self.lowest_available_blocks.is_storage_history_available(self.block_number) {
             return Err(ProviderError::StateAtBlockPruned(self.block_number))
         }
 
-        self.provider.with_rocksdb_tx(|rocks_tx_ref| {
-            let mut reader = EitherReader::new_storages_history(self.provider, rocks_tx_ref)?;
-            reader.storage_history_info(
-                address,
-                storage_key,
-                self.block_number,
-                self.lowest_available_blocks.storage_history_block_number,
-            )
-        })
+        // history key to search IntegerList of block number changesets.
+        let history_key = StorageShardedKey::new(address, storage_key, self.block_number);
+        self.history_info_lookup::<tables::StoragesHistory, _>(
+            history_key,
+            |key| key.address == address && key.sharded_key.key == storage_key,
+            self.lowest_available_blocks.storage_history_block_number,
+        )
     }
 
     /// Checks and returns `true` if distance to historical block exceeds the provided limit.
@@ -213,6 +204,25 @@ impl<'b, Provider: DBProvider + ChangeSetReader + BlockNumReader>
         Ok(HashedStorage::from_reverts(self.tx(), address, self.block_number)?)
     }
 
+    fn history_info_lookup<T, K>(
+        &self,
+        key: K,
+        key_filter: impl Fn(&K) -> bool,
+        lowest_available_block_number: Option<BlockNumber>,
+    ) -> ProviderResult<HistoryInfo>
+    where
+        T: Table<Key = K, Value = BlockNumberList>,
+    {
+        let mut cursor = self.tx().cursor_read::<T>()?;
+        history_info::<T, K, _>(
+            &mut cursor,
+            key,
+            self.block_number,
+            key_filter,
+            lowest_available_block_number,
+        )
+    }
+
     /// Set the lowest block number at which the account history is available.
     pub const fn with_lowest_available_account_history_block_number(
         mut self,
@@ -238,14 +248,8 @@ impl<Provider: DBProvider + BlockNumReader> HistoricalStateProviderRef<'_, Provi
     }
 }
 
-impl<
-        Provider: DBProvider
-            + BlockNumReader
-            + ChangeSetReader
-            + StorageSettingsCache
-            + RocksDBProviderFactory
-            + NodePrimitivesProvider,
-    > AccountReader for HistoricalStateProviderRef<'_, Provider>
+impl<Provider: DBProvider + BlockNumReader + ChangeSetReader> AccountReader
+    for HistoricalStateProviderRef<'_, Provider>
 {
     /// Get basic account information.
     fn basic_account(&self, address: &Address) -> ProviderResult<Option<Account>> {
@@ -291,7 +295,7 @@ impl<Provider: DBProvider + ChangeSetReader + BlockNumReader> StateRootProvider
     fn state_root(&self, hashed_state: HashedPostState) -> ProviderResult<B256> {
         let mut revert_state = self.revert_state()?;
         let hashed_state_sorted = hashed_state.into_sorted();
-        revert_state.extend_ref_and_sort(&hashed_state_sorted);
+        revert_state.extend_ref(&hashed_state_sorted);
         Ok(StateRoot::overlay_root(self.tx(), &revert_state)?)
     }
 
@@ -306,7 +310,7 @@ impl<Provider: DBProvider + ChangeSetReader + BlockNumReader> StateRootProvider
     ) -> ProviderResult<(B256, TrieUpdates)> {
         let mut revert_state = self.revert_state()?;
         let hashed_state_sorted = hashed_state.into_sorted();
-        revert_state.extend_ref_and_sort(&hashed_state_sorted);
+        revert_state.extend_ref(&hashed_state_sorted);
         Ok(StateRoot::overlay_root_with_updates(self.tx(), &revert_state)?)
     }
 
@@ -400,15 +404,8 @@ impl<Provider> HashedPostStateProvider for HistoricalStateProviderRef<'_, Provid
     }
 }
 
-impl<
-        Provider: DBProvider
-            + BlockNumReader
-            + BlockHashReader
-            + ChangeSetReader
-            + StorageSettingsCache
-            + RocksDBProviderFactory
-            + NodePrimitivesProvider,
-    > StateProvider for HistoricalStateProviderRef<'_, Provider>
+impl<Provider: DBProvider + BlockNumReader + BlockHashReader + ChangeSetReader> StateProvider
+    for HistoricalStateProviderRef<'_, Provider>
 {
     /// Get storage.
     fn storage(
@@ -498,7 +495,7 @@ impl<Provider: DBProvider + ChangeSetReader + BlockNumReader> HistoricalStatePro
 }
 
 // Delegates all provider impls to [HistoricalStateProviderRef]
-reth_storage_api::macros::delegate_provider_impls!(HistoricalStateProvider<Provider> where [Provider: DBProvider + BlockNumReader + BlockHashReader + ChangeSetReader + StorageSettingsCache + RocksDBProviderFactory + NodePrimitivesProvider]);
+reth_storage_api::macros::delegate_provider_impls!(HistoricalStateProvider<Provider> where [Provider: DBProvider + BlockNumReader + BlockHashReader + ChangeSetReader]);
 
 /// Lowest blocks at which different parts of the state are available.
 /// They may be [Some] if pruning is enabled.
@@ -528,32 +525,6 @@ impl LowestAvailableBlocks {
     }
 }
 
-/// Computes the rank and finds the next modification block in a history shard.
-///
-/// Given a `block_number`, this function returns:
-/// - `rank`: The number of entries strictly before `block_number` in the shard
-/// - `found_block`: The block number at position `rank` (i.e., the first block >= `block_number`
-///   where a modification occurred), or `None` if `rank` is out of bounds
-///
-/// The rank is adjusted when `block_number` exactly matches an entry in the shard,
-/// so that `found_block` always returns the modification at or after the target.
-///
-/// This logic is shared between MDBX cursor-based lookups and `RocksDB` iterator lookups.
-#[inline]
-pub fn compute_history_rank(
-    chunk: &reth_db_api::BlockNumberList,
-    block_number: BlockNumber,
-) -> (u64, Option<u64>) {
-    let mut rank = chunk.rank(block_number);
-    // `rank(block_number)` returns count of entries <= block_number.
-    // We want the first entry >= block_number, so if block_number is in the shard,
-    // we need to step back one position to point at it (not past it).
-    if rank.checked_sub(1).and_then(|r| chunk.select(r)) == Some(block_number) {
-        rank -= 1;
-    }
-    (rank, chunk.select(rank))
-}
-
 /// Checks if a previous shard lookup is needed to determine if we're before the first write.
 ///
 /// Returns `true` when `rank == 0` (first entry in shard) and the found block doesn't match
@@ -586,7 +557,16 @@ where
     // index, the first chunk for the next key will be returned so we filter out chunks that
     // have a different key.
     if let Some(chunk) = cursor.seek(key)?.filter(|(k, _)| key_filter(k)).map(|x| x.1) {
-        let (rank, found_block) = compute_history_rank(&chunk, block_number);
+        // Get the rank of the first entry before or equal to our block.
+        let mut rank = chunk.rank(block_number);
+
+        // Adjust the rank, so that we have the rank of the first entry strictly before our
+        // block (not equal to it).
+        if rank.checked_sub(1).and_then(|r| chunk.select(r)) == Some(block_number) {
+            rank -= 1;
+        }
+
+        let found_block = chunk.select(rank);
 
         // If our block is before the first entry in the index chunk and this first entry
         // doesn't equal to our block, it might be before the first write ever. To check, we
@@ -618,8 +598,7 @@ mod tests {
     use crate::{
         providers::state::historical::{HistoryInfo, LowestAvailableBlocks},
         test_utils::create_test_provider_factory,
-        AccountReader, HistoricalStateProvider, HistoricalStateProviderRef, RocksDBProviderFactory,
-        StateProvider,
+        AccountReader, HistoricalStateProvider, HistoricalStateProviderRef, StateProvider,
     };
     use alloy_primitives::{address, b256, Address, B256, U256};
     use reth_db_api::{
@@ -631,7 +610,6 @@ mod tests {
     use reth_primitives_traits::{Account, StorageEntry};
     use reth_storage_api::{
         BlockHashReader, BlockNumReader, ChangeSetReader, DBProvider, DatabaseProviderFactory,
-        NodePrimitivesProvider, StorageSettingsCache,
     };
     use reth_storage_errors::provider::ProviderError;
 
@@ -643,13 +621,7 @@ mod tests {
     const fn assert_state_provider<T: StateProvider>() {}
     #[expect(dead_code)]
     const fn assert_historical_state_provider<
-        T: DBProvider
-            + BlockNumReader
-            + BlockHashReader
-            + ChangeSetReader
-            + StorageSettingsCache
-            + RocksDBProviderFactory
-            + NodePrimitivesProvider,
+        T: DBProvider + BlockNumReader + BlockHashReader + ChangeSetReader,
     >() {
         assert_state_provider::<HistoricalStateProvider<T>>();
     }

crates/storage/provider/src/providers/state/overlay.rs

@@ -1,7 +1,6 @@
 use alloy_primitives::{BlockNumber, B256};
 use metrics::{Counter, Histogram};
 use parking_lot::RwLock;
-use reth_chain_state::LazyOverlay;
 use reth_db_api::DatabaseError;
 use reth_errors::{ProviderError, ProviderResult};
 use reth_metrics::Metrics;
@@ -54,35 +53,6 @@ struct Overlay {
     hashed_post_state: Arc<HashedPostStateSorted>,
 }
 
-/// Source of overlay data for [`OverlayStateProviderFactory`].
-///
-/// Either provides immediate pre-computed overlay data, or a lazy overlay that computes
-/// on first access.
-#[derive(Debug, Clone)]
-pub enum OverlaySource {
-    /// Immediate overlay with already-computed data.
-    Immediate {
-        /// Trie updates overlay.
-        trie: Arc<TrieUpdatesSorted>,
-        /// Hashed state overlay.
-        state: Arc<HashedPostStateSorted>,
-    },
-    /// Lazy overlay computed on first access.
-    Lazy(LazyOverlay),
-}
-
-impl OverlaySource {
-    /// Resolve the overlay source into (trie, state) tuple.
-    ///
-    /// For lazy overlays, this may block waiting for deferred data.
-    fn resolve(&self) -> (Arc<TrieUpdatesSorted>, Arc<HashedPostStateSorted>) {
-        match self {
-            Self::Immediate { trie, state } => (Arc::clone(trie), Arc::clone(state)),
-            Self::Lazy(lazy) => lazy.as_overlay(),
-        }
-    }
-}
-
 /// Factory for creating overlay state providers with optional reverts and overlays.
 ///
 /// This factory allows building an `OverlayStateProvider` whose DB state has been reverted to a
@@ -93,8 +63,10 @@ pub struct OverlayStateProviderFactory<F> {
     factory: F,
     /// Optional block hash for collecting reverts
     block_hash: Option<B256>,
-    /// Optional overlay source (lazy or immediate).
-    overlay_source: Option<OverlaySource>,
+    /// Optional trie overlay
+    trie_overlay: Option<Arc<TrieUpdatesSorted>>,
+    /// Optional hashed state overlay
+    hashed_state_overlay: Option<Arc<HashedPostStateSorted>>,
     /// Changeset cache handle for retrieving trie changesets
     changeset_cache: ChangesetCache,
     /// Metrics for tracking provider operations
@@ -110,7 +82,8 @@ impl<F> OverlayStateProviderFactory<F> {
         Self {
             factory,
             block_hash: None,
-            overlay_source: None,
+            trie_overlay: None,
+            hashed_state_overlay: None,
             changeset_cache,
             metrics: OverlayStateProviderMetrics::default(),
             overlay_cache: Default::default(),
@@ -124,59 +97,31 @@ impl<F> OverlayStateProviderFactory<F> {
         self
     }
 
-    /// Set the overlay source (lazy or immediate).
+    /// Set the trie overlay.
     ///
     /// This overlay will be applied on top of any reverts applied via `with_block_hash`.
-    pub fn with_overlay_source(mut self, source: Option<OverlaySource>) -> Self {
-        self.overlay_source = source;
+    pub fn with_trie_overlay(mut self, trie_overlay: Option<Arc<TrieUpdatesSorted>>) -> Self {
+        self.trie_overlay = trie_overlay;
         self
     }
 
-    /// Set a lazy overlay that will be computed on first access.
-    ///
-    /// Convenience method that wraps the lazy overlay in `OverlaySource::Lazy`.
-    pub fn with_lazy_overlay(mut self, lazy_overlay: Option<LazyOverlay>) -> Self {
-        self.overlay_source = lazy_overlay.map(OverlaySource::Lazy);
-        self
-    }
-
-    /// Set the hashed state overlay.
+    /// Set the hashed state overlay
     ///
     /// This overlay will be applied on top of any reverts applied via `with_block_hash`.
     pub fn with_hashed_state_overlay(
         mut self,
         hashed_state_overlay: Option<Arc<HashedPostStateSorted>>,
     ) -> Self {
-        if let Some(state) = hashed_state_overlay {
-            self.overlay_source = Some(OverlaySource::Immediate {
-                trie: Arc::new(TrieUpdatesSorted::default()),
-                state,
-            });
-        }
+        self.hashed_state_overlay = hashed_state_overlay;
         self
     }
 
     /// Extends the existing hashed state overlay with the given [`HashedPostStateSorted`].
-    ///
-    /// If no overlay exists, creates a new immediate overlay with the given state.
-    /// If a lazy overlay exists, it is resolved first then extended.
     pub fn with_extended_hashed_state_overlay(mut self, other: HashedPostStateSorted) -> Self {
-        match &mut self.overlay_source {
-            Some(OverlaySource::Immediate { state, .. }) => {
-                Arc::make_mut(state).extend_ref_and_sort(&other);
-            }
-            Some(OverlaySource::Lazy(lazy)) => {
-                // Resolve lazy overlay and convert to immediate with extension
-                let (trie, mut state) = lazy.as_overlay();
-                Arc::make_mut(&mut state).extend_ref_and_sort(&other);
-                self.overlay_source = Some(OverlaySource::Immediate { trie, state });
-            }
-            None => {
-                self.overlay_source = Some(OverlaySource::Immediate {
-                    trie: Arc::new(TrieUpdatesSorted::default()),
-                    state: Arc::new(other),
-                });
-            }
+        if let Some(overlay) = self.hashed_state_overlay.as_mut() {
+            Arc::make_mut(overlay).extend_ref(&other);
+        } else {
+            self.hashed_state_overlay = Some(Arc::new(other))
         }
         self
     }
@@ -191,19 +136,6 @@ where
         + DBProvider
         + BlockNumReader,
 {
-    /// Resolves the effective overlay (trie updates, hashed state).
-    ///
-    /// If an overlay source is set, it is resolved (blocking if lazy).
-    /// Otherwise, returns empty defaults.
-    fn resolve_overlays(&self) -> (Arc<TrieUpdatesSorted>, Arc<HashedPostStateSorted>) {
-        match &self.overlay_source {
-            Some(source) => source.resolve(),
-            None => {
-                (Arc::new(TrieUpdatesSorted::default()), Arc::new(HashedPostStateSorted::default()))
-            }
-        }
-    }
-
     /// Returns the block number for [`Self`]'s `block_hash` field, if any.
     fn get_requested_block_number(
         &self,
@@ -335,26 +267,26 @@ where
                 res
             };
 
-            // Resolve overlays (lazy or immediate) and extend reverts with them.
-            // If reverts are empty, use overlays directly to avoid cloning.
-            let (overlay_trie, overlay_state) = self.resolve_overlays();
-
-            let trie_updates = if trie_reverts.is_empty() {
-                overlay_trie
-            } else if !overlay_trie.is_empty() {
-                trie_reverts.extend_ref_and_sort(&overlay_trie);
-                Arc::new(trie_reverts)
-            } else {
-                Arc::new(trie_reverts)
+            // Extend with overlays if provided. If the reverts are empty we should just use the
+            // overlays directly, because `extend_ref` will actually clone the overlay.
+            let trie_updates = match self.trie_overlay.as_ref() {
+                Some(trie_overlay) if trie_reverts.is_empty() => Arc::clone(trie_overlay),
+                Some(trie_overlay) => {
+                    trie_reverts.extend_ref(trie_overlay);
+                    Arc::new(trie_reverts)
+                }
+                None => Arc::new(trie_reverts),
             };
 
-            let hashed_state_updates = if hashed_state_reverts.is_empty() {
-                overlay_state
-            } else if !overlay_state.is_empty() {
-                hashed_state_reverts.extend_ref_and_sort(&overlay_state);
-                Arc::new(hashed_state_reverts)
-            } else {
-                Arc::new(hashed_state_reverts)
+            let hashed_state_updates = match self.hashed_state_overlay.as_ref() {
+                Some(hashed_state_overlay) if hashed_state_reverts.is_empty() => {
+                    Arc::clone(hashed_state_overlay)
+                }
+                Some(hashed_state_overlay) => {
+                    hashed_state_reverts.extend_ref(hashed_state_overlay);
+                    Arc::new(hashed_state_reverts)
+                }
+                None => Arc::new(hashed_state_reverts),
             };
 
             trie_updates_total_len = trie_updates.total_len();
@@ -371,8 +303,13 @@ where
 
             (trie_updates, hashed_state_updates)
         } else {
-            // If no block_hash, use overlays directly (resolving lazy if set)
-            let (trie_updates, hashed_state) = self.resolve_overlays();
+            // If no block_hash, use overlays directly or defaults
+            let trie_updates =
+                self.trie_overlay.clone().unwrap_or_else(|| Arc::new(TrieUpdatesSorted::default()));
+            let hashed_state = self
+                .hashed_state_overlay
+                .clone()
+                .unwrap_or_else(|| Arc::new(HashedPostStateSorted::default()));
 
             retrieve_trie_reverts_duration = Duration::ZERO;
             retrieve_hashed_state_reverts_duration = Duration::ZERO;
@@ -400,9 +337,14 @@ where
     #[instrument(level = "debug", target = "providers::state::overlay", skip_all)]
     fn get_overlay(&self, provider: &F::Provider) -> ProviderResult<Overlay> {
         // If we have no anchor block configured then we will never need to get trie reverts, just
-        // return the in-memory overlay (resolving lazy overlay if set).
+        // return the in-memory overlay.
         if self.block_hash.is_none() {
-            let (trie_updates, hashed_post_state) = self.resolve_overlays();
+            let trie_updates =
+                self.trie_overlay.clone().unwrap_or_else(|| Arc::new(TrieUpdatesSorted::default()));
+            let hashed_post_state = self
+                .hashed_state_overlay
+                .clone()
+                .unwrap_or_else(|| Arc::new(HashedPostStateSorted::default()));
             return Ok(Overlay { trie_updates, hashed_post_state })
         }
 

crates/storage/provider/src/providers/static_file/manager.rs

@@ -594,7 +594,7 @@ impl<N: NodePrimitives> StaticFileProvider<N> {
                 continue
             }
 
-            for (i, receipt) in block.execution_outcome().receipts.iter().enumerate() {
+            for (i, receipt) in block.execution_outcome().receipts.iter().flatten().enumerate() {
                 w.append_receipt(first_tx + i as u64, receipt)?;
             }
         }
@@ -609,7 +609,7 @@ impl<N: NodePrimitives> StaticFileProvider<N> {
     ) -> ProviderResult<()> {
         for block in blocks {
             let block_number = block.recovered_block().number();
-            let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
+            let reverts = block.execution_outcome().bundle.reverts.to_plain_state_reverts();
 
             for account_block_reverts in reverts.accounts {
                 let changeset = account_block_reverts

crates/storage/provider/src/traits/rocksdb_provider.rs

@@ -1,5 +1,4 @@
-use crate::{either_writer::RocksTxRefArg, providers::RocksDBProvider};
-use reth_storage_errors::provider::ProviderResult;
+use crate::providers::RocksDBProvider;
 
 /// `RocksDB` provider factory.
 ///
@@ -14,21 +13,4 @@ pub trait RocksDBProviderFactory {
     /// commits, ensuring atomicity across all storage backends.
     #[cfg(all(unix, feature = "rocksdb"))]
     fn set_pending_rocksdb_batch(&self, batch: rocksdb::WriteBatchWithTransaction<true>);
-
-    /// Executes a closure with a `RocksDB` transaction for reading.
-    ///
-    /// This helper encapsulates all the cfg-gated `RocksDB` transaction handling for reads.
-    fn with_rocksdb_tx<F, R>(&self, f: F) -> ProviderResult<R>
-    where
-        F: FnOnce(RocksTxRefArg<'_>) -> ProviderResult<R>,
-    {
-        #[cfg(all(unix, feature = "rocksdb"))]
-        {
-            let rocksdb = self.rocksdb_provider();
-            let tx = rocksdb.tx();
-            f(&tx)
-        }
-        #[cfg(not(all(unix, feature = "rocksdb")))]
-        f(())
-    }
 }

crates/storage/storage-api/src/lib.rs

@@ -85,9 +85,7 @@ pub use primitives::*;
 mod block_indices;
 pub use block_indices::*;
 
-#[cfg(feature = "std")]
 mod block_writer;
-#[cfg(feature = "std")]
 pub use block_writer::*;
 
 mod state_writer;

crates/storage/storage-api/src/state_writer.rs

@@ -1,98 +1,23 @@
-use alloc::vec::Vec;
-use alloy_consensus::transaction::Either;
 use alloy_primitives::BlockNumber;
-use reth_execution_types::{BlockExecutionOutput, ExecutionOutcome};
+use reth_execution_types::ExecutionOutcome;
 use reth_storage_errors::provider::ProviderResult;
 use reth_trie_common::HashedPostStateSorted;
 use revm_database::{
     states::{PlainStateReverts, StateChangeset},
-    BundleState, OriginalValuesKnown,
+    OriginalValuesKnown,
 };
 
-/// A helper type used as input to [`StateWriter`] for writing execution outcome for one or many
-/// blocks.
-#[derive(Debug)]
-pub enum WriteStateInput<'a, R> {
-    /// A single block execution outcome.
-    Single {
-        /// The execution outcome.
-        outcome: &'a BlockExecutionOutput<R>,
-        /// Block number
-        block: BlockNumber,
-    },
-    /// Multiple block execution outcomes.
-    Multiple(&'a ExecutionOutcome<R>),
-}
-
-impl<'a, R> WriteStateInput<'a, R> {
-    /// Number of blocks in the execution outcome.
-    pub const fn len(&self) -> usize {
-        match self {
-            Self::Single { .. } => 1,
-            Self::Multiple(outcome) => outcome.len(),
-        }
-    }
-
-    /// Returns true if the execution outcome is empty.
-    pub const fn is_empty(&self) -> bool {
-        match self {
-            Self::Single { outcome, .. } => outcome.result.receipts.is_empty(),
-            Self::Multiple(outcome) => outcome.is_empty(),
-        }
-    }
-
-    /// Number of the first block.
-    pub const fn first_block(&self) -> BlockNumber {
-        match self {
-            Self::Single { block, .. } => *block,
-            Self::Multiple(outcome) => outcome.first_block(),
-        }
-    }
-
-    /// Number of the last block.
-    pub const fn last_block(&self) -> BlockNumber {
-        match self {
-            Self::Single { block, .. } => *block,
-            Self::Multiple(outcome) => outcome.last_block(),
-        }
-    }
-
-    /// Returns a reference to the [`BundleState`].
-    pub const fn state(&self) -> &BundleState {
-        match self {
-            Self::Single { outcome, .. } => &outcome.state,
-            Self::Multiple(outcome) => &outcome.bundle,
-        }
-    }
-
-    /// Returns an iterator over receipt sets for each block.
-    pub fn receipts(&self) -> impl Iterator<Item = &Vec<R>> {
-        match self {
-            Self::Single { outcome, .. } => {
-                Either::Left(core::iter::once(&outcome.result.receipts))
-            }
-            Self::Multiple(outcome) => Either::Right(outcome.receipts.iter()),
-        }
-    }
-}
-
-impl<'a, R> From<&'a ExecutionOutcome<R>> for WriteStateInput<'a, R> {
-    fn from(outcome: &'a ExecutionOutcome<R>) -> Self {
-        Self::Multiple(outcome)
-    }
-}
-
 /// A trait specifically for writing state changes or reverts
 pub trait StateWriter {
     /// Receipt type included into [`ExecutionOutcome`].
-    type Receipt: 'static;
+    type Receipt;
 
     /// Write the state and optionally receipts to the database.
     ///
     /// Use `config` to skip writing certain data types when they are written elsewhere.
-    fn write_state<'a>(
+    fn write_state(
         &self,
-        execution_outcome: impl Into<WriteStateInput<'a, Self::Receipt>>,
+        execution_outcome: &ExecutionOutcome<Self::Receipt>,
         is_value_known: OriginalValuesKnown,
         config: StateWriteConfig,
     ) -> ProviderResult<()>;

crates/transaction-pool/src/blobstore/tracker.rs

@@ -175,7 +175,8 @@ mod tests {
         );
 
         // Extract blocks from the chain
-        let chain: Chain = Chain::new(vec![block1, block2], Default::default(), BTreeMap::new());
+        let chain: Chain =
+            Chain::new(vec![block1, block2], Default::default(), BTreeMap::new(), BTreeMap::new());
         let blocks = chain.into_inner().0;
 
         // Add new chain blocks to the tracker

crates/trie/common/src/hashed_state.rs

@@ -621,9 +621,7 @@ impl HashedPostStateSorted {
 
     /// Extends this state with contents of another sorted state.
     /// Entries in `other` take precedence for duplicate keys.
-    ///
-    /// Sorts the accounts after extending. Sorts the storage after extending, for each account.
-    pub fn extend_ref_and_sort(&mut self, other: &Self) {
+    pub fn extend_ref(&mut self, other: &Self) {
         // Extend accounts
         extend_sorted_vec(&mut self.accounts, &other.accounts);
 
@@ -995,7 +993,6 @@ mod tests {
             nonce: 42,
             code_hash: B256::random(),
             code: Some(Bytecode::new_raw(Bytes::from(vec![1, 2]))),
-            account_id: None,
         };
 
         let mut storage = StorageWithOriginalValues::default();
@@ -1040,7 +1037,6 @@ mod tests {
             nonce: 1,
             code_hash: B256::random(),
             code: None,
-            account_id: None,
         };
 
         // Create hashed accounts with addresses.
@@ -1418,7 +1414,7 @@ mod tests {
             storages: B256Map::default(),
         };
 
-        state1.extend_ref_and_sort(&state2);
+        state1.extend_ref(&state2);
 
         // Check accounts are merged and sorted
         assert_eq!(state1.accounts.len(), 6);

crates/trie/common/src/lazy.rs

@@ -1,194 +0,0 @@
-//! Lazy initialization wrapper for trie data.
-//!
-//! Provides a no-std compatible [`LazyTrieData`] type for lazily initialized
-//! trie-related data containing sorted hashed state and trie updates.
-
-use crate::{updates::TrieUpdatesSorted, HashedPostStateSorted};
-use alloc::sync::Arc;
-use core::fmt;
-use reth_primitives_traits::sync::OnceLock;
-
-/// Container for sorted trie data: hashed state and trie updates.
-///
-/// This bundles both [`HashedPostStateSorted`] and [`TrieUpdatesSorted`] together
-/// for convenient passing and storage.
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
-pub struct SortedTrieData {
-    /// Sorted hashed post-state produced by execution.
-    pub hashed_state: Arc<HashedPostStateSorted>,
-    /// Sorted trie updates produced by state root computation.
-    pub trie_updates: Arc<TrieUpdatesSorted>,
-}
-
-impl SortedTrieData {
-    /// Creates a new [`SortedTrieData`] with the given values.
-    pub const fn new(
-        hashed_state: Arc<HashedPostStateSorted>,
-        trie_updates: Arc<TrieUpdatesSorted>,
-    ) -> Self {
-        Self { hashed_state, trie_updates }
-    }
-}
-
-/// Lazily initialized trie data containing sorted hashed state and trie updates.
-///
-/// This is a no-std compatible wrapper that supports two modes:
-/// 1. **Ready mode**: Data is available immediately (created via `ready()`)
-/// 2. **Deferred mode**: Data is computed on first access (created via `deferred()`)
-///
-/// In deferred mode, the computation runs on the first call to `get()`, `hashed_state()`,
-/// or `trie_updates()`, and results are cached for subsequent calls.
-///
-/// Cloning is cheap (Arc clone) and clones share the cached state.
-pub struct LazyTrieData {
-    /// Cached sorted trie data, computed on first access.
-    data: Arc<OnceLock<SortedTrieData>>,
-    /// Optional deferred computation function.
-    compute: Option<Arc<dyn Fn() -> SortedTrieData + Send + Sync>>,
-}
-
-impl Clone for LazyTrieData {
-    fn clone(&self) -> Self {
-        Self { data: Arc::clone(&self.data), compute: self.compute.clone() }
-    }
-}
-
-impl fmt::Debug for LazyTrieData {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("LazyTrieData")
-            .field("data", &if self.data.get().is_some() { "initialized" } else { "pending" })
-            .finish()
-    }
-}
-
-impl PartialEq for LazyTrieData {
-    fn eq(&self, other: &Self) -> bool {
-        self.get() == other.get()
-    }
-}
-
-impl Eq for LazyTrieData {}
-
-impl LazyTrieData {
-    /// Creates a new [`LazyTrieData`] that is already initialized with the given values.
-    pub fn ready(
-        hashed_state: Arc<HashedPostStateSorted>,
-        trie_updates: Arc<TrieUpdatesSorted>,
-    ) -> Self {
-        let data = OnceLock::new();
-        let _ = data.set(SortedTrieData::new(hashed_state, trie_updates));
-        Self { data: Arc::new(data), compute: None }
-    }
-
-    /// Creates a new [`LazyTrieData`] from pre-computed [`SortedTrieData`].
-    pub fn from_sorted(sorted: SortedTrieData) -> Self {
-        let data = OnceLock::new();
-        let _ = data.set(sorted);
-        Self { data: Arc::new(data), compute: None }
-    }
-
-    /// Creates a new [`LazyTrieData`] with a deferred computation function.
-    ///
-    /// The computation will run on the first call to `get()`, `hashed_state()`,
-    /// or `trie_updates()`. Results are cached for subsequent calls.
-    pub fn deferred(compute: impl Fn() -> SortedTrieData + Send + Sync + 'static) -> Self {
-        Self { data: Arc::new(OnceLock::new()), compute: Some(Arc::new(compute)) }
-    }
-
-    /// Returns a reference to the sorted trie data, computing if necessary.
-    ///
-    /// # Panics
-    ///
-    /// Panics if created via `deferred()` and the computation function was not provided.
-    pub fn get(&self) -> &SortedTrieData {
-        self.data.get_or_init(|| {
-            self.compute.as_ref().expect("LazyTrieData::get called before initialization")()
-        })
-    }
-
-    /// Returns a clone of the hashed state Arc.
-    ///
-    /// If not initialized, computes from the deferred source or panics.
-    pub fn hashed_state(&self) -> Arc<HashedPostStateSorted> {
-        Arc::clone(&self.get().hashed_state)
-    }
-
-    /// Returns a clone of the trie updates Arc.
-    ///
-    /// If not initialized, computes from the deferred source or panics.
-    pub fn trie_updates(&self) -> Arc<TrieUpdatesSorted> {
-        Arc::clone(&self.get().trie_updates)
-    }
-
-    /// Returns a clone of the [`SortedTrieData`].
-    ///
-    /// If not initialized, computes from the deferred source or panics.
-    pub fn sorted_trie_data(&self) -> SortedTrieData {
-        self.get().clone()
-    }
-}
-
-#[cfg(feature = "serde")]
-impl serde::Serialize for LazyTrieData {
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: serde::Serializer,
-    {
-        self.get().serialize(serializer)
-    }
-}
-
-#[cfg(feature = "serde")]
-impl<'de> serde::Deserialize<'de> for LazyTrieData {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        let data = SortedTrieData::deserialize(deserializer)?;
-        Ok(Self::from_sorted(data))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_lazy_ready_is_initialized() {
-        let lazy = LazyTrieData::ready(
-            Arc::new(HashedPostStateSorted::default()),
-            Arc::new(TrieUpdatesSorted::default()),
-        );
-        let _ = lazy.hashed_state();
-        let _ = lazy.trie_updates();
-    }
-
-    #[test]
-    fn test_lazy_clone_shares_state() {
-        let lazy1 = LazyTrieData::ready(
-            Arc::new(HashedPostStateSorted::default()),
-            Arc::new(TrieUpdatesSorted::default()),
-        );
-        let lazy2 = lazy1.clone();
-
-        // Both point to the same data
-        assert!(Arc::ptr_eq(&lazy1.hashed_state(), &lazy2.hashed_state()));
-        assert!(Arc::ptr_eq(&lazy1.trie_updates(), &lazy2.trie_updates()));
-    }
-
-    #[test]
-    fn test_lazy_deferred() {
-        let lazy = LazyTrieData::deferred(SortedTrieData::default);
-        assert!(lazy.hashed_state().is_empty());
-        assert!(lazy.trie_updates().is_empty());
-    }
-
-    #[test]
-    fn test_lazy_from_sorted() {
-        let sorted = SortedTrieData::default();
-        let lazy = LazyTrieData::from_sorted(sorted);
-        assert!(lazy.hashed_state().is_empty());
-        assert!(lazy.trie_updates().is_empty());
-    }
-}

crates/trie/common/src/lib.rs

@@ -11,10 +11,6 @@
 
 extern crate alloc;
 
-/// Lazy initialization wrapper for trie data.
-mod lazy;
-pub use lazy::{LazyTrieData, SortedTrieData};
-
 /// In-memory hashed state.
 mod hashed_state;
 pub use hashed_state::*;
@@ -59,9 +55,6 @@ pub use proofs::*;
 
 pub mod root;
 
-/// Incremental ordered trie root computation.
-pub mod ordered_root;
-
 /// Buffer for trie updates.
 pub mod updates;
 

crates/trie/common/src/ordered_root.rs

@@ -1,354 +0,0 @@
-//! Incremental ordered trie root computation.
-//!
-//! This module provides builders for computing ordered trie roots incrementally as items
-//! arrive, rather than requiring all items upfront. This is useful for receipt root
-//! calculation during block execution, where we know the total count but receive receipts
-//! one by one as transactions are executed.
-
-use crate::{HashBuilder, Nibbles, EMPTY_ROOT_HASH};
-use alloc::vec::Vec;
-use alloy_primitives::B256;
-use alloy_trie::root::adjust_index_for_rlp;
-use core::fmt;
-
-/// Error returned when using [`OrderedTrieRootEncodedBuilder`].
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum OrderedRootError {
-    /// Called `finalize()` before all items were pushed.
-    Incomplete {
-        /// The expected number of items.
-        expected: usize,
-        /// The number of items received.
-        received: usize,
-    },
-    /// Index is out of bounds.
-    IndexOutOfBounds {
-        /// The index that was provided.
-        index: usize,
-        /// The expected length.
-        len: usize,
-    },
-    /// Item at this index was already pushed.
-    DuplicateIndex {
-        /// The duplicate index.
-        index: usize,
-    },
-}
-
-impl OrderedRootError {
-    /// Returns `true` if the error is [`OrderedRootError::Incomplete`].
-    #[inline]
-    pub const fn is_incomplete(&self) -> bool {
-        matches!(self, Self::Incomplete { .. })
-    }
-
-    /// Returns `true` if the error is [`OrderedRootError::IndexOutOfBounds`].
-    #[inline]
-    pub const fn is_index_out_of_bounds(&self) -> bool {
-        matches!(self, Self::IndexOutOfBounds { .. })
-    }
-
-    /// Returns `true` if the error is [`OrderedRootError::DuplicateIndex`].
-    #[inline]
-    pub const fn is_duplicate_index(&self) -> bool {
-        matches!(self, Self::DuplicateIndex { .. })
-    }
-
-    /// Returns the index associated with the error, if any.
-    #[inline]
-    pub const fn index(&self) -> Option<usize> {
-        match self {
-            Self::Incomplete { .. } => None,
-            Self::IndexOutOfBounds { index, .. } | Self::DuplicateIndex { index } => Some(*index),
-        }
-    }
-}
-
-impl fmt::Display for OrderedRootError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            Self::Incomplete { expected, received } => {
-                write!(f, "incomplete: expected {expected} items, received {received}")
-            }
-            Self::IndexOutOfBounds { index, len } => {
-                write!(f, "index {index} out of bounds for length {len}")
-            }
-            Self::DuplicateIndex { index } => {
-                write!(f, "duplicate item at index {index}")
-            }
-        }
-    }
-}
-
-#[cfg(feature = "std")]
-impl std::error::Error for OrderedRootError {}
-
-/// A builder for computing ordered trie roots incrementally from pre-encoded items.
-///
-/// This builder allows pushing items one by one as they become available
-/// (e.g., receipts after each transaction execution), rather than requiring
-/// all items upfront.
-///
-/// # Use Case
-///
-/// When executing a block, the receipt root must be computed from all transaction
-/// receipts. With the standard `ordered_trie_root`, you must wait until all
-/// transactions are executed before computing the root. This builder enables
-/// **incremental computation** - you can start building the trie as soon as
-/// receipts become available, potentially in parallel with continued execution.
-///
-/// The builder requires knowing the total item count upfront (the number of
-/// transactions in the block), but items can be pushed in any order by index.
-///
-/// # How It Works
-///
-/// Items can be pushed in any order by specifying their index. The builder
-/// internally buffers items and flushes them to the underlying [`HashBuilder`]
-/// in the correct order for RLP key encoding (as determined by [`adjust_index_for_rlp`]).
-///
-/// # Memory
-///
-/// Each pushed item is stored in an internal buffer until it can be flushed.
-/// In the worst case (e.g., pushing index 0 last), all items except one will
-/// be buffered. For receipt roots, index 0 is typically flushed late due to
-/// RLP key ordering, so expect to buffer most items until near the end.
-///
-/// # Example
-///
-/// ```
-/// use reth_trie_common::ordered_root::OrderedTrieRootEncodedBuilder;
-///
-/// // Create a builder for 2 pre-encoded items
-/// let mut builder = OrderedTrieRootEncodedBuilder::new(2);
-///
-/// // Push pre-encoded items as they arrive (can be out of order)
-/// builder.push(1, b"encoded_item_1").unwrap();
-/// builder.push(0, b"encoded_item_0").unwrap();
-///
-/// // Finalize to get the root hash
-/// let root = builder.finalize().unwrap();
-/// ```
-#[derive(Debug)]
-pub struct OrderedTrieRootEncodedBuilder {
-    /// Total expected number of items.
-    len: usize,
-    /// Number of items received so far.
-    received: usize,
-    /// Next insertion loop counter (determines which adjusted index to flush next).
-    next_insert_i: usize,
-    /// Buffer for pending items, indexed by execution index.
-    pending: Vec<Option<Vec<u8>>>,
-    /// The underlying hash builder.
-    hb: HashBuilder,
-}
-
-impl OrderedTrieRootEncodedBuilder {
-    /// Creates a new builder for `len` pre-encoded items.
-    pub fn new(len: usize) -> Self {
-        Self {
-            len,
-            received: 0,
-            next_insert_i: 0,
-            pending: alloc::vec![None; len],
-            hb: HashBuilder::default(),
-        }
-    }
-
-    /// Pushes a pre-encoded item at the given index to the builder.
-    ///
-    /// Items can be pushed in any order. The builder will automatically
-    /// flush items to the underlying [`HashBuilder`] when they become
-    /// available in the correct order.
-    ///
-    /// # Errors
-    ///
-    /// - [`OrderedRootError::IndexOutOfBounds`] if `index >= len`
-    /// - [`OrderedRootError::DuplicateIndex`] if an item was already pushed at this index
-    #[inline]
-    pub fn push(&mut self, index: usize, bytes: &[u8]) -> Result<(), OrderedRootError> {
-        if index >= self.len {
-            return Err(OrderedRootError::IndexOutOfBounds { index, len: self.len });
-        }
-
-        if self.pending[index].is_some() {
-            return Err(OrderedRootError::DuplicateIndex { index });
-        }
-
-        self.push_unchecked(index, bytes);
-        Ok(())
-    }
-
-    /// Pushes a pre-encoded item at the given index without bounds or duplicate checking.
-    ///
-    /// This is a performance-critical method for callers that can guarantee:
-    /// - `index < len`
-    /// - No item has been pushed at this index before
-    ///
-    /// # Panics
-    ///
-    /// Panics in debug mode if `index >= len`.
-    #[inline]
-    pub fn push_unchecked(&mut self, index: usize, bytes: &[u8]) {
-        debug_assert!(index < self.len, "index {index} out of bounds for length {}", self.len);
-        debug_assert!(self.pending[index].is_none(), "duplicate item at index {index}");
-
-        self.pending[index] = Some(bytes.to_vec());
-        self.received += 1;
-
-        self.flush();
-    }
-
-    /// Attempts to flush pending items to the hash builder.
-    fn flush(&mut self) {
-        while self.next_insert_i < self.len {
-            let exec_index_needed = adjust_index_for_rlp(self.next_insert_i, self.len);
-
-            let Some(value) = self.pending[exec_index_needed].take() else {
-                break;
-            };
-
-            let index_buffer = alloy_rlp::encode_fixed_size(&exec_index_needed);
-            self.hb.add_leaf(Nibbles::unpack(&index_buffer), &value);
-
-            self.next_insert_i += 1;
-        }
-    }
-
-    /// Returns `true` if all items have been pushed.
-    #[inline]
-    pub const fn is_complete(&self) -> bool {
-        self.received == self.len
-    }
-
-    /// Returns the number of items pushed so far.
-    #[inline]
-    pub const fn pushed_count(&self) -> usize {
-        self.received
-    }
-
-    /// Returns the expected total number of items.
-    #[inline]
-    pub const fn expected_count(&self) -> usize {
-        self.len
-    }
-
-    /// Finalizes the builder and returns the trie root.
-    ///
-    /// # Errors
-    ///
-    /// Returns [`OrderedRootError::Incomplete`] if not all items have been pushed.
-    pub fn finalize(mut self) -> Result<B256, OrderedRootError> {
-        if self.len == 0 {
-            return Ok(EMPTY_ROOT_HASH);
-        }
-
-        if self.received != self.len {
-            return Err(OrderedRootError::Incomplete {
-                expected: self.len,
-                received: self.received,
-            });
-        }
-
-        debug_assert_eq!(self.next_insert_i, self.len, "not all items were flushed");
-
-        Ok(self.hb.root())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use alloy_trie::root::ordered_trie_root_encoded;
-
-    #[test]
-    fn test_ordered_encoded_builder_equivalence() {
-        for len in [0, 1, 2, 3, 10, 127, 128, 129, 130, 200] {
-            let items: Vec<Vec<u8>> =
-                (0..len).map(|i| format!("item_{i}_data").into_bytes()).collect();
-
-            let expected = ordered_trie_root_encoded(&items);
-
-            let mut builder = OrderedTrieRootEncodedBuilder::new(len);
-
-            for (i, item) in items.iter().enumerate() {
-                builder.push(i, item).unwrap();
-            }
-
-            let actual = builder.finalize().unwrap();
-            assert_eq!(
-                expected, actual,
-                "mismatch for len={len}: expected {expected:?}, got {actual:?}"
-            );
-        }
-    }
-
-    #[test]
-    fn test_ordered_builder_out_of_order() {
-        for len in [2, 3, 5, 10, 50] {
-            let items: Vec<Vec<u8>> =
-                (0..len).map(|i| format!("item_{i}_data").into_bytes()).collect();
-
-            let expected = ordered_trie_root_encoded(&items);
-
-            // Push in reverse order
-            let mut builder = OrderedTrieRootEncodedBuilder::new(len);
-            for i in (0..len).rev() {
-                builder.push(i, &items[i]).unwrap();
-            }
-            let actual = builder.finalize().unwrap();
-            assert_eq!(expected, actual, "mismatch for reverse order len={len}");
-
-            // Push odds first, then evens
-            let mut builder = OrderedTrieRootEncodedBuilder::new(len);
-            for i in (1..len).step_by(2) {
-                builder.push(i, &items[i]).unwrap();
-            }
-            for i in (0..len).step_by(2) {
-                builder.push(i, &items[i]).unwrap();
-            }
-            let actual = builder.finalize().unwrap();
-            assert_eq!(expected, actual, "mismatch for odd/even order len={len}");
-        }
-    }
-
-    #[test]
-    fn test_ordered_builder_empty() {
-        let builder = OrderedTrieRootEncodedBuilder::new(0);
-        assert!(builder.is_complete());
-        assert_eq!(builder.finalize().unwrap(), EMPTY_ROOT_HASH);
-    }
-
-    #[test]
-    fn test_ordered_builder_incomplete_error() {
-        let mut builder = OrderedTrieRootEncodedBuilder::new(3);
-
-        builder.push(0, b"item_0").unwrap();
-        builder.push(1, b"item_1").unwrap();
-
-        assert!(!builder.is_complete());
-        assert_eq!(
-            builder.finalize(),
-            Err(OrderedRootError::Incomplete { expected: 3, received: 2 })
-        );
-    }
-
-    #[test]
-    fn test_ordered_builder_index_errors() {
-        let mut builder = OrderedTrieRootEncodedBuilder::new(2);
-
-        assert_eq!(
-            builder.push(5, b"item"),
-            Err(OrderedRootError::IndexOutOfBounds { index: 5, len: 2 })
-        );
-
-        builder.push(0, b"item_0").unwrap();
-
-        assert_eq!(
-            builder.push(0, b"item_0_dup"),
-            Err(OrderedRootError::DuplicateIndex { index: 0 })
-        );
-
-        builder.push(1, b"item_1").unwrap();
-        assert!(builder.is_complete());
-    }
-}