fix(scanner): preserve compacted scan accounting and coverage

Commit isolated namespace child usage only after successful traversal, while retaining completed accounting on cancellation and resource deferral. Synchronize reduced root totals before the cancellation-prone flush boundary.

Treat skipped namespaces as incomplete coverage and invalidate legacy sampling baselines until a fresh verification pass succeeds. Keep provisional usage out of progress updates and retain the upstream symlink regression.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/scanner_folder.rs

@@ -2710,48 +2710,74 @@ impl FolderScanner {
 
                 (self.update_current_path)(&folder_item.name).await;
 
-                let isolate_child = isolate_compacted_children && child_mode == FolderScanMode::Namespace;
-                let compacted = into.compacted;
-                // This accumulator has no retained child tree. Save its totals so a
-                // failed child cannot contribute partial usage to a completed scan.
-                let rollback = (compacted && isolate_child).then(|| clone_cache_value(into));
-                let mut dst = None;
-                let result = if compacted {
-                    Box::pin(self.scan_folder_inner(ctx.clone(), folder_item.clone(), into, child_mode, isolate_child)).await
+                // Keep a compacted namespace child's accounting local until it
+                // succeeds. Ordinary child errors can then discard it without
+                // rolling back totals already committed by healthy siblings.
+                let isolate_compacted_child =
+                    into.compacted && isolate_compacted_children && child_mode == FolderScanMode::Namespace;
+                let outcome = if into.compacted && !isolate_compacted_child {
+                    // Keep the shared-accumulator path free of child summaries.
+                    Box::pin(self.scan_folder_inner(ctx.clone(), folder_item.clone(), into, child_mode, false)).await?
                 } else {
-                    let dst = dst.insert(DataUsageEntry::default());
-                    self.scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, dst, child_mode)
-                        .await
-                };
-                let outcome = match result {
-                    Ok(outcome) => outcome,
-                    Err(e) => {
-                        if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
-                            if let Some(dst) = &dst {
-                                self.preserve_partial_child_progress(&folder_item.parent, &h, into, dst).await;
+                    let mut dst = DataUsageEntry {
+                        compacted: isolate_compacted_child,
+                        ..Default::default()
+                    };
+                    // Provisional child usage must not escape through progress
+                    // updates before this accumulator is committed to its parent.
+                    let suspended_updates = if isolate_compacted_child {
+                        self.usage_updates.take()
+                    } else {
+                        None
+                    };
+                    let result = if isolate_compacted_child {
+                        Box::pin(self.scan_folder_inner(ctx.clone(), folder_item.clone(), &mut dst, child_mode, true)).await
+                    } else {
+                        self.scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst, child_mode)
+                            .await
+                    };
+                    if isolate_compacted_child {
+                        self.usage_updates = suspended_updates;
+                    }
+                    match result {
+                        Ok(outcome) => {
+                            if isolate_compacted_child {
+                                into.merge(&dst);
                             }
-                            return Err(e);
+                            outcome
                         }
-                        if child_mode != FolderScanMode::Namespace || (compacted && !isolate_child) {
-                            return Err(e);
+                        Err(e) => {
+                            if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
+                                if isolate_compacted_child {
+                                    // Cancellation and resource deferral keep completed
+                                    // work even though the child has not finished.
+                                    into.merge(&dst);
+                                } else {
+                                    self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
+                                        .await;
+                                }
+                                return Err(e);
+                            }
+                            if child_mode != FolderScanMode::Namespace {
+                                return Err(e);
+                            }
+                            warn!(
+                                target: "rustfs::scanner::folder",
+                                event = EVENT_SCANNER_FOLDER_STATE,
+                                component = LOG_COMPONENT_SCANNER,
+                                subsystem = LOG_SUBSYSTEM_FOLDER,
+                                state = "child_scan_failed",
+                                folder = %folder.name,
+                                child_folder = %folder_item.name,
+                                error = %e,
+                                "Scanner child folder scan failed"
+                            );
+                            // A skipped namespace is a coverage gap even when no
+                            // checkpointed forward sweep is active.
+                            self.coverage_gap = true;
+                            self.finish_raw_enumeration_parent(&folder_item.name);
+                            continue;
                         }
-                        if let Some(rollback) = rollback {
-                            *into = rollback;
-                        }
-                        warn!(
-                            target: "rustfs::scanner::folder",
-                            event = EVENT_SCANNER_FOLDER_STATE,
-                            component = LOG_COMPONENT_SCANNER,
-                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                            folder = %folder.name,
-                            child_folder = %folder_item.name,
-                            state = "child_scan_failed",
-                            error = %e,
-                            "Scanner child folder scan failed"
-                        );
-                        self.coverage_gap |= forward_sweep;
-                        self.finish_raw_enumeration_parent(&folder_item.name);
-                        continue;
                     }
                 };
                 self.finish_raw_enumeration_parent(&folder_item.name);
@@ -3243,6 +3269,15 @@ impl FolderScanner {
             let compact_self = self.new_cache.info.name != folder.name || self.old_cache.info.scan_progress.is_none();
             self.new_cache
                 .reduce_children_of(&this_hash, DATA_SCANNER_COMPACT_AT_CHILDREN, compact_self);
+            // The root accumulator is also used by interruption handling.
+            // Reduction may have removed all its old child links, so publish
+            // the aggregate there before any cancellation-prone flush wait.
+            if folder.parent.is_none()
+                && let Some(root) = self.new_cache.cache.get(&this_hash.0)
+                && root.compacted
+            {
+                into.clone_from(root);
+            }
             done_compact();
         }
 
@@ -3256,6 +3291,8 @@ impl FolderScanner {
         }
 
         if folder.parent.is_none() {
+            #[cfg(test)]
+            tests::checkpoint_fixture::deep_compacted::cancel_at_root_flush(&self.root, &self.new_cache, &ctx);
             wait_for_scanner_io(&ctx, self.sleeper.flush_pacing(&mut self.pacing_debt)).await?;
         }
         self.finish_size_reconciliation_batch();
@@ -3467,9 +3504,14 @@ pub(crate) async fn scan_data_folder_scoped(
                     // Retain observations, then verify from the beginning under
                     // the latest plan. A clean tail cannot certify an old prefix.
                     progress.started_plan = progress.requested_plan;
-                    new_cache.info.scan_plan_digest = None;
                 }
             }
+            if !new_cache.info.snapshot_complete {
+                // Legacy sweeps also reuse compacted children by sampling.
+                // Invalidate that baseline after gaps so the next preparation
+                // cannot certify an unvisited subtree from cached totals.
+                new_cache.info.scan_plan_digest = None;
+            }
             let had_scan_checkpoint = had_scan_checkpoint || new_cache.info.scan_checkpoint.is_some();
             new_cache.info.scan_resume_after = None;
             new_cache.info.scan_checkpoint = None;

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

@@ -21,7 +21,7 @@ use std::io::Cursor;
 use tokio::io::AsyncReadExt;
 
 mod completed_compaction;
-mod deep_compacted;
+pub(in crate::scanner_folder) mod deep_compacted;
 mod memory_budget;
 mod periodic_snapshot;
 mod segment_observation;

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

@@ -341,16 +341,352 @@ async fn wide_root_legacy_permission_failure_isolated_to_child() {
         )
         .await;
     std::fs::set_permissions(denied, std::fs::Permissions::from_mode(0o700)).expect("restore child access");
-    let ScannerDiskScanOutcome::Complete(wide) = wide.expect("wide legacy scanner must keep the same child-error isolation")
+    let ScannerDiskScanOutcome::Partial(wide) = wide.expect("wide legacy scanner must keep the same child-error isolation")
     else {
-        panic!("wide root should complete despite one unreadable prefix");
+        panic!("an unreadable prefix must prevent complete snapshot publication");
     };
+    assert!(!wide.info.snapshot_complete);
     let summary = wide.checked_flatten("bucket").expect("aggregate healthy child");
     assert!(summary.compacted);
     assert_eq!((summary.objects, summary.size), (1, 7));
     assert_eq!(wide.cache.len(), 1);
 }
 
