feat(engine): integrate hot account tracking into payload processor

Wire hot account tracking into the engine's payload processor for smart sparse trie preservation across block validations. Changes: - Add SharedHotAccounts (Arc<Mutex>) for cross-thread access - Record touched accounts in state_hook during block execution - Store TieredHotAccounts in PreservedSparseTrie - Call end_block() to rotate bloom filters before storing trie - Use prune_preserving() for hot account-aware pruning - Add HotAccountMetrics for observability The state hook records every account access, and prune_preserving uses this data to keep frequently-accessed accounts revealed in the trie, reducing proof fetching overhead on subsequent blocks. Amp-Thread-ID: https://ampcode.com/threads/T-019c11eb-6e7f-77ad-a330-c2fef46835fc Co-authored-by: Amp <amp@ampcode.com>
feat(trie): add hot account tracking and smart trie preservation
2026-04-30 03:01:58 -04:00 · 2026-01-31 03:53:55 +01:00 · 2026-01-31 03:53:48 +01:00
12 changed files with 1293 additions and 46 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11255,6 +11255,7 @@ dependencies = [
 "reth-trie",
 "reth-trie-common",
 "reth-trie-db",
+ "rustc-hash",
 "tracing",
 ]

--- a/crates/engine/tree/src/tree/payload_processor/mod.rs
+++ b/crates/engine/tree/src/tree/payload_processor/mod.rs
@@ -39,7 +39,9 @@ use reth_trie_parallel::{
    proof_task::{ProofTaskCtx, ProofWorkerHandle},
    root::ParallelStateRootError,
 };
-use reth_trie_sparse::{ClearedSparseStateTrie, RevealableSparseTrie, SparseStateTrie};
+use reth_trie_sparse::{
+    hot_accounts::TieredHotAccounts, ClearedSparseStateTrie, RevealableSparseTrie, SparseStateTrie,
+};
 use reth_trie_sparse_parallel::{ParallelSparseTrie, ParallelismThresholds};
 use std::{
    collections::BTreeMap,
@@ -61,7 +63,7 @@ pub mod prewarm;
 pub mod receipt_root_task;
 pub mod sparse_trie;

-use preserved_sparse_trie::{PreservedSparseTrie, SharedPreservedSparseTrie};
+use preserved_sparse_trie::{PreservedSparseTrie, SharedHotAccounts, SharedPreservedSparseTrie};

 /// Default parallelism thresholds to use with the [`ParallelSparseTrie`].
 ///
@@ -323,7 +325,7 @@ where
        let (state_root_tx, state_root_rx) = channel();

        // Spawn the sparse trie task using any stored trie and parallel trie configuration.
-        self.spawn_sparse_trie_task(
+        let hot_accounts = self.spawn_sparse_trie_task(
            sparse_trie_rx,
            proof_handle,
            state_root_tx,
@@ -337,6 +339,7 @@ where
            prewarm_handle,
            state_root: Some(state_root_rx),
            transactions: execution_rx,
+            hot_accounts: Some(hot_accounts),
            _span: span,
        }
    }
@@ -364,6 +367,7 @@ where
            prewarm_handle,
            state_root: None,
            transactions: execution_rx,
+            hot_accounts: None, // Cache-exclusive path doesn't use sparse trie
            _span: Span::current(),
        }
    }
@@ -511,6 +515,7 @@ where
    /// Spawns the [`SparseTrieTask`] for this payload processor.
    ///
    /// The trie is preserved when the new payload is a child of the previous one.
+    /// Returns the shared hot accounts tracker for use in the state hook.
    #[instrument(level = "debug", target = "engine::tree::payload_processor", skip_all)]
    fn spawn_sparse_trie_task(
        &self,
@@ -520,7 +525,7 @@ where
        from_multi_proof: CrossbeamReceiver<MultiProofMessage>,
        config: &TreeConfig,
        parent_state_root: B256,
-    ) {
+    ) -> SharedHotAccounts {
        let preserved_sparse_trie = self.sparse_state_trie.clone();
        let trie_metrics = self.trie_metrics.clone();
        let span = Span::current();
@@ -528,32 +533,34 @@ where
        let prune_depth = self.sparse_trie_prune_depth;
        let max_storage_tries = self.sparse_trie_max_storage_tries;

+        // Extract or create hot accounts BEFORE spawning the task so we can return a clone
+        let (sparse_state_trie, hot_accounts): (_, SharedHotAccounts) = preserved_sparse_trie
+            .take()
+            .map(|preserved| preserved.into_trie_for(parent_state_root))
+            .unwrap_or_else(|| {
+                debug!(
+                    target: "engine::tree::payload_processor",
+                    "Creating new sparse trie - no preserved trie available"
+                );
+                let default_trie = RevealableSparseTrie::blind_from(
+                    ParallelSparseTrie::default()
+                        .with_parallelism_thresholds(PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS),
+                );
+                let trie = SparseStateTrie::new()
+                    .with_accounts_trie(default_trie.clone())
+                    .with_default_storage_trie(default_trie)
+                    .with_updates(true);
+                let hot_accounts =
+                    Arc::new(parking_lot::Mutex::new(TieredHotAccounts::for_mainnet()));
+                (trie, hot_accounts)
+            });
+
+        // Clone for the caller before moving into spawned task
+        let hot_accounts_for_caller = hot_accounts.clone();
+
        self.executor.spawn_blocking(move || {
            let _enter = span.entered();

-            // Reuse a stored SparseStateTrie if available, applying continuation logic.
-            // If this payload's parent state root matches the preserved trie's anchor,
-            // we can reuse the pruned trie structure. Otherwise, we clear the trie but
-            // keep allocations.
-            let sparse_state_trie = preserved_sparse_trie
-                .take()
-                .map(|preserved| preserved.into_trie_for(parent_state_root))
-                .unwrap_or_else(|| {
-                    debug!(
-                        target: "engine::tree::payload_processor",
-                        "Creating new sparse trie - no preserved trie available"
-                    );
-                    let default_trie = RevealableSparseTrie::blind_from(
-                        ParallelSparseTrie::default().with_parallelism_thresholds(
-                            PARALLEL_SPARSE_TRIE_PARALLELISM_THRESHOLDS,
-                        ),
-                    );
-                    SparseStateTrie::new()
-                        .with_accounts_trie(default_trie.clone())
-                        .with_default_storage_trie(default_trie)
-                        .with_updates(true)
-                });
-
            let mut task = if disable_sparse_trie_as_cache {
                SpawnedSparseTrieTask::Cleared(SparseTrieTask::new(
                    sparse_trie_rx,
@@ -593,7 +600,7 @@ where
                    SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
                );
-                guard.store(PreservedSparseTrie::cleared(trie));
+                guard.store(PreservedSparseTrie::cleared(trie, hot_accounts));
                return;
            }

@@ -608,11 +615,12 @@ where
                    max_storage_tries,
                    SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
+                    &hot_accounts,
                );
                trie_metrics
                    .into_trie_for_reuse_duration_histogram
                    .record(start.elapsed().as_secs_f64());
-                guard.store(PreservedSparseTrie::anchored(trie, state_root));
+                guard.store(PreservedSparseTrie::anchored(trie, state_root, hot_accounts));
            } else {
                debug!(
                    target: "engine::tree::payload_processor",
@@ -622,9 +630,11 @@ where
                    SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                    SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
                );
-                guard.store(PreservedSparseTrie::cleared(trie));
+                guard.store(PreservedSparseTrie::cleared(trie, hot_accounts));
            }
        });
+
+        hot_accounts_for_caller
    }

    /// Updates the execution cache with the post-execution state from an inserted block.
@@ -692,6 +702,8 @@ pub struct PayloadHandle<Tx, Err, R> {
    transactions: mpsc::Receiver<Result<Tx, Err>>,
    /// Receiver for the state root
    state_root: Option<mpsc::Receiver<Result<StateRootComputeOutcome, ParallelStateRootError>>>,
+    /// Shared hot accounts tracker for recording touched accounts during execution.
+    hot_accounts: Option<SharedHotAccounts>,
    /// Span for tracing
    _span: Span,
 }
