feat(scanner): expose checkpoint and source work status (#3230)

* feat(scanner): expose checkpoint and source work status

* fix(scanner): count ignored checkpoints once per scan

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

crates/common/src/metrics.rs

@@ -228,6 +228,92 @@ impl Metric {
     }
 }
 
+const SCANNER_CHECKPOINT_EVENT_SET: &str = "set";
+const SCANNER_CHECKPOINT_EVENT_USED: &str = "used";
+const SCANNER_CHECKPOINT_EVENT_IGNORED: &str = "ignored";
+const SCANNER_CHECKPOINT_EVENT_STALE: &str = "stale";
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ScannerWorkSource {
+    Usage,
+    Lifecycle,
+    BucketReplication,
+    SiteReplication,
+    Heal,
+    Bitrot,
+    Alerts,
+}
+
+impl ScannerWorkSource {
+    const ALL: [Self; 7] = [
+        Self::Usage,
+        Self::Lifecycle,
+        Self::BucketReplication,
+        Self::SiteReplication,
+        Self::Heal,
+        Self::Bitrot,
+        Self::Alerts,
+    ];
+
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Usage => "usage",
+            Self::Lifecycle => "lifecycle",
+            Self::BucketReplication => "bucket_replication",
+            Self::SiteReplication => "site_replication",
+            Self::Heal => "heal",
+            Self::Bitrot => "bitrot",
+            Self::Alerts => "alerts",
+        }
+    }
+
+    fn all() -> &'static [Self] {
+        &Self::ALL
+    }
+
+    fn index(self) -> usize {
+        match self {
+            Self::Usage => 0,
+            Self::Lifecycle => 1,
+            Self::BucketReplication => 2,
+            Self::SiteReplication => 3,
+            Self::Heal => 4,
+            Self::Bitrot => 5,
+            Self::Alerts => 6,
+        }
+    }
+}
+
+#[derive(Debug, Default)]
+struct ScannerSourceWorkCounters {
+    checked: AtomicU64,
+    queued: AtomicU64,
+    executed: AtomicU64,
+    failed: AtomicU64,
+    skipped: AtomicU64,
+}
+
+impl ScannerSourceWorkCounters {
+    fn add(&self, checked: u64, queued: u64, executed: u64, failed: u64, skipped: u64) {
+        self.checked.fetch_add(checked, Ordering::Relaxed);
+        self.queued.fetch_add(queued, Ordering::Relaxed);
+        self.executed.fetch_add(executed, Ordering::Relaxed);
+        self.failed.fetch_add(failed, Ordering::Relaxed);
+        self.skipped.fetch_add(skipped, Ordering::Relaxed);
+    }
+
+    fn snapshot(&self, source: ScannerWorkSource) -> ScannerSourceWorkSnapshot {
+        ScannerSourceWorkSnapshot {
+            source: source.as_str().to_string(),
+            checked: self.checked.load(Ordering::Relaxed),
+            queued: self.queued.load(Ordering::Relaxed),
+            executed: self.executed.load(Ordering::Relaxed),
+            failed: self.failed.load(Ordering::Relaxed),
+            skipped: self.skipped.load(Ordering::Relaxed),
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // LockedLastMinuteLatency
 // ---------------------------------------------------------------------------
@@ -423,6 +509,12 @@ pub struct Metrics {
     scanner_cycle_max_directories: AtomicU64,
     scanner_bitrot_cycle_enabled: AtomicBool,
     scanner_bitrot_cycle_millis: AtomicU64,
+    scanner_checkpoint: Mutex<Option<ScannerCheckpointReport>>,
+    scanner_checkpoint_used: AtomicU64,
+    scanner_checkpoint_cleared: AtomicU64,
+    scanner_checkpoint_ignored: AtomicU64,
+    scanner_checkpoint_stale: AtomicU64,
+    scanner_source_work: Vec<ScannerSourceWorkCounters>,
     partial_scan_cycles: AtomicU64,
 }
 
@@ -495,6 +587,24 @@ pub struct ScannerTimedAction {
     pub bytes: u64,
 }
 
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ScannerCheckpointReport {
+    pub version: u16,
+    pub resume_after: String,
+    pub reason: String,
+    pub last_event: String,
+}
+
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ScannerSourceWorkSnapshot {
+    pub source: String,
+    pub checked: u64,
+    pub queued: u64,
+    pub executed: u64,
+    pub failed: u64,
+    pub skipped: u64,
+}
+
 #[derive(Clone, Debug, Default, Serialize, Deserialize)]
 pub struct ScannerLastMinute {
     pub actions: HashMap<String, ScannerTimedAction>,
@@ -614,6 +724,19 @@ pub struct ScannerMetricsReport {
     #[serde(default)]
     pub bitrot_cycle_seconds: f64,
     #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub scan_checkpoint: Option<ScannerCheckpointReport>,
+    #[serde(default)]
+    pub scan_checkpoint_used: u64,
+    #[serde(default)]
+    pub scan_checkpoint_cleared: u64,
+    #[serde(default)]
+    pub scan_checkpoint_ignored: u64,
+    #[serde(default)]
+    pub scan_checkpoint_stale: u64,
+    #[serde(default)]
+    pub source_work: Vec<ScannerSourceWorkSnapshot>,
+    #[serde(default)]
     pub partial_cycles: u64,
 }
 
@@ -661,6 +784,10 @@ fn duration_millis_saturated(duration: Duration) -> u64 {
     duration.as_millis().min(u64::MAX as u128) as u64
 }
 
+fn usize_to_u64_saturated(value: usize) -> u64 {
+    u64::try_from(value).unwrap_or(u64::MAX)
+}
+
 fn scaled_f64_to_u64_saturated(value: f64, scale: f64) -> u64 {
     if !value.is_finite() || value <= 0.0 {
         return 0;
@@ -786,6 +913,15 @@ impl Metrics {
             scanner_cycle_max_directories: AtomicU64::new(0),
             scanner_bitrot_cycle_enabled: AtomicBool::new(false),
             scanner_bitrot_cycle_millis: AtomicU64::new(0),
+            scanner_checkpoint: Mutex::new(None),
+            scanner_checkpoint_used: AtomicU64::new(0),
+            scanner_checkpoint_cleared: AtomicU64::new(0),
+            scanner_checkpoint_ignored: AtomicU64::new(0),
+            scanner_checkpoint_stale: AtomicU64::new(0),
+            scanner_source_work: ScannerWorkSource::all()
+                .iter()
+                .map(|_| ScannerSourceWorkCounters::default())
+                .collect(),
             partial_scan_cycles: AtomicU64::new(0),
         }
     }
@@ -806,6 +942,7 @@ impl Metrics {
         move |_custom: &HashMap<String, String>| {
             let duration = SystemTime::now().duration_since(start).unwrap_or_default();
             global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed);
+            global_metrics().record_source_work_for_metric(metric, 1);
             emit_otel_counter(metric_idx, 1);
             if metric_idx < Metric::LastRealtime as usize {
                 global_metrics().latency[metric_idx].add(duration);
@@ -821,6 +958,7 @@ impl Metrics {
         move |size: u64| {
             let duration = SystemTime::now().duration_since(start).unwrap_or_default();
             global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed);
+            global_metrics().record_source_work_for_metric(metric, 1);
             emit_otel_counter(metric_idx, 1);
             if metric_idx < Metric::LastRealtime as usize {
                 global_metrics().latency[metric_idx].add_size(duration, size);
@@ -836,6 +974,7 @@ impl Metrics {
         move || {
             let duration = SystemTime::now().duration_since(start).unwrap_or_default();
             global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed);
+            global_metrics().record_source_work_for_metric(metric, 1);
             emit_otel_counter(metric_idx, 1);
             if metric_idx < Metric::LastRealtime as usize {
                 global_metrics().latency[metric_idx].add(duration);
@@ -851,8 +990,10 @@ impl Metrics {
         Box::new(move |count: usize| {
             Box::new(move || {
                 let duration = SystemTime::now().duration_since(start).unwrap_or_default();
-                global_metrics().operations[metric_idx].fetch_add(count as u64, Ordering::Relaxed);
-                emit_otel_counter(metric_idx, count as u64);
+                let count = usize_to_u64_saturated(count);
+                global_metrics().operations[metric_idx].fetch_add(count, Ordering::Relaxed);
+                global_metrics().record_source_work_for_metric(metric, count);
+                emit_otel_counter(metric_idx, count);
                 if metric_idx < Metric::LastRealtime as usize {
                     global_metrics().latency[metric_idx].add(duration);
                 }
@@ -885,6 +1026,7 @@ impl Metrics {
     pub fn inc_time(metric: Metric, duration: Duration) {
         let metric_idx = metric as usize;
         global_metrics().operations[metric_idx].fetch_add(1, Ordering::Relaxed);
+        global_metrics().record_source_work_for_metric(metric, 1);
         emit_otel_counter(metric_idx, 1);
         if metric_idx < Metric::LastRealtime as usize {
             global_metrics().latency[metric_idx].add(duration);
@@ -921,6 +1063,94 @@ impl Metrics {
 
     pub fn record_scanner_ilm_action(&self, count: u64) {
         self.scanner_ilm_actions.fetch_add(count, Ordering::Relaxed);
+        self.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 0, count, 0, 0);
+    }
+
+    pub fn record_scanner_checkpoint_set(&self, version: u16, resume_after: impl Into<String>, reason: impl Into<String>) {
+        let checkpoint = ScannerCheckpointReport {
+            version,
+            resume_after: resume_after.into(),
+            reason: reason.into(),
+            last_event: SCANNER_CHECKPOINT_EVENT_SET.to_string(),
+        };
+        match self.scanner_checkpoint.lock() {
+            Ok(mut current) => *current = Some(checkpoint),
+            Err(poisoned) => *poisoned.into_inner() = Some(checkpoint),
+        }
+    }
+
+    pub fn record_scanner_checkpoint_used(&self) {
+        self.scanner_checkpoint_used.fetch_add(1, Ordering::Relaxed);
+        self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_USED);
+    }
+
+    pub fn record_scanner_checkpoint_cleared(&self) {
+        self.scanner_checkpoint_cleared.fetch_add(1, Ordering::Relaxed);
+        match self.scanner_checkpoint.lock() {
+            Ok(mut current) => *current = None,
+            Err(poisoned) => *poisoned.into_inner() = None,
+        }
+    }
+
+    pub fn record_scanner_checkpoint_ignored(&self) {
+        self.scanner_checkpoint_ignored.fetch_add(1, Ordering::Relaxed);
+        self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_IGNORED);
+    }
+
+    pub fn record_scanner_checkpoint_stale(&self) {
+        self.scanner_checkpoint_stale.fetch_add(1, Ordering::Relaxed);
+        self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_STALE);
+    }
+
+    pub fn record_scanner_source_work(
+        &self,
+        source: ScannerWorkSource,
+        checked: u64,
+        queued: u64,
+        executed: u64,
+        failed: u64,
+        skipped: u64,
+    ) {
+        if let Some(counters) = self.scanner_source_work.get(source.index()) {
+            counters.add(checked, queued, executed, failed, skipped);
+        }
+    }
+
+    fn update_scanner_checkpoint_event(&self, event: &str) {
+        match self.scanner_checkpoint.lock() {
+            Ok(mut current) => {
+                if let Some(checkpoint) = current.as_mut() {
+                    checkpoint.last_event = event.to_string();
+                }
+            }
+            Err(poisoned) => {
+                let mut current = poisoned.into_inner();
+                if let Some(checkpoint) = current.as_mut() {
+                    checkpoint.last_event = event.to_string();
+                }
+            }
+        }
+    }
+
+    fn record_source_work_for_metric(&self, metric: Metric, count: u64) {
+        match metric {
+            Metric::ScanObject | Metric::ScanFolder => {
+                self.record_scanner_source_work(ScannerWorkSource::Usage, count, 0, 0, 0, 0);
+            }
+            Metric::SaveUsage => {
+                self.record_scanner_source_work(ScannerWorkSource::Usage, 0, 0, count, 0, 0);
+            }
+            Metric::CheckReplication => {
+                self.record_scanner_source_work(ScannerWorkSource::BucketReplication, count, 0, 0, 0, 0);
+            }
+            Metric::HealCheck => {
+                self.record_scanner_source_work(ScannerWorkSource::Heal, count, 0, 0, 0, 0);
+            }
+            Metric::HealAbandonedObject => {
+                self.record_scanner_source_work(ScannerWorkSource::Heal, 0, 0, count, 0, 0);
+            }
+            _ => {}
+        }
     }
 
     pub fn record_scan_bucket_drive_start(&self) {
@@ -1327,6 +1557,22 @@ impl Metrics {
         m.cycle_max_directories = self.scanner_cycle_max_directories.load(Ordering::Relaxed);
         m.bitrot_cycle_enabled = self.scanner_bitrot_cycle_enabled.load(Ordering::Relaxed);
         m.bitrot_cycle_seconds = self.scanner_bitrot_cycle_millis.load(Ordering::Relaxed) as f64 / 1000.0;
+        m.scan_checkpoint = match self.scanner_checkpoint.lock() {
+            Ok(checkpoint) => checkpoint.clone(),
+            Err(poisoned) => poisoned.into_inner().clone(),
+        };
+        m.scan_checkpoint_used = self.scanner_checkpoint_used.load(Ordering::Relaxed);
+        m.scan_checkpoint_cleared = self.scanner_checkpoint_cleared.load(Ordering::Relaxed);
+        m.scan_checkpoint_ignored = self.scanner_checkpoint_ignored.load(Ordering::Relaxed);
+        m.scan_checkpoint_stale = self.scanner_checkpoint_stale.load(Ordering::Relaxed);
+        m.source_work = ScannerWorkSource::all()
+            .iter()
+            .filter_map(|source| {
+                self.scanner_source_work
+                    .get(source.index())
+                    .map(|counters| counters.snapshot(*source))
+            })
+            .collect();
         m.partial_cycles = self.partial_scan_cycles.load(Ordering::Relaxed);
 
         // Lifetime operation counts
@@ -1539,6 +1785,94 @@ mod tests {
         assert_eq!(report.current_disk_bucket_scans_active, 0);
     }
 
+    #[tokio::test]
+    async fn report_includes_scanner_checkpoint_status() {
+        let metrics = Metrics::new();
+        metrics.record_scanner_checkpoint_set(1, "bucket/child-a", "directories");
+        metrics.record_scanner_checkpoint_used();
+
+        let report = metrics.report().await;
+
+        let checkpoint = report.scan_checkpoint.as_ref().expect("scanner checkpoint should be visible");
+        assert_eq!(checkpoint.version, 1);
+        assert_eq!(checkpoint.resume_after, "bucket/child-a");
+        assert_eq!(checkpoint.reason, "directories");
+        assert_eq!(checkpoint.last_event, "used");
+        assert_eq!(report.scan_checkpoint_used, 1);
+        assert_eq!(report.scan_checkpoint_cleared, 0);
+
+        metrics.record_scanner_checkpoint_stale();
+        metrics.record_scanner_checkpoint_cleared();
+        let report = metrics.report().await;
+
+        assert!(report.scan_checkpoint.is_none());
+        assert_eq!(report.scan_checkpoint_used, 1);
+        assert_eq!(report.scan_checkpoint_stale, 1);
+        assert_eq!(report.scan_checkpoint_cleared, 1);
+    }
+
+    #[tokio::test]
+    async fn report_includes_scanner_source_work() {
+        let metrics = Metrics::new();
+        metrics.record_scanner_source_work(ScannerWorkSource::Usage, 3, 0, 1, 0, 0);
+        metrics.record_scanner_source_work(ScannerWorkSource::Lifecycle, 0, 2, 1, 1, 0);
+
+        let report = metrics.report().await;
+
+        let usage = report
+            .source_work
+            .iter()
+            .find(|work| work.source == ScannerWorkSource::Usage.as_str())
+            .expect("usage source work should be visible");
+        assert_eq!(usage.checked, 3);
+        assert_eq!(usage.executed, 1);
+
+        let lifecycle = report
+            .source_work
+            .iter()
+            .find(|work| work.source == ScannerWorkSource::Lifecycle.as_str())
+            .expect("lifecycle source work should be visible");
+        assert_eq!(lifecycle.queued, 2);
+        assert_eq!(lifecycle.executed, 1);
+        assert_eq!(lifecycle.failed, 1);
+    }
+
+    #[tokio::test]
+    async fn scanner_metrics_update_source_work() {
+        let metrics = Metrics::new();
+        metrics.record_source_work_for_metric(Metric::ScanObject, 3);
+        metrics.record_source_work_for_metric(Metric::ScanFolder, 2);
+        metrics.record_source_work_for_metric(Metric::SaveUsage, 1);
+        metrics.record_source_work_for_metric(Metric::CheckReplication, 4);
+        metrics.record_source_work_for_metric(Metric::HealCheck, 5);
+        metrics.record_source_work_for_metric(Metric::HealAbandonedObject, 6);
+
+        let report = metrics.report().await;
+
+        let usage = report
+            .source_work
+            .iter()
+            .find(|work| work.source == ScannerWorkSource::Usage.as_str())
+            .expect("usage source work should be visible");
+        assert_eq!(usage.checked, 5);
+        assert_eq!(usage.executed, 1);
+
+        let bucket_replication = report
+            .source_work
+            .iter()
+            .find(|work| work.source == ScannerWorkSource::BucketReplication.as_str())
+            .expect("bucket replication source work should be visible");
+        assert_eq!(bucket_replication.checked, 4);
+
+        let heal = report
+            .source_work
+            .iter()
+            .find(|work| work.source == ScannerWorkSource::Heal.as_str())
+            .expect("heal source work should be visible");
+        assert_eq!(heal.checked, 5);
+        assert_eq!(heal.executed, 6);
+    }
+
     #[tokio::test]
     async fn report_preserves_current_cycle_started_time() {
         let previous_init_time = *crate::globals::GLOBAL_INIT_TIME.read().await;