feat(DSM): add canister_log_retention_seconds metric (#9952)

## Summary

- Add `canister_log_retention_seconds` histogram to `SchedulerMetrics` —
wall-clock span between the oldest and newest records currently in a
canister's log buffer.
- Activity-gated observation at round finalization; cost scales with
activity, not canister count.
- Works on both log stores (branched on
`LOG_MEMORY_STORE_FEATURE_ENABLED`), giving directly-comparable
before/after-flip data on the same histogram. On the new path,
`first_timestamp` is cached in a transient `Option<u64>` on
`LogMemoryStore`, populated eagerly from every mutation site — no header
change, no checkpoint migration.
- Buckets: `decimal_buckets_with_zero(1, 6)` → 19 buckets, `[0, 10 s ..
~58 d]`.

## Stacked on

Depends on #9948. Base is `maksym/canister-log-metrics`; retarget to
`master` once #9948 lands.

---------

Co-authored-by: IDX GitHub Automation <infra+github-automation@dfinity.org>

Author: maksymar

rs/execution_environment/src/scheduler.rs

@@ -1495,6 +1495,19 @@ impl Scheduler for SchedulerImpl {
                 } else {
                     old_log.delta_log_sizes()
                 };
+                // Observe retention from whichever log store is active,
+                // only for canisters that appended this round.
+                let retention = if LOG_MEMORY_STORE_FEATURE_ENABLED {
+                    new_log
+                        .has_delta_log_sizes()
+                        .then(|| new_log.retention())
+                        .flatten()
+                } else {
+                    old_log
+                        .has_delta_log_sizes()
+                        .then(|| old_log.retention())
+                        .flatten()
+                };
                 if new_log.has_delta_log_sizes() || old_log.has_delta_log_sizes() {
                     // Only clone state if delta log sizes are not empty.
                     let canister = Arc::make_mut(canister);
@@ -1511,6 +1524,11 @@ impl Scheduler for SchedulerImpl {
                         .canister_log_delta_memory_usage
                         .observe(size as f64);
                 }
+                if let Some(retention) = retention {
+                    self.metrics
+                        .canister_log_retention
+                        .observe(retention.as_secs_f64());
+                }
 
                 // TODO(EXC-1124): Re-enable once the cycle balance check is fixed.
                 // cycles_out_sum += canister.system_state.queues().output_queue_cycles();

rs/execution_environment/src/scheduler/scheduler_metrics.rs

@@ -25,6 +25,7 @@ pub(crate) const SCHEDULER_CORES_INVARIANT_BROKEN: &str = "scheduler_cores_invar
 pub struct SchedulerMetrics {
     pub(super) canister_age: Histogram,
     pub(super) canister_log_delta_memory_usage: Histogram,
+    pub(super) canister_log_retention: Histogram,
     pub(super) canister_ingress_queue_latencies: Histogram,
     pub(super) compute_utilization_per_core: Histogram,
     pub(super) msg_execution_duration: Histogram,
@@ -91,6 +92,12 @@ impl SchedulerMetrics {
                 // 1 KiB (2^10) .. 8 MiB (2^23), plus zero — 15 total buckets (0 + 14 powers).
                 binary_buckets_with_zero(10, 23)
             ),
+            canister_log_retention: metrics_registry.histogram(
+                "canister_log_retention_seconds",
+                "Time span between the oldest and newest records in the canister log buffer, in seconds.",
+                // 10 s .. 5×10⁶ s (~58 d), plus zero — 19 total buckets (0 + 18 powers).
+                decimal_buckets_with_zero(1, 6),
+            ),
             canister_ingress_queue_latencies: metrics_registry.histogram(
                 "scheduler_canister_ingress_queue_latencies_seconds",
                 "Per-canister mean IC clock duration spent by messages in the ingress queue.",

rs/execution_environment/tests/canister_logging.rs

@@ -1794,6 +1794,39 @@ fn test_metric_canister_log_delta_memory_usage_bytes() {
     );
 }
 
+#[test]
+fn test_metric_canister_log_retention_seconds() {
+    // Observed by the scheduler at round finalization for canisters that
+    // appended log records this round. Retention is the wall-clock span
+    // between the oldest and newest records held in the buffer. Both log
+    // stores (old `CanisterLog` and new `LogMemoryStore`) compute it the
+    // same way — the assertions hold regardless of the feature flag.
+    const METRIC: &str = "canister_log_retention_seconds";
+    const TIME_ADVANCE: Duration = Duration::from_secs(60);
+    let env = setup_env();
+    let canister_id = create_and_install_canister(
+        &env,
+        CanisterSettingsArgsBuilder::new().build(),
+        wat_canister()
+            .update("test", wat_fn().debug_print(b"hello"))
+            .build_wasm(),
+    );
+    // Seed the buffer with a first record so retention is non-zero on the
+    // second observation.
+    let _ = env.execute_ingress(canister_id, "test", vec![]);
+    let before = fetch_histogram_stats(env.metrics_registry(), METRIC).unwrap();
+
+    // Advance simulated time and append another record.
+    env.advance_time(TIME_ADVANCE);
+    let _ = env.execute_ingress(canister_id, "test", vec![]);
+
+    let after = fetch_histogram_stats(env.metrics_registry(), METRIC).unwrap();
+    assert_eq!(after.count, before.count + 1);
+    let sample = after.sum - before.sum;
+    // The new sample should report at least the advanced wall-clock gap.
+    assert_le!(TIME_ADVANCE.as_secs_f64(), sample);
+}
+
 #[test]
 fn test_metric_canister_log_resize_duration_seconds() {
     // Observed at the resize call site in `CanisterManager::update_settings`

rs/replicated_state/src/canister_state/system_state/log_memory_store/mod.rs

@@ -20,6 +20,7 @@ use ic_validate_eq_derive::ValidateEq;
 use std::collections::VecDeque;
 use std::sync::Arc;
 use std::sync::OnceLock;
+use std::time::Duration;
 
 /// Upper bound on stored delta-log sizes used for metrics.
 /// Limits memory growth, 10k covers expected per-round
@@ -71,6 +72,17 @@ pub struct LogMemoryStore {
     #[validate_eq(Ignore)]
     header_cache: OnceLock<Option<Header>>,
 
+    /// Cached timestamp of the oldest live record. Makes the retention
+    /// metric's per-round observation (`max_timestamp − first_timestamp`)
+    /// an O(1) field read instead of a cold page-map read at `data_head`.
+    ///
+    /// Plain `Option<u64>` (not `OnceLock` like `header_cache`) because we
+    /// populate eagerly from every mutation site; no lazy init under
+    /// `&self` needed. Not persisted across checkpoints; rebuilt in
+    /// `new_inner`.
+    #[validate_eq(Ignore)]
+    first_timestamp_cache: Option<u64>,
+
     /// (!) No need to preserve across checkpoints.
     /// Tracks the size of each delta log appended during a round.
     /// Multiple logs can be appended in one round (e.g. heartbeat, timers, or message executions).
@@ -104,21 +116,33 @@ impl LogMemoryStore {
         maybe_page_map: Option<PageMap>,
         persistent_next_idx: u64,
     ) -> Self {
-        Self {
+        let maybe_page_map = if feature_flag == FlagStatus::Enabled {
+            maybe_page_map
+        } else {
+            None
+        };
+        let persistent_next_idx = if feature_flag == FlagStatus::Enabled {
+            persistent_next_idx
+        } else {
+            0
+        };
+        // Rebuild the first-timestamp cache from the ring buffer so the
+        // invariant holds immediately after `from_checkpoint`, without
+        // waiting for the next mutation to populate it.
+        let first_timestamp_cache = maybe_page_map
+            .clone()
+            .and_then(RingBuffer::load_checked)
+            .and_then(|rb| rb.first_timestamp(&rb.get_header()));
+        let store = Self {
             feature_flag,
-            maybe_page_map: if feature_flag == FlagStatus::Enabled {
-                maybe_page_map
-            } else {
-                None
-            },
-            persistent_next_idx: if feature_flag == FlagStatus::Enabled {
-                persistent_next_idx
-            } else {
-                0
-            },
+            maybe_page_map,
+            persistent_next_idx,
             header_cache: OnceLock::new(),
+            first_timestamp_cache,
             delta_log_sizes: VecDeque::new(),
-        }
+        };
+        debug_assert!(store.stats_ok());
+        store
     }
 
     /// Provides access to the underlying `PageMap`.
@@ -148,15 +172,21 @@ impl LogMemoryStore {
             self.save_ring_buffer(ring_buffer);
         } else {
             self.header_cache = OnceLock::new();
+            self.first_timestamp_cache = None;
+            debug_assert!(self.stats_ok());
         }
     }
 
-    /// Update page_map, header_cache and persistent_next_idx.
+    /// Update page_map, header_cache, first_timestamp_cache and persistent_next_idx.
     fn save_ring_buffer(&mut self, ring_buffer: RingBuffer) {
+        let header = ring_buffer.get_header();
+        let first_timestamp = ring_buffer.first_timestamp(&header);
         self.maybe_page_map = Some(ring_buffer.to_page_map());
-        self.header_cache = OnceLock::from(Some(ring_buffer.get_header()));
+        self.header_cache = OnceLock::from(Some(header));
+        self.first_timestamp_cache = first_timestamp;
         // Must come after header_cache update, since next_idx() reads from it.
         self.persistent_next_idx = self.next_idx();
+        debug_assert!(self.stats_ok());
     }
 
     /// Deallocates underlying memory.
@@ -165,6 +195,8 @@ impl LogMemoryStore {
         self.persistent_next_idx = self.next_idx();
         self.maybe_page_map = None;
         self.header_cache = OnceLock::new();
+        self.first_timestamp_cache = None;
+        debug_assert!(self.stats_ok());
     }
 
     /// Loads the ring buffer from the page map.
@@ -174,13 +206,63 @@ impl LogMemoryStore {
             .and_then(RingBuffer::load_checked)
     }
 
+    /// Invariant: populated caches must match what a fresh read of the ring
+    /// buffer would return. Intended to be called only via `debug_assert!`;
+    /// the `cfg!` guard keeps the body a no-op in release regardless of
+    /// caller, so it's fine for the check to be expensive.
+    fn stats_ok(&self) -> bool {
+        if !cfg!(debug_assertions) {
+            return true;
+        }
+        let ring_buffer = self.load_ring_buffer();
+        let actual_header = ring_buffer.as_ref().map(|rb| rb.get_header());
+        // `header_cache` is lazy: if populated, must match; if empty, skip.
+        if let Some(cached) = self.header_cache.get()
+            && *cached != actual_header
+        {
+            return false;
+        }
+        // `first_timestamp_cache` is kept eagerly in sync, so must always match.
+        let actual_first_ts = ring_buffer
+            .as_ref()
+            .zip(actual_header.as_ref())
+            .and_then(|(rb, h)| rb.first_timestamp(h));
+        if self.first_timestamp_cache != actual_first_ts {
+            return false;
+        }
+        true
+    }
+
     /// Returns the ring buffer header.
     fn get_header(&self) -> Option<Header> {
         *self
             .header_cache
             .get_or_init(|| self.load_ring_buffer().map(|rb| rb.get_header()))
     }
 
+    /// Returns the timestamp of the most recently appended record, or `None`
+    /// if the buffer is empty. O(1) via the cached header.
+    pub fn max_timestamp(&self) -> Option<u64> {
+        let header = self.get_header()?;
+        (header.data_size.get() > 0).then_some(header.max_timestamp)
+    }
+
+    /// Returns the timestamp of the oldest live record, or `None` if the
+    /// buffer is empty. O(1) field read — the cache is kept in sync with
+    /// the ring buffer by every mutation.
+    pub fn first_timestamp(&self) -> Option<u64> {
+        self.first_timestamp_cache
+    }
+
+    /// Returns the time span between the oldest and newest records, or
+    /// `None` if the buffer is empty. Returns `Duration::ZERO` when both
+    /// timestamps are equal (single record).
+    pub fn retention(&self) -> Option<Duration> {
+        let max = self.max_timestamp()?;
+        let first = self.first_timestamp()?;
+        Some(Duration::from_nanos(max.saturating_sub(first)))
+    }
+
     /// Returns the total allocated memory.
     pub fn memory_usage(&self) -> usize {
         self.total_virtual_memory_usage()
@@ -374,13 +456,15 @@ impl Clone for LogMemoryStore {
                 Some(val) => OnceLock::from(*val),
                 None => OnceLock::new(),
             },
+            first_timestamp_cache: self.first_timestamp_cache,
         }
     }
 }
 
 impl PartialEq for LogMemoryStore {
     fn eq(&self, other: &Self) -> bool {
-        // header_cache is a transient cache and should not be compared.
+        // header_cache and first_timestamp_cache are transient caches and
+        // should not be compared.
         self.feature_flag == other.feature_flag
             && self.maybe_page_map == other.maybe_page_map
             && self.persistent_next_idx == other.persistent_next_idx

rs/replicated_state/src/canister_state/system_state/log_memory_store/ring_buffer.rs

@@ -84,6 +84,21 @@ impl RingBuffer {
         self.io.load_header()
     }
 
+    /// Returns the timestamp of the oldest live record, or `None` if the
+    /// buffer is empty. Takes an already-loaded `header` to avoid a
+    /// redundant page-map read when the caller has one in hand.
+    ///
+    /// Reads the record header at `data_head` — the same access pattern used
+    /// by `load_index_table` to reconstruct its `front` entry.
+    pub fn first_timestamp(&self, header: &Header) -> Option<u64> {
+        if header.data_size.get() == 0 {
+            return None;
+        }
+        self.io
+            .load_record_without_content(header, header.data_head)
+            .map(|r| r.timestamp)
+    }
+
     /// Returns the data capacity of the ring buffer.
     #[cfg(test)]
     pub fn byte_capacity(&self) -> usize {

rs/replicated_state/src/canister_state/system_state/log_memory_store/tests/log_memory_store.rs

@@ -68,6 +68,56 @@ fn initialization_defaults() {
     assert_eq!(s.next_idx(), 0);
 }
 
+#[test]
+fn test_retention_across_lifecycle() {
+    use std::time::Duration;
+
+    // Empty store: no header, no retention.
+    let mut s = LogMemoryStore::new(TEST_LOG_MEMORY_STORE_FEATURE);
+    assert_eq!(s.first_timestamp(), None);
+    assert_eq!(s.max_timestamp(), None);
+    assert_eq!(s.retention(), None);
+
+    // Allocated but still empty: both timestamps report None.
+    s.resize_for_testing(TEST_LOG_MEMORY_LIMIT);
+    assert_eq!(s.first_timestamp(), None);
+    assert_eq!(s.max_timestamp(), None);
+    assert_eq!(s.retention(), None);
+
+    // Single record: retention is zero.
+    let mut delta = CanisterLog::default_delta();
+    delta.add_record(1_000_000_000, b"a".to_vec());
+    s.append_delta_log(&mut delta);
+    assert_eq!(s.first_timestamp(), Some(1_000_000_000));
+    assert_eq!(s.max_timestamp(), Some(1_000_000_000));
+    assert_eq!(s.retention(), Some(Duration::ZERO));
+
+    // Multiple records: retention spans first..last.
+    let mut delta = CanisterLog::new_delta_with_next_index(s.next_idx(), TEST_LOG_MEMORY_LIMIT);
+    delta.add_record(1_500_000_000, b"b".to_vec()); // t = 1.5 s
+    delta.add_record(61_000_000_000, b"c".to_vec()); // t = 61 s
+    s.append_delta_log(&mut delta);
+    assert_eq!(s.first_timestamp(), Some(1_000_000_000));
+    assert_eq!(s.max_timestamp(), Some(61_000_000_000));
+    assert_eq!(s.retention(), Some(Duration::from_secs(60)));
+
+    // Clear empties the buffer — both timestamps report None.
+    s.clear();
+    assert_eq!(s.first_timestamp(), None);
+    assert_eq!(s.max_timestamp(), None);
+    assert_eq!(s.retention(), None);
+
+    // Reappend, then deallocate — both caches go empty.
+    let mut delta = CanisterLog::new_delta_with_next_index(s.next_idx(), TEST_LOG_MEMORY_LIMIT);
+    delta.add_record(2_000_000_000, b"d".to_vec());
+    s.append_delta_log(&mut delta);
+    assert!(s.retention().is_some());
+    s.deallocate();
+    assert_eq!(s.first_timestamp(), None);
+    assert_eq!(s.max_timestamp(), None);
+    assert_eq!(s.retention(), None);
+}
+
 #[test]
 fn test_appending_to_uninitialized_store_is_no_op() {
     let mut s = LogMemoryStore::new(TEST_LOG_MEMORY_STORE_FEATURE);