feat(trie): add sparse trie cache hit rate metrics (#22767)

Co-authored-by: Alexey Shekhirin <5773434+shekhirin@users.noreply.github.com>

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

@@ -191,6 +191,15 @@ pub(crate) struct MultiProofTaskMetrics {
     /// Time spent waiting for preserved sparse trie cache to become available.
     pub sparse_trie_cache_wait_duration_histogram: Histogram,
 
+    /// Number of account leaf updates applied without needing a new proof (cache hits).
+    pub sparse_trie_account_cache_hits: Histogram,
+    /// Number of account leaf updates that required a new proof (cache misses).
+    pub sparse_trie_account_cache_misses: Histogram,
+    /// Number of storage leaf updates applied without needing a new proof (cache hits).
+    pub sparse_trie_storage_cache_hits: Histogram,
+    /// Number of storage leaf updates that required a new proof (cache misses).
+    pub sparse_trie_storage_cache_misses: Histogram,
+
     /// Retained memory of the preserved sparse trie cache in bytes.
     pub sparse_trie_retained_memory_bytes: Gauge,
     /// Number of storage tries retained in the preserved sparse trie cache.

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

@@ -92,6 +92,14 @@ pub(super) struct SparseTrieCacheTask<A = ParallelSparseTrie, S = ParallelSparse
     account_rlp_buf: Vec<u8>,
     /// Whether the last state update has been received.
     finished_state_updates: bool,
+    /// Accumulated account leaf update cache hits.
+    account_cache_hits: u64,
+    /// Accumulated account leaf update cache misses.
+    account_cache_misses: u64,
+    /// Accumulated storage leaf update cache hits.
+    storage_cache_hits: u64,
+    /// Accumulated storage leaf update cache misses.
+    storage_cache_misses: u64,
     /// Pending proof targets queued for dispatch to proof workers.
     pending_targets: PendingTargets,
     /// Number of pending execution/prewarming updates received but not yet passed to
@@ -142,6 +150,10 @@ where
             fetched_storage_targets: Default::default(),
             account_rlp_buf: Vec::with_capacity(TRIE_ACCOUNT_RLP_MAX_SIZE),
             finished_state_updates: Default::default(),
+            account_cache_hits: 0,
+            account_cache_misses: 0,
+            storage_cache_hits: 0,
+            storage_cache_misses: 0,
             pending_targets: Default::default(),
             pending_updates: Default::default(),
             metrics,
@@ -332,6 +344,15 @@ where
         self.metrics.sparse_trie_final_update_duration_histogram.record(end.duration_since(start));
         self.metrics.sparse_trie_total_duration_histogram.record(end.duration_since(now));
 
+        self.metrics.sparse_trie_account_cache_hits.record(self.account_cache_hits as f64);
+        self.metrics.sparse_trie_account_cache_misses.record(self.account_cache_misses as f64);
+        self.metrics.sparse_trie_storage_cache_hits.record(self.storage_cache_hits as f64);
+        self.metrics.sparse_trie_storage_cache_misses.record(self.storage_cache_misses as f64);
+        self.account_cache_hits = 0;
+        self.account_cache_misses = 0;
+        self.storage_cache_hits = 0;
+        self.storage_cache_misses = 0;
+
         Ok(StateRootComputeOutcome {
             state_root,
             trie_updates: Arc::new(trie_updates),
@@ -505,6 +526,7 @@ where
             let fetched = self.fetched_storage_targets.entry(*address).or_default();
             let mut targets = Vec::new();
 
+            let updates_len_before = updates.len();
             trie.update_leaves(updates, |path, min_len| match fetched.entry(path) {
                 Entry::Occupied(mut entry) => {
                     if min_len < *entry.get() {
@@ -517,6 +539,9 @@ where
                     targets.push(ProofV2Target::new(path).with_min_len(min_len));
                 }
             })?;
+            let updates_len_after = updates.len();
+            self.storage_cache_hits += (updates_len_before - updates_len_after) as u64;
+            self.storage_cache_misses += updates_len_after as u64;
 
             if !targets.is_empty() {
                 self.pending_targets.extend_storage_targets(address, targets);
@@ -562,7 +587,11 @@ where
             }
         })?;
 
-        Ok(account_updates.len() < updates_len_before)
+        let updates_len_after = account_updates.len();
+        self.account_cache_hits += (updates_len_before - updates_len_after) as u64;
+        self.account_cache_misses += updates_len_after as u64;
+
+        Ok(updates_len_after < updates_len_before)
     }
 
     /// Iterates through all storage tries for which all updates were processed, computes their