@@ -719,14 +731,38 @@ impl<Tx, Err, R: Send + Sync + 'static> PayloadHandle<Tx, Err, R> {
    /// Returns a state hook to be used to send state updates to this task.
    ///
    /// If a multiproof task is spawned the hook will notify it about new states.
+    /// Also records touched accounts in the hot accounts tracker for preservation.
    pub fn state_hook(&self) -> impl OnStateHook {
+        use alloy_primitives::keccak256;
+
        // convert the channel into a `StateHookSender` that emits an event on drop
        let to_multi_proof = self.to_multi_proof.clone().map(StateHookSender::new);
+        let hot_accounts = self.hot_accounts.clone();

        move |source: StateChangeSource, state: &EvmState| {
            if let Some(sender) = &to_multi_proof {
                let _ = sender.send(MultiProofMessage::StateUpdate(source.into(), state.clone()));
            }
+
+            // Record touched accounts in the hot accounts tracker
+            if let Some(ref tracker) = hot_accounts {
+                // Batch the addresses to minimize lock contention
+                let touched: Vec<_> = state
+                    .iter()
+                    .map(|(addr, account)| {
+                        let hashed = keccak256(addr);
+                        // Has code if storage is modified (implies contract) or if account has code
+                        let has_code = !account.storage.is_empty() ||
+                            account.info.code.as_ref().is_some_and(|c| !c.is_empty());
+                        (hashed, has_code)
+                    })
+                    .collect();
+
+                let mut guard = tracker.lock();
+                for (hashed_addr, has_code) in touched {
+                    guard.record(hashed_addr, has_code);
+                }
+            }
        }
    }

--- a/crates/engine/tree/src/tree/payload_processor/preserved_sparse_trie.rs
+++ b/crates/engine/tree/src/tree/payload_processor/preserved_sparse_trie.rs
@@ -2,7 +2,7 @@

 use alloy_primitives::B256;
 use parking_lot::Mutex;
-use reth_trie_sparse::SparseStateTrie;
+use reth_trie_sparse::{hot_accounts::TieredHotAccounts, SparseStateTrie};
 use reth_trie_sparse_parallel::ParallelSparseTrie;
 use std::sync::Arc;
 use tracing::debug;
@@ -10,6 +10,9 @@ use tracing::debug;
 /// Type alias for the sparse trie type used in preservation.
 pub(super) type SparseTrie = SparseStateTrie<ParallelSparseTrie, ParallelSparseTrie>;

+/// Shared handle to hot accounts tracker for recording touches across threads.
+pub(super) type SharedHotAccounts = Arc<Mutex<TieredHotAccounts>>;
+
 /// Shared handle to a preserved sparse trie that can be reused across payload validations.
 ///
 /// This is stored in [`PayloadProcessor`](super::PayloadProcessor) and cloned to pass to
@@ -57,11 +60,15 @@ pub(super) enum PreservedSparseTrie {
        /// The state root this trie represents (computed from the previous block).
        /// Used to verify continuity: new payload's `parent_state_root` must match this.
        state_root: B256,
+        /// Hot accounts tracker for trie update optimization (shared for cross-thread access).
+        hot_accounts: SharedHotAccounts,
    },
    /// Cleared trie with preserved allocations, ready for fresh use.
    Cleared {
        /// The sparse state trie with cleared data but preserved allocations.
        trie: SparseTrie,
+        /// Hot accounts tracker for trie update optimization (shared for cross-thread access).
+        hot_accounts: SharedHotAccounts,
    },
 }

