feat(scanner): expose authenticated dirty bucket snapshots (#7122)

* feat(scanner): add peer bucket dirty snapshots

* fix(scanner): keep dirty snapshot errors stable

* test(protos): satisfy dirty snapshot clippy

* fix(scanner): satisfy dirty snapshot clippy

---------

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

crates/scanner/src/lib.rs

@@ -89,14 +89,17 @@ pub use scanner::{
     scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_topology_digest,
 };
 pub use scanner_io::{
-    ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket,
-    record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state,
-    scanner_maintenance_generation,
+    ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,
+    acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change,
+    scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation,
 };
 pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
 use std::sync::atomic::{AtomicU64, Ordering};
 pub use storage_api::ScannerReplicationConfig as ReplicationConfig;
-pub use storage_api::scan::{SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION};
+pub use storage_api::scan::{
+    SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES,
+    SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
+};
 pub use workload_admission::set_scanner_workload_admission_snapshot_provider;
 
 static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0);

crates/scanner/src/scanner_io.rs

@@ -748,9 +748,9 @@ pub(crate) use cache::{
     current_cache_root_or_prepare_with_generation,
 };
 pub use dirty_usage::{
-    ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket,
-    record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state,
-    scanner_maintenance_generation,
+    ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageSnapshot, ScannerDirtyUsageState,
+    acknowledge_dirty_usage_generation, clear_dirty_usage_bucket, record_dirty_usage_bucket, record_scanner_maintenance_change,
+    scanner_activity_epoch, scanner_dirty_usage_snapshot, scanner_dirty_usage_state, scanner_maintenance_generation,
 };
 #[cfg(test)]
 pub(crate) use dirty_usage::{clear_dirty_usage_buckets_for_tests, dirty_usage_buckets_for_tests};

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -27,6 +27,25 @@ pub struct ScannerDirtyUsageState {
     pub pending: bool,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ScannerDirtyUsageBucket {
+    pub bucket: String,
+    pub generation: u64,
+}
+
+/// A point-in-time view of the local dirty bucket generations.
+///
+/// `complete == false` is an all-or-nothing overflow signal: `buckets` is
+/// empty and callers must fall back to the global dirty generation rather than
+/// treating a bounded prefix as authoritative.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ScannerDirtyUsageSnapshot {
+    pub generation: u64,
+    pub pending_bucket_count: u64,
+    pub complete: bool,
+    pub buckets: Vec<ScannerDirtyUsageBucket>,
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
 pub enum ScannerDirtyUsageAckError {
     #[error("scanner process instance changed before dirty usage acknowledgement")]
@@ -98,6 +117,35 @@ pub fn scanner_dirty_usage_state() -> ScannerDirtyUsageState {
     }
 }
 
+pub fn scanner_dirty_usage_snapshot(max_entries: usize) -> ScannerDirtyUsageSnapshot {
+    let (generation, pending_bucket_count, complete, mut buckets) = {
+        let dirty_buckets = dirty_usage_buckets();
+        let generation = DIRTY_USAGE_BUCKET_GENERATION.load(Ordering::Acquire);
+        let pending_bucket_count = usize_to_u64_saturated(dirty_buckets.len());
+        let complete = dirty_buckets.len() <= max_entries;
+        let buckets = if complete {
+            dirty_buckets
+                .iter()
+                .map(|(bucket, generation)| ScannerDirtyUsageBucket {
+                    bucket: bucket.clone(),
+                    generation: *generation,
+                })
+                .collect::<Vec<_>>()
+        } else {
+            Vec::new()
+        };
+        (generation, pending_bucket_count, complete, buckets)
+    };
+    buckets.sort_unstable_by(|left, right| left.bucket.cmp(&right.bucket));
+
+    ScannerDirtyUsageSnapshot {
+        generation,
+        pending_bucket_count,
+        complete,
+        buckets,
+    }
+}
+
 pub fn acknowledge_dirty_usage_generation(
     instance_id: &str,
     generation: u64,

crates/scanner/src/scanner_io/tests.rs

@@ -500,6 +500,51 @@ fn dirty_usage_generation_acknowledgement_preserves_newer_mutations() {
     clear_dirty_usage_buckets_for_tests();
 }
 
+#[test]
+#[serial]
+fn dirty_usage_snapshot_is_sorted_and_reports_its_cutoff() {
+    clear_dirty_usage_buckets_for_tests();
+    let empty = scanner_dirty_usage_snapshot(0);
+    assert_eq!(empty.pending_bucket_count, 0);
+    assert!(empty.complete);
+    assert!(empty.buckets.is_empty());
+
+    record_dirty_usage_bucket("videos");
+    record_dirty_usage_bucket("photos");
+    let expected_generation = scanner_dirty_usage_state().generation;
+
+    let snapshot = scanner_dirty_usage_snapshot(2);
+
+    assert_eq!(snapshot.generation, expected_generation);
+    assert_eq!(snapshot.pending_bucket_count, 2);
+    assert!(snapshot.complete);
+    assert_eq!(
+        snapshot
+            .buckets
+            .iter()
+            .map(|bucket| bucket.bucket.as_str())
+            .collect::<Vec<_>>(),
+        vec!["photos", "videos"]
+    );
+    assert!(snapshot.buckets.iter().all(|bucket| bucket.generation <= snapshot.generation));
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[test]
+#[serial]
+fn dirty_usage_snapshot_marks_truncated_results_incomplete() {
+    clear_dirty_usage_buckets_for_tests();
+    record_dirty_usage_bucket("archive");
+    record_dirty_usage_bucket("photos");
+
+    let snapshot = scanner_dirty_usage_snapshot(1);
+
+    assert_eq!(snapshot.pending_bucket_count, 2);
+    assert!(!snapshot.complete);
+    assert!(snapshot.buckets.is_empty(), "incomplete snapshots must not expose a partial bucket list");
+    clear_dirty_usage_buckets_for_tests();
+}
+
 #[test]
 #[serial]
 fn dirty_usage_generation_acknowledgement_rejects_stale_process_and_future_generation() {

crates/scanner/src/storage_api.rs

@@ -304,7 +304,10 @@ pub(crate) mod scan {
     };
     #[cfg(test)]
     pub(crate) use super::storage_contracts::{DeleteBucketOptions, MakeBucketOptions, ObjectIO};
-    pub use super::storage_contracts::{SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION};
+    pub use super::storage_contracts::{
+        SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES,
+        SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
+    };
 }
 
 pub(crate) mod scanner_io {