feat(scanner): observe cache read origins

crates/scanner/src/data_usage_define.rs

@@ -67,11 +67,12 @@ const METRIC_CACHE_SAVE_RETRY_TOTAL: &str = "rustfs_scanner_cache_save_retry_tot
 const METRIC_CACHE_SAVE_PAYLOAD_BYTES_TOTAL: &str = "rustfs_scanner_cache_save_payload_bytes_total";
 const METRIC_CACHE_SAVE_DURATION_SECONDS: &str = "rustfs_scanner_cache_save_duration_seconds";
 const METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL: &str = "rustfs_scanner_cache_backup_revision_failure_total";
+const METRIC_CACHE_READ_ORIGIN_TOTAL: &str = "rustfs_scanner_cache_read_origin_total";
 const LOG_COMPONENT_SCANNER: &str = "scanner";
 const LOG_SUBSYSTEM_CACHE: &str = "cache";
 const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state";
 const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state";
-static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
+static CACHE_METRICS_ONCE: Once = Once::new();
 
 pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1;
 pub const DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION: u16 = 1;
@@ -204,6 +205,54 @@ pub(crate) struct DataUsageCacheRevisions {
     backup: Option<DataUsageCacheRevision>,
 }
 
+/// The observed result of trying the logical primary cache slot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DataUsageCachePrimaryRead {
+    Missing,
+    Corrupt,
+}
+
+/// Process-local read provenance, not evidence of freshness or authority.
+/// Legacy objects may come from the old data-usage bucket. Empty may mean
+/// missing or corrupt objects; neither state certifies a complete snapshot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DataUsageCacheReadOrigin {
+    Primary {
+        legacy: bool,
+    },
+    Backup {
+        primary: DataUsageCachePrimaryRead,
+        legacy: bool,
+    },
+    Empty,
+}
+
+impl DataUsageCacheReadOrigin {
+    fn metric_label(self) -> &'static str {
+        match self {
+            Self::Primary { legacy: false } => "primary",
+            Self::Primary { legacy: true } => "legacy_primary",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: false,
+            } => "backup_primary_missing",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: true,
+            } => "legacy_backup_primary_missing",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: false,
+            } => "backup_primary_corrupt",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: true,
+            } => "legacy_backup_primary_corrupt",
+            Self::Empty => "empty",
+        }
+    }
+}
+
 pub static DATA_USAGE_BUCKET: LazyLock<String> =
     LazyLock::new(|| format!("{RUSTFS_META_BUCKET}{SLASH_SEPARATOR}{BUCKET_META_PREFIX}"));
 
@@ -1271,8 +1320,12 @@ impl DataUsageCache {
         Ok(())
     }
 
