Fix LFU tracking.

The reth_trie_common import was incorrectly gated behind #[cfg(feature = "std")],
breaking no_std/wasm/riscv builds. Restored the unconditional import as it was
on main.

Added frequency halving in the LFU cache to prevent the once-hot-always-hot
problem where stale entries accumulate unbounded frequency counts.

Also ran cargo +nightly fmt.

Amp-Thread-ID: https://ampcode.com/threads/T-019cadc3-58ed-745e-bae5-f85236d39783
Co-authored-by: Amp <amp@ampcode.com>

crates/engine/primitives/src/config.rs

@@ -26,10 +26,10 @@ pub const DEFAULT_RESERVED_CPU_CORES: usize = 1;
 /// Depth 4 means we keep roughly 16^4 = 65536 potential branch paths at most.
 pub const DEFAULT_SPARSE_TRIE_PRUNE_DEPTH: usize = 4;
 
-/// Default maximum number of storage tries to keep after pruning.
+/// Default LFU hot-slot capacity for sparse trie pruning.
 ///
-/// Storage tries beyond this limit are cleared (but allocations preserved).
-pub const DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES: usize = 100;
+/// Limits the number of `(address, slot)` pairs retained across prune cycles.
+pub const DEFAULT_SPARSE_TRIE_MAX_HOT_SLOTS: usize = 100;
 
 /// Default timeout for the state root task before spawning a sequential fallback.
 pub const DEFAULT_STATE_ROOT_TASK_TIMEOUT: Duration = Duration::from_secs(1);
