ci: run the stale backup regression

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,90 @@
-# 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 stale backup regression
 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-stale-backup-red-${{ 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-red-test:
+    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/scanner_folder/tests/checkpoint_fixture.rs' 'crates/scanner/src/scanner_folder/tests/erasure_probe.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" == '980a6b9770db6d5a10fbf52abbbd8c66db887ae9' ]]
+          [[ "$base_sha" == 'b4026b43b215a5d1f91c3e079e2be2f1bb8967f9' ]]
+          [[ "$source_tree" == '3879c0e6cefda7b9d48a7769ddd33774ab4bf6ec' ]]
+          [[ "$source_diff" == 'eac18e2b3901a750985f6165c68400bd5c9026404e1520f9dd489c3bc94483a9' ]]
+          [[ "$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 regression
+        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 .
+          test_name=scanner_folder::tests::checkpoint_fixture::corrupt_primary_cannot_promote_older_complete_backup_after_partial
+          log="$RUNNER_TEMP/scanner-stale-backup-red.log"
+          set +e
+          cargo test --locked -p rustfs-scanner --lib -- --exact "$test_name" >"$log" 2>&1
+          test_status=$?
+          set -e
+          tail -n 120 "$log"
+          [[ "$test_status" -eq 101 ]]
+          grep -Fxq -- 'running 1 test' "$log"
+          grep -Fxq -- "test $test_name ... FAILED" "$log"
+          grep -Fq -- 'checkpoint_fixture.rs:1167:' "$log"
+          grep -Fq -- 'an older backup must not certify an unverified failed subtree: snapshot_complete=true, failed_prefix_observed=false' "$log"
+          grep -Fq -- 'test result: FAILED. 0 passed; 1 failed;' "$log"
+          printf 'Confirmed the expected failure of the named test\n'
+      - name: Preserve test output
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
+        with:
+          name: scanner-stale-backup-red-${{ github.run_id }}
+          path: ${{ runner.temp }}/scanner-stale-backup-red.log
+          if-no-files-found: warn

crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs

@@ -49,6 +49,7 @@ struct FixtureStore {
     root: std::path::PathBuf,
     _owned_root: Option<tempfile::TempDir>,
     reject_save: AtomicBool,
+    reject_backup_read: AtomicBool,
 }
 
 impl FixtureStore {
@@ -58,6 +59,7 @@ impl FixtureStore {
             root: root.path().to_path_buf(),
             _owned_root: Some(root),
             reject_save: AtomicBool::new(false),
+            reject_backup_read: AtomicBool::new(false),
         })
     }
 
@@ -67,6 +69,7 @@ impl FixtureStore {
             root,
             _owned_root: None,
             reject_save: AtomicBool::new(false),
+            reject_backup_read: AtomicBool::new(false),
         })
     }
 
@@ -101,6 +104,9 @@ impl ObjectIO for FixtureStore {
         _headers: Self::HeaderMap,
         _options: &Self::ObjectOptions,
     ) -> crate::EcstoreResult<Self::GetObjectReader> {
+        if object.ends_with(".bkp") && self.reject_backup_read.load(Ordering::SeqCst) {
+            return Err(crate::EcstoreError::ErasureReadQuorum);
+        }
         let bytes = tokio::fs::read(self.path(object)).await.map_err(|error| {
             if error.kind() == std::io::ErrorKind::NotFound {
                 crate::EcstoreError::FileNotFound
@@ -889,6 +895,282 @@ async fn check_complete_sampling_resumption(resume_mode: HealScanMode) {
     panic!("sampling interruption must recover with the unchanged three-object budget");
 }
 
+#[tokio::test]
+#[serial]
+async fn corrupt_primary_cannot_promote_older_complete_backup_after_partial() {
+    temp_env::async_with_vars([(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, Some("16"))], async {
+        assert_eq!(data_usage_update_dir_cycles(), 16);
+        assert_corrupt_primary_cannot_promote_older_backup().await;
+    })
+    .await;
+}
+
+async fn assert_corrupt_primary_cannot_promote_older_backup() {
+    let (scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for index in 0..9 {
+        write_checkpoint_object(&root, &format!("prefix/{index:04}"), &[(None, 1)]).await;
+    }
+    write_checkpoint_object(&root, "z-good/healthy", &[(None, 1)]).await;
+    let _ordered = super::erasure_probe::OrderedDirectory::new(&root.join("bucket"));
+    let identity = crate::DataUsageScanIdentity {
+        tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
+        ..bound_checkpoint().1
+    };
+    let mut cache = DataUsageCache::default();
+    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+    cache.info.skip_healing = true;
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&parent, Default::default());
+    let baseline = scanner
+        .local_disk
+        .clone()
+        .nsscanner_disk(
+            budget.token(),
+            budget,
+            vec![scanner.local_disk.clone()],
+            cache,
+            None,
+            scan_options(HealScanMode::Normal),
+        )
+        .await
+        .expect("scan complete baseline");
+    let ScannerDiskScanOutcome::Complete(baseline) = baseline else {
+        panic!("baseline scan must be complete")
+    };
+    let prefix = hash_path("bucket/prefix");
+    assert!(baseline.is_compacted(&prefix), "fixture needs an older compacted subtree");
+
+    let store = FixtureStore::new();
+    let mut loaded = DataUsageCache::default();
+    let revisions = loaded
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("load initial cache revisions");
+    baseline
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("save complete baseline to both copies");
+    let backup = tokio::fs::read(store.root.join("backup"))
+        .await
+        .expect("read baseline backup");
+    assert!(
+        DataUsageCache::unmarshal(&backup)
+            .expect("decode baseline backup")
+            .info
+            .snapshot_complete
+    );
+
+    let sampling_cycles = data_usage_update_dir_cycles();
+    let sampling_cycle = (12..12 + sampling_cycles)
+        .find(|cycle| prefix.mod_(*cycle, sampling_cycles))
+        .expect("one prefix sampling cycle per rotation");
+    let recovery_cycle = (sampling_cycle + 1..sampling_cycle + 1 + sampling_cycles)
+        .find(|cycle| !prefix.mod_(*cycle, sampling_cycles))
+        .expect("another cycle skips the compacted prefix");
+    let mut partial_input = baseline.clone();
+    partial_input.prepare_bucket_checkpoint("bucket", u64::from(sampling_cycle), 7, SOURCE, PLAN, identity);
+    assert!(partial_input.info.scan_progress.is_none());
+    partial_input.info.skip_healing = true;
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        ScannerCycleBudgetConfig {
+            max_objects: Some(3),
+            ..Default::default()
+        },
+    );
+    let partial = scanner
+        .local_disk
+        .clone()
+        .nsscanner_disk(
+            budget.token(),
+            budget,
+            vec![scanner.local_disk.clone()],
+            partial_input,
+            None,
+            scan_options(HealScanMode::Normal),
+        )
+        .await
+        .expect("scan interrupted by object budget");
+    let ScannerDiskScanOutcome::Partial(partial) = partial else {
+        panic!("the newer sampled scan must stop as partial")
+    };
+    assert!(!partial.info.snapshot_complete);
+    assert!(partial.info.scan_checkpoint.is_some());
+
+    store.reject_backup_read.store(true, Ordering::SeqCst);
+    let revisions = loaded
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("load primary despite unavailable backup revision");
+    partial
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("persist newer partial while backup revision is unavailable");
+    store.reject_backup_read.store(false, Ordering::SeqCst);
+    let partial_bytes = tokio::fs::read(store.root.join("main")).await.expect("read newer partial");
+    let persisted_partial = DataUsageCache::unmarshal(&partial_bytes).expect("decode newer partial");
+    assert!(!persisted_partial.info.snapshot_complete);
+    assert_eq!(persisted_partial.info.next_cycle, u64::from(sampling_cycle));
+    assert_eq!(tokio::fs::read(store.root.join("backup")).await.expect("read older backup"), backup);
+
+    let mut latest = DataUsageCache::default();
+    store.reject_backup_read.store(true, Ordering::SeqCst);
+    let revisions = latest
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("reload newer partial before corruption");
+    store.reject_backup_read.store(false, Ordering::SeqCst);
+    let options = crate::scanner_io::DataUsageCacheReuseOptions {
+        require_source: true,
+        tier_registry_generation: Some(identity.tier_registry_generation),
+        checkpoint_identity: Some(identity),
+        checkpoint_plan_digest: None,
+    };
+    let latest_state = crate::scanner_io::current_cache_root_or_prepare_with_generation(
+        &mut latest,
+        "bucket",
+        SOURCE,
+        u64::from(recovery_cycle),
+        7,
+        PLAN,
+        options,
+    );
+    assert!(matches!(latest_state, crate::scanner_io::DataUsageCacheScanState::Prepared { .. }));
+    assert!(latest.info.scan_progress.is_some(), "the persisted partial requires a coverage sweep");
+    latest.info.skip_healing = true;
+    let _fault = super::erasure_probe::OrderedDirectory::fail_open(&root.join("bucket/prefix"), ErrorKind::PermissionDenied);
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&parent, Default::default());
+    let control_logs = CapturedLogs::default();
+    let control = control_logs
+        .capture(
+            tracing::Level::DEBUG,
+            scanner.local_disk.clone().nsscanner_disk(
+                budget.token(),
+                budget,
+                vec![scanner.local_disk.clone()],
+                latest,
+                None,
+                scan_options(HealScanMode::Normal),
+            ),
+        )
+        .await
+        .expect("scan the intact newer partial");
+    let ScannerDiskScanOutcome::Partial(control) = control else {
+        panic!("the intact partial must not complete across the injected directory failure")
+    };
+    assert!(!control.info.snapshot_complete);
+    assert_eq!(
+        control
+            .checked_flatten("bucket/z-good")
+            .expect("healthy sibling scanned")
+            .objects,
+        1
+    );
+    assert!(control_logs.contents().lines().any(|line| {
+        let event: serde_json::Value = serde_json::from_str(line).expect("scanner JSON event");
+        event["fields"]["state"] == "child_scan_failed" && event["fields"]["child_folder"] == "bucket/prefix"
+    }));
+
+    control
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("persist the partial that observed the failed subtree");
+    let partial_bytes = tokio::fs::read(store.root.join("main"))
+        .await
+        .expect("read failed-subtree partial");
+    let persisted_partial = DataUsageCache::unmarshal(&partial_bytes).expect("decode failed-subtree partial");
+    assert!(!persisted_partial.info.snapshot_complete);
+    assert!(
+        persisted_partial.info.scan_progress.is_some(),
+        "the failed subtree requires a verification sweep"
+    );
+    assert_eq!(tokio::fs::read(store.root.join("backup")).await.expect("read older backup"), backup);
+
+    let mut corrupt = partial_bytes;
+    corrupt[0] = 0xc1;
+    assert!(DataUsageCache::unmarshal(&corrupt).is_err(), "corrupt primary must not decode");
+    tokio::fs::write(store.root.join("main"), &corrupt)
+        .await
+        .expect("inject corruption into the newer primary");
+    let recovery_logs = CapturedLogs::default();
+    let mut recovered = DataUsageCache::default();
+    let recovery_result = match recovered.load_with_revisions(store.clone(), CACHE_NAME).await {
+        Err(error) => {
+            println!("backup_recovery load_error={error}");
+            None
+        }
+        Ok(_) => {
+            let state = crate::scanner_io::current_cache_root_or_prepare_with_generation(
+                &mut recovered,
+                "bucket",
+                SOURCE,
+                u64::from(recovery_cycle),
+                7,
+                PLAN,
+                options,
+            );
+            assert!(
+                matches!(state, crate::scanner_io::DataUsageCacheScanState::Prepared { .. }),
+                "an older backup must not be adopted as the current bucket root"
+            );
+            recovered.info.skip_healing = true;
+            let parent = CancellationToken::new();
+            let budget = ScannerCycleBudget::new(&parent, Default::default());
+            Some(
+                recovery_logs
+                    .capture(
+                        tracing::Level::DEBUG,
+                        scanner.local_disk.clone().nsscanner_disk(
+                            budget.token(),
+                            budget,
+                            vec![scanner.local_disk.clone()],
+                            recovered,
+                            None,
+                            scan_options(HealScanMode::Normal),
+                        ),
+                    )
+                    .await,
+            )
+        }
+    };
+    let saw_failure = recovery_logs.contents().lines().any(|line| {
+        let event: serde_json::Value = serde_json::from_str(line).expect("scanner JSON event");
+        event["fields"]["state"] == "child_scan_failed" && event["fields"]["child_folder"] == "bucket/prefix"
+    });
+    assert_eq!(
+        tokio::fs::read(store.root.join("main"))
+            .await
+            .expect("retain corrupted primary"),
+        corrupt
+    );
+    assert_eq!(
+        tokio::fs::read(store.root.join("backup"))
+            .await
+            .expect("retain original backup"),
+        backup
+    );
+    match recovery_result {
+        None => println!("backup_recovery load_rejected failed_prefix_observed={saw_failure}"),
+        Some(Err(error)) => println!("backup_recovery scan_error={error} failed_prefix_observed={saw_failure}"),
+        Some(Ok(ScannerDiskScanOutcome::Partial(cache) | ScannerDiskScanOutcome::NamespaceNotFound(cache))) => {
+            println!(
+                "backup_recovery incomplete snapshot_complete={} failed_prefix_observed={saw_failure}",
+                cache.info.snapshot_complete
+            );
+            assert!(!cache.info.snapshot_complete, "an unverified failed subtree must remain incomplete");
+        }
+        Some(Ok(ScannerDiskScanOutcome::Complete(cache))) => panic!(
+            "an older backup must not certify an unverified failed subtree: snapshot_complete={}, failed_prefix_observed={saw_failure}",
+            cache.info.snapshot_complete
+        ),
+    }
+}
+
 #[tokio::test]
 #[serial]
 async fn checkpoint_fixture_save_reload_resume_across_cycles_and_leaders() {

crates/scanner/src/scanner_folder/tests/erasure_probe.rs

@@ -15,10 +15,10 @@ enum DirectoryOrder {
 pub(in crate::scanner_folder) type OrderedEntries = VecDeque<std::io::Result<tokio::fs::DirEntry>>;
 static ORDERED_DIRECTORIES: LazyLock<Mutex<HashMap<String, DirectoryOrder>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
 
-struct OrderedDirectory(String);
+pub(super) struct OrderedDirectory(String);
 
 impl OrderedDirectory {
-    fn new(path: &std::path::Path) -> Self {
+    pub(super) fn new(path: &std::path::Path) -> Self {
         let path = path.to_string_lossy().into_owned();
         ORDERED_DIRECTORIES
             .lock()
@@ -34,7 +34,7 @@ impl OrderedDirectory {
             .insert(guard.0.clone(), DirectoryOrder::FailAfter(entries, kind));
         guard
     }
-    fn fail_open(path: &std::path::Path, kind: std::io::ErrorKind) -> Self {
+    pub(super) fn fail_open(path: &std::path::Path, kind: std::io::ErrorKind) -> Self {
         let guard = Self::new(path);
         ORDERED_DIRECTORIES
             .lock()