@@ -70,31 +77,37 @@ impl PreservedSparseTrie {
    ///
    /// The `state_root` is the computed state root from the trie, which becomes the
    /// anchor for determining if subsequent payloads can reuse this trie.
-    pub(super) const fn anchored(trie: SparseTrie, state_root: B256) -> Self {
-        Self::Anchored { trie, state_root }
+    pub(super) const fn anchored(
+        trie: SparseTrie,
+        state_root: B256,
+        hot_accounts: SharedHotAccounts,
+    ) -> Self {
+        Self::Anchored { trie, state_root, hot_accounts }
    }

    /// Creates a cleared preserved trie (allocations preserved, data cleared).
-    pub(super) const fn cleared(trie: SparseTrie) -> Self {
-        Self::Cleared { trie }
+    pub(super) const fn cleared(trie: SparseTrie, hot_accounts: SharedHotAccounts) -> Self {
+        Self::Cleared { trie, hot_accounts }
    }

-    /// Consumes self and returns the trie for reuse.
+    /// Consumes self and returns the trie and shared hot accounts for reuse.
    ///
    /// If the preserved trie is anchored and the parent state root matches, the pruned
    /// trie structure is reused directly. Otherwise, the trie is cleared but allocations
    /// are preserved to reduce memory overhead.
-    pub(super) fn into_trie_for(self, parent_state_root: B256) -> SparseTrie {
+    pub(super) fn into_trie_for(self, parent_state_root: B256) -> (SparseTrie, SharedHotAccounts) {
        match self {
-            Self::Anchored { trie, state_root } if state_root == parent_state_root => {
+            Self::Anchored { trie, state_root, hot_accounts }
+                if state_root == parent_state_root =>
+            {
                debug!(
                    target: "engine::tree::payload_processor",
                    %state_root,
                    "Reusing anchored sparse trie for continuation payload"
                );
-                trie
+                (trie, hot_accounts)
            }
-            Self::Anchored { mut trie, state_root } => {
+            Self::Anchored { mut trie, state_root, hot_accounts } => {
                debug!(
                    target: "engine::tree::payload_processor",
                    anchor_root = %state_root,
@@ -102,15 +115,16 @@ impl PreservedSparseTrie {
                    "Clearing anchored sparse trie - parent state root mismatch"
                );
                trie.clear();
-                trie
+                hot_accounts.lock().clear_dynamic();
+                (trie, hot_accounts)
            }
-            Self::Cleared { trie } => {
+            Self::Cleared { trie, hot_accounts } => {
                debug!(
                    target: "engine::tree::payload_processor",
                    %parent_state_root,
                    "Using cleared sparse trie with preserved allocations"
                );
-                trie
+                (trie, hot_accounts)
            }
        }
    }
--- a/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
+++ b/crates/engine/tree/src/tree/payload_processor/sparse_trie.rs
@@ -24,10 +24,13 @@ use reth_trie_parallel::{
 };
 use reth_trie_sparse::{
    errors::{SparseStateTrieResult, SparseTrieErrorKind},
+    hot_accounts::SmartPruneConfig,
    provider::{TrieNodeProvider, TrieNodeProviderFactory},
    ClearedSparseStateTrie, LeafUpdate, SerialSparseTrie, SparseStateTrie, SparseTrie,
    SparseTrieExt,
 };
+
+use super::preserved_sparse_trie::SharedHotAccounts;
 use revm_primitives::{hash_map::Entry, B256Map};
 use smallvec::SmallVec;
 use std::{
@@ -70,7 +73,11 @@ where
        max_storage_tries: usize,
        max_nodes_capacity: usize,
        max_values_capacity: usize,
+        hot_accounts: &SharedHotAccounts,
    ) -> SparseStateTrie<A, S> {
+        // Rotate hot account filters at end of block
+        hot_accounts.lock().end_block();
+
        match self {
            Self::Cleared(task) => task.into_cleared_trie(max_nodes_capacity, max_values_capacity),
            Self::Cached(task) => task.into_trie_for_reuse(
@@ -78,6 +85,7 @@ where
                max_storage_tries,
                max_nodes_capacity,
                max_values_capacity,
+                hot_accounts,
            ),
        }
    }
@@ -273,15 +281,20 @@ where

    /// Prunes and shrinks the trie for reuse in the next payload built on top of this one.
    ///
-    /// Should be called after the state root result has been sent.
+    /// Uses hot account-aware pruning to preserve frequently-accessed accounts and their
+    /// storage tries. Should be called after the state root result has been sent.
    pub(super) fn into_trie_for_reuse(
        mut self,
        prune_depth: usize,
        max_storage_tries: usize,
        max_nodes_capacity: usize,
        max_values_capacity: usize,
+        hot_accounts: &SharedHotAccounts,
    ) -> SparseStateTrie<A, S> {
-        self.trie.prune(prune_depth, max_storage_tries);
+        let guard = hot_accounts.lock();
+        let config = SmartPruneConfig::new(prune_depth, max_storage_tries, &guard);
+        self.trie.prune_preserving(&config);
+        drop(guard);
        self.trie.shrink_to(max_nodes_capacity, max_values_capacity);
        self.trie
    }
--- a/crates/trie/sparse-parallel/src/trie.rs
+++ b/crates/trie/sparse-parallel/src/trie.rs
@@ -1212,6 +1212,152 @@ impl SparseTrieExt for ParallelSparseTrie {
        nodes_converted
    }

+    #[cfg(feature = "std")]
+    fn prune_preserving(&mut self, config: &reth_trie_sparse::hot_accounts::SmartPruneConfig<'_>) {
+        // Decay heat for subtries not modified this cycle
+        self.subtrie_heat.decay_and_reset();
+
+        let max_depth = config.max_depth;
+
+        // DFS traversal to find nodes at max_depth that can be pruned.
+        // Hot accounts are preserved at full depth by skipping pruning for their paths.
+        let mut effective_pruned_roots = Vec::<(Nibbles, B256)>::new();
+        let mut stack: SmallVec<[(Nibbles, usize); 32]> = SmallVec::new();
+        stack.push((Nibbles::default(), 0));
+
+        while let Some((path, depth)) = stack.pop() {
+            // Skip traversal into hot lower subtries beyond max_depth.
+            if depth > max_depth &&
+                let SparseSubtrieType::Lower(idx) = SparseSubtrieType::from_path(&path) &&
+                self.subtrie_heat.is_hot(idx)
+            {
+                continue;
+            }
+
+            // Get children to visit from current node (immutable access)
+            let children: SmallVec<[Nibbles; 16]> = {
+                let Some(subtrie) = self.subtrie_for_path(&path) else { continue };
+                let Some(node) = subtrie.nodes.get(&path) else { continue };
+
+                match node {
+                    SparseNode::Empty | SparseNode::Hash(_) | SparseNode::Leaf { .. } => {
+                        SmallVec::new()
+                    }
+                    SparseNode::Extension { key, .. } => {
+                        let mut child = path;
+                        child.extend(key);
+                        SmallVec::from_buf_and_len([child; 16], 1)
+                    }
+                    SparseNode::Branch { state_mask, .. } => {
+                        let mut children = SmallVec::new();
+                        let mut mask = state_mask.get();
+                        while mask != 0 {
+                            let nibble = mask.trailing_zeros() as u8;
+                            mask &= mask - 1;
+                            let mut child = path;
+                            child.push_unchecked(nibble);
+                            children.push(child);
+                        }
+                        children
+                    }
+                }
+            };
+
+            // Process children - either continue traversal or prune
+            for child in children {
+                if depth == max_depth {
+                    // Check if this child path leads to a hot account.
+                    // For account tries, we can check if any hot account's nibble path
+                    // starts with this child path.
+                    if Self::path_leads_to_hot_account(&child, config) {
+                        // Don't prune hot account paths - continue traversal
+                        stack.push((child, depth + 1));
+                        continue;
+                    }
+
+                    // Check if child has a computed hash and replace inline
+                    let hash = self
+                        .subtrie_for_path(&child)
+                        .and_then(|s| s.nodes.get(&child))
+                        .filter(|n| !n.is_hash())
+                        .and_then(|n| n.hash());
+
+                    if let Some(hash) = hash &&
+                        let Some(subtrie) = self.subtrie_for_path_mut_untracked(&child)
+                    {
+                        subtrie.nodes.insert(child, SparseNode::Hash(hash));
+                        effective_pruned_roots.push((child, hash));
+                    }
+                } else {
+                    stack.push((child, depth + 1));
+                }
+            }
+        }
+
+        if effective_pruned_roots.is_empty() {
+            return;
+        }
+
+        // Sort roots by subtrie type (upper first), then by path for efficient partitioning.
+        effective_pruned_roots.sort_unstable_by(|(path_a, _), (path_b, _)| {
+            let subtrie_type_a = SparseSubtrieType::from_path(path_a);
+            let subtrie_type_b = SparseSubtrieType::from_path(path_b);
+            subtrie_type_a.cmp(&subtrie_type_b).then(path_a.cmp(path_b))
+        });
+
+        // Split off upper subtrie roots (they come first due to sorting)
+        let num_upper_roots = effective_pruned_roots
+            .iter()
+            .position(|(p, _)| !SparseSubtrieType::path_len_is_upper(p.len()))
+            .unwrap_or(effective_pruned_roots.len());
+
+        let roots_upper = &effective_pruned_roots[..num_upper_roots];
+        let roots_lower = &effective_pruned_roots[num_upper_roots..];
+
+        // Upper prune roots that are prefixes of lower subtrie root paths cause the entire
+        // subtrie to be cleared (preserving allocations for reuse).
+        if !roots_upper.is_empty() {
+            for subtrie in &mut self.lower_subtries {
+                let should_clear = subtrie.as_revealed_ref().is_some_and(|s| {
+                    let search_idx = roots_upper.partition_point(|(root, _)| root <= &s.path);
+                    search_idx > 0 && s.path.starts_with(&roots_upper[search_idx - 1].0)
+                });
+                if should_clear {
+                    subtrie.clear();
+                }
+            }
+        }
+
+        // Upper subtrie: prune nodes and values
+        self.upper_subtrie.nodes.retain(|p, _| !is_strict_descendant_in(roots_upper, p));
+        self.upper_subtrie.inner.values.retain(|p, _| {
+            !starts_with_pruned_in(roots_upper, p) && !starts_with_pruned_in(roots_lower, p)
+        });
+
+        // Process lower subtries using chunk_by to group roots by subtrie
+        for roots_group in roots_lower.chunk_by(|(path_a, _), (path_b, _)| {
+            SparseSubtrieType::from_path(path_a) == SparseSubtrieType::from_path(path_b)
+        }) {
+            let subtrie_idx = path_subtrie_index_unchecked(&roots_group[0].0);
+
+            let Some(subtrie) = self.lower_subtries[subtrie_idx].as_revealed_mut() else {
+                continue;
+            };
+
+            subtrie.nodes.retain(|p, _| !is_strict_descendant_in(roots_group, p));
+            subtrie.inner.values.retain(|p, _| !starts_with_pruned_in(roots_group, p));
+        }
+
+        // Branch node masks pruning
+        self.branch_node_masks.retain(|p, _| {
+            if SparseSubtrieType::path_len_is_upper(p.len()) {
+                !starts_with_pruned_in(roots_upper, p)
+            } else {
+                !starts_with_pruned_in(roots_lower, p) && !starts_with_pruned_in(roots_upper, p)
+            }
+        });
+    }
+
    fn update_leaves(
        &mut self,
        updates: &mut alloy_primitives::map::B256Map<reth_trie_sparse::LeafUpdate>,
@@ -1333,6 +1479,49 @@ impl ParallelSparseTrie {
        B256::from(bytes)
    }

+    /// Checks if a trie path leads to a hot account that should be preserved.
+    ///
+    /// For paths shorter than 64 nibbles (partial account paths), we check if any
+    /// hot account's hashed address starts with this path prefix. This is done
+    /// efficiently by checking if the path prefix matches any Tier A/B accounts
+    /// or if the dynamic filters indicate a hot account in this subtrie.
+    #[cfg(feature = "std")]
+    fn path_leads_to_hot_account(
+        path: &Nibbles,
+        config: &reth_trie_sparse::hot_accounts::SmartPruneConfig<'_>,
+    ) -> bool {
+        // For complete account paths (64 nibbles), check directly
+        if path.len() >= 64 {
+            let hashed = B256::from_slice(&path.pack()[..32]);
+            return config.hot_accounts.should_preserve(&hashed);
+        }
+
+        // For partial paths, check if any known hot account has this prefix
+        let hot_accounts = config.hot_accounts;
+
+        // Check Tier A: system contracts and major contracts
+        for addr in hot_accounts.tier_a_iter() {
+            let account_path = Nibbles::unpack(*addr);
+            if account_path.starts_with(path) {
+                return true;
+            }
+        }
+
+        // Check Tier B: known builders and fee recipients
+        for addr in hot_accounts.tier_b_iter() {
+            let account_path = Nibbles::unpack(*addr);
+            if account_path.starts_with(path) {
+                return true;
+            }
+        }
+
+        // For dynamic (Tier C) accounts, we can't efficiently check partial paths
+        // since bloom filters don't support prefix queries. Conservative approach:
+        // don't preserve based on dynamic accounts for partial paths.
+        // This means only statically-known hot accounts get full preservation.
+        false
+    }
+
    /// Rolls back a partial update by removing the value, removing any inserted nodes,
    /// removing any inserted branch masks, and restoring any modified original node.
    /// This ensures `update_leaf` is atomic - either it succeeds completely or leaves the trie
--- a/crates/trie/sparse/Cargo.toml
+++ b/crates/trie/sparse/Cargo.toml
@@ -26,6 +26,7 @@ alloy-rlp.workspace = true
 # misc
 auto_impl.workspace = true
 rayon = { workspace = true, optional = true }
+rustc-hash = { workspace = true, optional = true }

 # metrics
 reth-metrics = { workspace = true, optional = true }
@@ -55,6 +56,7 @@ rand_08.workspace = true
 default = ["std", "metrics"]
 std = [
    "dep:rayon",
+    "dep:rustc-hash",
    "alloy-primitives/std",
    "alloy-rlp/std",
    "alloy-trie/std",
--- a/crates/trie/sparse/src/hot_accounts.rs
+++ b/crates/trie/sparse/src/hot_accounts.rs
@@ -0,0 +1,809 @@
+//! Hot account tracking for smart trie preservation.
+//!
+//! This module provides data structures for tracking frequently-accessed ("hot") accounts
+//! across blocks. Hot accounts are preserved in the sparse trie between blocks to avoid
+//! redundant proof fetching.
+//!
+//! # Architecture
+//!
+//! The tracking uses a tiered approach:
+//! - **Tier A (Always)**: System contracts and major defi contracts (static, known)
+//! - **Tier B (Likely)**: Builder addresses, fee recipients (semi-static)
+//! - **Tier C (Dynamic)**: Accounts detected as hot via rotating bloom filters
+//!
+//! # Example
+//!
+//! ```ignore
+//! let mut tracker = TieredHotAccounts::for_mainnet();
+//!
+//! // During block execution, record touched accounts
+//! tracker.record(hashed_address, /* has_code */ true);
+//!
+//! // At end of block, rotate filters
+//! tracker.end_block();
+//!
+//! // Check if account should be preserved
+//! if tracker.should_preserve(&hashed_address) {
+//!     // Keep in trie
+//! }
+//! ```
+
+use alloy_primitives::{map::HashSet, B256};
+use core::hash::{Hash, Hasher};
+use rustc_hash::FxHasher;
+use std::collections::VecDeque;
+
+/// Number of hash functions to use in bloom filter.
+/// Using 3 hash functions is optimal for ~1% false positive rate.
+const BLOOM_HASH_COUNT: usize = 3;
+
+/// Default bloom filter size in bits (8KB = 65536 bits).
+/// At 1% FPR with 3 hash functions, this supports ~6000 elements.
+const DEFAULT_BLOOM_BITS: usize = 65536;
+
+/// Default number of blocks to track in rotating filter.
+const DEFAULT_HISTORY_BLOCKS: usize = 32;
+
+/// Default threshold: account is "hot" if seen in this many recent blocks.
+const DEFAULT_HOT_THRESHOLD: usize = 8;
+
+/// Maximum number of recent fee recipients to track (prevents unbounded growth).
+const MAX_RECENT_FEE_RECIPIENTS: usize = 128;
+
+/// A simple bloom filter backed by a bit vector.
+///
+/// Uses multiple hash functions derived from the input to set/check bits.
+/// False positives are possible, false negatives are not.
+#[derive(Debug, Clone)]
+pub struct BloomFilter {
+    /// Bit storage as u64 words.
+    bits: Vec<u64>,
+    /// Number of bits (may not be exactly `bits.len() * 64`).
+    num_bits: usize,
+    /// Number of hash functions.
+    hash_count: usize,
+}
+
+impl Default for BloomFilter {
+    fn default() -> Self {
+        Self::new(DEFAULT_BLOOM_BITS)
+    }
+}
+
+impl BloomFilter {
+    /// Creates a new bloom filter with the specified number of bits.
+    pub fn new(num_bits: usize) -> Self {
+        let num_words = num_bits.div_ceil(64);
+        Self { bits: vec![0u64; num_words], num_bits, hash_count: BLOOM_HASH_COUNT }
+    }
+
+    /// Creates a bloom filter sized for an expected number of elements at ~1% FPR.
+    pub fn with_capacity(expected_elements: usize) -> Self {
+        // Optimal bits = -n * ln(p) / (ln(2)^2) where p = 0.01
+        // Simplified: bits ≈ n * 10 for 1% FPR
+        let num_bits = (expected_elements * 10).max(1024).next_power_of_two();
+        Self::new(num_bits)
+    }
+
+    /// Inserts an element into the bloom filter.
+    pub fn insert(&mut self, value: &B256) {
+        for i in 0..self.hash_count {
+            let idx = self.hash_index(value, i);
+            let word_idx = idx / 64;
+            let bit_idx = idx % 64;
+            self.bits[word_idx] |= 1u64 << bit_idx;
+        }
+    }
+
+    /// Checks if an element is possibly in the filter.
+    ///
+    /// Returns `true` if the element might be present (could be false positive).
+    /// Returns `false` if the element is definitely not present.
+    pub fn contains(&self, value: &B256) -> bool {
+        (0..self.hash_count).all(|i| {
+            let idx = self.hash_index(value, i);
+            let word_idx = idx / 64;
+            let bit_idx = idx % 64;
+            (self.bits[word_idx] & (1u64 << bit_idx)) != 0
+        })
+    }
+
+    /// Clears all bits in the filter.
+    pub fn clear(&mut self) {
+        self.bits.fill(0);
+    }
+
+    /// Returns the number of bits set in the filter.
+    pub fn count_ones(&self) -> usize {
+        self.bits.iter().map(|w| w.count_ones() as usize).sum()
+    }
+
+    /// Returns the total number of bits in the filter.
+    pub const fn len(&self) -> usize {
+        self.num_bits
+    }
+
+    /// Returns true if no bits are set.
+    pub fn is_empty(&self) -> bool {
+        self.bits.iter().all(|&w| w == 0)
+    }
+
+    /// Computes the bit index for a value using hash function `i`.
+    #[inline]
+    fn hash_index(&self, value: &B256, i: usize) -> usize {
+        // Use different portions of the B256 as hash seeds
+        // B256 is 32 bytes, we can extract multiple u64s
+        let bytes = value.as_slice();
+
+        // Combine bytes with index to create different hash functions
+        let mut hasher = FxHasher::default();
+        bytes.hash(&mut hasher);
+        (i as u64).hash(&mut hasher);
+        let hash = hasher.finish();
+
+        (hash as usize) % self.num_bits
+    }
+}
+
+/// A rotating bloom filter that tracks elements across multiple time periods (blocks).
+///
+/// Maintains a ring buffer of bloom filters, one per block. Elements are considered
+/// "hot" if they appear in at least `hot_threshold` of the recent blocks.
+#[derive(Debug)]
+pub struct RotatingBloomFilter {
+    /// Current block's filter (being filled).
+    current: BloomFilter,
+    /// Historical filters (most recent first).
+    history: VecDeque<BloomFilter>,
+    /// Maximum number of historical blocks to track.
+    history_size: usize,
+    /// Minimum appearances in recent blocks to be considered hot.
+    hot_threshold: usize,
+    /// Expected elements per block (for sizing new filters).
+    expected_elements: usize,
+}
+
+impl Default for RotatingBloomFilter {
+    fn default() -> Self {
+        Self::new(DEFAULT_HISTORY_BLOCKS, DEFAULT_HOT_THRESHOLD, 1000)
+    }
+}
+
+impl RotatingBloomFilter {
+    /// Creates a new rotating bloom filter.
+    ///
+    /// # Arguments
+    /// * `history_size` - Number of historical blocks to track
+    /// * `hot_threshold` - Minimum appearances to be considered hot
+    /// * `expected_elements` - Expected elements per block (for filter sizing)
+    pub fn new(history_size: usize, hot_threshold: usize, expected_elements: usize) -> Self {
+        Self {
+            current: BloomFilter::with_capacity(expected_elements),
+            history: VecDeque::with_capacity(history_size),
+            history_size,
+            hot_threshold,
+            expected_elements,
+        }
+    }
+
+    /// Records an element as seen in the current block.
+    pub fn insert(&mut self, value: B256) {
+        self.current.insert(&value);
+    }
+
+    /// Rotates the filter at the end of a block.
+    ///
+    /// The current filter becomes history, and a new empty filter is created.
+    pub fn end_block(&mut self) {
+        // Determine size for next filter based on current usage
+        let next_expected = self.current.count_ones().max(self.expected_elements);
+
+        // Move current to history
+        let prev = std::mem::replace(&mut self.current, BloomFilter::with_capacity(next_expected));
+        self.history.push_front(prev);
+
+        // Trim old history
+        while self.history.len() > self.history_size {
+            self.history.pop_back();
+        }
+    }
+
+    /// Checks if an element is "hot" (seen frequently in recent blocks).
+    pub fn is_hot(&self, value: &B256) -> bool {
+        let count = self.history.iter().filter(|f| f.contains(value)).count();
+        count >= self.hot_threshold
+    }
+
+    /// Checks if an element was seen in the current block.
+    pub fn seen_this_block(&self, value: &B256) -> bool {
+        self.current.contains(value)
+    }
+
+    /// Returns the number of historical blocks being tracked.
+    pub fn history_len(&self) -> usize {
+        self.history.len()
+    }
+
+    /// Handles a reorg by invalidating recent history.
+    ///
+    /// Removes the most recent `depth` blocks from history.
+    pub fn handle_reorg(&mut self, depth: usize) {
+        for _ in 0..depth.min(self.history.len()) {
+            self.history.pop_front();
+        }
+        self.current.clear();
+    }
+
+    /// Clears all state.
+    pub fn clear(&mut self) {
+        self.current.clear();
+        self.history.clear();
+    }
+
+    /// Returns approximate memory usage in bytes.
+    pub fn memory_usage(&self) -> usize {
+        let current_bytes = self.current.len() / 8;
+        let history_bytes: usize = self.history.iter().map(|f| f.len() / 8).sum();
+        current_bytes + history_bytes
+    }
+}
+
+/// Hotness level for an account.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub enum Hotness {
+    /// Cold: not tracked as hot.
+    Cold = 0,
+    /// Dynamic: detected as hot via bloom filter.
+    Dynamic = 1,
+    /// Likely: known builder/fee recipient.
+    Likely = 2,
+    /// Always: system contract or major defi.
+    Always = 3,
+}
+
+/// A storage slot key combining account and slot hashes.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct StorageSlotKey {
+    /// Hashed account address.
+    pub account: B256,
+    /// Hashed storage slot.
+    pub slot: B256,
+}
+
+impl StorageSlotKey {
+    /// Creates a new storage slot key.
+    pub const fn new(account: B256, slot: B256) -> Self {
+        Self { account, slot }
+    }
+
+    /// Combines account and slot into a single B256 for bloom filter insertion.
+    fn combined(&self) -> B256 {
+        let mut combined = [0u8; 32];
+        for i in 0..16 {
+            combined[i] = self.account.0[i] ^ self.slot.0[i];
+            combined[i + 16] = self.account.0[i + 16] ^ self.slot.0[i + 16];
+        }
+        B256::from(combined)
+    }
+}
+
+/// Tracks hot storage slots for smart preservation.
+///
+/// Maintains:
+/// - Predictable slots for system contracts (computed per block)
+/// - Dynamic tracking of frequently-accessed slots via bloom filter
+#[derive(Debug)]
+pub struct HotStorageSlots {
+    /// Bloom filter for dynamically-detected hot slots.
+    dynamic_slots: RotatingBloomFilter,
+}
+
+impl Default for HotStorageSlots {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl HotStorageSlots {
+    /// Creates a new hot storage slots tracker.
+    pub fn new() -> Self {
+        Self { dynamic_slots: RotatingBloomFilter::new(32, 4, 1000) }
+    }
+
+    /// Records a storage slot access.
+    pub fn record(&mut self, account: B256, slot: B256) {
+        let key = StorageSlotKey::new(account, slot);
+        self.dynamic_slots.insert(key.combined());
+    }
+
+    /// Checks if a storage slot is hot.
+    pub fn is_hot(&self, account: &B256, slot: &B256) -> bool {
+        let key = StorageSlotKey::new(*account, *slot);
+        self.dynamic_slots.is_hot(&key.combined())
+    }
+
+    /// Rotates filters at end of block.
+    pub fn end_block(&mut self) {
+        self.dynamic_slots.end_block();
+    }
+
+    /// Handles a reorg by invalidating recent history.
+    pub fn handle_reorg(&mut self, depth: usize) {
+        self.dynamic_slots.handle_reorg(depth);
+    }
+
+    /// Clears all state.
+    pub fn clear(&mut self) {
+        self.dynamic_slots.clear();
+    }
+
+    /// Returns predictable storage slots for system contracts at a given block.
+    ///
+    /// These slots are deterministically computed based on block number/timestamp.
+    pub fn predictable_system_slots(block_number: u64, timestamp: u64) -> Vec<StorageSlotKey> {
+        use alloy_primitives::keccak256;
+
+        // EIP-4788 Beacon Roots contract
+        let beacon_roots_addr =
+            keccak256(alloy_primitives::address!("000F3df6D732807Ef1319fB7B8bB8522d0Beac02"));
+        // Ring buffer slots: timestamp % 8191 (timestamp slot) and timestamp % 8191 + 8191 (root
+        // slot)
+        let beacon_ts_slot = B256::from(alloy_primitives::U256::from(timestamp % 8191));
+        let beacon_root_slot = B256::from(alloy_primitives::U256::from(timestamp % 8191 + 8191));
+
+        // EIP-2935 Block Hash History contract
+        let history_addr =
+            keccak256(alloy_primitives::address!("0000F90827F1C53a10cb7A02335B175320002935"));
+        // Ring buffer slot: (block_number - 1) % 8191
+        let history_slot =
+            B256::from(alloy_primitives::U256::from((block_number.saturating_sub(1)) % 8191));
+
+        vec![
+            StorageSlotKey::new(beacon_roots_addr, beacon_ts_slot),
+            StorageSlotKey::new(beacon_roots_addr, beacon_root_slot),
+            StorageSlotKey::new(history_addr, history_slot),
+        ]
+    }
+}
+
+impl Hotness {
+    /// Returns true if this account should be preserved.
+    pub const fn should_preserve(&self) -> bool {
+        !matches!(self, Self::Cold)
+    }
+
+    /// Returns true if storage trie should be preserved.
+    pub const fn should_preserve_storage(&self) -> bool {
+        matches!(self, Self::Always | Self::Likely)
+    }
+}
+
+/// Tiered hot account tracking.
+///
+/// Tracks hot accounts using a combination of:
+/// - Static known-hot addresses (system contracts, major defi)
+/// - Semi-static addresses (builders, fee recipients)
+/// - Dynamic detection via rotating bloom filters
+#[derive(Debug)]
+pub struct TieredHotAccounts {
+    // === Tier A: Always hot (known contracts) ===
+    /// System contract addresses (hashed).
+    system_contracts: HashSet<B256>,
+    /// Major defi contract addresses (hashed).
+    major_contracts: HashSet<B256>,
+
+    // === Tier B: Likely hot (semi-static) ===
+    /// Known builder/searcher addresses (hashed).
+    known_builders: HashSet<B256>,
+    /// Recent fee recipients as a ring buffer (most recent first).
+    /// Capped at `MAX_RECENT_FEE_RECIPIENTS` to prevent unbounded growth.
+    recent_fee_recipients: VecDeque<B256>,
+    /// Set for O(1) membership checks of recent fee recipients.
+    recent_fee_recipients_set: HashSet<B256>,
+
+    // === Tier C: Dynamic hot (runtime detected) ===
+    /// Rotating bloom filter for EOAs.
+    eoa_filter: RotatingBloomFilter,
+    /// Rotating bloom filter for contracts (separate because they have storage).
+    contract_filter: RotatingBloomFilter,
+    /// Bloom filter to track which addresses are contracts.
+    is_contract: BloomFilter,
+
+    // === Configuration ===
+    /// Configuration for the hot account tracker.
+    #[allow(dead_code)]
+    config: HotAccountConfig,
+}
+
+/// Configuration for smart pruning with hot account preservation.
+#[derive(Debug)]
+pub struct SmartPruneConfig<'a> {
+    /// Maximum depth to keep in account trie (for non-hot accounts).
+    pub max_depth: usize,
+    /// Maximum number of storage tries to keep.
+    pub max_storage_tries: usize,
+    /// Hot account tracker.
+    pub hot_accounts: &'a TieredHotAccounts,
+}
+
+impl<'a> SmartPruneConfig<'a> {
+    /// Creates a new smart prune configuration.
+    pub const fn new(
+        max_depth: usize,
+        max_storage_tries: usize,
+        hot_accounts: &'a TieredHotAccounts,
+    ) -> Self {
+        Self { max_depth, max_storage_tries, hot_accounts }
+    }
+}
+
+/// Configuration for hot account tracking.
+#[derive(Debug, Clone)]
+pub struct HotAccountConfig {
+    /// Number of historical blocks to track.
+    pub history_blocks: usize,
+    /// Threshold for EOAs to be considered hot.
+    pub eoa_hot_threshold: usize,
+    /// Threshold for contracts to be considered hot (lower, as contracts are more valuable).
+    pub contract_hot_threshold: usize,
+    /// Expected accounts per block.
+    pub expected_accounts_per_block: usize,
+    /// Maximum memory usage for filters (bytes).
+    pub max_memory_bytes: usize,
+}
+
+impl Default for HotAccountConfig {
+    fn default() -> Self {
+        Self {
+            history_blocks: 32,
+            eoa_hot_threshold: 8,
+            contract_hot_threshold: 4,
+            expected_accounts_per_block: 500,
+            max_memory_bytes: 256 * 1024, // 256KB
+        }
+    }
+}
+
+impl Default for TieredHotAccounts {
+    fn default() -> Self {
+        Self::new(HotAccountConfig::default())
+    }
+}
+
+impl TieredHotAccounts {
+    /// Creates a new hot account tracker with the given configuration.
+    pub fn new(config: HotAccountConfig) -> Self {
+        Self {
+            system_contracts: HashSet::default(),
+            major_contracts: HashSet::default(),
+            known_builders: HashSet::default(),
+            recent_fee_recipients: VecDeque::with_capacity(MAX_RECENT_FEE_RECIPIENTS),
+            recent_fee_recipients_set: HashSet::default(),
+            eoa_filter: RotatingBloomFilter::new(
+                config.history_blocks,
+                config.eoa_hot_threshold,
+                config.expected_accounts_per_block,
+            ),
+            contract_filter: RotatingBloomFilter::new(
+                config.history_blocks,
+                config.contract_hot_threshold,
+                config.expected_accounts_per_block / 4, // Fewer contracts than EOAs
+            ),
+            is_contract: BloomFilter::with_capacity(10_000),
+            config,
+        }
+    }
+
+    /// Creates a tracker configured for Ethereum mainnet.
+    ///
+    /// Includes known system contracts and major defi protocols.
+    pub fn for_mainnet() -> Self {
+        use alloy_primitives::keccak256;
+
+        let mut tracker = Self::new(HotAccountConfig::default());
+
+        // Tier A: System contracts (addresses from EIPs)
+        // EIP-4788: Beacon roots
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "000F3df6D732807Ef1319fB7B8bB8522d0Beac02"
+        )));
+        // EIP-2935: Block hash history
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "0000F90827F1C53a10cb7A02335B175320002935"
+        )));
+        // EIP-7002: Withdrawal requests
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "00000961Ef480Eb55e80D19ad83579A64c007002"
+        )));
+        // EIP-7251: Consolidation requests
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "0000BBdDc7CE488642fb579F8B00f3a590007251"
+        )));
+
+        // Tier A: Major defi contracts
+        // WETH
+        tracker.add_major_contract(keccak256(alloy_primitives::address!(
+            "C02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
+        )));
+        // USDC
+        tracker.add_major_contract(keccak256(alloy_primitives::address!(
+            "A0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
+        )));
+        // USDT
+        tracker.add_major_contract(keccak256(alloy_primitives::address!(
+            "dAC17F958D2ee523a2206206994597C13D831ec7"
+        )));
+        // Uniswap V3 Router
+        tracker.add_major_contract(keccak256(alloy_primitives::address!(
+            "E592427A0AEce92De3Edee1F18E0157C05861564"
+        )));
+        // Uniswap Universal Router
+        tracker.add_major_contract(keccak256(alloy_primitives::address!(
+            "3fC91A3afd70395Cd496C647d5a6CC9D4B2b7FAD"
+        )));
+
+        tracker
+    }
+
+    /// Creates a tracker configured for Optimism.
+    pub fn for_optimism() -> Self {
+        use alloy_primitives::keccak256;
+
+        let mut tracker = Self::new(HotAccountConfig::default());
+
+        // L1 Block contract
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "4200000000000000000000000000000000000015"
+        )));
+        // L2 Cross Domain Messenger
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "4200000000000000000000000000000000000007"
+        )));
+        // Gas Price Oracle
+        tracker.add_system_contract(keccak256(alloy_primitives::address!(
+            "420000000000000000000000000000000000000F"
+        )));
+
+        tracker
+    }
+
+    /// Creates a tracker configured for Base.
+    pub fn for_base() -> Self {
+        // Base uses the same system contracts as Optimism
+        Self::for_optimism()
+    }
+
+    /// Adds a system contract address (Tier A - always hot).
+    pub fn add_system_contract(&mut self, hashed_address: B256) {
+        self.system_contracts.insert(hashed_address);
+        self.is_contract.insert(&hashed_address);
+    }
+
+    /// Adds a major defi contract address (Tier A - always hot).
+    pub fn add_major_contract(&mut self, hashed_address: B256) {
+        self.major_contracts.insert(hashed_address);
+        self.is_contract.insert(&hashed_address);
+    }
+
+    /// Adds a known builder address (Tier B - likely hot).
+    pub fn add_builder(&mut self, hashed_address: B256) {
+        self.known_builders.insert(hashed_address);
+    }
+
+    /// Records an account as touched in the current block.
+    ///
+    /// # Arguments
+    /// * `hashed_address` - The keccak256 hash of the account address
+    /// * `has_code` - Whether the account has contract code
+    pub fn record(&mut self, hashed_address: B256, has_code: bool) {
+        if has_code {
+            self.is_contract.insert(&hashed_address);
+            self.contract_filter.insert(hashed_address);
+        } else {
+            self.eoa_filter.insert(hashed_address);
+        }
+    }
+
+    /// Records a fee recipient for the current block.
+    ///
+    /// Fee recipients are automatically promoted to Tier B after being seen.
+    /// The ring buffer maintains a fixed-size history to prevent unbounded growth.
+    pub fn record_fee_recipient(&mut self, hashed_address: B256) {
+        // Skip if already in the set
+        if self.recent_fee_recipients_set.contains(&hashed_address) {
+            return;
+        }
+
+        // Add to ring buffer and set
+        self.recent_fee_recipients.push_front(hashed_address);
+        self.recent_fee_recipients_set.insert(hashed_address);
+
+        // Evict oldest if over capacity
+        if self.recent_fee_recipients.len() > MAX_RECENT_FEE_RECIPIENTS &&
+            let Some(evicted) = self.recent_fee_recipients.pop_back()
+        {
+            self.recent_fee_recipients_set.remove(&evicted);
+        }
+    }
+
+    /// Call at the end of each block to rotate filters.
+    pub fn end_block(&mut self) {
+        self.eoa_filter.end_block();
+        self.contract_filter.end_block();
+    }
+
+    /// Returns the hotness level of an account.
+    pub fn hotness(&self, hashed_address: &B256) -> Hotness {
+        // Tier A: Always hot
+        if self.system_contracts.contains(hashed_address) ||
+            self.major_contracts.contains(hashed_address)
+        {
+            return Hotness::Always;
+        }
+
+        // Tier B: Likely hot
+        if self.known_builders.contains(hashed_address) ||
+            self.recent_fee_recipients_set.contains(hashed_address)
+        {
+            return Hotness::Likely;
+        }
+
+        // Tier C: Dynamic detection
+        let is_contract = self.is_contract.contains(hashed_address);
+        if is_contract {
+            if self.contract_filter.is_hot(hashed_address) {
+                return Hotness::Dynamic;
+            }
+        } else if self.eoa_filter.is_hot(hashed_address) {
+            return Hotness::Dynamic;
+        }
+
+        Hotness::Cold
+    }
+
+    /// Returns true if the account should be preserved in the trie.
+    pub fn should_preserve(&self, hashed_address: &B256) -> bool {
+        self.hotness(hashed_address).should_preserve()
+    }
+
+    /// Returns true if the account's storage trie should be preserved.
+    pub fn should_preserve_storage(&self, hashed_address: &B256) -> bool {
+        self.hotness(hashed_address).should_preserve_storage()
+    }
+
+    /// Handles a chain reorg by invalidating recent history.
+    pub fn handle_reorg(&mut self, depth: usize) {
+        self.eoa_filter.handle_reorg(depth);
+        self.contract_filter.handle_reorg(depth);
+    }
+
+    /// Clears all dynamic tracking state.
+    ///
+    /// Static Tier A (system/major contracts) is preserved.
+    /// Tier B (builders) is preserved. Recent fee recipients and dynamic filters are cleared.
+    pub fn clear_dynamic(&mut self) {
+        self.eoa_filter.clear();
+        self.contract_filter.clear();
+        self.is_contract.clear();
+        self.recent_fee_recipients.clear();
+        self.recent_fee_recipients_set.clear();
+    }
+
+    /// Returns approximate memory usage in bytes.
+    pub fn memory_usage(&self) -> usize {
+        self.eoa_filter.memory_usage() +
+            self.contract_filter.memory_usage() +
+            self.is_contract.len() / 8 +
+            self.system_contracts.len() * 32 +
+            self.major_contracts.len() * 32 +
+            self.known_builders.len() * 32 +
+            self.recent_fee_recipients.len() * 32 +
+            self.recent_fee_recipients_set.len() * 32
+    }
+
+    /// Returns an iterator over Tier A addresses (system contracts and major contracts).
+    ///
+    /// These are the highest priority accounts that should always be preserved.
+    pub fn tier_a_iter(&self) -> impl Iterator<Item = &B256> {
+        self.system_contracts.iter().chain(self.major_contracts.iter())
+    }
+
+    /// Returns an iterator over Tier B addresses (known builders and recent fee recipients).
+    ///
+    /// These are semi-static accounts that are likely to be accessed.
+    pub fn tier_b_iter(&self) -> impl Iterator<Item = &B256> {
+        self.known_builders.iter().chain(self.recent_fee_recipients.iter())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_primitives::B256;
+
+    #[test]
+    fn bloom_filter_basic() {
+        let mut filter = BloomFilter::new(1024);
+
+        let value1 = B256::repeat_byte(0x01);
+        let value2 = B256::repeat_byte(0x02);
+        let value3 = B256::repeat_byte(0x03);
+
+        assert!(!filter.contains(&value1));
+        assert!(!filter.contains(&value2));
+
+        filter.insert(&value1);
+        filter.insert(&value2);
+
+        assert!(filter.contains(&value1));
+        assert!(filter.contains(&value2));
+        // value3 might have false positive, but likely not with a sparse filter
+        let _ = filter.contains(&value3); // Just exercise the code path
+
+        filter.clear();
+        assert!(!filter.contains(&value1));
+    }
+
+    #[test]
+    fn rotating_filter_hot_detection() {
+        let mut filter = RotatingBloomFilter::new(10, 3, 100);
+
+        let hot_addr = B256::repeat_byte(0xAA);
+        let cold_addr = B256::repeat_byte(0xBB);
+
+        // Insert hot_addr in multiple blocks
+        for _ in 0..5 {
+            filter.insert(hot_addr);
+            filter.end_block();
+        }
+
+        // Insert cold_addr only once
+        filter.insert(cold_addr);
+        filter.end_block();
+
+        assert!(filter.is_hot(&hot_addr));
+        assert!(!filter.is_hot(&cold_addr));
+    }
+
+    #[test]
+    fn tiered_hot_accounts() {
+        let mut tracker = TieredHotAccounts::for_mainnet();
+
+        // System contract should always be hot
+        let system_addr = alloy_primitives::keccak256(alloy_primitives::address!(
+            "000F3df6D732807Ef1319fB7B8bB8522d0Beac02"
+        ));
+        assert_eq!(tracker.hotness(&system_addr), Hotness::Always);
+
+        // Random address should be cold
+        let random = B256::repeat_byte(0xFF);
+        assert_eq!(tracker.hotness(&random), Hotness::Cold);
+
+        // Record address multiple times to make it hot
+        for _ in 0..10 {
+            tracker.record(random, false);
+            tracker.end_block();
+        }
+        assert_eq!(tracker.hotness(&random), Hotness::Dynamic);
+    }
+
+    #[test]
+    fn reorg_handling() {
+        let mut filter = RotatingBloomFilter::new(10, 3, 100);
+
+        let addr = B256::repeat_byte(0xCC);
+
+        // Build up history
+        for _ in 0..5 {
+            filter.insert(addr);
+            filter.end_block();
+        }
+
+        assert!(filter.is_hot(&addr));
+
+        // Reorg removes recent history
+        filter.handle_reorg(10);
+
+        assert!(!filter.is_hot(&addr));
+    }
+}
--- a/crates/trie/sparse/src/lib.rs
+++ b/crates/trie/sparse/src/lib.rs
@@ -16,6 +16,14 @@ pub use traits::*;

 pub mod provider;

