feat(engine): add proof worker handle infrastructure to SparseTrieTask

Add infrastructure for SparseTrieTask to dispatch proof requests directly
to proof workers:

- Add proof_worker_handle field for dispatching proof requests
- Add proof_result_rx for receiving proof results
- Add with_proof_worker_handle() builder method

This prepares for the sparse-trie-as-cache optimization where the sparse
trie generates proof targets based on revealed nodes.

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

@@ -8,7 +8,7 @@ use crate::tree::{
     },
     payload_processor::{
         prewarm::{PrewarmCacheTask, PrewarmContext, PrewarmMode, PrewarmTaskEvent},
-        sparse_trie::StateRootComputeOutcome,
+        sparse_trie::{SparseTrieMessage, StateRootComputeOutcome},
     },
     sparse_trie::SparseTrieTask,
     StateProviderBuilder, TreeConfig,
@@ -19,7 +19,7 @@ use alloy_evm::{block::StateChangeSource, ToTxEnv};
 use alloy_primitives::B256;
 use crossbeam_channel::Sender as CrossbeamSender;
 use executor::WorkloadExecutor;
-use multiproof::{SparseTrieUpdate, *};
+use multiproof::*;
 use parking_lot::RwLock;
 use prewarm::PrewarmMetrics;
 use rayon::prelude::*;
@@ -483,7 +483,7 @@ where
     #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
     fn spawn_sparse_trie_task<BPF>(
         &self,
-        sparse_trie_rx: mpsc::Receiver<SparseTrieUpdate>,
+        sparse_trie_rx: mpsc::Receiver<SparseTrieMessage>,
         proof_worker_handle: BPF,
         state_root_tx: mpsc::Sender<Result<StateRootComputeOutcome, ParallelStateRootError>>,
     ) where

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

@@ -1,6 +1,8 @@
 //! Multiproof task related functionality.
 
-use crate::tree::payload_processor::bal::bal_to_hashed_post_state;
+use crate::tree::payload_processor::{
+    bal::bal_to_hashed_post_state, sparse_trie::SparseTrieMessage,
+};
 use alloy_eip7928::BlockAccessList;
 use alloy_evm::block::StateChangeSource;
 use alloy_primitives::{keccak256, map::HashSet, B256};
@@ -21,7 +23,7 @@ use reth_trie_parallel::{
     },
 };
 use revm_primitives::map::{hash_map, B256Map};
-use std::{collections::BTreeMap, sync::Arc, time::Instant};
+use std::{sync::Arc, time::Instant};
 use tracing::{debug, error, instrument, trace};
 
 /// Source of state changes, either from EVM execution or from a Block Access List.
@@ -119,51 +121,8 @@ pub(super) enum MultiProofMessage {
     FinishedStateUpdates,
 }
 
-/// Handle to track proof calculation ordering.
-#[derive(Debug, Default)]
-struct ProofSequencer {
-    /// The next proof sequence number to be produced.
-    next_sequence: u64,
-    /// The next sequence number expected to be delivered.
-    next_to_deliver: u64,
-    /// Buffer for out-of-order proofs and corresponding state updates
-    pending_proofs: BTreeMap<u64, SparseTrieUpdate>,
-}
-
-impl ProofSequencer {
-    /// Gets the next sequence number and increments the counter
-    const fn next_sequence(&mut self) -> u64 {
-        let seq = self.next_sequence;
-        self.next_sequence += 1;
-        seq
-    }
-
-    /// 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> {
-        if sequence >= self.next_to_deliver {
-            self.pending_proofs.insert(sequence, update);
-        }
-
-        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
-    pub(crate) fn has_pending(&self) -> bool {
-        !self.pending_proofs.is_empty()
-    }
-}
+// Re-export ProofSequencer from sparse_trie module
+use super::sparse_trie::ProofSequencer;
 
 /// A wrapper for the sender that signals completion when dropped.
 ///
