ci: prepare scanner cache origin preflight

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,107 @@
-# Copyright 2024 RustFS Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-name: Windows Filesystem Tests
-
+name: Scanner cache read origin preflight
 on:
-  push:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
-  pull_request:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
-
 permissions:
   contents: read
-
 concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
-
+  group: scanner-cache-origin-${{ github.ref }}
+  cancel-in-progress: false
 env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  RUST_BACKTRACE: 1
-
+  CARGO_TERM_COLOR: never
+  CARGO_INCREMENTAL: '0'
+  CARGO_BUILD_JOBS: '1'
+  RUST_BACKTRACE: '0'
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
+  NO_PROXY: 127.0.0.1,localhost
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Check production Windows dependencies
-        shell: pwsh
-        run: cargo check -p rustfs-ecstore --lib
-
-      - name: Test guarded rename publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
-
-      - name: Test multipart parent durability barriers
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
-
-      - name: Test Windows path conversion
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test Windows volume and disk-space probes
-        shell: pwsh
-        run: |
-          # Map a drive letter to the temp drive's administrative share so the
-          # remote-volume probe also covers mapped network drives.
-          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
-          net use R: "\\localhost\$drive`$" /persistent:no
-          if ($LASTEXITCODE -eq 0) {
-            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
-          } else {
-            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
-          }
-          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
-          $status = $LASTEXITCODE
-          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
-          exit $status
-
-      - name: Test Windows handle guards
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
-
-      - name: Test endpoint path parsing
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
-
-      - name: Test startup temporary-directory cleanup
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
-
-      - name: Test object key Windows admission
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
-
-      - name: Test Windows file-name rules
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test fresh format publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
-
-  # The native Windows main thread only gets the 1 MiB PE stack reserve, while
-  # Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
-  # thread that calls `block_on`, so a debug server once overflowed before it
-  # listened on Windows only. Start a real server to keep that covered.
-  startup-smoke:
-    name: Startup Smoke
-    runs-on: windows-latest
+  scoped-tests:
+    runs-on: ubuntu-latest
     timeout-minutes: 90
     steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+      - name: Checkout
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
         with:
+          fetch-depth: 3
           persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
+          clean: false
+      - name: Verify source snapshot
+        shell: bash
+        run: |
+          set -euo pipefail
+          workflow_sha=$(git rev-parse HEAD)
+          source_sha=$(git rev-parse HEAD^)
+          base_sha=$(git rev-parse HEAD^^)
+          source_tree=$(git rev-parse "${source_sha}^{tree}")
+          source_diff=$(git diff --full-index --binary "$base_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          source_files=$(git diff --name-only "$base_sha" "$source_sha")
+          workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
+          expected_files=$(printf '%s\n' 'crates/scanner/src/data_usage_define.rs' 'crates/scanner/src/data_usage_define/persistence.rs' 'crates/scanner/src/data_usage_define/tests.rs' 'crates/scanner/src/data_usage_define/tests/save_metrics.rs')
+          printf 'head=%s source=%s base=%s tree=%s diff=%s\n' "$workflow_sha" "$source_sha" "$base_sha" "$source_tree" "$source_diff"
+          [[ "$workflow_sha" == "$GITHUB_SHA" ]]
+          [[ "$source_sha" == '495e1d3b873cdd2be2f9ffd6eb76e095010832b9' ]]
+          [[ "$base_sha" == '7a3bf8cdbd781116a0970585b995ade668c0e662' ]]
+          [[ "$source_tree" == 'c8bd60069f74095d5de1efaee916e177041a9e2e' ]]
+          [[ "$source_diff" == '0a433216abecfc91416729e77cf5b16b45ffb04400701216b6aaf92321eeaf82' ]]
+          [[ "$source_files" == "$expected_files" ]]
+          [[ "$workflow_files" == '.github/workflows/windows-filesystem.yml' ]]
+          [[ -z "$(git status --porcelain)" ]]
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential pkg-config libssl-dev ripgrep
+      - name: Install protoc
+        uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6
         with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Build debug server binary
-        shell: pwsh
-        run: cargo build --locked -p rustfs --bin rustfs
-
-      - name: Print version
-        shell: pwsh
+          version: '35.1'
+          repo-token: ${{ github.token }}
+      - name: Install flatc
+        uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b
+        with:
+          version: '25.12.19'
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8
+        with:
+          toolchain: stable
+      - name: Run the named tests
+        shell: bash
         run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
-
-      - name: Start server and wait for readiness
-        shell: pwsh
-        env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
-        run: |
-          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
-          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
-          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
-          New-Item -ItemType Directory -Force -Path $volume | Out-Null
-          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
-            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
-            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
-            -NoNewWindow -PassThru
-          # Cache the process handle now; otherwise ExitCode reads back empty.
-          $null = $server.Handle
-          $ready = $false
-          try {
-            for ($i = 0; $i -lt 180; $i++) {
-              if ($server.HasExited) { break }
-              try {
-                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
-                if ($response.StatusCode -eq 200) { $ready = $true; break }
-              } catch {}
-              Start-Sleep -Seconds 1
-            }
-          } finally {
-            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
-            $server.WaitForExit()
-            Write-Host '--- server stdout (tail) ---'
-            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
-            Write-Host '--- server stderr (tail) ---'
-            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
-          }
-          if (-not $ready) {
-            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
-          }
+          set -euo pipefail
+          df -h .
+          output_dir="$RUNNER_TEMP/scanner-cache-read-origin"
+          mkdir -p "$output_dir"
+          tests=(
+            data_usage_define::tests::cache_read_origin_reports_logical_slot_and_legacy_location
+            data_usage_define::tests::save_metrics::read_origin_counts_successful_loads_with_fixed_labels
+            data_usage_define::tests::corrupt_primary_cache_loads_valid_backup
+            data_usage_define::tests::corrupt_cache_without_valid_backup_rebuilds_with_cas_revisions
+            data_usage_define::tests::backup_cache_load_fails_closed_without_main_revision_quorum
+            data_usage_define::tests::recovered_main_cache_wins_over_stale_backup
+            data_usage_define::tests::usage_cache_wire_format_is_pinned
+          )
+          printf '%s\n' "${tests[@]}" > "$output_dir/selection.txt"
+          : > "$output_dir/results.txt"
+          overall=0
+          for test_name in "${tests[@]}"; do
+            log="$output_dir/${test_name//::/_}.log"
+            set +e
+            cargo test --locked -p rustfs-scanner --lib -- --exact "$test_name" > "$log" 2>&1
+            status=$?
+            set -e
+            selected=$(grep -Fxc 'running 1 test' "$log" || true)
+            passed=$(grep -Fxc "test $test_name ... ok" "$log" || true)
+            summary=$(grep -Fc 'test result: ok. 1 passed; 0 failed;' "$log" || true)
+            printf 'test=%s exit=%s selected=%s passed=%s summary=%s\n' "$test_name" "$status" "$selected" "$passed" "$summary" | tee -a "$output_dir/results.txt"
+            tail -n 35 "$log"
+            if [[ "$status" -ne 0 || "$selected" -ne 1 || "$passed" -ne 1 || "$summary" -ne 1 ]]; then
+              overall=1
+              break
+            fi
+          done
+          exit "$overall"
+      - name: Preserve scoped test logs
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
+        with:
+          name: scanner-cache-origin-${{ github.run_id }}
+          path: ${{ runner.temp }}/scanner-cache-read-origin/
+          if-no-files-found: warn

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,