+#[cfg(feature = "std")]
+pub mod hot_accounts;
+#[cfg(feature = "std")]
+pub use hot_accounts::{
+    BloomFilter, HotAccountConfig, HotStorageSlots, Hotness, RotatingBloomFilter, SmartPruneConfig,
+    StorageSlotKey, TieredHotAccounts,
+};
+
 #[cfg(feature = "metrics")]
 mod metrics;

--- a/crates/trie/sparse/src/metrics.rs
+++ b/crates/trie/sparse/src/metrics.rs
@@ -1,6 +1,9 @@
 //! Metrics for the sparse state trie

-use reth_metrics::{metrics::Histogram, Metrics};
+use reth_metrics::{
+    metrics::{Counter, Histogram},
+    Metrics,
+};

 /// Metrics for the sparse state trie
 #[derive(Default, Debug)]
@@ -73,3 +76,14 @@ pub(crate) struct SparseStateTrieInnerMetrics {
    /// Histogram of total storage nodes, including those that were skipped.
    pub(crate) multiproof_total_storage_nodes: Histogram,
 }
+
+/// Metrics for hot account tracking during pruning
+#[allow(dead_code)]
+#[derive(Metrics)]
+#[metrics(scope = "sparse_trie_hot_accounts")]
+pub(crate) struct HotAccountMetrics {
+    /// Number of storage tries preserved due to hot account status
+    pub(crate) hot_storage_tries_preserved: Counter,
+    /// Number of storage tries evicted (not hot)
+    pub(crate) cold_storage_tries_evicted: Counter,
+}
--- a/crates/trie/sparse/src/state.rs
+++ b/crates/trie/sparse/src/state.rs
@@ -1073,6 +1073,54 @@ where
            },
        );
    }
