feat(evm): add witness oracle metrics

crates/evm/Cargo.toml

@@ -21,6 +21,8 @@ alloy-rpc-types-debug.workspace = true
 auto_impl.workspace = true
 crossbeam-channel.workspace = true
 futures.workspace = true
+metrics.workspace = true
+metrics-derive.workspace = true
 parking_lot.workspace = true
 eyre.workspace = true
 op-alloy-consensus.workspace = true

crates/evm/src/cache.rs

@@ -5,13 +5,29 @@ use std::{collections::BTreeMap, sync::Arc};
 use alloy_rpc_types_debug::ExecutionWitness;
 use parking_lot::Mutex;
 
+use crate::metrics::WitnessMetrics;
+
 /// Default number of execution witnesses to retain in memory.
 pub const DEFAULT_WITNESS_CAP: usize = 1024;
 
+/// A cached witness alongside the number of bytes it retains.
+#[derive(Debug)]
+struct CachedWitness {
+    witness: Arc<ExecutionWitness>,
+    bytes: usize,
+}
+
+/// Cache contents guarded by a single lock, with a running total of retained bytes.
+#[derive(Debug, Default)]
+struct Inner {
+    witnesses: BTreeMap<u64, CachedWitness>,
+    bytes: usize,
+}
+
 /// A bounded, reorg-safe in-memory ring buffer of [`ExecutionWitness`]es keyed by block number.
 #[derive(Debug)]
 pub struct WitnessCache {
-    inner: Mutex<BTreeMap<u64, Arc<ExecutionWitness>>>,
+    inner: Mutex<Inner>,
     depth: usize,
 }
 