-    fn ensure_cache_save_metrics_registered() {
-        CACHE_SAVE_METRICS_ONCE.call_once(|| {
+    fn ensure_cache_metrics_registered() {
+        CACHE_METRICS_ONCE.call_once(|| {
+            describe_counter!(
+                METRIC_CACHE_READ_ORIGIN_TOTAL,
+                "Successful scanner cache loads by observed logical origin; does not prove cache freshness."
+            );
             describe_counter!(
                 METRIC_CACHE_SAVE_ATTEMPT_TOTAL,
                 "Total scanner data usage cache save attempts by result and cache type."

crates/scanner/src/data_usage_define/persistence.rs

@@ -70,6 +70,7 @@ struct DataUsageCacheLoadResult {
     cache: DataUsageCache,
     main_revision: DataUsageCacheRevision,
     backup_revision: Option<DataUsageCacheRevision>,
+    origin: DataUsageCacheReadOrigin,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -97,6 +98,7 @@ impl DataUsageCache {
     /// If the object is not found, a nil error with empty data usage cache is returned.
     pub async fn load<S: ScannerObjectIO>(&mut self, store: Arc<S>, name: &str) -> StorageResult<()> {
         let loaded = Self::load_cache(store, name).await?;
+        Self::record_cache_read_origin(loaded.origin);
         *self = loaded.cache;
         Ok(())
     }
@@ -106,6 +108,17 @@ impl DataUsageCache {
         store: Arc<S>,
         name: &str,
     ) -> StorageResult<DataUsageCacheRevisions> {
+        self.load_with_revisions_and_origin(store, name)
+            .await
+            .map(|(revisions, _)| revisions)
+    }
+
+    /// Origin is an observation of this load, not proof of current coverage.
+    pub(crate) async fn load_with_revisions_and_origin<S: ScannerObjectIO>(
+        &mut self,
+        store: Arc<S>,
+        name: &str,
+    ) -> StorageResult<(DataUsageCacheRevisions, DataUsageCacheReadOrigin)> {
         let backup_name = format!("{name}.bkp");
         let backup_path = path_join_buf(&[BUCKET_META_PREFIX, &backup_name]);
         let loaded = Self::load_cache(store.clone(), name).await?;
@@ -131,9 +144,16 @@ impl DataUsageCache {
             },
         };
         let main = loaded.main_revision;
+        let origin = loaded.origin;
+        Self::record_cache_read_origin(origin);
         *self = loaded.cache;
 
-        Ok(DataUsageCacheRevisions { main, backup })
+        Ok((DataUsageCacheRevisions { main, backup }, origin))
+    }
+
+    fn record_cache_read_origin(origin: DataUsageCacheReadOrigin) {
+        Self::ensure_cache_metrics_registered();
+        counter!(METRIC_CACHE_READ_ORIGIN_TOTAL, "origin" => origin.metric_label()).increment(1);
     }
 
     pub(crate) async fn read_revisions<S: ScannerObjectIO>(store: Arc<S>, name: &str) -> StorageResult<DataUsageCacheRevisions> {
@@ -156,10 +176,12 @@ impl DataUsageCache {
                     cache,
                     revision: Some(main_revision),
                 } => {
+                    let legacy = matches!(&main_revision, DataUsageCacheRevision::Missing);
                     return Ok(DataUsageCacheLoadResult {
                         cache: *cache,
                         main_revision,
                         backup_revision: None,
+                        origin: DataUsageCacheReadOrigin::Primary { legacy },
                     });
                 }
                 DataUsageCacheLoadAttempt::Loaded { revision: None, .. } => {
@@ -193,10 +215,17 @@ impl DataUsageCache {
                         let main_revision = main_revision.ok_or_else(|| {
                             Error::other(format!("scanner cache main revision is unavailable while loading backup: {name}"))
                         })?;
+                        let primary = if main_state == DataUsageCacheLoadState::Corrupt {
+                            DataUsageCachePrimaryRead::Corrupt
+                        } else {
+                            DataUsageCachePrimaryRead::Missing
+                        };
+                        let legacy = matches!(&backup_revision, DataUsageCacheRevision::Missing);
                         return Ok(DataUsageCacheLoadResult {
                             cache: *cache,
                             main_revision,
                             backup_revision: Some(backup_revision),
+                            origin: DataUsageCacheReadOrigin::Backup { primary, legacy },
                         });
                     }
                     DataUsageCacheLoadState::Loaded
@@ -220,6 +249,7 @@ impl DataUsageCache {
                         main_revision: main_revision
                             .ok_or_else(|| Error::other(format!("scanner cache missing state has no revision: {name}")))?,
                         backup_revision,
+                        origin: DataUsageCacheReadOrigin::Empty,
                     });
                 }
                 (DataUsageCacheLoadState::Corrupt, DataUsageCacheLoadState::Missing)
@@ -239,6 +269,7 @@ impl DataUsageCache {
                         main_revision: main_revision
                             .ok_or_else(|| Error::other(format!("scanner cache corrupt state has no revision: {name}")))?,
                         backup_revision,
+                        origin: DataUsageCacheReadOrigin::Empty,
                     });
                 }
                 _ => {}
@@ -518,7 +549,7 @@ impl DataUsageCache {
         revision: Option<DataUsageCacheRevision>,
         expected_epoch: Option<u64>,
     ) -> StorageResult<()> {
-        Self::ensure_cache_save_metrics_registered();
+        Self::ensure_cache_metrics_registered();
         let path_type = Self::cache_path_type(path);
         let path = path.to_string();
         let sha256hex = (revision.is_some() && !buf.is_empty())

crates/scanner/src/data_usage_define/tests.rs

@@ -168,8 +168,10 @@ impl AsyncRead for PrefixThenErrorReader {
 #[derive(Debug)]
 struct CacheReadStore {
     main: CacheReadBody,
+    main_missing: bool,
     backup: Option<Vec<u8>>,
     legacy: Option<Vec<u8>>,
+    legacy_backup: Option<Vec<u8>>,
     puts: AtomicUsize,
     #[cfg(feature = "diagnostic-read-version-caller")]
     entry_sources: Mutex<Vec<Option<crate::storage_api::ecstore_diagnostic::ReadVersionEntrySource>>>,
@@ -185,8 +187,10 @@ impl CacheReadStore {
     fn new(main: CacheReadBody, backup: Option<Vec<u8>>) -> Self {
         Self {
             main,
+            main_missing: false,
             backup,
             legacy: None,
+            legacy_backup: None,
             puts: AtomicUsize::new(0),
             #[cfg(feature = "diagnostic-read-version-caller")]
             entry_sources: Mutex::new(Vec::new()),
@@ -239,8 +243,12 @@ impl ObjectIO for CacheReadStore {
             .await
             .push(crate::storage_api::ecstore_diagnostic::current_read_version_entry_source());
         if bucket == DATA_USAGE_BUCKET.as_str() {
-            return self
-                .legacy
+            let legacy = if object.ends_with(".bkp") {
+                &self.legacy_backup
+            } else {
+                &self.legacy
+            };
+            return legacy
                 .clone()
                 .map(|data| Self::reader(CacheReadBody::Bytes(data), "legacy-etag"))
                 .ok_or(Error::FileNotFound);
@@ -252,6 +260,9 @@ impl ObjectIO for CacheReadStore {
         let main_path = path_join_buf(&[BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME]);
         let backup_path = format!("{main_path}.bkp");
         if object == main_path {
+            if self.main_missing {
+                return Err(Error::FileNotFound);
+            }
             return Ok(Self::reader(self.main.clone(), "main-etag"));
         }
         if object == backup_path {
@@ -712,6 +723,143 @@ async fn corrupt_primary_cache_loads_valid_backup() {
     assert_eq!(store.puts.load(Ordering::SeqCst), 0);
 }
 
+#[tokio::test]
+async fn cache_read_origin_reports_logical_slot_and_legacy_location() {
+    let mut primary = DataUsageCache::default();
+    primary.info.name = "primary".to_owned();
+    let mut backup = DataUsageCache::default();
+    backup.info.name = "backup".to_owned();
+    backup.info.snapshot_complete = true;
+    let primary_bytes = primary.marshal_msg().expect("serialize primary");
+    let backup_bytes = backup.marshal_msg().expect("serialize backup");
+    let empty_bytes = DataUsageCache::default().marshal_msg().expect("serialize empty cache");
+    let corrupt = b"not-msgpack".to_vec();
+
+    let cases = [
+        (
+            "primary",
+            Some(primary_bytes.clone()),
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Primary { legacy: false },
+            primary_bytes.clone(),
+        ),
+        (
+            "legacy primary",
+            None,
+            None,
+            Some(primary_bytes.clone()),
+            None,
+            DataUsageCacheReadOrigin::Primary { legacy: true },
+            primary_bytes,
+        ),
+        (
+            "backup after missing",
+            None,
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: false,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "legacy backup after missing",
+            None,
+            None,
+            None,
+            Some(backup_bytes.clone()),
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: true,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "backup after corrupt",
+            Some(corrupt.clone()),
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: false,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "legacy backup after corrupt",
+            Some(corrupt.clone()),
+            None,
+            None,
+            Some(backup_bytes.clone()),
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: true,
+            },
+            backup_bytes,
+        ),
+        (
+            "missing objects",
+            None,
+            None,
+            None,
+            None,
+            DataUsageCacheReadOrigin::Empty,
+            empty_bytes.clone(),
+        ),
+        (
+            "corrupt primary without backup",
+            Some(corrupt),
+            None,
+            None,
+            None,
+            DataUsageCacheReadOrigin::Empty,
+            empty_bytes,
+        ),
+    ];
+
+    for (name, main, current_backup, legacy_primary, legacy_backup, expected_origin, expected_bytes) in cases {
+        let has_main = main.is_some();
+        let has_backup = current_backup.is_some();
+        let mut fixture = CacheReadStore::new(CacheReadBody::Bytes(main.unwrap_or_default()), current_backup);
+        fixture.main_missing = !has_main;
+        fixture.legacy = legacy_primary;
+        fixture.legacy_backup = legacy_backup;
+        let store = Arc::new(fixture);
+        let mut loaded = DataUsageCache::default();
+        let (revisions, origin) = loaded
+            .load_with_revisions_and_origin(store.clone(), DATA_USAGE_CACHE_NAME)
+            .await
+            .expect("fixture cache load");
+
+        assert_eq!(origin, expected_origin, "origin for {name}");
+        assert_eq!(loaded.marshal_msg().expect("re-encode cache"), expected_bytes, "cache body for {name}");
+        assert_eq!(
+            revisions.main,
+            if has_main {
+                DataUsageCacheRevision::Etag("main-etag".to_owned())
+            } else {
+                DataUsageCacheRevision::Missing
+            },
+            "primary CAS revision for {name}"
+        );
+        assert_eq!(
+            revisions.backup,
+            Some(if has_backup {
+                DataUsageCacheRevision::Etag("backup-etag".to_owned())
+            } else {
+                DataUsageCacheRevision::Missing
+            }),
+            "backup CAS revision for {name}"
+        );
+        assert_eq!(store.puts.load(Ordering::SeqCst), 0, "reading must not write for {name}");
+    }
+}
+
 #[tokio::test]
 async fn corrupt_cache_without_valid_backup_rebuilds_with_cas_revisions() {
     for backup in [None, Some(b"also-not-msgpack".to_vec())] {

crates/scanner/src/data_usage_define/tests/save_metrics.rs

@@ -7,17 +7,18 @@ use std::sync::atomic::AtomicU64;
 
 const PAYLOAD_BYTES: &str = "rustfs_scanner_cache_save_payload_bytes_total";
 const ENCODED_BYTES: &str = "rustfs_scanner_cache_serialized_bytes_total";
+const READ_ORIGIN: &str = "rustfs_scanner_cache_read_origin_total";
 
 #[derive(Default)]
-struct SaveRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
+struct CacheRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
 
-impl metrics::Recorder for SaveRecorder {
+impl metrics::Recorder for CacheRecorder {
     fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
     fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
     fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
 
     fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
-        if ![PAYLOAD_BYTES, ENCODED_BYTES].contains(&key.name()) {
+        if ![PAYLOAD_BYTES, ENCODED_BYTES, READ_ORIGIN].contains(&key.name()) {
             return metrics::Counter::noop();
         }
         metrics::Counter::from_arc(self.0.lock().expect("save recorder").entry(key.clone()).or_default().clone())
@@ -36,7 +37,7 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
     let store = Arc::new(store);
     let cache = DataUsageCache::default();
     let encoded = rmp_serde::to_vec(&cache).expect("independent encoded body");
-    let recorder = SaveRecorder::default();
+    let recorder = CacheRecorder::default();
     let save = async {
         match revisions {
             Some(revisions) => {
@@ -85,6 +86,49 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
     );
 }
 
+#[tokio::test(start_paused = true)]
+async fn read_origin_counts_successful_loads_with_fixed_labels() {
+    let bytes = DataUsageCache::default().marshal_msg().expect("serialize backup");
+    let corrupt_primary = CacheReadStore::new(CacheReadBody::Bytes(b"not-msgpack".to_vec()), Some(bytes.clone()));
+    let mut legacy_backup = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
+    legacy_backup.main_missing = true;
+    legacy_backup.legacy_backup = Some(bytes);
+    let mut empty = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
+    empty.main_missing = true;
+    let recorder = CacheRecorder::default();
+
+    for (store, with_revisions) in [(corrupt_primary, false), (legacy_backup, true), (empty, true)] {
+        let store = Arc::new(store);
+        let mut loaded = DataUsageCache::default();
+        let read = async {
+            if with_revisions {
+                loaded.load_with_revisions(store, DATA_USAGE_CACHE_NAME).await.map(|_| ())
+            } else {
+                loaded.load(store, DATA_USAGE_CACHE_NAME).await
+            }
+        };
+        let mut read = std::pin::pin!(read);
+        std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || read.as_mut().poll(cx)))
+            .await
+            .expect("fixture load");
+    }
+
+    let unavailable = Arc::new(BackupFallbackStore::new(Vec::new(), false));
+    let mut loaded = DataUsageCache::default();
+    let mut failed_read = std::pin::pin!(loaded.load_with_revisions(unavailable, DATA_USAGE_CACHE_NAME));
+    let error = std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || failed_read.as_mut().poll(cx)))
+        .await
+        .expect_err("an unreadable primary must not count as a fallback");
+    assert!(matches!(error, Error::ErasureReadQuorum));
+
+    let rows = recorder.0.lock().expect("cache recorder");
+    assert_eq!(rows.len(), 3, "one fixed origin label per successful load");
+    for origin in ["backup_primary_corrupt", "legacy_backup_primary_missing", "empty"] {
+        let key = metrics::Key::from_parts(READ_ORIGIN, vec![metrics::Label::new("origin", origin)]);
+        assert_eq!(rows[&key].load(Ordering::Relaxed), 1, "one observation for {origin}");
+    }
+}
+
 fn revisions(backup: bool) -> DataUsageCacheRevisions {
     DataUsageCacheRevisions {
         main: DataUsageCacheRevision::Missing,