+// Select one fixture disk so unrelated concurrent scans cannot trigger this
+// cancellation. The hook runs at the root flush boundary, after compaction.
+static CANCEL_ROOT_FLUSH: Mutex<Option<std::path::PathBuf>> = Mutex::new(None);
+
+pub(in crate::scanner_folder) fn cancel_at_root_flush(root: &str, cache: &DataUsageCache, ctx: &CancellationToken) {
+    let mut selected = CANCEL_ROOT_FLUSH.lock().expect("root flush cancellation lock");
+    if selected.as_deref() == Some(std::path::Path::new(root)) {
+        let cached = cache.root().expect("root cached before flush");
+        assert!(cached.compacted, "fixture must reach post-reduction compaction");
+        assert!(cached.children.is_empty());
+        *selected = None;
+        ctx.cancel();
+    }
+}
+
+struct RootFlushCancellationGuard;
+
+impl Drop for RootFlushCancellationGuard {
+    fn drop(&mut self) {
+        *CANCEL_ROOT_FLUSH.lock().expect("clear root flush cancellation") = None;
+    }
+}
+
+async fn wide_legacy_root() -> (FolderScanner, std::path::PathBuf, TestGuard, DataUsageCache) {
+    let (scanner, root) = build_test_scanner().await;
+    let cleanup = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    write_checkpoint_object(&root, "a-healthy/object", &[(Some(Uuid::from_u128(101)), 7)]).await;
+    write_checkpoint_object(&root, "z-healthy/object", &[(Some(Uuid::from_u128(102)), 11)]).await;
+    for index in 0..DATA_SCANNER_COMPACT_AT_FOLDERS - 2 {
+        std::fs::create_dir(root.join("bucket").join(format!("empty-{index:05}"))).expect("create wide root child");
+    }
+    let mut cache = DataUsageCache::default();
+    cache.prepare_for_scan("bucket", 11, 7, SOURCE, PLAN, false);
+    cache.info.skip_healing = true;
+    assert!(cache.info.scan_progress.is_none());
+    (scanner, root, cleanup, cache)
+}
+
+#[cfg(unix)]
+#[tokio::test]
+#[serial]
+async fn wide_root_legacy_namespace_error_isolated_to_child() {
+    assert!(!crate::scanner_is_erasure().await, "single-drive Namespace scan");
+    let (scanner, root, _cleanup, cache) = wide_legacy_root().await;
+    // The error occurs while resolving a symlink inside the child, after its
+    // directory opened successfully. An open-only permission check misses it.
+    let regular = root.join("regular-file");
+    std::fs::write(&regular, b"payload").expect("write non-directory symlink target");
+    for broken in ["m-broken", "nested/broken"] {
+        let directory = root.join("bucket").join(broken);
+        std::fs::create_dir_all(&directory).expect("create failing namespace child");
+        let link = directory.join("bad-link");
+        symlink(regular.join("child"), &link).expect("create NotADirectory symlink");
+        assert_eq!(
+            std::fs::metadata(link).expect_err("invalid symlink target").kind(),
+            ErrorKind::NotADirectory
+        );
+    }
+    write_checkpoint_object(&root, "nested/healthy/object", &[(Some(Uuid::from_u128(103)), 13)]).await;
+    write_checkpoint_object(&root, "m-partial", &[(Some(Uuid::from_u128(104)), 101)]).await;
+    // Fail enumeration after this child's metadata contributed usage. Simply
+    // catching the error with a shared accumulator would retain its 101 bytes.
+    let _partial =
+        erasure_probe::OrderedDirectory::failing(&scanner.local_disk.path().join("bucket/m-partial"), 1, ErrorKind::Other);
+    let (outcome, budget) = scan(&scanner.local_disk, cache, HealScanMode::Normal, 5).await;
+    let ScannerDiskScanOutcome::Partial(cache) = outcome else {
+        panic!("namespace errors must prevent complete snapshot publication without aborting healthy siblings")
+    };
+    assert!(!cache.info.snapshot_complete);
+    assert_eq!(budget.progress().0, 4, "healthy siblings and the failing child must read metadata");
+    let total = cache.checked_flatten("bucket").expect("valid compacted root");
+    assert!(total.compacted);
+    assert_eq!((total.objects, total.versions, total.size), (3, 3, 31));
+    assert_eq!(cache.cache.len(), 1, "child isolation must retain aggregate-only storage");
+    let mut reloaded = save_reload(&FixtureStore::new(), &cache).await;
+    assert_eq!(reloaded.checked_flatten("bucket").expect("persisted aggregate").objects, 3);
+    drop(_partial);
+    for broken in ["m-broken", "nested/broken"] {
+        std::fs::remove_dir_all(root.join("bucket").join(broken)).expect("remove failed namespace fixture");
+    }
+    assert_eq!(
+        reloaded.prepare_for_scan("bucket", 12, 7, SOURCE, PLAN, false),
+        crate::DataUsageCachePrepareOutcome::Reset
+    );
+    reloaded.info.skip_healing = true;
+    let (outcome, budget) = scan(&scanner.local_disk, reloaded, HealScanMode::Normal, 5).await;
+    let ScannerDiskScanOutcome::Complete(verified) = outcome else {
+        panic!("a clean retry must rebuild complete coverage")
+    };
+    assert!(verified.info.snapshot_complete);
+    assert_eq!(
+        budget.progress().0,
+        4,
+        "retry must revisit every object including the previously failed child"
+    );
+    let total = verified.checked_flatten("bucket").expect("verified complete aggregate");
+    assert_eq!((total.objects, total.versions, total.size), (4, 4, 132));
+}
+
+#[tokio::test]
+#[serial]
+async fn wide_root_legacy_child_cancellation_preserves_partial_accounting() {
+    let (scanner, _root, _cleanup, cache) = wide_legacy_root().await;
+    let (outcome, budget) = scan(&scanner.local_disk, cache, HealScanMode::Normal, 1).await;
+    assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Objects));
+    assert_eq!(budget.progress().0, 1);
+    let ScannerDiskScanOutcome::Partial(cache) = outcome else {
+        panic!("cancellation inside a compacted namespace child must propagate")
+    };
+    let total = cache.checked_flatten("bucket").expect("partial compacted accounting");
+    assert!(total.compacted);
+    assert_eq!((total.objects, total.versions), (1, 1));
+    assert!([7, 11].contains(&total.size), "completed child-local bytes must survive cancellation");
+    let reloaded = save_reload(&FixtureStore::new(), &cache).await;
+    assert_eq!(reloaded.checked_flatten("bucket").expect("persisted partial accounting").size, total.size);
+}
+
+#[tokio::test]
+#[serial]
+async fn wide_root_reused_children_cancellation_preserves_compacted_accounting() {
+    let (scanner, root) = build_test_scanner().await;
+    let _cleanup = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let mut cache = DataUsageCache::default();
+    cache.prepare_for_scan("bucket", 11, 7, SOURCE, PLAN, false);
+    cache.info.skip_healing = true;
+    cache.replace("bucket", "", DataUsageEntry::default());
+    let children = DATA_SCANNER_COMPACT_AT_CHILDREN + 1;
+    for index in 0..children {
+        let name = format!("bucket/clean-{index:05}");
+        std::fs::create_dir_all(root.join(&name)).expect("create reusable root child");
+        cache.replace(
+            &name,
+            "bucket",
+            DataUsageEntry {
+                compacted: true,
+                objects: 1,
+                versions: 2,
+                size: 7,
+                ..Default::default()
+            },
+        );
+    }
+    assert_eq!(cache.root().expect("wide baseline root").children.len(), children);
+    *CANCEL_ROOT_FLUSH.lock().expect("install root flush cancellation") = Some(scanner.local_disk.path());
+    let _cancel_guard = RootFlushCancellationGuard;
+    let budget = ScannerCycleBudget::new(&CancellationToken::new(), Default::default());
+    let mut options = super::scan_options(HealScanMode::Normal);
+    options.prefix_scan_scope = ScannerBucketPrefixScanScope::from_dirty_top_level_entries(HashSet::from(["dirty".into()]));
+    let outcome = scanner
+        .local_disk
+        .clone()
+        .nsscanner_disk(budget.token(), budget.clone(), vec![scanner.local_disk.clone()], cache, None, options)
+        .await
+        .expect("cancelled root flush must return a partial cache");
+    assert!(CANCEL_ROOT_FLUSH.lock().expect("flush hook observed").is_none());
+    assert_eq!(budget.progress().0, 0, "clean children must be reused without metadata reads");
+    let ScannerDiskScanOutcome::Partial(cache) = outcome else {
+        panic!("flush cancellation must propagate")
+    };
+    assert!(!cache.info.snapshot_complete);
+    let total = cache
+        .checked_flatten("bucket")
+        .expect("cancellation must not leave dangling root links");
+    assert!(total.compacted);
+    assert!(total.children.is_empty());
+    assert_eq!((total.objects, total.versions, total.size), (children, children * 2, children * 7));
+    assert_eq!(cache.cache.len(), 1);
+    let reloaded = save_reload(&FixtureStore::new(), &cache).await;
+    assert_eq!(reloaded.checked_flatten("bucket").expect("persisted root accounting").objects, children);
+}
+
+#[tokio::test]
+#[serial]
+async fn wide_root_legacy_child_resource_deferral_preserves_partial_accounting() {
+    use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+    use std::num::NonZeroU16;
+
+    let (mut scanner, _root, _cleanup, cache) = wide_legacy_root().await;
+    scanner.old_cache = cache.clone();
+    scanner.new_cache.info = cache.info;
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    let admission = NodeAdmission::new(NonZeroU16::MIN);
+    admission.update(
+        &mut FeedbackController::new(),
+        Some(FeedbackObservation {
+            generation: 1,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        }),
+    );
+    scanner.node_admission = Some(admission.clone());
+    let (sender, mut receiver) = mpsc::channel(16);
+    scanner.usage_updates = Some(ScannerUsageUpdates::new(sender, "bucket".into()));
+    let deferred = Arc::new(AtomicBool::new(false));
+    let observed = deferred.clone();
+    scanner.update_current_path = Arc::new(move |path| {
+        if path == "bucket/a-healthy/object" {
+            // Defer the next directory after this object's metadata completed.
+            observed.store(true, Ordering::SeqCst);
+        } else if observed.load(Ordering::SeqCst) {
+            admission.update(&mut FeedbackController::new(), None);
+        }
+        Box::pin(async {})
+    });
+    let mut total = DataUsageEntry::default();
+    let result = scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut total,
+        )
+        .await;
+    assert!(deferred.load(Ordering::SeqCst), "first metadata object must be reached");
+    assert!(
+        matches!(result, Err(ScannerError::ResourceDeferred)),
+        "resource pressure must propagate: {result:?}"
+    );
+    assert!(total.compacted);
+    assert_eq!((total.objects, total.versions, total.size), (1, 1, 7));
+    assert!(scanner.usage_updates.is_some(), "resource deferral must restore the progress owner");
+    let committed = receiver.try_recv().expect("completed sibling progress");
+    assert_eq!((committed.objects, committed.versions, committed.size), (1, 1, 7));
+}
+
+#[tokio::test]
+#[serial]
+async fn wide_root_legacy_usage_updates_exclude_uncommitted_child_accounting() {
+    temp_env::async_with_vars([(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, Some("1"))], async {
+        let (mut scanner, root, _cleanup, cache) = wide_legacy_root().await;
+        write_checkpoint_object(&root, "000-partial", &[(Some(Uuid::from_u128(105)), 101)]).await;
+        let _partial = erasure_probe::OrderedDirectory::failing(&root.join("bucket/000-partial"), 1, ErrorKind::Other);
+        scanner.old_cache = cache.clone();
+        scanner.new_cache.info = cache.info;
+        scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+        let (sender, mut receiver) = mpsc::channel(16);
+        scanner.usage_updates = Some(ScannerUsageUpdates::new(sender, "bucket".into()));
+        scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: "bucket".into(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut DataUsageEntry::default(),
+            )
+            .await
+            .expect("failed namespace must not abort healthy siblings");
+        let committed = scanner
+            .new_cache
+            .checked_flatten("bucket")
+            .expect("canonical committed aggregate");
+        assert_eq!((committed.objects, committed.versions, committed.size), (2, 2, 18));
+        let first = receiver.try_recv().expect("committed usage progress");
+        assert_eq!(
+            (first.objects, first.versions, first.size),
+            (1, 1, 7),
+            "provisional 101-byte child must never be published"
+        );
+        assert!(receiver.try_recv().is_err(), "minute interval still bounds updates");
+        let updates = scanner
+            .usage_updates
+            .as_mut()
+            .expect("requested usage owner must be restored");
+        let total = updates.cache.checked_flatten("bucket").expect("connected committed progress");
+        assert_eq!((total.objects, total.versions, total.size), (2, 2, 18));
+        assert_eq!(
+            updates.cache.cache.len(),
+            1,
+            "compacted progress must not retain provisional child entries"
+        );
+        updates.last_update = SystemTime::UNIX_EPOCH;
+        scanner.send_update().await;
+        let last = receiver.try_recv().expect("final committed aggregate");
+        assert_eq!((last.objects, last.versions, last.size), (2, 2, 18));
+    })
+    .await;
+}
+
+#[tokio::test]
+#[serial]
+async fn legacy_namespace_gap_requires_fresh_verification_before_sampling() {
+    let (scanner, root) = build_test_scanner().await;
+    let _cleanup = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for (object, version, size) in [("prefix/000-partial", 111, 101), ("prefix/healthy", 112, 7)] {
+        write_checkpoint_object(&root, object, &[(Some(Uuid::from_u128(version)), size)]).await;
+    }
+    let partial = erasure_probe::OrderedDirectory::failing(
+        &scanner.local_disk.path().join("bucket/prefix/000-partial"),
+        1,
+        ErrorKind::Other,
+    );
+    let mut cache = DataUsageCache::default();
+    cache.prepare_for_scan("bucket", 11, 7, SOURCE, PLAN, false);
+    cache.info.skip_healing = true;
+    let (outcome, budget) = scan(&scanner.local_disk, cache, HealScanMode::Normal, 3).await;
+    let ScannerDiskScanOutcome::Partial(cache) = outcome else {
+        panic!("a narrow root must also refuse to certify skipped namespace coverage")
+    };
+    assert_eq!(budget.progress().0, 2, "failed child must reach provisional metadata accounting");
+    assert!(!cache.info.snapshot_complete);
+    assert!(cache.find(PREFIX).expect("incomplete aggregate").compacted);
+    assert_eq!(
+        cache.info.scan_plan_digest, None,
+        "incomplete coverage must invalidate sampled baseline reuse"
+    );
+    let mut reloaded = save_reload(&FixtureStore::new(), &cache).await;
+    drop(partial);
+    assert_eq!(
+        reloaded.prepare_for_scan("bucket", 12, 7, SOURCE, PLAN, false),
+        crate::DataUsageCachePrepareOutcome::Reset,
+        "a legacy retry must not sample an incomplete compacted subtree"
+    );
+    reloaded.info.skip_healing = true;
+    let (outcome, budget) = scan(&scanner.local_disk, reloaded, HealScanMode::Normal, 3).await;
+    let ScannerDiskScanOutcome::Complete(cache) = outcome else {
+        panic!("clean retry must restore complete coverage")
+    };
+    assert!(cache.info.snapshot_complete);
+    assert_eq!(budget.progress().0, 2, "both metadata objects must be rechecked");
+    let total = cache.checked_flatten("bucket").expect("verified retry accounting");
+    assert_eq!((total.objects, total.versions, total.size), (2, 2, 108));
+}
+
 #[cfg(unix)]
 #[tokio::test]
 #[serial]