@@ -540,8 +499,8 @@ pub(super) struct MultiProofTask {
     tx: CrossbeamSender<MultiProofMessage>,
     /// Receiver for proof results directly from workers.
     proof_result_rx: CrossbeamReceiver<ProofResultMessage>,
-    /// Sender for state updates emitted by this type.
-    to_sparse_trie: std::sync::mpsc::Sender<SparseTrieUpdate>,
+    /// Sender for messages to the sparse trie task.
+    to_sparse_trie: std::sync::mpsc::Sender<SparseTrieMessage>,
     /// Proof targets that have been already fetched.
     fetched_proof_targets: MultiProofTargets,
     /// Tracks keys which have been added and removed throughout the entire block.
@@ -563,7 +522,7 @@ impl MultiProofTask {
     /// `proof_result_rx`.
     pub(super) fn new(
         proof_worker_handle: ProofWorkerHandle,
-        to_sparse_trie: std::sync::mpsc::Sender<SparseTrieUpdate>,
+        to_sparse_trie: std::sync::mpsc::Sender<SparseTrieMessage>,
         chunk_size: Option<usize>,
         tx: CrossbeamSender<MultiProofMessage>,
         rx: CrossbeamReceiver<MultiProofMessage>,
@@ -868,7 +827,9 @@ impl MultiProofTask {
                                 sequence_number,
                                 SparseTrieUpdate { state, multiproof: Default::default() },
                             ) {
-                                let _ = self.to_sparse_trie.send(combined_update);
+                                let _ = self
+                                    .to_sparse_trie
+                                    .send(SparseTrieMessage::ProofUpdate(combined_update));
                             }
                         }
                         Ok(other_msg) => {
@@ -1000,7 +961,8 @@ impl MultiProofTask {
                     sequence_number,
                     SparseTrieUpdate { state, multiproof: Default::default() },
                 ) {
-                    let _ = self.to_sparse_trie.send(combined_update);
+                    let _ =
+                        self.to_sparse_trie.send(SparseTrieMessage::ProofUpdate(combined_update));
                 }
 
                 if self.is_done(batch_metrics, ctx) {
@@ -1107,7 +1069,9 @@ impl MultiProofTask {
                                     if let Some(combined_update) =
                                         self.on_proof(proof_result.sequence_number, update)
                                     {
-                                        let _ = self.to_sparse_trie.send(combined_update);
+                                        let _ = self
+                                            .to_sparse_trie
+                                            .send(SparseTrieMessage::ProofUpdate(combined_update));
                                     }
                                 }
                                 Err(error) => {
@@ -1379,7 +1343,7 @@ mod tests {
         let mut sequencer = ProofSequencer::default();
         let proof1 = MultiProof::default();
         let proof2 = MultiProof::default();
-        sequencer.next_sequence = 2;
+        sequencer.set_next_sequence(2);
 
         let ready = sequencer.add_proof(0, SparseTrieUpdate::from_multiproof(proof1).unwrap());
         assert_eq!(ready.len(), 1);
@@ -1396,7 +1360,7 @@ mod tests {
         let proof1 = MultiProof::default();
         let proof2 = MultiProof::default();
         let proof3 = MultiProof::default();
-        sequencer.next_sequence = 3;
+        sequencer.set_next_sequence(3);
 
         let ready = sequencer.add_proof(2, SparseTrieUpdate::from_multiproof(proof3).unwrap());
         assert_eq!(ready.len(), 0);
@@ -1416,7 +1380,7 @@ mod tests {
         let mut sequencer = ProofSequencer::default();
         let proof1 = MultiProof::default();
         let proof3 = MultiProof::default();
-        sequencer.next_sequence = 3;
+        sequencer.set_next_sequence(3);
 
         let ready = sequencer.add_proof(0, SparseTrieUpdate::from_multiproof(proof1).unwrap());
         assert_eq!(ready.len(), 1);
@@ -1444,7 +1408,7 @@ mod tests {
     fn test_add_proof_batch_processing() {
         let mut sequencer = ProofSequencer::default();
         let proofs: Vec<_> = (0..5).map(|_| MultiProof::default()).collect();
-        sequencer.next_sequence = 5;
+        sequencer.set_next_sequence(5);
 
         sequencer.add_proof(4, SparseTrieUpdate::from_multiproof(proofs[4].clone()).unwrap());
         sequencer.add_proof(2, SparseTrieUpdate::from_multiproof(proofs[2].clone()).unwrap());

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

@@ -3,8 +3,12 @@
 use crate::tree::payload_processor::multiproof::{MultiProofTaskMetrics, SparseTrieUpdate};
 use alloy_primitives::B256;
 use rayon::iter::{ParallelBridge, ParallelIterator};
-use reth_trie::{updates::TrieUpdates, Nibbles};
-use reth_trie_parallel::root::ParallelStateRootError;
+use crossbeam_channel::Receiver as CrossbeamReceiver;
+use reth_trie::{updates::TrieUpdates, HashedPostState, Nibbles};
+use reth_trie_parallel::{
+    proof_task::{ProofResultMessage, ProofWorkerHandle},
+    root::ParallelStateRootError,
+};
 use reth_trie_sparse::{
     errors::{SparseStateTrieResult, SparseTrieErrorKind},
     provider::{TrieNodeProvider, TrieNodeProviderFactory},
@@ -12,11 +16,86 @@ use reth_trie_sparse::{
 };
 use smallvec::SmallVec;
 use std::{
+    collections::BTreeMap,
     sync::mpsc,
     time::{Duration, Instant},
 };
 use tracing::{debug, debug_span, instrument, trace};
 
+/// Message types for the sparse trie task.
+#[derive(Debug)]
+pub(super) enum SparseTrieMessage {
+    /// Already computed proof update (existing flow from `MultiProofTask`).
+    ProofUpdate(SparseTrieUpdate),
+    /// State update needing proof target generation (new flow for sparse trie caching).
+    #[allow(dead_code)]
+    StateUpdate {
+        /// The sequence number for ordering.
+        sequence_number: u64,
+        /// The hashed post state to process.
+        state: HashedPostState,
+    },
+}
+
+/// Handle to track proof calculation ordering.
+///
+/// The `ProofSequencer` ensures that proofs are processed in the correct order,
+/// buffering out-of-order proofs until all preceding proofs have been received.
+#[derive(Debug, Default)]
+pub(super) struct ProofSequencer {
+    /// The next proof sequence number to be produced.
+    next_sequence: u64,
+    /// The next sequence number expected to be delivered.
+    next_to_deliver: u64,
+    /// Buffer for out-of-order proofs and corresponding state updates
+    pending_proofs: BTreeMap<u64, SparseTrieUpdate>,
+}
+
+impl ProofSequencer {
+    /// Gets the next sequence number and increments the counter
+    pub(super) const fn next_sequence(&mut self) -> u64 {
+        let seq = self.next_sequence;
+        self.next_sequence += 1;
+        seq
+    }
+
+    /// Adds a proof with the corresponding state update and returns all sequential proofs and state
+    /// updates if we have a continuous sequence
+    pub(super) fn add_proof(
+        &mut self,
+        sequence: u64,
+        update: SparseTrieUpdate,
+    ) -> Vec<SparseTrieUpdate> {
+        if sequence >= self.next_to_deliver {
+            self.pending_proofs.insert(sequence, update);
+        }
+
+        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
+    pub(super) fn has_pending(&self) -> bool {
+        !self.pending_proofs.is_empty()
+    }
+
+    /// Sets the next sequence number (for testing purposes).
+    #[cfg(test)]
+    pub(super) const fn set_next_sequence(&mut self, seq: u64) {
+        self.next_sequence = seq;
+    }
+}
+
 /// A task responsible for populating the sparse trie.
 pub(super) struct SparseTrieTask<BPF, A = SerialSparseTrie, S = SerialSparseTrie>
 where
@@ -24,13 +103,23 @@ where
     BPF::AccountNodeProvider: TrieNodeProvider + Send + Sync,
     BPF::StorageNodeProvider: TrieNodeProvider + Send + Sync,
 {
-    /// Receives updates from the state root task.
-    pub(super) updates: mpsc::Receiver<SparseTrieUpdate>,
+    /// Receives messages from the state root task (either proof updates or state updates).
+    pub(super) messages: mpsc::Receiver<SparseTrieMessage>,
     /// `SparseStateTrie` used for computing the state root.
     pub(super) trie: SparseStateTrie<A, S>,
     pub(super) metrics: MultiProofTaskMetrics,
     /// Trie node provider factory.
     blinded_provider_factory: BPF,
+    /// Proof sequencer for ordering state updates.
+    #[allow(dead_code)]
+    proof_sequencer: ProofSequencer,
+    /// Handle to the proof worker pools for dispatching proof requests.
+    /// When set, the sparse trie task can generate proof targets and dispatch them to workers.
+    #[allow(dead_code)]
+    proof_worker_handle: Option<ProofWorkerHandle>,
+    /// Receiver for proof results from workers when using direct proof dispatch.
+    #[allow(dead_code)]
+    proof_result_rx: Option<CrossbeamReceiver<ProofResultMessage>>,
 }
 
 impl<BPF, A, S> SparseTrieTask<BPF, A, S>
@@ -43,12 +132,32 @@ where
 {
     /// Creates a new sparse trie, pre-populating with a [`ClearedSparseStateTrie`].
     pub(super) fn new_with_cleared_trie(
-        updates: mpsc::Receiver<SparseTrieUpdate>,
+        messages: mpsc::Receiver<SparseTrieMessage>,
         blinded_provider_factory: BPF,
         metrics: MultiProofTaskMetrics,
         sparse_state_trie: ClearedSparseStateTrie<A, S>,
     ) -> Self {
-        Self { updates, metrics, trie: sparse_state_trie.into_inner(), blinded_provider_factory }
+        Self {
+            messages,
+            metrics,
+            trie: sparse_state_trie.into_inner(),
+            blinded_provider_factory,
+            proof_sequencer: ProofSequencer::default(),
+            proof_worker_handle: None,
+            proof_result_rx: None,
+        }
+    }
+
+    /// Sets the proof worker handle for dispatching proof requests to workers.
+    #[allow(dead_code)]
+    pub(super) fn with_proof_worker_handle(
+        mut self,
+        handle: ProofWorkerHandle,
+        proof_result_rx: CrossbeamReceiver<ProofResultMessage>,
+    ) -> Self {
+        self.proof_worker_handle = Some(handle);
+        self.proof_result_rx = Some(proof_result_rx);
+        self
     }
 
     /// Runs the sparse trie task to completion.
@@ -82,35 +191,49 @@ where
 
         let mut num_iterations = 0;
 
-        while let Ok(mut update) = self.updates.recv() {
+        while let Ok(message) = self.messages.recv() {
             num_iterations += 1;
-            let mut num_updates = 1;
             let _enter =
-                debug_span!(target: "engine::tree::payload_processor::sparse_trie", "drain updates")
+                debug_span!(target: "engine::tree::payload_processor::sparse_trie", "process message")
                     .entered();
-            while let Ok(next) = self.updates.try_recv() {
-                update.extend(next);
-                num_updates += 1;
+
+            match message {
+                SparseTrieMessage::ProofUpdate(mut update) => {
+                    let mut num_updates = 1;
+                    while let Ok(next) = self.messages.try_recv() {
+                        match next {
+                            SparseTrieMessage::ProofUpdate(next_update) => {
+                                update.extend(next_update);
+                                num_updates += 1;
+                            }
+                            SparseTrieMessage::StateUpdate { sequence_number, state } => {
+                                self.handle_state_update(sequence_number, state)?;
+                            }
+                        }
+                    }
+
+                    debug!(
+                        target: "engine::root",
+                        num_updates,
+                        account_proofs = update.multiproof.account_subtree.len(),
+                        storage_proofs = update.multiproof.storages.len(),
+                        "Updating sparse trie"
+                    );
+
+                    let elapsed =
+                        update_sparse_trie(&mut self.trie, update, &self.blinded_provider_factory)
+                            .map_err(|e| {
+                                ParallelStateRootError::Other(format!(
+                                    "could not calculate state root: {e:?}"
+                                ))
+                            })?;
+                    self.metrics.sparse_trie_update_duration_histogram.record(elapsed);
+                    trace!(target: "engine::root", ?elapsed, num_iterations, "Root calculation completed");
+                }
+                SparseTrieMessage::StateUpdate { sequence_number, state } => {
+                    self.handle_state_update(sequence_number, state)?;
+                }
             }
-            drop(_enter);
-
-            debug!(
-                target: "engine::root",
-                num_updates,
-                account_proofs = update.multiproof.account_subtree.len(),
-                storage_proofs = update.multiproof.storages.len(),
-                "Updating sparse trie"
-            );
-
-            let elapsed =
-                update_sparse_trie(&mut self.trie, update, &self.blinded_provider_factory)
-                    .map_err(|e| {
-                        ParallelStateRootError::Other(format!(
-                            "could not calculate state root: {e:?}"
-                        ))
-                    })?;
-            self.metrics.sparse_trie_update_duration_histogram.record(elapsed);
-            trace!(target: "engine::root", ?elapsed, num_iterations, "Root calculation completed");
         }
 
         debug!(target: "engine::root", num_iterations, "All proofs processed, ending calculation");
@@ -127,6 +250,31 @@ where
 
         Ok(StateRootComputeOutcome { state_root, trie_updates })
     }
+
+    /// Handles a state update by generating proof targets from the sparse trie's knowledge
+    /// of revealed nodes and processing the state.
+    ///
+    /// This is a placeholder for the sparse-trie-as-cache optimization. Currently, it logs
+    /// the state update information but doesn't generate proof targets since that requires
+    /// access to the internal trie nodes which is only available for `SerialSparseTrie`.
+    fn handle_state_update(
+        &self,
+        sequence_number: u64,
+        state: HashedPostState,
+    ) -> Result<(), ParallelStateRootError> {
+        let num_accounts = state.accounts.len();
+        let num_storage_updates: usize = state.storages.values().map(|s| s.storage.len()).sum();
+
+        debug!(
+            target: "engine::root",
+            sequence_number,
+            num_accounts,
+            num_storage_updates,
+            "Received state update for sparse trie caching (proof target generation pending)"
+        );
+
+        Ok(())
+    }
 }
 
 /// Outcome of the state root computation, including the state root itself with

crates/trie/sparse-parallel/src/lib.rs

@@ -10,5 +10,8 @@ pub use trie::*;
 mod lower;
 use lower::*;
 
+mod state;
+pub use state::*;
+
 #[cfg(feature = "metrics")]
 mod metrics;

crates/trie/sparse-parallel/src/state.rs

@@ -0,0 +1,45 @@
+//! Extensions for `SparseStateTrie` when using `ParallelSparseTrie`.
+
+use crate::ParallelSparseTrie;
+use alloy_primitives::B256;
+use reth_trie_sparse::SparseStateTrie;
+
+/// Extension trait for `SparseStateTrie<ParallelSparseTrie, ParallelSparseTrie>` that provides
+/// optimized proof target generation methods.
+pub trait ParallelSparseStateTrieExt {
+    /// Generate account proof targets using the parallel trie's optimized method.
+    ///
+    /// For each key, finds the deepest revealed node and returns a `(key, min_len)` pair.
+    /// Keys that are already fully revealed (leaf exists) are excluded.
+    fn generate_account_proof_targets_optimized(&self, keys: &[B256]) -> Vec<(B256, u8)>;
+
+    /// Generate storage proof targets using the parallel trie's optimized method.
+    ///
+    /// For each storage key, finds the deepest revealed node and returns a `(key, min_len)` pair.
+    /// Keys that are already fully revealed (leaf exists) are excluded.
+    fn generate_storage_proof_targets_optimized(
+        &self,
+        account: B256,
+        storage_keys: &[B256],
+    ) -> Vec<(B256, u8)>;
+}
+
+impl ParallelSparseStateTrieExt for SparseStateTrie<ParallelSparseTrie, ParallelSparseTrie> {
+    fn generate_account_proof_targets_optimized(&self, keys: &[B256]) -> Vec<(B256, u8)> {
+        match self.state_trie_ref() {
+            None => keys.iter().map(|key| (*key, 0u8)).collect(),
+            Some(trie) => trie.generate_proof_targets(keys),
+        }
+    }
+
+    fn generate_storage_proof_targets_optimized(
+        &self,
+        account: B256,
+        storage_keys: &[B256],
+    ) -> Vec<(B256, u8)> {
+        match self.storage_trie_ref(&account) {
+            None => storage_keys.iter().map(|key| (*key, 0u8)).collect(),
+            Some(trie) => trie.generate_proof_targets(storage_keys),
+        }
+    }
+}

crates/trie/sparse-parallel/src/trie.rs

@@ -1600,6 +1600,81 @@ impl ParallelSparseTrie {
             self.lower_subtries[index] = LowerSparseSubtrie::Revealed(subtrie);
         }
     }
+
+    /// Find the deepest revealed depth for a path in the parallel sparse trie.
+    ///
+    /// Walks the trie from root following the path until:
+    /// - We hit a Hash node (blinded) - return the current depth
+    /// - We hit an Empty node or missing node - return the current depth
+    /// - We reach the end of the path - return the full path length
+    pub fn find_deepest_revealed_depth(&self, full_path: &Nibbles) -> u8 {
+        let mut current = Nibbles::default();
+        let mut deepest_revealed = 0u8;
+
+        while current.len() < full_path.len() {
+            let subtrie = if SparseSubtrieType::path_len_is_upper(current.len()) {
+                Some(self.upper_subtrie.as_ref())
+            } else {
+                self.lower_subtrie_for_path(&current)
+            };
+
+            let Some(subtrie) = subtrie else {
+                break;
+            };
+
+            match subtrie.nodes.get(&current) {
+                Some(SparseNode::Empty | SparseNode::Hash(_)) | None => break,
+                Some(SparseNode::Leaf { key, .. }) => {
+                    let leaf_path_len = current.len() + key.len();
+                    deepest_revealed = leaf_path_len.min(64) as u8;
+                    break;
+                }
+                Some(SparseNode::Extension { key, .. }) => {
+                    deepest_revealed = current.len() as u8;
+                    current.extend(key);
+                    if full_path.len() < current.len() || !full_path.starts_with(&current) {
+                        break;
+                    }
+                }
+                Some(SparseNode::Branch { state_mask, .. }) => {
+                    deepest_revealed = current.len() as u8;
+                    let next_nibble = full_path.get_unchecked(current.len());
+                    if !state_mask.is_bit_set(next_nibble) {
+                        break;
+                    }
+                    current.push_unchecked(next_nibble);
+                }
+            }
+        }
+
+        deepest_revealed
+    }
+
+    /// Generate proof targets for the given keys.
+    ///
+    /// For each key, finds the deepest revealed node and returns a `(key, min_len)` pair.
+    /// Keys that are fully revealed (have a leaf) are excluded.
+    pub fn generate_proof_targets(&self, keys: &[B256]) -> Vec<(B256, u8)> {
+        let mut targets = Vec::with_capacity(keys.len());
+
+        for key in keys {
+            let path = Nibbles::unpack(key);
+
+            // Check if already fully revealed as a leaf in either upper or lower subtrie
+            let has_leaf = std::iter::once(self.upper_subtrie.as_ref())
+                .chain(self.lower_subtrie_for_path(&path))
+                .any(|subtrie| subtrie.inner.values.contains_key(&path));
+
+            if has_leaf {
+                continue;
+            }
+
+            let min_len = self.find_deepest_revealed_depth(&path);
+            targets.push((*key, min_len));
+        }
+
+        targets
+    }
 }
 
 /// This is a subtrie of the [`ParallelSparseTrie`] that contains a map from path to sparse trie
@@ -6968,4 +7043,129 @@ mod tests {
 
         assert_eq!(branch_0x3_update, &expected_branch);
     }
+
+    #[test]
+    fn test_generate_proof_targets_empty_trie() {
+        // Default trie is revealed with an empty root
+        let trie = ParallelSparseTrie::default();
+
+        let key1 = B256::repeat_byte(0x11);
+        let key2 = B256::repeat_byte(0x22);
+
+        // Since root is Empty (not a leaf for these keys), both should need proofs
+        let targets = trie.generate_proof_targets(&[key1, key2]);
+        assert_eq!(targets.len(), 2);
+        assert_eq!(targets[0], (key1, 0));
+        assert_eq!(targets[1], (key2, 0));
+    }
+
+    #[test]
+    fn test_generate_proof_targets_revealed_leaf() {
+        let mut trie = ParallelSparseTrie::default();
+
+        let key1 = B256::repeat_byte(0x00);
+        let nibbles = Nibbles::unpack(key1);
+        let leaf_node = create_leaf_node(nibbles.to_vec(), 1);
+
+        // Reveal root as the leaf
+        trie = trie.with_root(leaf_node, None, false).unwrap();
+
+        // Key1 is fully revealed as a leaf, should be excluded from targets
+        let targets = trie.generate_proof_targets(&[key1]);
+        assert!(targets.is_empty());
+    }
+
+    #[test]
+    fn test_generate_proof_targets_partial_reveal() {
+        let mut trie = ParallelSparseTrie::default();
+
+        // Create a branch node at root with blinded children at nibbles 0 and 1
+        let key_0 = B256::repeat_byte(0x00);
+        let key_1 = B256::repeat_byte(0x11);
+
+        let child_hash_0 = B256::repeat_byte(0xaa);
+        let child_hash_1 = B256::repeat_byte(0xbb);
+        let branch_node = create_branch_node_with_children(
+            &[0x0, 0x1],
+            vec![RlpNode::word_rlp(&child_hash_0), RlpNode::word_rlp(&child_hash_1)],
+        );
+
+        trie = trie.with_root(branch_node, None, false).unwrap();
+
+        // Both keys have blinded children, so both need proofs
+        let targets = trie.generate_proof_targets(&[key_0, key_1]);
+        assert_eq!(targets.len(), 2);
+        // Branch at depth 0 is revealed, children at depth 1 are blinded
+        assert_eq!(targets[0], (key_0, 0));
+        assert_eq!(targets[1], (key_1, 0));
+
+        // A key at nibble 2 has no child in the branch
+        let key_2 = B256::repeat_byte(0x22);
+        let targets = trie.generate_proof_targets(&[key_2]);
+        assert_eq!(targets.len(), 1);
+        assert_eq!(targets[0], (key_2, 0));
+    }
+
+    #[test]
+    fn test_find_deepest_revealed_depth_extension() {
+        let mut trie = ParallelSparseTrie::default();
+
+        // Create extension -> blinded child structure
+        let child_hash = B256::repeat_byte(0xbb);
+        let ext_node = create_extension_node([0x1, 0x2, 0x3], child_hash);
+
+        trie = trie.with_root(ext_node, None, false).unwrap();
+
+        // For a path that follows the extension but then hits the blinded child
+        let following_path = Nibbles::from_nibbles([0x1, 0x2, 0x3, 0x4, 0x5]);
+        let depth = trie.find_deepest_revealed_depth(&following_path);
+        // Extension is at root (depth 0), after following it we're at depth 3,
+        // then we hit the blinded hash, so deepest revealed is 0 (the extension itself)
+        assert_eq!(depth, 0);
+
+        // For a path that diverges from the extension
+        let diverging_path = Nibbles::from_nibbles([0x1, 0x2, 0x4, 0x0, 0x0]);
+        let depth = trie.find_deepest_revealed_depth(&diverging_path);
+        // Extension at depth 0, but path diverges so we stop there
+        assert_eq!(depth, 0);
+
+        // For a path that doesn't match at all
+        let non_matching_path = Nibbles::from_nibbles([0x2, 0x0, 0x0, 0x0, 0x0]);
+        let depth = trie.find_deepest_revealed_depth(&non_matching_path);
+        // Extension at depth 0 doesn't match, so depth is 0
+        assert_eq!(depth, 0);
+    }
+
+    #[test]
+    fn test_find_deepest_revealed_depth_branch() {
+        let mut trie = ParallelSparseTrie::default();
+
+        // Create branch with blinded children at nibbles 0 and 1
+        let child_hash_0 = B256::repeat_byte(0xaa);
+        let child_hash_1 = B256::repeat_byte(0xbb);
+
+        let branch_node = create_branch_node_with_children(
+            &[0x0, 0x1],
+            vec![RlpNode::word_rlp(&child_hash_0), RlpNode::word_rlp(&child_hash_1)],
+        );
+
+        trie = trie.with_root(branch_node, None, false).unwrap();
+
+        // Path going through nibble 0 - hits blinded hash at depth 1
+        let path_0 = Nibbles::from_nibbles([0x0, 0x0, 0x0, 0x0]);
+        let depth = trie.find_deepest_revealed_depth(&path_0);
+        // Branch at depth 0 is revealed, child at depth 1 is blinded
+        assert_eq!(depth, 0);
+
+        // Path going through nibble 1 - hits blinded hash at depth 1
+        let path_1 = Nibbles::from_nibbles([0x1, 0x0, 0x0, 0x0]);
+        let depth = trie.find_deepest_revealed_depth(&path_1);
+        assert_eq!(depth, 0);
+
+        // Path going through nibble 2 - no child at this nibble
+        let path_2 = Nibbles::from_nibbles([0x2, 0x0, 0x0, 0x0]);
+        let depth = trie.find_deepest_revealed_depth(&path_2);
+        // Branch at depth 0 doesn't have child at nibble 2
+        assert_eq!(depth, 0);
+    }
 }

crates/trie/sparse/src/state.rs

@@ -806,6 +806,207 @@ where
         storage_trie.remove_leaf(slot, provider)?;
         Ok(())
     }
+
+    /// Generate proof targets for the given account keys (generic fallback).
+    ///
+    /// This generic implementation returns full proofs (`min_len=0`) for all keys,
+    /// excluding keys that are already revealed as leaves.
+    ///
+    /// For an optimized implementation that calculates partial proof depths based on
+    /// revealed nodes, see the `SerialSparseTrie` specialization.
+    pub fn generate_account_proof_targets(&self, keys: &[B256]) -> Vec<(B256, u8)> {
+        let mut targets = Vec::with_capacity(keys.len());
+
+        for key in keys {
+            let path = Nibbles::unpack(key);
+
+            // Check if already fully revealed as a leaf using the trait interface
+            if let Some(trie) = self.state.as_revealed_ref() &&
+                trie.get_leaf_value(&path).is_some()
+            {
+                continue;
+            }
+
+            // For generic types, we can't walk the node structure, so request full proof
+            targets.push((*key, 0u8));
+        }
+
+        targets
+    }
+
+    /// Generate storage proof targets for the given storage keys of an account (generic fallback).
+    ///
+    /// This generic implementation returns full proofs (`min_len=0`) for all keys,
+    /// excluding keys that are already revealed as leaves.
+    ///
+    /// For an optimized implementation that calculates partial proof depths based on
+    /// revealed nodes, see the `SerialSparseTrie` specialization.
+    pub fn generate_storage_proof_targets(
+        &self,
+        account: B256,
+        storage_keys: &[B256],
+    ) -> Vec<(B256, u8)> {
+        let mut targets = Vec::with_capacity(storage_keys.len());
+
+        let storage_trie = self.storage.tries.get(&account).and_then(|t| t.as_revealed_ref());
+
+        for key in storage_keys {
+            let path = Nibbles::unpack(key);
+
+            // Check if already fully revealed as a leaf using the trait interface
+            if let Some(trie) = storage_trie &&
+                trie.get_leaf_value(&path).is_some()
+            {
+                continue;
+            }
+
+            // For generic types, we can't walk the node structure, so request full proof
+            targets.push((*key, 0u8));
+        }
+
+        targets
+    }
+}
+
+/// Implementation of optimized proof v2 target generation for `SparseStateTrie` using
+/// `SerialSparseTrie`.
+///
+/// These methods allow generating partial proof targets based on what's already revealed
+/// in the sparse trie, enabling efficient incremental proof fetching with accurate `min_len`.
+impl SparseStateTrie<SerialSparseTrie, SerialSparseTrie> {
+    /// Generate proof targets for the given account keys with optimized `min_len` calculation.
+    ///
+    /// For each key, walks down the account trie to find the deepest revealed node.
+    /// Returns a vector of `(key, min_len)` pairs where:
+    /// - `key` is the account key that needs a proof
+    /// - `min_len` is the depth of the deepest already-revealed node (in nibbles)
+    ///
+    /// Keys that are already fully revealed (leaf exists) are excluded from the result.
+    ///
+    /// This allows generating partial proofs - we only need to fetch nodes we don't already have.
+    pub fn generate_account_proof_targets_optimized(&self, keys: &[B256]) -> Vec<(B256, u8)> {
+        let Some(trie) = self.state.as_revealed_ref() else {
+            // If the trie is blind, we need full proofs for all keys
+            return keys.iter().map(|key| (*key, 0u8)).collect();
+        };
+
+        let nodes = trie.nodes_ref();
+        let mut targets = Vec::with_capacity(keys.len());
+
+        for key in keys {
+            let path = Nibbles::unpack(key);
+
+            // Check if already fully revealed as a leaf
+            if trie.get_leaf_value(&path).is_some() {
+                continue;
+            }
+
+            let min_len = Self::find_deepest_revealed_depth_in_nodes(nodes, &path);
+            targets.push((*key, min_len));
+        }
+
+        targets
+    }
+
+    /// Generate storage proof targets for the given storage keys of an account with optimized
+    /// `min_len` calculation.
+    ///
+    /// For each storage key, walks down the storage trie to find the deepest revealed node.
+    /// Returns a vector of `(key, min_len)` pairs where:
+    /// - `key` is the storage key that needs a proof
+    /// - `min_len` is the depth of the deepest already-revealed node (in nibbles)
+    ///
+    /// Keys that are already fully revealed (leaf exists) are excluded from the result.
+    pub fn generate_storage_proof_targets_optimized(
+        &self,
+        account: B256,
+        storage_keys: &[B256],
+    ) -> Vec<(B256, u8)> {
+        let Some(trie) = self.storage.tries.get(&account).and_then(|t| t.as_revealed_ref()) else {
+            // If the storage trie is blind or doesn't exist, we need full proofs for all keys
+            return storage_keys.iter().map(|key| (*key, 0u8)).collect();
+        };
+
+        let nodes = trie.nodes_ref();
+        let mut targets = Vec::with_capacity(storage_keys.len());
+
+        for key in storage_keys {
+            let path = Nibbles::unpack(key);
+
+            // Check if already fully revealed as a leaf
+            if trie.get_leaf_value(&path).is_some() {
+                continue;
+            }
+
+            let min_len = Self::find_deepest_revealed_depth_in_nodes(nodes, &path);
+            targets.push((*key, min_len));
+        }
+
+        targets
+    }
+
+    /// Helper to find the deepest revealed depth for a path in a set of trie nodes.
+    ///
+    /// Walks the trie from root following the path until:
+    /// - We hit a Hash node (blinded) - return the current depth
+    /// - We hit an Empty node or missing node - return the current depth
+    /// - We reach the end of the path - return the full path length
+    fn find_deepest_revealed_depth_in_nodes(
+        nodes: &alloy_primitives::map::HashMap<Nibbles, crate::SparseNode>,
+        full_path: &Nibbles,
+    ) -> u8 {
+        use crate::SparseNode;
+
+        let mut current = Nibbles::default();
+        let mut deepest_revealed = 0u8;
+
+        while current.len() < full_path.len() {
+            match nodes.get(&current) {
+                Some(SparseNode::Empty) | None => {
+                    // No node at this path
+                    break;
+                }
+                Some(SparseNode::Hash(_)) => {
+                    // Hit a blinded node - stop here
+                    break;
+                }
+                Some(SparseNode::Leaf { key, .. }) => {
+                    // Found a leaf - the path is revealed up to current + key
+                    let leaf_path_len = current.len() + key.len();
+                    deepest_revealed = leaf_path_len.min(64) as u8;
+                    break;
+                }
+                Some(SparseNode::Extension { key, .. }) => {
+                    // Extension node - advance by the extension key
+                    deepest_revealed = current.len() as u8;
+                    current.extend(key);
+
+                    // Check if the path diverges
+                    if full_path.len() < current.len() || !full_path.starts_with(&current) {
+                        break;
+                    }
+                }
+                Some(SparseNode::Branch { state_mask, .. }) => {
+                    deepest_revealed = current.len() as u8;
+
+                    // Get the next nibble in the path
+                    let nibble = full_path.get_unchecked(current.len());
+
+                    // Check if branch has a child at this nibble
+                    if !state_mask.is_bit_set(nibble) {
+                        // No child at this nibble - path diverges here
+                        break;
+                    }
+
+                    // Continue down the branch
+                    current.push_unchecked(nibble);
+                    deepest_revealed = current.len() as u8;
+                }
+            }
+        }
+
+        deepest_revealed
+    }
 }
 
 /// The fields of [`SparseStateTrie`] related to storage tries. This is kept separate from the rest
@@ -1375,4 +1576,67 @@ mod tests {
             }
         );
     }
+
+    #[test]
+    fn generate_account_proof_targets_blind_trie() {
+        let sparse = SparseStateTrie::<SerialSparseTrie>::default();
+
+        let key1 = B256::repeat_byte(0x11);
+        let key2 = B256::repeat_byte(0x22);
+
+        // Blind trie should return all keys with min_len 0
+        let targets = sparse.generate_account_proof_targets(&[key1, key2]);
+        assert_eq!(targets.len(), 2);
+        assert_eq!(targets[0], (key1, 0));
+        assert_eq!(targets[1], (key2, 0));
+    }
+
+    #[test]
+    fn generate_account_proof_targets_revealed_leaf() {
+        let mut sparse = SparseStateTrie::<SerialSparseTrie>::default();
+
+        let key1 = B256::repeat_byte(0x00);
+        let leaf_value = alloy_rlp::encode(TrieAccount::default());
+        let leaf =
+            alloy_rlp::encode(TrieNode::Leaf(LeafNode::new(Nibbles::unpack(key1), leaf_value)));
+
+        let multiproof = MultiProof {
+            account_subtree: ProofNodes::from_iter([(Nibbles::default(), leaf.into())]),
+            ..Default::default()
+        };
+
+        sparse.reveal_decoded_multiproof(multiproof.try_into().unwrap()).unwrap();
+
+        // Key1 is fully revealed, so it should not be in the targets
+        let targets = sparse.generate_account_proof_targets(&[key1]);
+        assert!(targets.is_empty());
+    }
+
+    #[test]
+    fn generate_account_proof_targets_partial_reveal() {
+        let mut sparse = SparseStateTrie::<SerialSparseTrie>::default();
+
+        // Create a branch node at root with child at nibble 0
+        let key_revealed = B256::repeat_byte(0x00);
+        let key_unrevealed = B256::repeat_byte(0x10); // Different first nibble
+
+        let leaf_value = alloy_rlp::encode(TrieAccount::default());
+        let leaf = alloy_rlp::encode(TrieNode::Leaf(LeafNode::new(
+            Nibbles::unpack(key_revealed),
+            leaf_value,
+        )));
+
+        // Reveal only the path to key_revealed
+        let multiproof = MultiProof {
+            account_subtree: ProofNodes::from_iter([(Nibbles::default(), leaf.into())]),
+            ..Default::default()
+        };
+
+        sparse.reveal_decoded_multiproof(multiproof.try_into().unwrap()).unwrap();
+
+        // key_unrevealed should need a proof from the root
+        let targets = sparse.generate_account_proof_targets(&[key_unrevealed]);
+        assert_eq!(targets.len(), 1);
+        // min_len will be 0 since we haven't revealed any nodes on the path to key_unrevealed
+    }
 }