+
+    /// Prunes the trie while preserving hot accounts.
+    ///
+    /// This method combines depth-based pruning with hot account preservation:
+    /// - Hot accounts (Tier A/B/C) are preserved at full depth
+    /// - Cold accounts are pruned aggressively or evicted entirely
+    /// - Storage tries for hot accounts are preserved
+    ///
+    /// # Arguments
+    ///
+    /// * `config` - Configuration for smart pruning with hot account tracking
+    ///
+    /// # Preconditions
+    ///
+    /// Node hashes must be computed via `root()` before calling this method.
+    #[cfg(feature = "std")]
+    #[instrument(target = "trie::sparse", skip_all)]
+    pub fn prune_preserving(&mut self, config: &crate::hot_accounts::SmartPruneConfig<'_>) {
+        let fn_start = std::time::Instant::now();
+
+        // Prune state and storage tries in parallel
+        rayon::join(
+            || {
+                // Prune account trie with hot account awareness
+                if let Some(trie) = self.state.as_revealed_mut() {
+                    trie.prune_preserving(config);
+                }
+                // Keep revealed paths only for hot accounts
+                self.revealed_account_paths.retain(|path| {
+                    if path.len() >= 64 {
+                        let hashed = B256::from_slice(&path.pack()[..32]);
+                        config.hot_accounts.should_preserve(&hashed)
+                    } else {
+                        false
+                    }
+                });
+            },
+            || {
+                self.storage.prune_preserving(config);
+            },
+        );
+
+        debug!(
+            target: "trie::sparse",
+            elapsed = ?fn_start.elapsed(),
+            "SparseStateTrie::prune_preserving completed"
+        );
+    }
 }

 /// The fields of [`SparseStateTrie`] related to storage tries. This is kept separate from the rest