@@ -382,9 +718,10 @@ async fn wide_root_legacy_symlink_lookup_failure_isolated_to_child() {
     cache.prepare_for_scan("bucket", 11, 7, SOURCE, PLAN, false);
     cache.info.skip_healing = true;
     let (outcome, _) = scan(&scanner.local_disk, cache, HealScanMode::Normal, 100).await;
-    let ScannerDiskScanOutcome::Complete(wide) = outcome else {
-        panic!("a failed child must not interrupt the wide root");
+    let ScannerDiskScanOutcome::Partial(wide) = outcome else {
+        panic!("a failed child must preserve sibling progress without certifying complete coverage");
     };
+    assert!(!wide.info.snapshot_complete);
     let summary = wide.checked_flatten("bucket").expect("aggregate healthy child");
     assert!(summary.compacted);
     assert_eq!((summary.objects, summary.size), (1, 7));

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

@@ -15,7 +15,7 @@ enum DirectoryOrder {
 pub(in crate::scanner_folder) type OrderedEntries = VecDeque<std::io::Result<native_io::DirEntry>>;
 static ORDERED_DIRECTORIES: LazyLock<Mutex<HashMap<String, DirectoryOrder>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
 
-struct OrderedDirectory(String);
+pub(in crate::scanner_folder) struct OrderedDirectory(String);
 
 impl OrderedDirectory {
     fn new(path: &std::path::Path) -> Self {
@@ -26,7 +26,7 @@ impl OrderedDirectory {
             .insert(path.clone(), DirectoryOrder::PartLast);
         Self(path)
     }
-    fn failing(path: &std::path::Path, entries: usize, kind: std::io::ErrorKind) -> Self {
+    pub(in crate::scanner_folder) fn failing(path: &std::path::Path, entries: usize, kind: std::io::ErrorKind) -> Self {
         let guard = Self::new(path);
         ORDERED_DIRECTORIES
             .lock()