@@ -131,8 +131,8 @@ pub struct TreeConfig {
     disable_cache_metrics: bool,
     /// Depth for sparse trie pruning after state root computation.
     sparse_trie_prune_depth: usize,
-    /// Maximum number of storage tries to retain after pruning.
-    sparse_trie_max_storage_tries: usize,
+    /// LFU hot-slot capacity: max `(address, slot)` pairs retained across prune cycles.
+    sparse_trie_max_hot_slots: usize,
     /// Whether to fully disable sparse trie cache pruning between blocks.
     disable_sparse_trie_cache_pruning: bool,
     /// Timeout for the state root task before spawning a sequential fallback computation.
@@ -165,7 +165,7 @@ impl Default for TreeConfig {
             allow_unwind_canonical_header: false,
             disable_cache_metrics: false,
             sparse_trie_prune_depth: DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
-            sparse_trie_max_storage_tries: DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
+            sparse_trie_max_hot_slots: DEFAULT_SPARSE_TRIE_MAX_HOT_SLOTS,
             disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout: Some(DEFAULT_STATE_ROOT_TASK_TIMEOUT),
         }
@@ -196,7 +196,7 @@ impl TreeConfig {
         allow_unwind_canonical_header: bool,
         disable_cache_metrics: bool,
         sparse_trie_prune_depth: usize,
-        sparse_trie_max_storage_tries: usize,
+        sparse_trie_max_hot_slots: usize,
         state_root_task_timeout: Option<Duration>,
     ) -> Self {
         Self {
@@ -220,7 +220,7 @@ impl TreeConfig {
             allow_unwind_canonical_header,
             disable_cache_metrics,
             sparse_trie_prune_depth,
-            sparse_trie_max_storage_tries,
+            sparse_trie_max_hot_slots,
             disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout,
         }
@@ -471,14 +471,14 @@ impl TreeConfig {
         self
     }
 
-    /// Returns the maximum number of storage tries to retain after pruning.
-    pub const fn sparse_trie_max_storage_tries(&self) -> usize {
-        self.sparse_trie_max_storage_tries
+    /// Returns the LFU hot-slot capacity for sparse trie pruning.
+    pub const fn sparse_trie_max_hot_slots(&self) -> usize {
+        self.sparse_trie_max_hot_slots
     }
 
-    /// Setter for maximum storage tries to retain.
-    pub const fn with_sparse_trie_max_storage_tries(mut self, max_tries: usize) -> Self {
-        self.sparse_trie_max_storage_tries = max_tries;
+    /// Setter for LFU hot-slot capacity.
+    pub const fn with_sparse_trie_max_hot_slots(mut self, max_hot_slots: usize) -> Self {
+        self.sparse_trie_max_hot_slots = max_hot_slots;
         self
     }
 

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

@@ -133,8 +133,8 @@ where
     sparse_state_trie: SharedPreservedSparseTrie,
     /// Sparse trie prune depth.
     sparse_trie_prune_depth: usize,
-    /// Maximum storage tries to retain after pruning.
-    sparse_trie_max_storage_tries: usize,
+    /// LFU hot-slot capacity: max storage slots retained across prune cycles.
+    sparse_trie_max_hot_slots: usize,
     /// Whether sparse trie cache pruning is fully disabled.
     disable_sparse_trie_cache_pruning: bool,
     /// Whether to disable cache metrics recording.
@@ -170,7 +170,7 @@ where
             precompile_cache_map,
             sparse_state_trie: SharedPreservedSparseTrie::default(),
             sparse_trie_prune_depth: config.sparse_trie_prune_depth(),
-            sparse_trie_max_storage_tries: config.sparse_trie_max_storage_tries(),
+            sparse_trie_max_hot_slots: config.sparse_trie_max_hot_slots(),
             disable_sparse_trie_cache_pruning: config.disable_sparse_trie_cache_pruning(),
             disable_cache_metrics: config.disable_cache_metrics(),
         }
@@ -559,7 +559,7 @@ where
         let preserved_sparse_trie = self.sparse_state_trie.clone();
         let trie_metrics = self.trie_metrics.clone();
         let prune_depth = self.sparse_trie_prune_depth;
-        let max_storage_tries = self.sparse_trie_max_storage_tries;
+        let max_hot_slots = self.sparse_trie_max_hot_slots;
         let disable_cache_pruning = self.disable_sparse_trie_cache_pruning;
         let executor = self.executor.clone();
 
@@ -644,7 +644,7 @@ where
                 let start = Instant::now();
                 let (trie, deferred) = task.into_trie_for_reuse(
                     prune_depth,
-                    max_storage_tries,
+                    max_hot_slots,
                     SPARSE_TRIE_MAX_NODES_SHRINK_CAPACITY,
                     SPARSE_TRIE_MAX_VALUES_SHRINK_CAPACITY,
                     disable_cache_pruning,

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

@@ -190,7 +190,7 @@ where
     pub(super) fn into_trie_for_reuse(
         self,
         prune_depth: usize,
-        max_storage_tries: usize,
+        max_hot_slots: usize,
         max_nodes_capacity: usize,
         max_values_capacity: usize,
         disable_pruning: bool,
@@ -199,7 +199,7 @@ where
         let Self { mut trie, .. } = self;
         trie.commit_updates(updates);
         if !disable_pruning {
-            trie.prune(prune_depth, max_storage_tries);
+            trie.prune(prune_depth, max_hot_slots);
             trie.shrink_to(max_nodes_capacity, max_values_capacity);
         }
         let deferred = trie.take_deferred_drops();
@@ -387,6 +387,8 @@ where
     fn on_hashed_state_update(&mut self, hashed_state_update: HashedPostState) {
         for (address, storage) in hashed_state_update.storages {
             for (slot, value) in storage.storage {
+                self.trie.record_hot_storage_slot(address, slot);
+
                 let encoded = if value.is_zero() {
                     Vec::new()
                 } else {

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

@@ -273,7 +273,7 @@ async fn test_sparse_trie_reuse_across_blocks() -> eyre::Result<()> {
     let tree_config = TreeConfig::default()
         .with_legacy_state_root(false)
         .with_sparse_trie_prune_depth(2)
-        .with_sparse_trie_max_storage_tries(100);
+        .with_sparse_trie_max_hot_slots(100);
 
     let (mut nodes, _wallet) = setup_engine::<EthereumNode>(
         1,

crates/node/core/src/args/engine.rs

@@ -2,7 +2,7 @@
 
 use clap::{builder::Resettable, Args};
 use reth_engine_primitives::{
-    TreeConfig, DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE, DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
+    TreeConfig, DEFAULT_MULTIPROOF_TASK_CHUNK_SIZE, DEFAULT_SPARSE_TRIE_MAX_HOT_SLOTS,
     DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
 };
 use std::{sync::OnceLock, time::Duration};
@@ -40,7 +40,7 @@ pub struct DefaultEngineValues {
     prewarming_threads: Option<usize>,
     cache_metrics_disabled: bool,
     sparse_trie_prune_depth: usize,
-    sparse_trie_max_storage_tries: usize,
+    sparse_trie_max_hot_slots: usize,
     disable_sparse_trie_cache_pruning: bool,
     state_root_task_timeout: Option<String>,
 }
@@ -179,9 +179,9 @@ impl DefaultEngineValues {
         self
     }
 
-    /// Set the maximum number of storage tries to retain after sparse trie pruning by default
-    pub const fn with_sparse_trie_max_storage_tries(mut self, v: usize) -> Self {
-        self.sparse_trie_max_storage_tries = v;
+    /// Set the LFU hot-slot capacity for sparse trie pruning by default
+    pub const fn with_sparse_trie_max_hot_slots(mut self, v: usize) -> Self {
+        self.sparse_trie_max_hot_slots = v;
         self
     }
 
@@ -221,7 +221,7 @@ impl Default for DefaultEngineValues {
             prewarming_threads: None,
             cache_metrics_disabled: false,
             sparse_trie_prune_depth: DEFAULT_SPARSE_TRIE_PRUNE_DEPTH,
-            sparse_trie_max_storage_tries: DEFAULT_SPARSE_TRIE_MAX_STORAGE_TRIES,
+            sparse_trie_max_hot_slots: DEFAULT_SPARSE_TRIE_MAX_HOT_SLOTS,
             disable_sparse_trie_cache_pruning: false,
             state_root_task_timeout: Some("1s".to_string()),
         }
@@ -353,9 +353,9 @@ pub struct EngineArgs {
     #[arg(long = "engine.sparse-trie-prune-depth", default_value_t = DefaultEngineValues::get_global().sparse_trie_prune_depth)]
     pub sparse_trie_prune_depth: usize,
 
-    /// Maximum number of storage tries to retain after sparse trie pruning.
-    #[arg(long = "engine.sparse-trie-max-storage-tries", default_value_t = DefaultEngineValues::get_global().sparse_trie_max_storage_tries)]
-    pub sparse_trie_max_storage_tries: usize,
+    /// LFU hot-slot capacity: max storage slots retained across sparse trie prune cycles.
+    #[arg(long = "engine.sparse-trie-max-hot-slots", alias = "engine.sparse-trie-max-storage-tries", default_value_t = DefaultEngineValues::get_global().sparse_trie_max_hot_slots)]
+    pub sparse_trie_max_hot_slots: usize,
 
     /// Fully disable sparse trie cache pruning. When set, the cached sparse trie is preserved
     /// without any node pruning or storage trie eviction between blocks. Useful for benchmarking
@@ -403,7 +403,7 @@ impl Default for EngineArgs {
             prewarming_threads,
             cache_metrics_disabled,
             sparse_trie_prune_depth,
-            sparse_trie_max_storage_tries,
+            sparse_trie_max_hot_slots,
             disable_sparse_trie_cache_pruning,
             state_root_task_timeout,
         } = DefaultEngineValues::get_global().clone();
@@ -432,7 +432,7 @@ impl Default for EngineArgs {
             prewarming_threads,
             cache_metrics_disabled,
             sparse_trie_prune_depth,
-            sparse_trie_max_storage_tries,
+            sparse_trie_max_hot_slots,
             disable_sparse_trie_cache_pruning,
             state_root_task_timeout: state_root_task_timeout
                 .as_deref()
@@ -463,7 +463,7 @@ impl EngineArgs {
             .with_unwind_canonical_header(self.allow_unwind_canonical_header)
             .without_cache_metrics(self.cache_metrics_disabled)
             .with_sparse_trie_prune_depth(self.sparse_trie_prune_depth)
-            .with_sparse_trie_max_storage_tries(self.sparse_trie_max_storage_tries)
+            .with_sparse_trie_max_hot_slots(self.sparse_trie_max_hot_slots)
             .with_disable_sparse_trie_cache_pruning(self.disable_sparse_trie_cache_pruning)
             .with_state_root_task_timeout(self.state_root_task_timeout.filter(|d| !d.is_zero()))
     }
@@ -516,7 +516,7 @@ mod tests {
             prewarming_threads: Some(4),
             cache_metrics_disabled: true,
             sparse_trie_prune_depth: 10,
-            sparse_trie_max_storage_tries: 100,
+            sparse_trie_max_hot_slots: 100,
             disable_sparse_trie_cache_pruning: true,
             state_root_task_timeout: Some(Duration::from_secs(2)),
         };
@@ -552,7 +552,7 @@ mod tests {
             "--engine.disable-cache-metrics",
             "--engine.sparse-trie-prune-depth",
             "10",
-            "--engine.sparse-trie-max-storage-tries",
+            "--engine.sparse-trie-max-hot-slots",
             "100",
             "--engine.disable-sparse-trie-cache-pruning",
             "--engine.state-root-task-timeout",

crates/trie/sparse/src/parallel.rs

@@ -125,9 +125,6 @@ pub struct ParallelSparseTrie {
     update_actions_buffers: Vec<Vec<SparseTrieUpdatesAction>>,
     /// Thresholds controlling when parallelism is enabled for different operations.
     parallelism_thresholds: ParallelismThresholds,
-    /// Tracks heat of lower subtries for smart pruning decisions.
-    /// Hot subtries are skipped during pruning to keep frequently-used data revealed.
-    subtrie_heat: SubtrieModifications,
     /// Metrics for the parallel sparse trie.
     #[cfg(feature = "metrics")]
     metrics: crate::metrics::ParallelSparseTrieMetrics,
@@ -151,7 +148,6 @@ impl Default for ParallelSparseTrie {
             branch_node_masks: BranchNodeMasksMap::default(),
             update_actions_buffers: Vec::default(),
             parallelism_thresholds: Default::default(),
-            subtrie_heat: SubtrieModifications::default(),
             #[cfg(feature = "metrics")]
             metrics: Default::default(),
             #[cfg(feature = "trie-debug")]
@@ -308,7 +304,6 @@ impl SparseTrie for ParallelSparseTrie {
                     continue;
                 }
                 self.lower_subtries[idx].reveal(&node.path);
-                self.subtrie_heat.mark_modified(idx);
                 self.lower_subtries[idx].as_revealed_mut().expect("just revealed").reveal_node(
                     node.path,
                     &node.node,
@@ -1004,7 +999,6 @@ impl SparseTrie for ParallelSparseTrie {
         }
         self.prefix_set = PrefixSetMut::all();
         self.updates = self.updates.is_some().then(SparseTrieUpdates::wiped);
-        self.subtrie_heat.clear();
     }
 
     fn clear(&mut self) {
@@ -1016,7 +1010,6 @@ impl SparseTrie for ParallelSparseTrie {
         self.prefix_set.clear();
         self.updates = None;
         self.branch_node_masks.clear();
-        self.subtrie_heat.clear();
         #[cfg(feature = "trie-debug")]
         self.debug_recorder.reset();
         // `update_actions_buffers` doesn't need to be cleared; we want to reuse the Vecs it has
@@ -1153,198 +1146,106 @@ impl SparseTrie for ParallelSparseTrie {
         upper_count + lower_count
     }
 
-    fn prune(&mut self, max_depth: usize) -> usize {
+    fn prune(&mut self, max_depth: usize, retained_leaves: &[Nibbles]) -> usize {
         #[cfg(feature = "trie-debug")]
         self.debug_recorder.reset();
 
-        // Decay heat for subtries not modified this cycle
-        self.subtrie_heat.decay_and_reset();
+        let mut retained_leaves = retained_leaves.to_vec();
+        retained_leaves.sort_unstable();
+        retained_leaves.dedup();
 
-        // DFS traversal to find nodes at max_depth that can be pruned.
-        // Collects "effective pruned roots" - children of nodes at max_depth with computed hashes.
-        // We replace nodes with Hash stubs inline during traversal.
         let mut effective_pruned_roots = Vec::<Nibbles>::new();
         let mut stack: SmallVec<[(Nibbles, usize); 32]> = SmallVec::new();
         stack.push((Nibbles::default(), 0));
 
-        // DFS traversal: pop path and depth, skip if subtrie or node not found.
         while let Some((path, depth)) = stack.pop() {
-            // Skip traversal into hot lower subtries beyond max_depth.
-            // At max_depth, we still need to process the node to convert children to hashes.
-            // This keeps frequently-modified subtries revealed to avoid expensive re-reveals.
-            if depth > max_depth &&
-                let SparseSubtrieType::Lower(idx) = SparseSubtrieType::from_path(&path) &&
-                self.subtrie_heat.is_hot(idx)
-            {
-                continue;
-            }
-
             let Some(subtrie) = self.subtrie_for_path_mut_untracked(&path) else { continue };
             let Some(node) = subtrie.nodes.get_mut(&path) else { continue };
 
             match node {
                 SparseNode::Empty | SparseNode::Leaf { .. } => {}
                 SparseNode::Extension { key, state, .. } => {
-                    // For extension nodes at max depth, collapse both extension and its child
-                    // branch to preserve invariant of all extension nodes children being revealed.
-                    if depth == max_depth {
-                        let Some(hash) = state.cached_hash() else { continue };
-                        subtrie.nodes.remove(&path);
+                    let mut child = path;
+                    child.extend(key);
 
-                        let parent_path = path.slice(0..path.len() - 1);
-                        let SparseNode::Branch { blinded_mask, blinded_hashes, .. } =
-                            subtrie.nodes.get_mut(&parent_path).unwrap()
-                        else {
-                            panic!("expected branch node at path {parent_path:?}");
-                        };
-
-                        let nibble = path.last().unwrap();
-                        blinded_mask.set_bit(nibble);
-                        blinded_hashes[nibble as usize] = hash;
-
-                        effective_pruned_roots.push(path);
-                    } else {
-                        let mut child = path;
-                        child.extend(key);
+                    // Within the safe zone or on a retained path — keep traversing.
+                    if depth < max_depth || has_retained_descendant(&retained_leaves, &child) {
                         stack.push((child, depth + 1));
+                        continue;
                     }
+
+                    // Root extension has no parent branch edge to blind; keep it as-is.
+                    if path.is_empty() {
+                        continue;
+                    }
+
+                    let Some(hash) = state.cached_hash() else { continue };
+                    subtrie.nodes.remove(&path);
+
+                    let parent_path = path.slice(0..path.len() - 1);
+                    let SparseNode::Branch { blinded_mask, blinded_hashes, .. } =
+                        subtrie.nodes.get_mut(&parent_path).unwrap()
+                    else {
+                        panic!("expected branch node at path {parent_path:?}");
+                    };
+
+                    let nibble = path.last().unwrap();
+                    blinded_mask.set_bit(nibble);
+                    blinded_hashes[nibble as usize] = hash;
+                    effective_pruned_roots.push(path);
                 }
                 SparseNode::Branch { state_mask, blinded_mask, blinded_hashes, .. } => {
-                    // For branch nodes at max depth, collapse all children onto them,
-                    if depth == max_depth {
-                        let mut blinded_mask = *blinded_mask;
-                        let mut blinded_hashes = blinded_hashes.clone();
-                        for nibble in state_mask.iter() {
-                            if blinded_mask.is_bit_set(nibble) {
-                                continue;
-                            }
-                            let mut child = path;
-                            child.push_unchecked(nibble);
-
-                            let Entry::Occupied(entry) = self
-                                .subtrie_for_path_mut_untracked(&child)
-                                .unwrap()
-                                .nodes
-                                .entry(child)
-                            else {
-                                panic!("expected node at path {child:?}");
-                            };
-
-                            let Some(hash) = entry.get().cached_hash() else {
-                                continue;
-                            };
-                            entry.remove();
-                            blinded_mask.set_bit(nibble);
-                            blinded_hashes[nibble as usize] = hash;
-                            effective_pruned_roots.push(child);
+                    let mut blinded_mask = *blinded_mask;
+                    let mut blinded_hashes = blinded_hashes.clone();
+                    for nibble in state_mask.iter() {
+                        if blinded_mask.is_bit_set(nibble) {
+                            continue;
                         }
 
-                        let SparseNode::Branch {
-                            blinded_mask: old_blinded_mask,
-                            blinded_hashes: old_blinded_hashes,
-                            ..
-                        } = self
-                            .subtrie_for_path_mut_untracked(&path)
-                            .unwrap()
-                            .nodes
-                            .get_mut(&path)
-                            .unwrap()
-                        else {
-                            unreachable!("expected branch node at path {path:?}");
-                        };
-                        *old_blinded_mask = blinded_mask;
-                        *old_blinded_hashes = blinded_hashes;
-                    } else {
-                        for nibble in state_mask.iter() {
-                            if blinded_mask.is_bit_set(nibble) {
-                                continue;
-                            }
-                            let mut child = path;
-                            child.push_unchecked(nibble);
+                        let mut child = path;
+                        child.push_unchecked(nibble);
+
+                        // Within the safe zone or on a retained path — keep traversing.
+                        if depth < max_depth || has_retained_descendant(&retained_leaves, &child) {
                             stack.push((child, depth + 1));
+                            continue;
                         }
+
+                        let Entry::Occupied(entry) =
+                            self.subtrie_for_path_mut_untracked(&child).unwrap().nodes.entry(child)
+                        else {
+                            panic!("expected node at path {child:?}");
+                        };
+
+                        let Some(hash) = entry.get().cached_hash() else {
+                            continue;
+                        };
+                        entry.remove();
+                        blinded_mask.set_bit(nibble);
+                        blinded_hashes[nibble as usize] = hash;
+                        effective_pruned_roots.push(child);
                     }
+
+                    let SparseNode::Branch {
+                        blinded_mask: old_blinded_mask,
+                        blinded_hashes: old_blinded_hashes,
+                        ..
+                    } = self
+                        .subtrie_for_path_mut_untracked(&path)
+                        .unwrap()
+                        .nodes
+                        .get_mut(&path)
+                        .unwrap()
+                    else {
+                        unreachable!("expected branch node at path {path:?}");
+                    };
+                    *old_blinded_mask = blinded_mask;
+                    *old_blinded_hashes = blinded_hashes;
                 }
             }
         }
 
-        if effective_pruned_roots.is_empty() {
-            return 0;
-        }
-
-        let nodes_converted = effective_pruned_roots.len();
-
-        // 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..];
-
-        debug_assert!(
-            {
-                let mut all_roots: Vec<_> = effective_pruned_roots.clone();
-                all_roots.sort_unstable();
-                all_roots.windows(2).all(|w| !w[1].starts_with(&w[0]))
-            },
-            "prune roots must be prefix-free"
-        );
-
-        // 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])
-                });
-                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]);
-
-            // Skip unrevealed/blinded subtries - nothing to prune
-            let Some(subtrie) = self.lower_subtries[subtrie_idx].as_revealed_mut() else {
-                continue;
-            };
-
-            // Retain only nodes/values not descended from any pruned root.
-            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)
-            }
-        });
-
-        nodes_converted
+        Self::finalize_pruned_roots(self, effective_pruned_roots)
     }
 
     fn update_leaves(
@@ -1555,6 +1456,86 @@ impl ParallelSparseTrie {
         Self::default().with_root(root, masks, retain_updates)
     }
 
+    fn finalize_pruned_roots(&mut self, mut effective_pruned_roots: Vec<Nibbles>) -> usize {
+        if effective_pruned_roots.is_empty() {
+            return 0;
+        }
+
+        let nodes_converted = effective_pruned_roots.len();
+
+        // 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..];
+
+        debug_assert!(
+            {
+                let mut all_roots: Vec<_> = effective_pruned_roots.clone();
+                all_roots.sort_unstable();
+                all_roots.windows(2).all(|w| !w[1].starts_with(&w[0]))
+            },
+            "prune roots must be prefix-free"
+        );
+
+        // 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])
+                });
+                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]);
+
+            // Skip unrevealed/blinded subtries - nothing to prune
+            let Some(subtrie) = self.lower_subtries[subtrie_idx].as_revealed_mut() else {
+                continue;
+            };
+
+            // Retain only nodes/values not descended from any pruned root.
+            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)
+            }
+        });
+
+        nodes_converted
+    }
+
     /// Returns a reference to the lower `SparseSubtrie` for the given path, or None if the
     /// path belongs to the upper trie, or if the lower subtrie for the path doesn't exist or is
     /// blinded.
@@ -1576,7 +1557,6 @@ impl ParallelSparseTrie {
             SparseSubtrieType::Upper => None,
             SparseSubtrieType::Lower(idx) => {
                 self.lower_subtries[idx].reveal(path);
-                self.subtrie_heat.mark_modified(idx);
                 Some(self.lower_subtries[idx].as_revealed_mut().expect("just revealed"))
             }
         }
@@ -1610,8 +1590,7 @@ impl ParallelSparseTrie {
         }
     }
 
-    /// Returns a mutable reference to a subtrie without marking it as modified.
-    /// Used for internal operations like pruning that shouldn't affect heat tracking.
+    /// Returns a mutable reference to a subtrie without revealing it if blinded.
     fn subtrie_for_path_mut_untracked(&mut self, path: &Nibbles) -> Option<&mut SparseSubtrie> {
         if SparseSubtrieType::path_len_is_upper(path.len()) {
             Some(&mut self.upper_subtrie)
@@ -2179,7 +2158,6 @@ impl ParallelSparseTrie {
             }
 
             self.lower_subtries[index] = LowerSparseSubtrie::Revealed(subtrie);
-            self.subtrie_heat.mark_modified(index);
         }
     }
 
@@ -2334,68 +2312,6 @@ impl SubtriesBitmap {
         debug_assert!(idx < NUM_LOWER_SUBTRIES);
         self.0.bit(idx)
     }
-
-    /// Clears all modification flags.
-    #[inline]
-    const fn clear(&mut self) {
-        self.0 = U256::ZERO;
-    }
-}
-
-/// Tracks heat (modification frequency) for each of the 256 lower subtries.
-///
-/// Heat is used to avoid pruning frequently-modified subtries, which would cause
-/// expensive re-reveal operations on subsequent updates.
-///
-/// - Heat is incremented by 2 when a subtrie is modified
-/// - Heat decays by 1 each prune cycle for subtries not modified that cycle
-/// - Subtries with heat > 0 are considered "hot" and skipped during pruning
-#[derive(Clone, PartialEq, Eq, Debug)]
-struct SubtrieModifications {
-    /// Heat level (0-255) for each of the 256 lower subtries.
-    heat: [u8; NUM_LOWER_SUBTRIES],
-    /// Tracks which subtries were modified in the current cycle.
-    modified: SubtriesBitmap,
-}
-
-impl Default for SubtrieModifications {
-    fn default() -> Self {
-        Self { heat: [0; NUM_LOWER_SUBTRIES], modified: SubtriesBitmap::default() }
-    }
-}
-
-impl SubtrieModifications {
-    /// Marks a subtrie as modified, incrementing its heat by 1.
-    #[inline]
-    fn mark_modified(&mut self, idx: usize) {
-        debug_assert!(idx < NUM_LOWER_SUBTRIES);
-        self.modified.set(idx);
-        self.heat[idx] = self.heat[idx].saturating_add(1);
-    }
-
-    /// Returns whether a subtrie is currently hot (heat > 0).
-    #[inline]
-    fn is_hot(&self, idx: usize) -> bool {
-        debug_assert!(idx < NUM_LOWER_SUBTRIES);
-        self.heat[idx] > 0
-    }
-
-    /// Decays heat for subtries not modified this cycle and resets modification tracking.
-    /// Called at the start of each prune cycle.
-    fn decay_and_reset(&mut self) {
-        for (idx, heat) in self.heat.iter_mut().enumerate() {
-            if !self.modified.get(idx) {
-                *heat = heat.saturating_sub(1);
-            }
-        }
-        self.modified.clear();
-    }
-
-    /// Clears all heat tracking state.
-    const fn clear(&mut self) {
-        self.heat = [0; NUM_LOWER_SUBTRIES];
-        self.modified.clear();
-    }
 }
 
 /// This is a subtrie of the [`ParallelSparseTrie`] that contains a map from path to sparse trie
@@ -3508,6 +3424,18 @@ fn is_strict_descendant_in(roots: &[Nibbles], path: &Nibbles) -> bool {
     false
 }
 
+/// Returns true if any retained leaf path has `prefix` as a prefix.
+///
+/// The `retained` slice must be sorted.
+fn has_retained_descendant(retained: &[Nibbles], prefix: &Nibbles) -> bool {
+    if retained.is_empty() {
+        return false;
+    }
+    debug_assert!(retained.windows(2).all(|w| w[0] <= w[1]), "retained must be sorted by path");
+    let idx = retained.partition_point(|path| path < prefix);
+    idx < retained.len() && retained[idx].starts_with(prefix)
+}
+
 /// Checks if `path` starts with any root in a sorted slice (inclusive).
 ///
 /// Uses binary search to find the candidate root that could be a prefix.
@@ -7726,7 +7654,7 @@ mod tests {
         let root_before = parallel.root();
 
         // Prune at depth 0: the children of root become pruned roots
-        parallel.prune(0);
+        parallel.prune(0, &[]);
 
         let root_after = parallel.root();
         assert_eq!(root_before, root_after, "root hash must be preserved");
@@ -7745,10 +7673,6 @@ mod tests {
 
     #[test]
     fn test_prune_at_various_depths() {
-        // Test depths 0 and 1, which are in the Upper subtrie (no heat tracking).
-        // Depth 2 is the boundary where Lower subtries start (UPPER_TRIE_MAX_DEPTH=2),
-        // and with `depth >= max_depth` heat check, hot Lower subtries at depth 2
-        // are protected from pruning traversal.
         for max_depth in [0, 1] {
             let provider = DefaultTrieNodeProvider;
             let mut trie = ParallelSparseTrie::default();
@@ -7771,12 +7695,7 @@ mod tests {
             let root_before = trie.root();
             let nodes_before = trie.size_hint();
 
-            // Prune multiple times to allow heat to fully decay.
-            // Heat starts at 1 and decays by 1 each cycle for unmodified subtries,
-            // so we need 2 prune cycles: 1→0, then actual prune.
-            for _ in 0..2 {
-                trie.prune(max_depth);
-            }
+            trie.prune(max_depth, &[]);
 
             let root_after = trie.root();
             assert_eq!(root_before, root_after, "root hash should be preserved after prune");
@@ -7797,7 +7716,7 @@ mod tests {
     #[test]
     fn test_prune_empty_trie() {
         let mut trie = ParallelSparseTrie::default();
-        trie.prune(2);
+        trie.prune(2, &[]);
         let root = trie.root();
         assert_eq!(root, EMPTY_ROOT_HASH, "empty trie should have empty root hash");
     }
@@ -7821,7 +7740,7 @@ mod tests {
         }
 
         let root_before = trie.root();
-        trie.prune(1);
+        trie.prune(1, &[]);
         let root_after = trie.root();
         assert_eq!(root_before, root_after, "root hash must be preserved after prune");
     }
@@ -7842,7 +7761,7 @@ mod tests {
         let root_before = trie.root();
         let nodes_before = trie.size_hint();
 
-        trie.prune(0);
+        trie.prune(0, &[]);
 
         let root_after = trie.root();
         assert_eq!(root_before, root_after, "root hash should be preserved");
@@ -7868,7 +7787,7 @@ mod tests {
         trie.root();
         let nodes_before = trie.size_hint();
 
-        trie.prune(100);
+        trie.prune(100, &[]);
 
         assert_eq!(nodes_before, trie.size_hint(), "deep prune should have no effect");
     }
@@ -7894,12 +7813,7 @@ mod tests {
         .unwrap();
 
         let root_before = trie.root();
-        // Prune multiple times to allow heat to fully decay.
-        // Heat starts at 1 and decays by 1 each cycle for unmodified subtries,
-        // so we need 2 prune cycles: 1→0, then actual prune.
-        for _ in 0..2 {
-            trie.prune(1);
-        }
+        trie.prune(1, &[]);
 
         assert_eq!(root_before, trie.root(), "root hash should be preserved");
         // Root + branch
@@ -7921,7 +7835,7 @@ mod tests {
 
         let root_before = trie.root();
 
-        trie.prune(0);
+        trie.prune(0, &[]);
 
         assert_eq!(root_before, trie.root(), "root hash must be preserved after pruning");
     }
@@ -7945,7 +7859,7 @@ mod tests {
         }
 
         let root_before = trie.root();
-        trie.prune(0);
+        trie.prune(0, &[]);
         assert_eq!(root_before, trie.root(), "root hash must be preserved");
     }
 
@@ -7972,12 +7886,7 @@ mod tests {
 
         let root_before = trie.root();
 
-        // Prune multiple times to allow heat to fully decay.
-        // Heat starts at 1 and decays by 1 each cycle for unmodified subtries.
-        let mut total_pruned = 0;
-        for _ in 0..2 {
-            total_pruned += trie.prune(1);
-        }
+        let total_pruned = trie.prune(1, &[]);
 
         assert!(total_pruned > 0, "should have pruned some nodes");
         assert_eq!(root_before, trie.root(), "root hash should be preserved");
@@ -7987,6 +7896,73 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_prune_retained_leaves_keeps_only_hot_paths() {
+        let provider = DefaultTrieNodeProvider;
+        let mut trie = ParallelSparseTrie::default();
+
+        let key_keep = pad_nibbles_right(Nibbles::from_nibbles([0x1, 0x2, 0x3, 0x4]));
+        let key_drop_1 = pad_nibbles_right(Nibbles::from_nibbles([0x5, 0x2, 0x3, 0x4]));
+        let key_drop_2 = pad_nibbles_right(Nibbles::from_nibbles([0x9, 0x2, 0x3, 0x4]));
+
+        let value = large_account_value();
+        trie.update_leaf(key_keep, value.clone(), &provider).unwrap();
+        trie.update_leaf(key_drop_1, value.clone(), &provider).unwrap();
+        trie.update_leaf(key_drop_2, value, &provider).unwrap();
+
+        let root_before = trie.root();
+
+        let pruned = trie.prune(0, &[key_keep]);
+        assert!(pruned > 0, "expected some nodes to be pruned");
+        assert_eq!(root_before, trie.root(), "root hash should be preserved after LFU prune");
+
+        assert!(trie.get_leaf_value(&key_keep).is_some(), "retained key must remain revealed");
+        assert!(trie.get_leaf_value(&key_drop_1).is_none(), "non-retained key should be pruned");
+        assert!(trie.get_leaf_value(&key_drop_2).is_none(), "non-retained key should be pruned");
+    }
+
+    #[test]
+    fn test_prune_depth_plus_retained_leaves() {
+        // Tests the combined behaviour: nodes within max_depth are always kept,
+        // nodes beyond max_depth are kept only on retained paths.
+        let provider = DefaultTrieNodeProvider;
+        let mut trie = ParallelSparseTrie::default();
+
+        let value = large_account_value();
+
+        // 4 top-level branches, each with 4 children → depth-2 trie
+        let mut all_keys = Vec::new();
+        for i in 0..4u8 {
+            for j in 0..4u8 {
+                let key = pad_nibbles_right(Nibbles::from_nibbles([i, j, 0x1, 0x2, 0x3, 0x4]));
+                trie.update_leaf(key, value.clone(), &provider).unwrap();
+                all_keys.push(key);
+            }
+        }
+
+        let root_before = trie.root();
+        let nodes_before = trie.size_hint();
+
+        // Retain one key under branch [0] — its full path should survive.
+        // max_depth=1 means depth-0 (root) is the safe zone.
+        // At depth >= 1: only the retained path is kept, everything else is blinded.
+        let retained_key = all_keys[0]; // [0, 0, ...]
+        trie.prune(1, &[retained_key]);
+
+        assert_eq!(root_before, trie.root(), "root hash must be preserved");
+        assert!(trie.size_hint() < nodes_before, "should have pruned nodes");
+
+        // The retained key's value must still be readable.
+        assert!(trie.get_leaf_value(&retained_key).is_some(), "retained leaf must survive");
+
+        // A sibling key under a different top-level branch must be blinded.
+        let dropped_key = all_keys[4]; // [1, 0, ...]
+        assert!(
+            trie.get_leaf_value(&dropped_key).is_none(),
+            "non-retained leaf beyond max_depth should be pruned"
+        );
+    }
+
     #[test]
     fn test_prune_max_depth_overflow() {
         // Verify that max_depth > 255 is not truncated (was u8, now usize)
@@ -8008,7 +7984,7 @@ mod tests {
         let nodes_before = trie.size_hint();
 
         // If depth were truncated to u8, 300 would become 44 and might prune something
-        trie.prune(300);
+        trie.prune(300, &[]);
 
         assert_eq!(
             nodes_before,
@@ -8044,7 +8020,7 @@ mod tests {
         let root_before = trie.root();
 
         // Prune at depth 0 - upper roots become prefixes of lower subtrie paths
-        trie.prune(0);
+        trie.prune(0, &[]);
 
         let root_after = trie.root();
         assert_eq!(root_before, root_after, "root hash should be preserved");

crates/trie/sparse/src/state.rs

@@ -5,23 +5,19 @@ use crate::{
     traits::SparseTrie as SparseTrieTrait,
     ParallelSparseTrie, RevealableSparseTrie,
 };
-use alloc::vec::Vec;
+use alloc::{collections::BTreeMap, vec::Vec};
 use alloy_primitives::{
-    map::{B256Map, B256Set, HashSet},
+    map::{B256Map, HashSet},
     B256,
 };
 use alloy_rlp::{Decodable, Encodable};
 use reth_execution_errors::{SparseStateTrieErrorKind, SparseStateTrieResult, SparseTrieErrorKind};
 use reth_primitives_traits::Account;
-#[cfg(feature = "std")]
-use reth_primitives_traits::FastInstant as Instant;
 use reth_trie_common::{
     updates::{StorageTrieUpdates, TrieUpdates},
     BranchNodeMasks, DecodedMultiProof, MultiProof, Nibbles, ProofTrieNodeV2, TrieAccount,
     TrieNodeV2, EMPTY_ROOT_HASH, TRIE_ACCOUNT_RLP_MAX_SIZE,
 };
-#[cfg(feature = "std")]
-use tracing::debug;
 use tracing::{instrument, trace};
 
 /// Holds data that should be dropped after any locks are released.
@@ -56,6 +52,8 @@ pub struct SparseStateTrie<
     account_rlp_buf: Vec<u8>,
     /// Holds data that should be dropped after final state root is calculated.
     deferred_drops: DeferredDrops,
+    /// Global LFU tracker for hot `(address, slot)` storage entries.
+    hot_slots_lfu: HotSlotsLfu,
     /// Metrics for the sparse state trie.
     #[cfg(feature = "metrics")]
     metrics: crate::metrics::SparseStateTrieMetrics,
@@ -75,6 +73,7 @@ where
             skip_proof_node_filtering: false,
             account_rlp_buf: Vec::with_capacity(TRIE_ACCOUNT_RLP_MAX_SIZE),
             deferred_drops: DeferredDrops::default(),
+            hot_slots_lfu: HotSlotsLfu::default(),
             #[cfg(feature = "metrics")]
             metrics: Default::default(),
         }
@@ -216,6 +215,12 @@ where
             .is_some_and(|slots| slots.contains(&Nibbles::unpack(slot)))
     }
 
+    /// Records a storage slot access/update in the global LFU tracker.
+    #[inline]
+    pub fn record_hot_storage_slot(&mut self, account: B256, slot: B256) {
+        self.hot_slots_lfu.touch(account, slot);
+    }
+
     /// Returns reference to bytes representing leaf value for the target account.
     pub fn get_account_value(&self, account: &B256) -> Option<&Vec<u8>> {
         self.state.as_revealed_ref()?.get_leaf_value(&Nibbles::unpack(account))
@@ -316,8 +321,6 @@ where
         {
             for (account, storage_proofs) in multiproof.storage_proofs {
                 self.reveal_storage_v2_proof_nodes(account, storage_proofs)?;
-                // Mark this storage trie as hot (accessed this tick)
-                self.storage.modifications.mark_accessed(account);
             }
 
             Ok(())
@@ -363,8 +366,6 @@ where
             for (account, result, revealed_nodes, trie, bufs) in results {
                 self.storage.revealed_paths.insert(account, revealed_nodes);
                 self.storage.tries.insert(account, trie);
-                // Mark this storage trie as hot (accessed this tick)
-                self.storage.modifications.mark_accessed(account);
                 if let Ok(_metric_values) = result {
                     #[cfg(feature = "metrics")]
                     {
@@ -846,10 +847,12 @@ where
         self.storage.shrink_to(storage_nodes, storage_values);
     }
 
-    /// Prunes the account trie and selected storage tries to reduce memory usage.
+    /// Prunes account/storage tries according to global LFU hot-slot retention.
     ///
-    /// Storage tries not in the top `max_storage_tries` by revealed node count are cleared
-    /// entirely.
+    /// - Top LFU `(address, slot)` entries are retained up to `max_hot_slots`.
+    /// - Account trie retains only paths needed for retained addresses.
+    /// - Storage tries retain only paths needed for retained slots.
+    /// - All other revealed paths are pruned to hash stubs or fully evicted.
     ///
     /// # Preconditions
     ///
@@ -865,21 +868,28 @@ where
         name = "SparseStateTrie::prune",
         target = "trie::sparse",
         skip_all,
-        fields(%max_depth, %max_storage_tries)
+        fields(%max_depth, %max_hot_slots)
     )]
-    pub fn prune(&mut self, max_depth: usize, max_storage_tries: usize) {
-        // Prune state and storage tries in parallel
+    pub fn prune(&mut self, max_depth: usize, max_hot_slots: usize) {
+        self.hot_slots_lfu.decay_and_evict(max_hot_slots);
+        let retained = self.hot_slots_lfu.retained_slots_by_address();
+        let mut retained_account_paths: Vec<Nibbles> =
+            retained.keys().copied().map(Nibbles::unpack).collect();
+
+        // Prune account and storage tries in parallel using the same LFU-selected set.
         rayon::join(
             || {
                 if let Some(trie) = self.state.as_revealed_mut() {
-                    trie.prune(max_depth);
+                    trie.prune(max_depth, &retained_account_paths);
                 }
                 self.revealed_account_paths.clear();
             },
             || {
-                self.storage.prune(max_depth, max_storage_tries);
+                self.storage.prune_by_retained_slots(max_depth, retained);
             },
         );
+
+        retained_account_paths.clear();
     }
 
     /// Commits the [`TrieUpdates`] to the sparse trie.
@@ -912,59 +922,27 @@ struct StorageTries<S = ParallelSparseTrie> {
     cleared_revealed_paths: Vec<HashSet<Nibbles>>,
     /// A default cleared trie instance, which will be cloned when creating new tries.
     default_trie: RevealableSparseTrie<S>,
-    /// Tracks access patterns and modification state of storage tries for smart pruning decisions.
-    modifications: StorageTrieModifications,
 }
 
 #[cfg(feature = "std")]
 impl<S: SparseTrieTrait> StorageTries<S> {
-    /// Prunes and evicts storage tries.
+    /// Prunes storage tries using LFU-retained slots.
     ///
-    /// Keeps the top `max_storage_tries` by a score combining size and heat.
-    /// Evicts lower-scored tries entirely, prunes kept tries to `max_depth`.
-    fn prune(&mut self, max_depth: usize, max_storage_tries: usize) {
-        let fn_start = Instant::now();
-        let mut stats =
-            StorageTriesPruneStats { total_tries_before: self.tries.len(), ..Default::default() };
-
-        // Update heat for accessed tries
-        self.modifications.update_and_reset();
-
-        // Collect (address, size, score) for all tries
-        // Score = size * heat_multiplier
-        // Hot tries (high heat) get boosted weight
-        let mut trie_info: Vec<(B256, usize, usize)> = self
+    /// Tries without retained slots are evicted entirely. Tries with retained slots are pruned
+    /// using both the depth limit and retained slot paths.
+    fn prune_by_retained_slots(
+        &mut self,
+        max_depth: usize,
+        mut retained_slots: B256Map<Vec<Nibbles>>,
+    ) {
+        let addresses_to_evict: Vec<B256> = self
             .tries
-            .iter()
-            .map(|(address, trie)| {
-                let size = match trie {
-                    RevealableSparseTrie::Blind(_) => return (*address, 0, 0),
-                    RevealableSparseTrie::Revealed(t) => t.size_hint(),
-                };
-                let heat = self.modifications.heat(address);
-                // Heat multiplier: 1 (cold) to 3 (very hot, heat >= 4)
-                let heat_multiplier = 1 + (heat.min(4) / 2) as usize;
-                (*address, size, size * heat_multiplier)
-            })
+            .keys()
+            .filter(|address| !retained_slots.contains_key(*address))
+            .copied()
             .collect();
 
-        // Use O(n) selection to find top max_storage_tries by score
-        if trie_info.len() > max_storage_tries {
-            trie_info
-                .select_nth_unstable_by(max_storage_tries.saturating_sub(1), |a, b| b.2.cmp(&a.2));
-            trie_info.truncate(max_storage_tries);
-        }
-        let tries_to_keep: B256Map<usize> =
-            trie_info.iter().map(|(address, size, _)| (*address, *size)).collect();
-        stats.tries_to_keep = tries_to_keep.len();
-
-        // Collect keys to evict
-        let tries_to_clear: Vec<B256> =
-            self.tries.keys().filter(|addr| !tries_to_keep.contains_key(*addr)).copied().collect();
-        stats.tries_to_evict = tries_to_clear.len();
-
-        // Evict storage tries that exceeded limit, saving cleared allocations for reuse
-        for address in &tries_to_clear {
+        for address in &addresses_to_evict {
             if let Some(mut trie) = self.tries.remove(address) {
                 trie.clear();
                 self.cleared_tries.push(trie);
@@ -973,58 +951,16 @@ impl<S: SparseTrieTrait> StorageTries<S> {
                 paths.clear();
                 self.cleared_revealed_paths.push(paths);
             }
-            self.modifications.remove(address);
         }
 
-        // Prune storage tries that are kept, but only if:
-        // - They haven't been pruned since last access
-        // - They're large enough to be worth pruning
-        const MIN_SIZE_TO_PRUNE: usize = 1000;
-        let prune_start = Instant::now();
-        for (address, size) in &tries_to_keep {
-            if *size < MIN_SIZE_TO_PRUNE {
-                stats.skipped_small += 1;
-                continue; // Small tries aren't worth the DFS cost
-            }
-            let Some(heat_state) = self.modifications.get_mut(address) else {
-                continue; // No heat state = not tracked
-            };
-            // Only prune if backlog >= 2 (skip every other cycle)
-            if heat_state.prune_backlog < 2 {
-                stats.skipped_recently_pruned += 1;
-                continue; // Recently pruned, skip this cycle
-            }
+        for (address, slots) in &mut retained_slots {
             if let Some(trie) = self.tries.get_mut(address).and_then(|t| t.as_revealed_mut()) {
-                trie.prune(max_depth);
-                heat_state.prune_backlog = 0; // Reset backlog after prune
-                stats.pruned_count += 1;
+                trie.prune(max_depth, slots);
             }
-        }
-        stats.prune_elapsed = prune_start.elapsed();
-
-        // Clear revealed_paths for kept tries
-        for hash in tries_to_keep.keys() {
-            if let Some(paths) = self.revealed_paths.get_mut(hash) {
+            if let Some(paths) = self.revealed_paths.get_mut(address) {
                 paths.clear();
             }
         }
-
-        stats.total_tries_after = self.tries.len();
-        stats.total_elapsed = fn_start.elapsed();
-
-        debug!(
-            target: "trie::sparse",
-            before = stats.total_tries_before,
-            after = stats.total_tries_after,
-            kept = stats.tries_to_keep,
-            evicted = stats.tries_to_evict,
-            pruned = stats.pruned_count,
-            skipped_small = stats.skipped_small,
-            skipped_recent = stats.skipped_recently_pruned,
-            ?stats.prune_elapsed,
-            ?stats.total_elapsed,
-            "StorageTries::prune completed"
-        );
     }
 }
 
@@ -1040,7 +976,6 @@ impl<S: SparseTrieTrait> StorageTries<S> {
             set.clear();
             set
         }));
-        self.modifications.clear();
     }
 
     /// Shrinks the capacity of all storage tries to the given total sizes.
@@ -1123,93 +1058,68 @@ impl<S: SparseTrieTrait + Clone> StorageTries<S> {
     }
 }
 
-/// Statistics from a storage tries prune operation.
+/// Key for identifying a storage slot in the global LFU cache.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+struct HotSlotKey {
+    address: B256,
+    slot: B256,
+}
+
+/// Global LFU tracker for hot storage slots across all storage tries.
 #[derive(Debug, Default)]
-#[allow(dead_code)]
-struct StorageTriesPruneStats {
-    total_tries_before: usize,
-    total_tries_after: usize,
-    tries_to_keep: usize,
-    tries_to_evict: usize,
-    pruned_count: usize,
-    skipped_small: usize,
-    skipped_recently_pruned: usize,
-    prune_elapsed: core::time::Duration,
-    total_elapsed: core::time::Duration,
+struct HotSlotsLfu {
+    capacity: usize,
+    frequencies: BTreeMap<HotSlotKey, u32>,
 }
 
-/// Per-trie access tracking and prune state.
-///
-/// Tracks how frequently a storage trie is accessed and when it was last pruned,
-/// enabling smart pruning decisions that preserve frequently-used tries.
-#[derive(Debug, Clone, Copy, Default)]
-#[allow(dead_code)]
-struct TrieModificationState {
-    /// Access frequency level (0-255). Incremented each cycle the trie is accessed.
-    /// Used for prioritizing which tries to keep during pruning.
-    heat: u8,
-    /// Prune backlog - cycles since last prune. Incremented each cycle,
-    /// reset to 0 when pruned. Used to decide when pruning is needed.
-    prune_backlog: u8,
-}
+impl HotSlotsLfu {
+    /// Sets LFU capacity, decays frequencies, and trims entries if needed.
+    fn decay_and_evict(&mut self, capacity: usize) {
+        self.capacity = capacity;
 
-/// Tracks access patterns and modification state of storage tries for smart pruning decisions.
-///
-/// Access-based tracking is more accurate than simple generation counting because it tracks
-/// actual access patterns rather than administrative operations (take/insert).
-///
-/// - Access frequency is incremented when a storage proof is revealed (accessed)
-/// - Access frequency decays each prune cycle for tries not accessed that cycle
-/// - Tries with higher access frequency are prioritized for preservation during pruning
-#[derive(Debug, Default)]
-struct StorageTrieModifications {
-    /// Access frequency and prune state per storage trie address.
-    state: B256Map<TrieModificationState>,
-    /// Tracks which tries were accessed in the current cycle (between prune calls).
-    accessed_this_cycle: B256Set,
-}
+        // Halve all frequencies so stale entries gradually decay to zero.
+        self.frequencies.retain(|_, freq| {
+            *freq >>= 1;
+            *freq > 0
+        });
 
-#[allow(dead_code)]
-impl StorageTrieModifications {
-    /// Marks a storage trie as accessed this cycle.
-    /// Heat and `prune_backlog` are updated in [`Self::update_and_reset`].
-    #[inline]
-    fn mark_accessed(&mut self, address: B256) {
-        self.accessed_this_cycle.insert(address);
-    }
-
-    /// Returns mutable reference to the heat state for a storage trie.
-    #[inline]
-    fn get_mut(&mut self, address: &B256) -> Option<&mut TrieModificationState> {
-        self.state.get_mut(address)
-    }
-
-    /// Returns the heat level for a storage trie (0 if not tracked).
-    #[inline]
-    fn heat(&self, address: &B256) -> u8 {
-        self.state.get(address).map_or(0, |s| s.heat)
-    }
-
-    /// Updates heat and prune backlog for accessed tries.
-    /// Called at the start of each prune cycle.
-    fn update_and_reset(&mut self) {
-        for address in self.accessed_this_cycle.drain() {
-            let entry = self.state.entry(address).or_default();
-            entry.heat = entry.heat.saturating_add(1);
-            entry.prune_backlog = entry.prune_backlog.saturating_add(1);
+        while self.frequencies.len() > self.capacity {
+            let Some(evict_key) = self
+                .frequencies
+                .iter()
+                .min_by_key(|(key, frequency)| (**frequency, **key))
+                .map(|(key, _)| *key)
+            else {
+                break;
+            };
+            self.frequencies.remove(&evict_key);
         }
     }
 
-    /// Removes tracking for a specific address (when trie is evicted).
-    fn remove(&mut self, address: &B256) {
-        self.state.remove(address);
-        self.accessed_this_cycle.remove(address);
+    /// Records a storage slot touch.
+    fn touch(&mut self, address: B256, slot: B256) {
+        let key = HotSlotKey { address, slot };
+        if let Some(frequency) = self.frequencies.get_mut(&key) {
+            *frequency = frequency.saturating_add(1);
+            return;
+        }
+
+        self.frequencies.insert(key, 1);
     }
 
-    /// Clears all heat tracking state.
-    fn clear(&mut self) {
-        self.state.clear();
-        self.accessed_this_cycle.clear();
+    /// Returns retained slots grouped by address.
+    fn retained_slots_by_address(&self) -> B256Map<Vec<Nibbles>> {
+        let mut grouped = B256Map::<Vec<Nibbles>>::default();
+        for key in self.frequencies.keys() {
+            grouped.entry(key.address).or_default().push(Nibbles::unpack(key.slot));
+        }
+
+        for slots in grouped.values_mut() {
+            slots.sort_unstable();
+            slots.dedup();
+        }
+
+        grouped
     }
 }
 

crates/trie/sparse/src/traits.rs

@@ -288,10 +288,17 @@ pub trait SparseTrie: Sized + Debug + Send + Sync {
     /// during pruning. Larger values indicate larger tries that are more valuable to preserve.
     fn size_hint(&self) -> usize;
 
-    /// Replaces nodes beyond `max_depth` with hash stubs and removes their descendants.
+    /// Prunes the trie by replacing revealed subtrees with hash stubs.
     ///
+    /// Nodes within `max_depth` are never pruned. Beyond `max_depth`, subtrees are blinded
+    /// unless they are on the path to a retained leaf:
+    ///
+    /// - `depth < max_depth` → always kept
+    /// - `depth >= max_depth` → kept only if `retained_leaves` contains a descendant
+    ///
+    /// When `retained_leaves` is empty, all nodes beyond `max_depth` are blinded.
     /// Depth counts nodes traversed (not nibbles), so extension nodes count as 1 depth
-    /// regardless of key length. `max_depth == 0` prunes all children of the root node.
+    /// regardless of key length.
     ///
     /// # Preconditions
     ///
@@ -306,7 +313,7 @@ pub trait SparseTrie: Sized + Debug + Send + Sync {
     /// # Returns
     ///
     /// The number of nodes converted to hash stubs.
-    fn prune(&mut self, max_depth: usize) -> usize;
+    fn prune(&mut self, max_depth: usize, retained_leaves: &[Nibbles]) -> usize;
 
     /// Takes the debug recorder out of this trie, replacing it with an empty one.
     ///

docs/vocs/docs/pages/cli/reth/node.mdx

@@ -991,8 +991,8 @@ Engine:
 
           [default: 4]
 
-      --engine.sparse-trie-max-storage-tries <SPARSE_TRIE_MAX_STORAGE_TRIES>
-          Maximum number of storage tries to retain after sparse trie pruning
+      --engine.sparse-trie-max-hot-slots <SPARSE_TRIE_MAX_HOT_SLOTS>
+          LFU hot-slot capacity: max storage slots retained across sparse trie prune cycles
 
           [default: 100]