@@ -1204,6 +1252,106 @@ impl<S: SparseTrieTrait + SparseTrieExt> StorageTries<S> {
            "StorageTries::prune completed"
        );
    }
+
+    /// Prunes storage tries while preserving hot accounts.
+    ///
+    /// Hot accounts (as determined by the tracker) are preserved entirely.
+    /// Cold accounts are subject to normal pruning/eviction.
+    fn prune_preserving(&mut self, config: &crate::hot_accounts::SmartPruneConfig<'_>) {
+        let fn_start = std::time::Instant::now();
+
+        // Update heat for accessed tries
+        self.modifications.update_and_reset();
+
+        // Score tries: hot accounts get infinite score, cold accounts get size*heat
+        let trie_info: Vec<(B256, usize, bool)> = self
+            .tries
+            .iter()
+            .map(|(address, trie)| {
+                let is_hot = config.hot_accounts.should_preserve_storage(address);
+                let size = match trie {
+                    RevealableSparseTrie::Blind(_) => 0,
+                    RevealableSparseTrie::Revealed(t) => t.size_hint(),
+                };
+                (*address, size, is_hot)
+            })
+            .collect();
+
+        // Separate hot and cold tries
+        let hot_addrs: HashSet<B256> =
+            trie_info.iter().filter(|(_, _, hot)| *hot).map(|(addr, _, _)| *addr).collect();
+
+        let mut cold_tries: Vec<(B256, usize)> = trie_info
+            .iter()
+            .filter(|(_, _, hot)| !*hot)
+            .map(|(addr, size, _)| {
+                let heat = self.modifications.heat(addr);
+                let heat_multiplier = 1 + (heat.min(4) / 2) as usize;
+                (*addr, size * heat_multiplier)
+            })
+            .collect();
+
+        // Keep top N cold tries by score
+        let cold_slots = config.max_storage_tries.saturating_sub(hot_addrs.len());
+        if cold_tries.len() > cold_slots {
+            cold_tries.select_nth_unstable_by(cold_slots.saturating_sub(1), |a, b| b.1.cmp(&a.1));
+            cold_tries.truncate(cold_slots);
+        }
+        let cold_to_keep: HashSet<B256> = cold_tries.iter().map(|(addr, _)| *addr).collect();
+
+        // Evict cold tries that didn't make the cut
+        let tries_to_evict: Vec<B256> = self
+            .tries
+            .keys()
+            .filter(|addr| !hot_addrs.contains(*addr) && !cold_to_keep.contains(*addr))
+            .copied()
+            .collect();
+
+        for address in &tries_to_evict {
+            if let Some(mut trie) = self.tries.remove(address) {
+                trie.clear();
+                self.cleared_tries.push(trie);
+            }
+            if let Some(mut paths) = self.revealed_paths.remove(address) {
+                paths.clear();
+                self.cleared_revealed_paths.push(paths);
+            }
+            self.modifications.remove(address);
+        }
+
+        // Prune cold tries that were kept (hot tries are not pruned)
+        const MIN_SIZE_TO_PRUNE: usize = 1000;
+        for (address, size) in &cold_tries {
+            if *size < MIN_SIZE_TO_PRUNE {
+                continue;
+            }
+            if let Some(heat_state) = self.modifications.get_mut(address) {
+                if heat_state.prune_backlog < 2 {
+                    continue;
+                }
+                if let Some(trie) = self.tries.get_mut(address).and_then(|t| t.as_revealed_mut()) {
+                    trie.prune(config.max_depth);
+                    heat_state.prune_backlog = 0;
+                }
+            }
+        }
+
+        // Clear revealed_paths for cold tries only
+        for addr in &cold_to_keep {
+            if let Some(paths) = self.revealed_paths.get_mut(addr) {
+                paths.clear();
+            }
+        }
+
+        debug!(
+            target: "trie::sparse",
+            hot_preserved = hot_addrs.len(),
+            cold_kept = cold_to_keep.len(),
+            evicted = tries_to_evict.len(),
+            elapsed = ?fn_start.elapsed(),
+            "StorageTries::prune_preserving completed"
+        );
+    }
 }

 impl<S: SparseTrieTrait> StorageTries<S> {
--- a/crates/trie/sparse/src/traits.rs
+++ b/crates/trie/sparse/src/traits.rs
@@ -293,6 +293,18 @@ pub trait SparseTrieExt: SparseTrie {
    /// The number of nodes converted to hash stubs.
    fn prune(&mut self, max_depth: usize) -> usize;

+    /// Prunes the trie while preserving hot accounts.
+    ///
+    /// Similar to `prune`, but applies different depth limits based on account hotness:
+    /// - Hot accounts (Tier A/B/C) are preserved at full depth
+    /// - Cold accounts are pruned to `max_depth`
+    ///
+    /// # Arguments
+    ///
+    /// * `config` - Configuration containing hot account tracker and depth settings
+    #[cfg(feature = "std")]
+    fn prune_preserving(&mut self, config: &crate::hot_accounts::SmartPruneConfig<'_>);
+
    /// Applies leaf updates to the sparse trie.
    ///
    /// When a [`LeafUpdate::Changed`] is successfully applied, it is removed from the
--- a/typos.toml
+++ b/typos.toml
@@ -38,3 +38,4 @@ Iy = "Iy" # Part of base64 encoded ENR
 flate = "flate" # zlib-flate is a valid tool name
 Pn = "Pn" # Part of UPnP (Universal Plug and Play)
 BA = "BA" # Part of BAL - Block Access List (EIP-7928)
+FPR = "FPR" # False Positive Rate (bloom filter terminology)