@@ -35,33 +51,71 @@ impl WitnessCache {
     #[must_use]
     pub fn with_depth(depth: usize) -> Self {
         Self {
-            inner: Mutex::new(BTreeMap::new()),
+            inner: Mutex::new(Inner::default()),
             depth: depth.max(1),
         }
     }
 
     /// Inserts (or replaces) the execution witness for `block_number`, evicting the lowest block(s)
     /// once `depth` is exceeded.
     pub fn insert(&self, block_number: u64, witness: ExecutionWitness) {
+        let metrics = WitnessMetrics::get();
+        let bytes = retained_bytes(&witness);
         let mut inner = self.inner.lock();
-        inner.insert(block_number, Arc::new(witness));
-        while inner.len() > self.depth {
-            inner.pop_first();
+
+        if let Some(replaced) = inner.witnesses.insert(
+            block_number,
+            CachedWitness {
+                witness: Arc::new(witness),
+                bytes,
+            },
+        ) {
+            inner.bytes -= replaced.bytes;
+        }
+        inner.bytes += bytes;
+
+        let mut evicted = 0u64;
+        while inner.witnesses.len() > self.depth {
+            let Some((_, dropped)) = inner.witnesses.pop_first() else {
+                break;
+            };
+            inner.bytes -= dropped.bytes;
+            evicted += 1;
+        }
+
+        metrics.inserted.increment(1);
+        metrics.evicted.increment(evicted);
+        metrics.witness_bytes.record(bytes as f64);
+        metrics.cache_bytes.set(inner.bytes as f64);
+        metrics.cache_len.set(inner.witnesses.len() as f64);
+        if let (Some((oldest, _)), Some((newest, _))) = (
+            inner.witnesses.first_key_value(),
+            inner.witnesses.last_key_value(),
+        ) {
+            metrics.cache_oldest_block.set(*oldest as f64);
+            metrics.cache_newest_block.set(*newest as f64);
         }
     }
 
     /// Returns the execution witness for `block_number`, if cached. Zero-copy: clones only the
     /// [`Arc`].
     #[must_use]
     pub fn get(&self, block_number: u64) -> Option<Arc<ExecutionWitness>> {
-        self.inner.lock().get(&block_number).cloned()
+        self.inner
+            .lock()
+            .witnesses
+            .get(&block_number)
+            .map(|entry| Arc::clone(&entry.witness))
     }
 
     /// Returns the lowest and highest cached block numbers, if any.
     #[must_use]
     pub fn bounds(&self) -> Option<(u64, u64)> {
         let inner = self.inner.lock();
-        Some((*inner.first_key_value()?.0, *inner.last_key_value()?.0))
+        Some((
+            *inner.witnesses.first_key_value()?.0,
+            *inner.witnesses.last_key_value()?.0,
+        ))
     }
 
     /// Collects the execution witnesses for the contiguous, inclusive L2 range
@@ -72,19 +126,40 @@ impl WitnessCache {
     /// only the per-block [`Arc`]s are cloned, never the witness data.
     #[must_use]
     pub fn range(&self, start_block: u64, end_block: u64) -> Option<Vec<Arc<ExecutionWitness>>> {
+        let metrics = WitnessMetrics::get();
         if end_block < start_block {
+            metrics.range_miss.increment(1);
             return None;
         }
+
         let inner = self.inner.lock();
         let witnesses: Vec<_> = inner
+            .witnesses
             .range(start_block..=end_block)
-            .map(|(_, witness)| Arc::clone(witness))
+            .map(|(_, entry)| Arc::clone(&entry.witness))
             .collect();
+        drop(inner);
+
         // Every block in `[start_block, end_block]` must be present.
-        (witnesses.len() as u64 == end_block - start_block + 1).then_some(witnesses)
+        let requested = end_block - start_block + 1;
+        let missing = requested - witnesses.len() as u64;
+        if missing > 0 {
+            metrics.range_miss.increment(1);
+            metrics.range_missing_blocks.increment(missing);
+            return None;
+        }
+        metrics.range_hit.increment(1);
+        Some(witnesses)
     }
 }
 
+/// Returns the number of witness-data bytes an [`ExecutionWitness`] retains.
+fn retained_bytes(witness: &ExecutionWitness) -> usize {
+    let total =
+        |entries: &[alloy_primitives::Bytes]| entries.iter().map(|b| b.len()).sum::<usize>();
+    total(&witness.state) + total(&witness.codes) + total(&witness.keys) + total(&witness.headers)
+}
+
 #[cfg(test)]
 mod tests {
     use alloy_primitives::Bytes;
@@ -128,6 +203,31 @@ mod tests {
         assert_eq!(cache.bounds(), Some((10, 10)));
     }
 
+    /// A witness whose `state` entry is exactly `len` bytes.
+    fn sized_witness(len: usize) -> ExecutionWitness {
+        ExecutionWitness {
+            state: vec![Bytes::from(vec![0u8; len])],
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn tracks_retained_bytes_across_replace_and_evict() {
+        let cache = WitnessCache::with_depth(2);
+        cache.insert(1, witness(1));
+        cache.insert(2, witness(2));
+        assert_eq!(cache.inner.lock().bytes, 2);
+
+        // Replacing a block swaps its contribution instead of double-counting it.
+        cache.insert(2, sized_witness(4));
+        assert_eq!(cache.inner.lock().bytes, 5);
+
+        // Evicting block 1 releases exactly its contribution.
+        cache.insert(3, witness(3));
+        assert_eq!(cache.inner.lock().bytes, 5);
+        assert_eq!(cache.bounds(), Some((2, 3)));
+    }
+
     #[test]
     fn range_requires_every_block() {
         let cache = WitnessCache::with_depth(16);

crates/evm/src/collector.rs

@@ -7,7 +7,9 @@ use reth_provider::CanonStateSubscriptions;
 use reth_revm::witness::ExecutionWitnessRecord;
 use reth_tasks::TaskExecutor;
 
-use crate::{BlockExecutionWitness, ExecutionWitnessHandle, ProviderBounds};
+use crate::{
+    BlockExecutionWitness, ExecutionWitnessHandle, ProviderBounds, metrics::WitnessMetrics,
+};
 
 /// Spawns the live witness collector as a critical task.
 pub fn spawn_witness_collector<P>(
@@ -67,12 +69,15 @@ pub fn spawn_witness_collector<P>(
                     });
                     match result {
                         Ok(witness) => cache.insert(block_number, witness),
-                        Err(err) => tracing::error!(
-                            target: "world_chain::witness",
-                            block_number,
-                            %err,
-                            "failed to assemble execution witness; skipping",
-                        ),
+                        Err(err) => {
+                            WitnessMetrics::get().assembly_failed.increment(1);
+                            tracing::error!(
+                                target: "world_chain::witness",
+                                block_number,
+                                %err,
+                                "failed to assemble execution witness; skipping",
+                            );
+                        }
                     }
                 }
             });

crates/evm/src/execution/executor.rs

@@ -7,7 +7,7 @@ use reth_revm::{State, witness::ExecutionWitnessRecord};
 use revm::context::Block;
 use tracing::error;
 
-use crate::BlockExecutionWitness;
+use crate::{BlockExecutionWitness, metrics::WitnessMetrics};
 
 /// A [`BlockExecutor`] that delegates to an inner executor and, on
 #[derive(Debug)]
@@ -76,9 +76,16 @@ where
                 record,
             };
 
-            let _ = sender.try_send(captured).inspect_err(|e| {
-                error!(target: "world_chain::witness", %block_number, %e, "failed to send captured witness");
-            });
+            let metrics = WitnessMetrics::get();
+            match sender.try_send(captured) {
+                Ok(()) => metrics.captured.increment(1),
+                Err(e) => {
+                    // A dropped witness leaves a permanent hole: any range spanning this block is
+                    // unservable until it is evicted.
+                    metrics.dropped.increment(1);
+                    error!(target: "world_chain::witness", %block_number, %e, "failed to send captured witness");
+                }
+            }
         }
 
         self.inner.finish()

crates/evm/src/metrics.rs

@@ -1,5 +1,7 @@
 use auto_impl::auto_impl;
-use std::time::Duration;
+use metrics::{Counter, Gauge, Histogram};
+use metrics_derive::Metrics;
+use std::{sync::LazyLock, time::Duration};
 
 /// General trait to collect metrics around flashblock execution.
 ///
@@ -25,3 +27,46 @@ pub enum PayloadBuildStage {
     StateRoot,
     BlockAssembly,
 }
+
+/// Process-wide metrics for the live pre-image witness oracle (`--witness.collect`).
+#[derive(Clone, Metrics)]
+#[metrics(scope = "world_chain.witness")]
+pub struct WitnessMetrics {
+    /// Witnesses captured from the block-import executor and handed to the collector.
+    pub captured: Counter,
+    /// Witnesses dropped at capture time because the collector channel was full or closed.
+    pub dropped: Counter,
+    /// Witnesses the collector failed to assemble, leaving a permanent hole in the cache.
+    pub assembly_failed: Counter,
+    /// Witnesses inserted into the cache.
+    pub inserted: Counter,
+    /// Witnesses evicted from the cache by the ring-buffer depth bound.
+    pub evicted: Counter,
+    /// Size of each cached witness, in bytes.
+    pub witness_bytes: Histogram,
+    /// Bytes of witness data currently retained by the cache.
+    pub cache_bytes: Gauge,
+    /// Witnesses currently retained by the cache.
+    pub cache_len: Gauge,
+    /// Lowest block number currently cached.
+    pub cache_oldest_block: Gauge,
+    /// Highest block number currently cached.
+    pub cache_newest_block: Gauge,
+    /// Range lookups served in full from the cache.
+    pub range_hit: Counter,
+    /// Range lookups rejected because at least one block in the range was missing.
+    pub range_miss: Counter,
+    /// Blocks absent from the cache, summed over every rejected range lookup.
+    pub range_missing_blocks: Counter,
+}
+
+impl WitnessMetrics {
+    /// Returns the process-wide witness metrics, registered on first use.
+    ///
+    /// Registration is lazy so the handles resolve against the Prometheus recorder installed at
+    /// node startup; the first use is a block import, long after that.
+    pub fn get() -> &'static Self {
+        static METRICS: LazyLock<WitnessMetrics> = LazyLock::new(WitnessMetrics::default);
+        &METRICS
+    }
+}