fix(scanner): avoid placeholder retention and resume checkpoints (#8460)

* fix(scanner): bound old-cache lookups and recover checkpoint saves

Keep missing child lookups read-only and preserve the checkpoint receiver
after ordinary persistence errors. Reload on stale CAS or indeterminate
revision refresh, and report confirmed periodic saves separately.

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

* fix(scanner): remove redundant checkpoint error conversion

Return the existing StorageError directly so strict Clippy passes.

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

* docs(scanner): clarify checkpoint persistence guarantees

Distinguish a confirmed periodic save from validated restart coverage.
Keep adoption and subtree-skip requirements explicit.

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

* fix(scanner): preserve resumable periodic coverage

Materialize active ancestor accounting in bounded periodic snapshots and
seal completed coverage independently of raw enumeration state. Preserve
durable coverage during early resumed enumeration and exclude unfinished
out-of-order heal work from periodic and cancellation receipts.

Reuse ancestor frame slots to avoid growing child-set copies on descent,
retain compacted accumulator safety, and report checkpoint adoption resets.

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

* fix(scanner): reject covered active ancestors in checkpoints

Check every suspended ancestor alongside the current folder before
certifying completed coverage. An unfinished covered ancestor must not
become a resumable subtree merely because its descendant sorts later.

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

* fix(scanner): retain an incomplete root for missing namespaces

Create the empty bucket root explicitly at the missing-namespace boundary.
Preserve nonempty durable caches and keep namespace results unpublished.

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner-metrics/src/metrics.rs

@@ -18,7 +18,7 @@ use jiff::Timestamp;
 use rustfs_heal_contracts::heal_channel::HealScanMode;
 use serde::{Deserialize, Serialize};
 use std::{
-    collections::{BTreeSet, HashMap},
+    collections::{BTreeMap, BTreeSet, HashMap},
     fmt::Display,
     future::Future,
     pin::Pin,
@@ -994,6 +994,8 @@ pub struct Metrics {
     scanner_bitrot_cycle_enabled: AtomicBool,
     scanner_bitrot_cycle_millis: AtomicU64,
     scanner_checkpoint: Mutex<Option<ScannerCheckpointReport>>,
+    scanner_checkpoints_persisted: Mutex<BTreeMap<(usize, usize), ScannerPersistedCheckpointReport>>,
+    scanner_checkpoint_save_failures: AtomicU64,
     scanner_checkpoint_used: AtomicU64,
     scanner_checkpoint_cleared: AtomicU64,
     scanner_checkpoint_ignored: AtomicU64,
@@ -1137,6 +1139,15 @@ pub struct ScannerCheckpointReport {
     pub last_event: String,
 }
 
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct ScannerPersistedCheckpointReport {
+    pub pool_index: usize,
+    pub set_index: usize,
+    pub bucket: String,
+    pub saved_unix_secs: u64,
+    pub checkpoint: Option<ScannerCheckpointReport>,
+}
+
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 pub struct ScannerSourceWorkSnapshot {
     pub source: String,
@@ -1499,6 +1510,10 @@ pub struct ScannerMetricsReport {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub scan_checkpoint: Option<ScannerCheckpointReport>,
     #[serde(default)]
+    pub scan_checkpoints_persisted: Vec<ScannerPersistedCheckpointReport>,
+    #[serde(default)]
+    pub scan_checkpoint_save_failures: u64,
+    #[serde(default)]
     pub scan_checkpoint_used: u64,
     #[serde(default)]
     pub scan_checkpoint_cleared: u64,
@@ -2046,6 +2061,8 @@ impl Metrics {
             scanner_bitrot_cycle_enabled: AtomicBool::new(false),
             scanner_bitrot_cycle_millis: AtomicU64::new(0),
             scanner_checkpoint: Mutex::new(None),
+            scanner_checkpoints_persisted: Mutex::new(BTreeMap::new()),
+            scanner_checkpoint_save_failures: AtomicU64::new(0),
             scanner_checkpoint_used: AtomicU64::new(0),
             scanner_checkpoint_cleared: AtomicU64::new(0),
             scanner_checkpoint_ignored: AtomicU64::new(0),
@@ -2353,6 +2370,34 @@ impl Metrics {
         }
     }
 
+    /// Keep only the latest confirmed periodic save per set on this process.
+    pub fn record_scanner_checkpoint_saved(
+        &self,
+        pool_index: usize,
+        set_index: usize,
+        bucket: &str,
+        checkpoint: Option<ScannerCheckpointReport>,
+    ) {
+        let mut saved = self
+            .scanner_checkpoints_persisted
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        saved.insert(
+            (pool_index, set_index),
+            ScannerPersistedCheckpointReport {
+                pool_index,
+                set_index,
+                bucket: bucket.to_string(),
+                saved_unix_secs: unix_now_secs(),
+                checkpoint,
+            },
+        );
+    }
+
+    pub fn record_scanner_checkpoint_save_failed(&self) {
+        self.scanner_checkpoint_save_failures.fetch_add(1, Ordering::Relaxed);
+    }
+
     pub fn record_scanner_checkpoint_used(&self) {
         self.scanner_checkpoint_used.fetch_add(1, Ordering::Relaxed);
         self.update_scanner_checkpoint_event(SCANNER_CHECKPOINT_EVENT_USED);
@@ -3512,6 +3557,14 @@ impl Metrics {
             Ok(checkpoint) => checkpoint.clone(),
             Err(poisoned) => poisoned.into_inner().clone(),
         };
+        m.scan_checkpoints_persisted = self
+            .scanner_checkpoints_persisted
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .values()
+            .cloned()
+            .collect();
+        m.scan_checkpoint_save_failures = self.scanner_checkpoint_save_failures.load(Ordering::Relaxed);
         m.scan_checkpoint_used = self.scanner_checkpoint_used.load(Ordering::Relaxed);
         m.scan_checkpoint_cleared = self.scanner_checkpoint_cleared.load(Ordering::Relaxed);
         m.scan_checkpoint_ignored = self.scanner_checkpoint_ignored.load(Ordering::Relaxed);
@@ -3975,6 +4028,37 @@ mod tests {
         assert_eq!(report.scan_checkpoint_cleared, 1);
     }
 
+    #[tokio::test]
+    async fn report_distinguishes_persisted_checkpoints_from_memory_and_failures() {
+        let metrics = Metrics::new();
+        let saved = ScannerCheckpointReport {
+            version: 1,
+            resume_after: "bucket/first".to_string(),
+            reason: "objects".to_string(),
+            last_event: "saved".to_string(),
+        };
+        metrics.record_scanner_checkpoint_saved(0, 1, "bucket", Some(saved.clone()));
+        metrics.record_scanner_checkpoint_set(1, "bucket/second", "objects");
+        metrics.record_scanner_checkpoint_save_failed();
+        let report = metrics.report().await;
+        assert_eq!(report.scan_checkpoint_save_failures, 1);
+        assert_eq!(report.scan_checkpoint.expect("in-memory cursor").resume_after, "bucket/second");
+        assert_eq!(report.scan_checkpoints_persisted.len(), 1);
+        assert_eq!(report.scan_checkpoints_persisted[0].checkpoint, Some(saved));
+        assert_eq!(report.scan_checkpoints_persisted[0].set_index, 1);
+        assert!(report.scan_checkpoints_persisted[0].saved_unix_secs > 0);
+        metrics.record_scanner_checkpoint_saved(0, 1, "other-bucket", None);
+        metrics.record_scanner_checkpoint_saved(0, 2, "bucket", None);
+        let report = metrics.report().await;
+        assert_eq!(
+            report.scan_checkpoints_persisted.len(),
+            2,
+            "retain at most one save per set, not per bucket"
+        );
+        assert_eq!(report.scan_checkpoints_persisted[0].bucket, "other-bucket");
+        assert!(report.scan_checkpoints_persisted[0].checkpoint.is_none());
+    }
+
     #[tokio::test]
     async fn report_includes_scanner_source_work() {
         let metrics = Metrics::new();

crates/scanner/src/data_usage_define.rs

@@ -234,7 +234,7 @@ pub static DATA_USAGE_BLOOM_RECOVERY_PATH: LazyLock<String> =
 pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
     LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}.background-heal.json"));
 
-const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
+pub(crate) const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
 
 /// Scanner-side accounting on the shared [`SizeSummary`].
 ///
@@ -948,6 +948,42 @@ impl DataUsageCache {
             return self.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, true);
         }
         if !reusable {
+            let had_resume_state = self.info.scan_checkpoint.is_some()
+                || self.info.scan_resume_after.is_some()
+                || self.info.scan_coverage_receipt.is_some()
+                || self.info.scan_raw_enumeration_cursor.is_some()
+                || self.info.scan_raw_enumeration_page_index.is_some();
+            if had_resume_state {
+                let reason = if name == DATA_USAGE_ROOT {
+                    "invalid_bucket_scope"
+                } else if !identity.is_valid() {
+                    "invalid_identity"
+                } else if self.info.name != name {
+                    "bucket_mismatch"
+                } else if self.info.source != Some(source) {
+                    "source_mismatch"
+                } else if self.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT {
+                    "key_format_mismatch"
+                } else if self.info.scan_identity != Some(identity)
+                    || self.info.tier_registry_generation != Some(identity.tier_registry_generation)
+                {
+                    "identity_mismatch"
+                } else if self.info.leader_epoch != leader_epoch && !cross_epoch_checkpoint {
+                    "unvalidated_handoff"
+                } else {
+                    "incomplete_cache_scope"
+                };
+                warn!(
+                    target: "rustfs::scanner::data_usage",
+                    event = EVENT_SCANNER_CACHE_LOAD_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_CACHE,
+                    state = "checkpoint_reset",
+                    bucket = %name,
+                    reason,
+                    "Scanner bucket checkpoint cannot be adopted"
+                );
+            }
             let keep_debts = self.info.name == name
                 && self
                     .info
@@ -1221,8 +1257,8 @@ impl DataUsageCache {
         self.cache.get(&hash_path(path).key())
     }
 
-    pub fn find_children_copy(&mut self, h: DataUsageHash) -> DataUsageHashMap {
-        self.cache.entry(h.string()).or_default().children.clone()
+    pub fn find_children_copy(&self, h: &DataUsageHash) -> DataUsageHashMap {
+        self.cache.get(&h.0).map(|entry| entry.children.clone()).unwrap_or_default()
     }
 
     pub fn flatten(&self, root: &DataUsageEntry) -> DataUsageEntry {

crates/scanner/src/data_usage_define/tests.rs

@@ -2491,12 +2491,32 @@ fn checked_flatten_accepts_depth_limit_and_rejects_deeper_tree() {
 }
 
 #[test]
-fn test_find_children_copy_preserves_missing_entry_behavior() {
-    let mut cache = DataUsageCache::default();
+fn test_find_children_copy_does_not_insert_missing_entries() {
+    let cache = DataUsageCache::default();
     let missing_hash = hash_path("missing");
 
-    assert!(cache.find_children_copy(missing_hash.clone()).is_empty());
-    assert!(cache.cache.contains_key(&missing_hash.key()));
+    assert!(cache.find_children_copy(&missing_hash).is_empty());
+    assert!(cache.cache.is_empty());
+    assert_eq!(cache.cache.capacity(), 0);
+}
+
+#[test]
+fn test_find_children_copy_preserves_existing_children() {
+    let hash = hash_path("bucket");
+    let children = HashSet::from(["bucket/child".to_string()]);
+    let mut cache = DataUsageCache::default();
+    cache.cache.insert(
+        hash.key(),
+        DataUsageEntry {
+            children: children.clone(),
+            ..Default::default()
+        },
+    );
+    let mut copied = cache.find_children_copy(&hash);
+    assert_eq!(copied, children);
+    copied.clear();
+    assert_eq!(cache.cache[&hash.0].children, children);
+    assert_eq!(cache.cache.len(), 1);
 }
 
 #[test]

crates/scanner/src/remote_scanner/stream.rs

@@ -853,7 +853,7 @@ async fn scan_and_persist_local_bucket(
                 )
                 .await
                 {
-                    ScannerCheckpointPersistResult::Saved => {
+                    ScannerCheckpointPersistResult::Saved | ScannerCheckpointPersistResult::Skipped => {
                         if guard.is_lock_lost() {
                             scan_ctx.cancel();
                             let _ = tokio::time::timeout(NS_SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, scan.as_mut()).await;
@@ -869,8 +869,12 @@ async fn scan_and_persist_local_bucket(
                             "remote namespace scanner cache fence changed during checkpoint save",
                         ));
                     }
-                    ScannerCheckpointPersistResult::Failed(_error) => {
-                        checkpoint_channel_closed = true;
+                    ScannerCheckpointPersistResult::RetryBucket(error) => {
+                        scan_ctx.cancel();
+                        let _ = tokio::time::timeout(NS_SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, scan.as_mut()).await;
+                        return Err(RemoteScannerServerError::retry_bucket(format!(
+                            "remote namespace scanner checkpoint needs reload: {error}",
+                        )));
                     }
                 }
             }

crates/scanner/src/scanner_folder.rs

@@ -21,8 +21,8 @@ use std::time::{Duration, Instant, SystemTime};
 use crate::ReplTargetSizeSummary;
 use crate::data_usage_define::{
     DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
-    DataUsageRawEnumerationCursor, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal,
-    PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path,
+    DataUsageRawEnumerationCursor, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, MAX_DATA_USAGE_CACHE_DEPTH,
+    PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path,
 };
 use crate::error::ScannerError;
 use crate::raw_page_index::{RawEnumerationPageIndex, RawEnumerationPageWriter};
@@ -426,6 +426,10 @@ fn folder_resume_match(folder_name: &str, resume_after: &str) -> Option<FolderRe
         .map(|_| FolderResumeMatch::Descendant)
 }
 
+fn folder_is_covered_by_frontier(folder: &str, frontier: &str) -> bool {
+    folder <= frontier && !matches!(folder_resume_match(folder, frontier), Some(FolderResumeMatch::Descendant))
+}
+
 fn order_items_for_resume<T, F>(items: &mut [T], resume_after: Option<&str>, name: F) -> FolderResumeOrder
 where
     F: Fn(&T) -> &str,
@@ -704,6 +708,12 @@ impl ScannerBucketPrefixScanScope {
 }
 
 /// Folder scanner for scanning directory structures
+#[derive(Default)]
+struct ScannerCheckpointAncestor {
+    hash: DataUsageHash,
+    entry: DataUsageEntry,
+}
+
 pub struct FolderScanner {
     root: String,
     old_cache: DataUsageCache,
@@ -728,6 +738,8 @@ pub struct FolderScanner {
 
     updates: Option<mpsc::Sender<DataUsageEntry>>,
     checkpoint_tx: Option<mpsc::Sender<DataUsageCache>>,
+    checkpoint_ancestors: Vec<ScannerCheckpointAncestor>,
+    checkpoint_depth: usize,
     last_update: SystemTime,
     checkpoint_objects: u64,
     last_checkpoint_objects: u64,
@@ -1210,10 +1222,42 @@ impl FolderScanner {
             .map_or((None, None), |progress| (progress.cursor(), progress.page_index()))
     }
 
-    fn maybe_send_checkpoint(&mut self) {
+    async fn scan_child_folder(
+        &mut self,
+        ctx: CancellationToken,
+        folder: CachedFolder,
+        parent_hash: &mut DataUsageHash,
+        parent_entry: &mut DataUsageEntry,
+        child_entry: &mut DataUsageEntry,
+    ) -> Result<(), ScannerError> {
+        if self.checkpoint_tx.is_none() || self.old_cache.info.scan_progress.is_none() {
+            return Box::pin(self.scan_folder(ctx, folder, child_entry)).await;
+        }
+        let depth = self.checkpoint_depth;
+        if depth >= MAX_DATA_USAGE_CACHE_DEPTH {
+            return Err(ScannerError::Other("scanner checkpoint ancestor depth limit exceeded".to_string()));
+        }
+        if self.checkpoint_ancestors.len() == depth {
+            self.checkpoint_ancestors.push(ScannerCheckpointAncestor::default());
+        }
+        // Reuse the empty frame slots: moving suspended accumulators avoids
+        // cloning a growing child set or allocating histogram vectors per child.
+        std::mem::swap(parent_hash, &mut self.checkpoint_ancestors[depth].hash);
+        std::mem::swap(parent_entry, &mut self.checkpoint_ancestors[depth].entry);
+        self.checkpoint_depth += 1;
+        let result = Box::pin(self.scan_folder(ctx, folder, child_entry)).await;
+        self.checkpoint_depth -= 1;
+        // Restore the parent before cancellation/error handling stitches partial work.
+        std::mem::swap(parent_hash, &mut self.checkpoint_ancestors[depth].hash);
+        std::mem::swap(parent_entry, &mut self.checkpoint_ancestors[depth].entry);
+        result
+    }
+
+    fn maybe_send_checkpoint(&mut self, folder: &CachedFolder, hash: &DataUsageHash, into: &DataUsageEntry) {
         let Some(checkpoint_tx) = self.checkpoint_tx.clone() else { return };
         let elapsed = self.last_checkpoint_at.elapsed();
-        if self.new_cache.info.scan_progress.is_none()
+        if into.compacted
+            || self.new_cache.info.scan_progress.is_none()
             || elapsed < SCANNER_CHECKPOINT_MIN_INTERVAL
             || (self.checkpoint_objects.saturating_sub(self.last_checkpoint_objects) < SCANNER_CHECKPOINT_OBJECT_INTERVAL
                 && elapsed < SCANNER_CHECKPOINT_INTERVAL)
@@ -1233,34 +1277,51 @@ impl FolderScanner {
         }
         let Ok(permit) = checkpoint_tx.try_reserve() else { return };
 
+        // Invalid partial graphs must not trigger a fresh O(N) clone on every object.
+        self.last_checkpoint_objects = self.checkpoint_objects;
+        self.last_checkpoint_at = Instant::now();
+        // Heal revisits are not sorted with the forward sweep. An unfinished
+        // earlier subtree, including a suspended ancestor, must not be signed
+        // as completed prefix coverage.
+        if self.coverage_frontier.as_deref().is_some_and(|frontier| {
+            folder_is_covered_by_frontier(&folder.name, frontier)
+                || self.checkpoint_ancestors[..self.checkpoint_depth]
+                    .iter()
+                    .any(|ancestor| folder_is_covered_by_frontier(&ancestor.hash.0, frontier))
+        }) {
+            return;
+        }
         let mut snapshot = self.new_cache.clone();
+        let mut parent = None;
+        for ancestor in &self.checkpoint_ancestors[..self.checkpoint_depth] {
+            snapshot.replace_hashed(&ancestor.hash, &parent, &ancestor.entry);
+            parent = Some(ancestor.hash.clone());
+        }
+        snapshot.replace_hashed(hash, &folder.parent, into);
+        if snapshot.checked_flatten_complete_scope(&snapshot.info.name).is_none() {
+            return;
+        }
         snapshot.info.last_update = Some(SystemTime::now());
         snapshot.info.snapshot_complete = false;
         let (cursor, page_index) = self.peek_raw_enumeration_resume_state();
-        if cursor.is_some() || page_index.is_some() {
-            snapshot.info.scan_raw_enumeration_cursor = cursor;
-            snapshot.info.scan_raw_enumeration_page_index = page_index;
-            snapshot.info.scan_resume_after = None;
-            snapshot.info.scan_checkpoint = None;
-            snapshot.info.scan_coverage_receipt = None;
-        } else if snapshot.seal_scan_frontier(self.coverage_frontier.as_deref()).is_err() {
+        snapshot.info.scan_raw_enumeration_cursor = cursor;
+        snapshot.info.scan_raw_enumeration_page_index = page_index;
+        if snapshot.seal_scan_frontier(self.coverage_frontier.as_deref()).is_err() {
             return;
         }
-        if snapshot.root().is_none()
-            && snapshot.info.scan_raw_enumeration_cursor.is_none()
-            && snapshot.info.scan_raw_enumeration_page_index.is_none()
-        {
+        // A resumed readdir can run before covered subtrees are copied. Keep
+        // its durable receipt until this snapshot can represent that coverage.
+        if self.coverage_frontier.is_some() && snapshot.info.scan_coverage_receipt.is_none() {
             return;
         }
+        // The receipt was just sealed from this snapshot; do not hash it twice.
         if snapshot.info.scan_raw_enumeration_cursor.is_none()
             && snapshot.info.scan_raw_enumeration_page_index.is_none()
-            && snapshot.validated_scan_frontier().is_none()
+            && snapshot.info.scan_coverage_receipt.is_none()
         {
             return;
         }
         permit.send(snapshot);
-        self.last_checkpoint_objects = self.checkpoint_objects;
-        self.last_checkpoint_at = Instant::now();
     }
 
     fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) {
@@ -1297,6 +1358,15 @@ impl FolderScanner {
         child_entry: &DataUsageEntry,
     ) {
         if data_usage_entry_has_progress(child_entry) {
+            // A cancelled out-of-order heal revisit is not completed prefix work.
+            if self
+                .coverage_frontier
+                .as_deref()
+                .is_some_and(|frontier| folder_is_covered_by_frontier(&child_hash.0, frontier))
+            {
+                self.coverage_frontier = None;
+                self.coverage_gap = true;
+            }
             let mut child_entry = child_entry.clone();
             self.carry_forward_old_children(child_hash, &mut child_entry);
             self.record_scan_resume_hint_if_not_ancestor(&child_hash.key());
@@ -1513,7 +1583,7 @@ impl FolderScanner {
             return Err(ScannerError::Other("Operation cancelled".to_string()));
         }
 
-        let this_hash = hash_path(&folder.name);
+        let mut this_hash = hash_path(&folder.name);
         // Store initial compaction state.
         let was_compacted = into.compacted;
 
@@ -1526,7 +1596,7 @@ impl FolderScanner {
 
             let mut abandoned_children: DataUsageHashMap = HashSet::new();
             if !into.compacted {
-                abandoned_children = self.old_cache.find_children_copy(this_hash.clone());
+                abandoned_children = self.old_cache.find_children_copy(&this_hash);
             }
 
             debug!(
@@ -1651,7 +1721,7 @@ impl FolderScanner {
                     }
                 }
                 self.record_raw_enumeration_entry(&folder.name, &file_name);
-                self.maybe_send_checkpoint();
+                self.maybe_send_checkpoint(&folder, &this_hash, into);
                 let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE;
 
                 let file_path = entry.path().to_string_lossy().to_string();
@@ -1958,7 +2028,7 @@ impl FolderScanner {
                 object_count += 1;
                 self.budget.record_object_scanned();
                 self.checkpoint_objects = self.checkpoint_objects.saturating_add(1);
-                self.maybe_send_checkpoint();
+                self.maybe_send_checkpoint(&folder, &this_hash, into);
 
                 timer.sleep().await;
 
@@ -2159,10 +2229,9 @@ impl FolderScanner {
                 let h = hash_path(&folder_item.name);
                 if forward_sweep
                     && !into.compacted
-                    && forward_resume_after.as_deref().is_some_and(|resume| {
-                        folder_item.name.as_str() <= resume
-                            && !matches!(folder_resume_match(&folder_item.name, resume), Some(FolderResumeMatch::Descendant))
-                    })
+                    && forward_resume_after
+                        .as_deref()
+                        .is_some_and(|resume| folder_is_covered_by_frontier(&folder_item.name, resume))
                     && self.old_cache.find(&folder_item.name).is_some()
                 {
                     self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
@@ -2233,8 +2302,10 @@ impl FolderScanner {
                     let mut dst = DataUsageEntry::default();
 
                     // Use Box::pin for recursive async call
-                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
-                    if let Err(e) = fut.await {
+                    if let Err(e) = self
+                        .scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst)
+                        .await
+                    {
                         if ctx.is_cancelled() {
                             self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
                                 .await;
@@ -2258,7 +2329,7 @@ impl FolderScanner {
 
                     into.add_child(&h);
                     self.record_completed_child(&folder_item.name, dst.failed_objects == 0);
-                    self.maybe_send_checkpoint();
+                    self.maybe_send_checkpoint(&folder, &this_hash, into);
                     // We scanned a folder, optionally send update.
                     self.update_cache.delete_recursive(&h);
                     self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
@@ -2635,8 +2706,10 @@ impl FolderScanner {
                         let h = hash_path(&folder_item.name);
 
                         // Use Box::pin for recursive async call
-                        let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
-                        if let Err(e) = fut.await {
+                        if let Err(e) = self
+                            .scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst)
+                            .await
+                        {
                             if ctx.is_cancelled() {
                                 self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
                                     .await;
@@ -2662,7 +2735,7 @@ impl FolderScanner {
                         self.update_cache.delete_recursive(&h);
                         self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
                         self.send_update().await;
-                        self.maybe_send_checkpoint();
+                        self.maybe_send_checkpoint(&folder, &this_hash, into);
                     }
                 }
             }
@@ -2803,6 +2876,9 @@ pub(crate) async fn scan_data_folder_scoped(
     cache.info.tier_registry_generation = Some(tier_registry.generation);
 
     let resume_frontier = cache.validated_scan_frontier().map(str::to_owned);
+    if resume_frontier.is_some() {
+        global_metrics().record_scanner_checkpoint_used();
+    }
     let bucket_name = cache.info.name.clone();
     let next_cycle = cache.info.next_cycle;
     let had_scan_checkpoint = cache.info.scan_checkpoint.is_some();
@@ -2834,6 +2910,8 @@ pub(crate) async fn scan_data_folder_scoped(
         disks_quorum,
         updates,
         checkpoint_tx,
+        checkpoint_ancestors: Vec::new(),
+        checkpoint_depth: 0,
         last_update: SystemTime::UNIX_EPOCH,
         checkpoint_objects: 0,
         last_checkpoint_objects: 0,
@@ -2988,6 +3066,13 @@ pub(crate) async fn scan_data_folder_scoped(
             }
             if matches!(&e, ScannerError::Io(io) if io.kind() == ErrorKind::NotFound) {
                 let mut partial_cache = scanner.old_cache;
+                // An empty input still needs a root for the non-publishable
+                // partial result; retain nonempty durable caches unchanged.
+                if partial_cache.cache.is_empty() {
+                    partial_cache
+                        .cache
+                        .insert(hash_path(&bucket_name).0, DataUsageEntry::default());
+                }
                 partial_cache.info.last_update = Some(SystemTime::now());
                 partial_cache.info.next_cycle = next_cycle;
                 partial_cache.info.snapshot_complete = false;

crates/scanner/src/scanner_folder/tests.rs

@@ -343,6 +343,8 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         disks_quorum: 0,
         updates: None,
         checkpoint_tx: None,
+        checkpoint_ancestors: Vec::new(),
+        checkpoint_depth: 0,
         last_update: SystemTime::UNIX_EPOCH,
         checkpoint_objects: 0,
         last_checkpoint_objects: 0,
@@ -409,14 +411,21 @@ async fn periodic_checkpoint_emits_at_object_threshold_without_wall_clock_wait()
         .seal_scan_frontier(Some("bucket/a"))
         .expect("fixture frontier");
     assert_eq!(scanner.new_cache.validated_scan_frontier(), Some("bucket/a"));
+    let checkpoint_folder = CachedFolder {
+        name: "bucket".to_string(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+    let checkpoint_hash = hash_path("bucket");
+    let checkpoint_entry = scanner.new_cache.root().expect("fixture root");
 
-    scanner.maybe_send_checkpoint();
+    scanner.maybe_send_checkpoint(&checkpoint_folder, &checkpoint_hash, &checkpoint_entry);
 
     scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL * 2;
     scanner.last_checkpoint_at = Instant::now()
         .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
         .expect("test instant subtraction");
-    scanner.maybe_send_checkpoint();
+    scanner.maybe_send_checkpoint(&checkpoint_folder, &checkpoint_hash, &checkpoint_entry);
     assert_eq!(
         scanner.last_checkpoint_objects, SCANNER_CHECKPOINT_OBJECT_INTERVAL,
         "a full checkpoint queue must reject before cloning or advancing progress"
@@ -427,11 +436,73 @@ async fn periodic_checkpoint_emits_at_object_threshold_without_wall_clock_wait()
     assert!(!checkpoint.info.snapshot_complete);
     assert_eq!(scanner.last_checkpoint_objects, SCANNER_CHECKPOINT_OBJECT_INTERVAL);
     assert!(checkpoint_rx.try_recv().is_err(), "checkpoint queue remains bounded");
+    scanner
+        .new_cache
+        .cache
+        .insert("bucket/orphan".to_string(), DataUsageEntry::default());
+    scanner.maybe_send_checkpoint(&checkpoint_folder, &checkpoint_hash, &checkpoint_entry);
+    assert!(
+        checkpoint_rx.try_recv().is_err(),
+        "unreachable observations must not become resumable snapshots"
+    );
+    assert_eq!(
+        scanner.last_checkpoint_objects,
+        SCANNER_CHECKPOINT_OBJECT_INTERVAL * 2,
+        "an invalid attempt must still consume the producer interval"
+    );
+    scanner.new_cache.cache.remove("bucket/orphan");
+    scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL * 3;
+    scanner.maybe_send_checkpoint(&checkpoint_folder, &checkpoint_hash, &checkpoint_entry);
+    assert!(checkpoint_rx.try_recv().is_err(), "a repaired graph still waits for the minimum interval");
+
     tokio::fs::remove_dir_all(temp_dir)
         .await
         .expect("remove test scanner directory");
 }
 
+#[tokio::test]
+async fn checkpoint_ancestor_depth_limit_preserves_suspended_accounting() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir),
+    };
+    let (tx, _rx) = mpsc::channel(1);
+    scanner.checkpoint_tx = Some(tx);
+    scanner.old_cache.info.scan_progress = Some(crate::DataUsageScanProgress {
+        started_plan: crate::DataUsageScanPlanDigest([1; 32]),
+        requested_plan: crate::DataUsageScanPlanDigest([1; 32]),
+    });
+    scanner.checkpoint_depth = MAX_DATA_USAGE_CACHE_DEPTH;
+    scanner.checkpoint_ancestors = (0..MAX_DATA_USAGE_CACHE_DEPTH)
+        .map(|_| ScannerCheckpointAncestor::default())
+        .collect();
+    let mut hash = hash_path("bucket/parent");
+    let mut parent = DataUsageEntry {
+        objects: 3,
+        size: 7,
+        ..Default::default()
+    };
+    let mut child = DataUsageEntry::default();
+    let error = scanner
+        .scan_child_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket/parent/child".to_string(),
+                parent: Some(hash.clone()),
+                object_heal_prob_div: 1,
+            },
+            &mut hash,
+            &mut parent,
+            &mut child,
+        )
+        .await
+        .expect_err("unsupported depth must stop before recursion");
+    assert!(matches!(error, ScannerError::Other(message) if message.contains("depth limit")));
+    assert_eq!(hash, hash_path("bucket/parent"));
+    assert_eq!((parent.objects, parent.size), (3, 7));
+    assert_eq!(scanner.checkpoint_depth, MAX_DATA_USAGE_CACHE_DEPTH);
+}
+
 struct TestGuard {
     temp_dir: Option<std::path::PathBuf>,
 }
@@ -1658,6 +1729,51 @@ fn metadata_for_object_version(bucket: &str, object: &str, version_id: Option<Uu
     meta.marshal_msg().expect("test metadata should marshal")
 }
 
+#[tokio::test]
+#[serial]
+async fn test_scan_folder_keeps_old_cache_bounded_by_loaded_entries() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
+    scanner.old_cache.info.name = "bucket".to_string();
+    scanner.new_cache.info.name = "bucket".to_string();
+    scanner.update_cache.info.name = "bucket".to_string();
+    scanner.is_erasure_mode = true;
+    for directory in 0..16 {
+        for object in 0..32 {
+            write_test_object_metadata(&temp_dir, "bucket", &format!("dir-{directory:03}/obj-{object:03}")).await;
+        }
+    }
+    let initial_capacity = scanner.old_cache.cache.capacity();
+    let mut into = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut into,
+        )
+        .await
+        .expect("scan object directory tree");
+    assert_eq!(
+        scanner
+            .new_cache
+            .size_recursive(&hash_path("bucket").0)
+            .expect("scanned root totals")
+            .objects,
+        512,
+        "all object directories must still be counted",
+    );
+    assert!(scanner.old_cache.cache.is_empty(), "lookups must not retain visited object placeholders");
+    assert_eq!(
+        scanner.old_cache.cache.capacity(),
+        initial_capacity,
+        "old cache must not allocate a growing table"
+    );
+}
+
 async fn write_test_object_metadata(root: &std::path::Path, bucket: &str, object: &str) {
     write_test_object_metadata_bytes(root, bucket, object, &metadata_for_object(bucket, object)).await;
 }
@@ -3277,6 +3393,62 @@ async fn scan_data_folder_missing_scan_root_returns_partial() {
     assert_eq!(root.size, 0);
 }
 
+#[tokio::test]
+#[serial]
+async fn scan_data_folder_missing_scan_root_preserves_disconnected_durable_cache() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    tokio::fs::remove_dir_all(&temp_dir).await.expect("remove scanner root");
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir),
+    };
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&parent, Default::default());
+    let mut cache = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: "missing-bucket".to_string(),
+            next_cycle: 9,
+            snapshot_complete: true,
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    let orphan = hash_path("missing-bucket/orphan");
+    cache.cache.insert(
+        orphan.0.clone(),
+        DataUsageEntry {
+            objects: 7,
+            size: 70,
+            ..Default::default()
+        },
+    );
+    let retained_key_ptr = cache.cache.keys().next().expect("durable orphan key").as_ptr();
+    let result = scan_data_folder(
+        budget.token(),
+        budget,
+        vec![scanner.local_disk.clone()],
+        scanner.local_disk,
+        cache,
+        None,
+        HealScanMode::Normal,
+        SCANNER_SLEEPER.clone(),
+    )
+    .await;
+    let partial = match result {
+        Err(ScannerError::NamespaceNotFoundCache(partial)) => partial,
+        other => panic!("missing namespace must remain incomplete, got {other:?}"),
+    };
+    assert!(!partial.info.snapshot_complete);
+    assert_eq!(partial.info.next_cycle, 9);
+    assert_eq!(partial.cache.len(), 1);
+    assert_eq!(partial.cache.keys().next().expect("retained orphan key").as_ptr(), retained_key_ptr);
+    assert!(
+        partial.root().is_none(),
+        "a disconnected durable cache must not gain an invented zero root"
+    );
+    assert!(partial.checked_flatten_complete_scope("missing-bucket").is_none());
+    assert_eq!((partial.cache[&orphan.0].objects, partial.cache[&orphan.0].size), (7, 70));
+}
+
 #[tokio::test]
 #[serial]
 async fn test_scan_data_folder_resume_hint_orders_across_new_and_existing_folders() {

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

@@ -21,6 +21,7 @@ use std::io::Cursor;
 use tokio::io::AsyncReadExt;
 
 mod deep_compacted;
+mod periodic_snapshot;
 mod segment_observation;
 
 const CACHE_NAME: &str = "bucket/checkpoint-fixture.bin";

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

@@ -0,0 +1,794 @@
+// Copyright 2026 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.
+
+//! Periodic snapshots captured while nested traversal frames are still active.
+
+use super::*;
+use tokio::sync::Notify;
+
+const OBJECTS: u64 = 8;
+
+fn prepare_scanner(scanner: &mut FolderScanner, cache: DataUsageCache) {
+    scanner.resume_frontier = cache.validated_scan_frontier().map(str::to_owned);
+    scanner.coverage_frontier = scanner.resume_frontier.clone();
+    scanner.new_cache = DataUsageCache {
+        info: cache.info.clone(),
+        ..Default::default()
+    };
+    scanner.update_cache = DataUsageCache {
+        info: cache.info.clone(),
+        ..Default::default()
+    };
+    scanner.old_cache = cache;
+    scanner.is_erasure_mode = true;
+    scanner.skip_heal.store(true, Ordering::SeqCst);
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+}
+
+async fn nested_periodic_snapshot(
+    corrupt_first: bool,
+    root_scalars: (usize, usize),
+) -> (DataUsageCache, crate::DataUsageScanIdentity, FolderScanner, TestGuard) {
+    let (mut scanner, root) = build_test_scanner().await;
+    let guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for branch in ["a", "b"] {
+        for object in 0..4 {
+            write_checkpoint_object(&root, &format!("prefix/{branch}/{object:04}"), &[(None, 1)]).await;
+        }
+    }
+    if corrupt_first {
+        write_test_object_metadata_bytes(&root, "bucket", "prefix/a/0000", b"not-valid-filemeta").await;
+    }
+    let (_, identity) = bound_checkpoint();
+    let mut cache = DataUsageCache::default();
+    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+    prepare_scanner(&mut scanner, cache);
+    let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel(1);
+    scanner.checkpoint_tx = Some(checkpoint_tx);
+    scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL - 2;
+    scanner.last_checkpoint_at = Instant::now()
+        .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+        .expect("seed elapsed checkpoint interval");
+    let paused = Arc::new(Notify::new());
+    let release = Arc::new(Notify::new());
+    let pause_path = if corrupt_first {
+        "bucket/prefix/a/0003"
+    } else {
+        "bucket/prefix/a/0002"
+    };
+    scanner.update_current_path = Arc::new({
+        let paused = paused.clone();
+        let release = release.clone();
+        move |path: &str| {
+            let pause_here = path == pause_path;
+            let paused = paused.clone();
+            let release = release.clone();
+            Box::pin(async move {
+                if pause_here {
+                    paused.notify_one();
+                    release.notified().await;
+                }
+            })
+        }
+    });
+    let parent = CancellationToken::new();
+    scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
+    let budget = scanner.budget.clone();
+    let mut entry = DataUsageEntry {
+        objects: root_scalars.0,
+        size: root_scalars.1,
+        ..Default::default()
+    };
+    let snapshot = {
+        let scan = scanner.scan_folder(
+            parent.clone(),
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut entry,
+        );
+        tokio::pin!(scan);
+        let snapshot = tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                snapshot = checkpoint_rx.recv() => snapshot.expect("real walk emits periodic snapshot"),
+                result = &mut scan => panic!("walk finished before periodic snapshot: {result:?}"),
+            }
+        })
+        .await
+        .expect("bounded nested checkpoint emission");
+        tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                () = paused.notified() => {},
+                result = &mut scan => panic!("nested ancestors unwound before capture: {result:?}"),
+            }
+        })
+        .await
+        .expect("walk pauses before third healthy object");
+        assert_eq!(budget.progress().0, 2, "capture must precede ancestor unwind and further object reads");
+        assert!(!parent.is_cancelled(), "periodic capture must not use cancellation linking");
+        snapshot
+    };
+    assert!(!snapshot.info.snapshot_complete);
+    assert!(
+        snapshot.validated_raw_enumeration_cursor().is_some() || snapshot.validated_raw_enumeration_page_index().is_some(),
+        "enumeration progress must coexist with completed subtree coverage"
+    );
+    let totals = snapshot
+        .checked_flatten_complete_scope("bucket")
+        .expect("periodic snapshot must connect every active ancestor to the bucket root");
+    assert_eq!(
+        totals.objects,
+        2 + root_scalars.0,
+        "snapshot must retain suspended ancestor and current frame object accounting"
+    );
+    assert_eq!(
+        totals.size,
+        2 + root_scalars.1,
+        "suspended ancestor and current frame scalar bytes must survive snapshot construction"
+    );
+    let store = FixtureStore::new();
+    let revisions = DataUsageCache::default()
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("initial periodic fixture revisions");
+    snapshot
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("persist real mid-walk periodic snapshot");
+    let mut loaded = DataUsageCache::default();
+    loaded
+        .load(store.clone(), CACHE_NAME)
+        .await
+        .expect("reload real periodic snapshot");
+    assert_eq!(loaded.info.scan_coverage_receipt, snapshot.info.scan_coverage_receipt);
+    assert_eq!(loaded.info.scan_raw_enumeration_cursor, snapshot.info.scan_raw_enumeration_cursor);
+    assert_eq!(
+        loaded
+            .checked_flatten_complete_scope("bucket")
+            .expect("reloaded complete scope")
+            .objects,
+        2 + root_scalars.0
+    );
+    assert_eq!(
+        loaded.prepare_bucket_checkpoint("bucket", 11, 8, SOURCE, PLAN, identity),
+        crate::DataUsageCachePrepareOutcome::Reused,
+        "a higher leader must adopt the persisted periodic snapshot"
+    );
+    (loaded, identity, scanner, guard)
+}
+
+async fn finish_resumed_walk(cache: DataUsageCache, original: &FolderScanner) -> (DataUsageCache, u64, TestGuard) {
+    let (mut scanner, unused_root) = build_test_scanner().await;
+    let guard = TestGuard {
+        temp_dir: Some(unused_root),
+    };
+    scanner.root = original.root.clone();
+    scanner.local_disk = original.local_disk.clone();
+    prepare_scanner(&mut scanner, cache);
+    let parent = CancellationToken::new();
+    scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
+    let mut entry = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            parent,
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut entry,
+        )
+        .await
+        .expect("finish resumed nested periodic walk");
+    let visits = scanner.budget.progress().0;
+    (scanner.new_cache, visits, guard)
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_resumes_after_leader_handoff_with_exact_totals() {
+    let (loaded, _, original, _guard) = nested_periodic_snapshot(false, (0, 0)).await;
+    assert_eq!(loaded.validated_scan_frontier(), Some("bucket/prefix/a/0000"));
+    let (finished, visits, _unused_guard) = finish_resumed_walk(loaded, &original).await;
+    assert_eq!(visits, OBJECTS - 1, "the completed nested sibling must skip its metadata read");
+    let totals = finished
+        .checked_flatten_complete_scope("bucket")
+        .expect("finished connected root");
+    assert_eq!(u64::try_from(totals.objects).expect("object count"), OBJECTS);
+    assert_eq!(u64::try_from(totals.size).expect("byte count"), OBJECTS);
+    assert_eq!(totals.failed_objects, 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_keeps_failed_prefix_outside_resume_frontier() {
+    let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await;
+    assert!(
+        loaded.validated_scan_frontier().is_none(),
+        "an earlier corrupt object must block subtree skipping"
+    );
+    assert!(!loaded.info.failed_objects.is_empty(), "persisted failure debt must survive handoff");
+    let (finished, visits, _unused_guard) = finish_resumed_walk(loaded, &original).await;
+    assert_eq!(visits, OBJECTS - 1, "every healthy object must be re-read after the coverage gap");
+    let totals = finished
+        .checked_flatten_complete_scope("bucket")
+        .expect("failed sweep still has connected accounting");
+    assert_eq!(u64::try_from(totals.objects).expect("healthy count"), OBJECTS - 1);
+    assert_eq!(u64::try_from(totals.size).expect("healthy bytes"), OBJECTS - 1);
+    assert!(!finished.info.failed_objects.is_empty(), "unrepaired metadata must remain pending");
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_preserves_suspended_ancestor_scalars_through_reload() {
+    let (loaded, _, _, _guard) = nested_periodic_snapshot(false, (3, 21)).await;
+    let root = loaded.root().expect("reloaded root accumulator");
+    assert_eq!(root.objects, 3, "suspended root object accounting must not become an empty topology node");
+    assert_eq!(root.size, 21, "suspended root bytes must remain independent of descendant totals");
+    let totals = loaded
+        .checked_flatten_complete_scope("bucket")
+        .expect("reloaded ancestor accounting");
+    assert_eq!(totals.objects, 5);
+    assert_eq!(totals.size, 23);
+    assert_eq!(loaded.validated_scan_frontier(), Some("bucket/prefix/a/0000"));
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_keeps_durable_frontier_across_early_repeated_handoffs() {
+    let (cache, identity, original, _guard) = nested_periodic_snapshot(false, (0, 0)).await;
+    let store = FixtureStore::new();
+    let revisions = DataUsageCache::default()
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("initial repeated-handoff revisions");
+    cache
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("save the inherited valid subtree frontier");
+    let (mut scanner, unused_root) = build_test_scanner().await;
+    let _unused_guard = TestGuard {
+        temp_dir: Some(unused_root),
+    };
+    scanner.root = original.root.clone();
+    scanner.local_disk = original.local_disk.clone();
+    prepare_scanner(&mut scanner, cache);
+    let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel(1);
+    scanner.checkpoint_tx = Some(checkpoint_tx);
+    scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL;
+    scanner.last_checkpoint_at = Instant::now()
+        .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+        .expect("early resumed emitter cadence");
+    let folder = CachedFolder {
+        name: "bucket".to_string(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+    scanner.record_raw_enumeration_entry("bucket", "prefix");
+    scanner.maybe_send_checkpoint(&folder, &hash_path("bucket"), &DataUsageEntry::default());
+    assert!(
+        matches!(checkpoint_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+        "an early raw-only snapshot must not overwrite inherited completed coverage"
+    );
+    assert_eq!(scanner.last_checkpoint_objects, SCANNER_CHECKPOINT_OBJECT_INTERVAL);
+    let mut inherited = DataUsageCache::default();
+    let revisions = inherited
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("reload unchanged durable frontier after deferred early emission");
+    assert_eq!(inherited.validated_scan_frontier(), Some("bucket/prefix/a/0000"));
+    assert_eq!(
+        inherited.prepare_bucket_checkpoint("bucket", 11, 9, SOURCE, PLAN, identity),
+        crate::DataUsageCachePrepareOutcome::Reused
+    );
+    assert_eq!(inherited.validated_scan_frontier(), Some("bucket/prefix/a/0000"));
+    prepare_scanner(&mut scanner, inherited);
+    scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL - 2;
+    scanner.last_checkpoint_objects = 0;
+    scanner.last_checkpoint_at = Instant::now()
+        .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+        .expect("later useful resumed emitter cadence");
+    let parent = CancellationToken::new();
+    scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
+    let mut entry = DataUsageEntry::default();
+    let candidate = {
+        let scan = scanner.scan_folder(parent, folder, &mut entry);
+        tokio::pin!(scan);
+        tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                checkpoint = checkpoint_rx.recv() => checkpoint.expect("resumed walk can later emit useful coverage"),
+                result = &mut scan => {
+                    result.expect("later resumed walk completes");
+                    checkpoint_rx.try_recv().expect("completed walk emitted a useful periodic checkpoint")
+                }
+            }
+        })
+        .await
+        .expect("bounded later useful emission")
+    };
+    assert_eq!(candidate.validated_scan_frontier(), Some("bucket/prefix/a/0001"));
+    assert_eq!(
+        candidate
+            .checked_flatten_complete_scope("bucket")
+            .expect("later snapshot connected accounting")
+            .objects,
+        3,
+        "inherited completed object and both newly read objects must survive"
+    );
+    candidate
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("persist later coverage without losing the inherited prefix");
+    let mut latest = DataUsageCache::default();
+    latest
+        .load(store, CACHE_NAME)
+        .await
+        .expect("reload later periodic checkpoint");
+    assert_eq!(
+        latest.prepare_bucket_checkpoint("bucket", 11, 10, SOURCE, PLAN, identity),
+        crate::DataUsageCachePrepareOutcome::Reused
+    );
+    assert_eq!(latest.validated_scan_frontier(), Some("bucket/prefix/a/0001"));
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_defers_unfinished_paths_at_or_before_completed_frontier() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard { temp_dir: Some(root) };
+    let (_, identity) = bound_checkpoint();
+    let mut cache = DataUsageCache::default();
+    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+    cache.replace("bucket", "", DataUsageEntry::default());
+    cache.replace(
+        "bucket/z",
+        "bucket",
+        DataUsageEntry {
+            objects: 3,
+            size: 9,
+            ..Default::default()
+        },
+    );
+    cache
+        .seal_scan_frontier(Some("bucket/z"))
+        .expect("completed later sibling coverage");
+    prepare_scanner(&mut scanner, cache.clone());
+    scanner.new_cache = cache;
+    let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel(1);
+    scanner.checkpoint_tx = Some(checkpoint_tx);
+    let partial = DataUsageEntry {
+        objects: 2,
+        size: 5,
+        ..Default::default()
+    };
+    for active in ["bucket/a", "bucket/z"] {
+        scanner.record_raw_enumeration_entry(active, STORAGE_FORMAT_FILE);
+        scanner.checkpoint_objects += SCANNER_CHECKPOINT_OBJECT_INTERVAL;
+        scanner.last_checkpoint_at = Instant::now()
+            .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+            .expect("unordered active path cadence");
+        scanner.maybe_send_checkpoint(
+            &CachedFolder {
+                name: active.to_string(),
+                parent: Some(hash_path("bucket")),
+                object_heal_prob_div: 1,
+            },
+            &hash_path(active),
+            &partial,
+        );
+        assert!(
+            matches!(checkpoint_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+            "unfinished path {active} must not become certified completed coverage"
+        );
+        assert_eq!(
+            scanner.last_checkpoint_objects, scanner.checkpoint_objects,
+            "deferred paths still consume cadence"
+        );
+    }
+    scanner.new_cache.replace("bucket/a", "bucket", partial);
+    let root_entry = scanner.new_cache.root().expect("completed subtree owner");
+    scanner.checkpoint_objects += SCANNER_CHECKPOINT_OBJECT_INTERVAL;
+    scanner.last_checkpoint_at = Instant::now()
+        .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+        .expect("completed owner cadence");
+    scanner.maybe_send_checkpoint(
+        &CachedFolder {
+            name: "bucket".to_string(),
+            parent: None,
+            object_heal_prob_div: 1,
+        },
+        &hash_path("bucket"),
+        &root_entry,
+    );
+    let checkpoint = checkpoint_rx.try_recv().expect("completed subtree owner may safely emit");
+    assert_eq!(checkpoint.validated_scan_frontier(), Some("bucket/z"));
+    let totals = checkpoint
+        .checked_flatten_complete_scope("bucket")
+        .expect("both completed sibling links remain reachable");
+    assert_eq!(totals.objects, 5);
+    assert_eq!(totals.size, 14);
+    assert_eq!(checkpoint.cache.len(), 3);
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_checks_suspended_ancestors_before_certifying_frontier() {
+    for (ancestor, frontier, may_emit) in [
+        ("bucket/z", "bucket/z", false),
+        ("bucket/a", "bucket/a-", false),
+        ("bucket/a", "bucket/a/done", true),
+    ] {
+        let (mut scanner, root) = build_test_scanner().await;
+        let _guard = TestGuard { temp_dir: Some(root) };
+        let (_, identity) = bound_checkpoint();
+        let mut cache = DataUsageCache::default();
+        cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+        cache.replace("bucket", "", DataUsageEntry::default());
+        cache.replace(ancestor, "bucket", DataUsageEntry::default());
+        let completed = format!("{ancestor}/done");
+        cache.replace(
+            &completed,
+            ancestor,
+            DataUsageEntry {
+                objects: 3,
+                size: 9,
+                ..Default::default()
+            },
+        );
+        let sibling_frontier = frontier != ancestor && frontier != completed;
+        if sibling_frontier {
+            cache.replace(
+                frontier,
+                "bucket",
+                DataUsageEntry {
+                    objects: 3,
+                    size: 9,
+                    ..Default::default()
+                },
+            );
+        }
+        cache.seal_scan_frontier(Some(frontier)).expect("initial completed coverage");
+        prepare_scanner(&mut scanner, cache);
+        scanner.new_cache.replace("bucket", "", DataUsageEntry::default());
+        scanner
+            .new_cache
+            .copy_with_children(&scanner.old_cache, &hash_path(&completed), &Some(hash_path(ancestor)));
+        if sibling_frontier {
+            scanner
+                .new_cache
+                .copy_with_children(&scanner.old_cache, &hash_path(frontier), &Some(hash_path("bucket")));
+        }
+        let root_entry = scanner.new_cache.root().expect("suspended root owner").clone();
+        let mut ancestor_entry = scanner.new_cache.find(ancestor).expect("suspended subtree owner").clone();
+        ancestor_entry.objects = 1;
+        ancestor_entry.size = 2;
+        scanner.checkpoint_ancestors = vec![
+            ScannerCheckpointAncestor {
+                hash: hash_path("bucket"),
+                entry: root_entry,
+            },
+            ScannerCheckpointAncestor {
+                hash: hash_path(ancestor),
+                entry: ancestor_entry,
+            },
+        ];
+        scanner.checkpoint_depth = 2;
+        let current = format!("{ancestor}/next");
+        assert!(
+            current.as_str() > frontier,
+            "the current-folder-only guard must not already reject this fixture"
+        );
+        scanner.record_raw_enumeration_entry(&current, STORAGE_FORMAT_FILE);
+        let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel(1);
+        scanner.checkpoint_tx = Some(checkpoint_tx);
+        scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL;
+        scanner.last_checkpoint_at = Instant::now()
+            .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+            .expect("suspended ancestor emitter cadence");
+        scanner.maybe_send_checkpoint(
+            &CachedFolder {
+                name: current.clone(),
+                parent: Some(hash_path(ancestor)),
+                object_heal_prob_div: 1,
+            },
+            &hash_path(&current),
+            &DataUsageEntry {
+                objects: 2,
+                size: 5,
+                ..Default::default()
+            },
+        );
+        assert_eq!(scanner.last_checkpoint_objects, SCANNER_CHECKPOINT_OBJECT_INTERVAL);
+        if may_emit {
+            let snapshot = checkpoint_rx
+                .try_recv()
+                .expect("an ancestor containing the frontier may safely emit");
+            assert_eq!(snapshot.validated_scan_frontier(), Some(frontier));
+            let totals = snapshot
+                .checked_flatten_complete_scope("bucket")
+                .expect("all suspended and current accounting connected");
+            assert_eq!(totals.objects, 6);
+            assert_eq!(totals.size, 16);
+            assert_eq!(snapshot.cache.len(), 4);
+        } else {
+            assert!(
+                matches!(checkpoint_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+                "unfinished ancestor {ancestor} must not be certified through {frontier}"
+            );
+        }
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_partial_preservation_invalidates_covered_child_but_keeps_inner_frontier() {
+    for (child, frontier, frontier_inside_child, expected_objects) in [
+        ("bucket/a", "bucket/z", false, 5),
+        ("bucket/z", "bucket/z", false, 2),
+        ("bucket/parent", "bucket/parent/done", true, 5),
+    ] {
+        let (mut scanner, root) = build_test_scanner().await;
+        let _guard = TestGuard { temp_dir: Some(root) };
+        let (_, identity) = bound_checkpoint();
+        let mut cache = DataUsageCache::default();
+        cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+        cache.replace("bucket", "", DataUsageEntry::default());
+        if frontier_inside_child {
+            cache.replace(child, "bucket", DataUsageEntry::default());
+        }
+        cache.replace(
+            frontier,
+            if frontier_inside_child { child } else { "bucket" },
+            DataUsageEntry {
+                objects: 3,
+                size: 9,
+                ..Default::default()
+            },
+        );
+        cache
+            .seal_scan_frontier(Some(frontier))
+            .expect("initial completed child coverage");
+        prepare_scanner(&mut scanner, cache.clone());
+        scanner.new_cache = cache;
+        let mut parent_entry = scanner.new_cache.root().expect("partial child owner");
+        let child_entry = DataUsageEntry {
+            objects: 2,
+            size: 5,
+            ..Default::default()
+        };
+        scanner
+            .preserve_partial_child_progress(&Some(hash_path("bucket")), &hash_path(child), &mut parent_entry, &child_entry)
+            .await;
+        let retained_frontier = scanner.coverage_frontier.clone();
+        scanner.new_cache.replace_hashed(&hash_path("bucket"), &None, &parent_entry);
+        scanner
+            .new_cache
+            .seal_scan_frontier(retained_frontier.as_deref())
+            .expect("seal the final partial cache after child preservation");
+        let totals = scanner
+            .new_cache
+            .checked_flatten_complete_scope("bucket")
+            .expect("preserved partial child and completed siblings remain connected");
+        assert_eq!(totals.objects, expected_objects, "partial child {child} must be counted once");
+        assert_eq!(totals.size, if expected_objects == 2 { 5 } else { 14 });
+        if frontier_inside_child {
+            assert!(!scanner.coverage_gap);
+            assert_eq!(scanner.new_cache.validated_scan_frontier(), Some(frontier));
+        } else {
+            assert!(
+                scanner.coverage_gap,
+                "unfinished covered child {child} must block further frontier advancement"
+            );
+            assert!(retained_frontier.is_none());
+            assert!(scanner.new_cache.info.scan_coverage_receipt.is_none());
+            assert!(scanner.new_cache.validated_scan_frontier().is_none());
+        }
+    }
+}
+
+async fn assert_nested_frame_restoration(max_objects: Option<u64>) {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for branch in ["a", "b"] {
+        for object in 0..4 {
+            write_checkpoint_object(&root, &format!("prefix/{branch}/{object:04}"), &[(None, 1)]).await;
+        }
+    }
+    let (_, identity) = bound_checkpoint();
+    let mut cache = DataUsageCache::default();
+    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+    prepare_scanner(&mut scanner, cache);
+    let (checkpoint_tx, _checkpoint_rx) = mpsc::channel(1);
+    scanner.checkpoint_tx = Some(checkpoint_tx);
+    let parent = CancellationToken::new();
+    scanner.budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        ScannerCycleBudgetConfig {
+            max_objects,
+            ..Default::default()
+        },
+    );
+    let mut entry = DataUsageEntry {
+        objects: usize::from(max_objects.is_some()) * 3,
+        size: usize::from(max_objects.is_some()) * 21,
+        ..Default::default()
+    };
+    let result = scanner
+        .scan_folder(
+            scanner.budget.token(),
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut entry,
+        )
+        .await;
+    match max_objects {
+        Some(_) => {
+            assert!(result.is_err(), "object budget must interrupt the actual nested walk");
+            assert_eq!(scanner.budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects));
+            assert!(
+                scanner.budget.token().is_cancelled(),
+                "partial preservation must follow actual cancellation"
+            );
+            assert_eq!(
+                entry.objects, 3,
+                "suspended root object accounting must be restored before returning the error"
+            );
+            assert_eq!(
+                entry.size, 21,
+                "suspended root byte accounting must be restored before cancellation linking"
+            );
+        }
+        None => result.expect("complete real nested walk with periodic sender enabled"),
+    }
+    let totals = scanner
+        .new_cache
+        .checked_flatten_complete_scope("bucket")
+        .expect("complete nested accounting");
+    let expected = max_objects.unwrap_or(OBJECTS);
+    assert_eq!(u64::try_from(totals.objects).expect("object count"), expected);
+    assert_eq!(u64::try_from(totals.size).expect("byte count"), expected);
+    assert_eq!(scanner.checkpoint_depth, 0, "every completed child must restore its parent frame");
+    assert_eq!(
+        scanner.checkpoint_ancestors.len(),
+        3,
+        "sibling count must not increase depth-bounded spare storage"
+    );
+    for spare in &scanner.checkpoint_ancestors {
+        assert!(spare.hash.0.is_empty(), "reusable frame must release its suspended path");
+        assert!(
+            !data_usage_root_has_progress(&spare.entry),
+            "completed frame must not retain ancestor accounting"
+        );
+        assert!(!spare.entry.compacted);
+        assert!(spare.entry.obj_sizes.is_empty());
+        assert!(spare.entry.obj_versions.is_empty());
+        assert!(spare.entry.tier_accounting_proof.is_none());
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_reuses_empty_ancestor_frames_after_complete_walk() {
+    assert_nested_frame_restoration(None).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_restores_ancestor_frames_before_cancellation_linking() {
+    assert_nested_frame_restoration(Some(2)).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_adoption_records_checkpoint_used_on_real_scan_startup() {
+    let (scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    tokio::fs::create_dir_all(root.join("bucket/static"))
+        .await
+        .expect("create the real cached subtree directory");
+    let (mut cache, _) = bound_checkpoint();
+    cache.info.skip_healing = true;
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        ScannerCycleBudgetConfig {
+            max_directories: Some(1),
+            ..Default::default()
+        },
+    );
+    let before = global_metrics().report().await.scan_checkpoint_used;
+    let completed = scan_data_folder(
+        budget.token(),
+        budget.clone(),
+        vec![scanner.local_disk.clone()],
+        scanner.local_disk,
+        cache,
+        None,
+        HealScanMode::Normal,
+        DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
+    )
+    .await
+    .expect("validated frontier must skip the child within a root-only directory budget");
+    let after = global_metrics().report().await.scan_checkpoint_used;
+    assert_eq!(after, before + 1, "adopting a verified forward-sweep frontier must be observable");
+    assert!(completed.info.snapshot_complete);
+    assert_eq!(
+        completed
+            .checked_flatten_complete_scope("bucket")
+            .expect("completed adopted scope")
+            .objects,
+        3
+    );
+    assert_eq!(budget.progress().1, 1, "adoption must avoid entering the completed subtree");
+    assert_eq!(budget.reason(), None);
+}
+
+#[tokio::test]
+#[serial]
+async fn periodic_nested_snapshot_does_not_publish_shared_compacted_accumulator_under_child_path() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for object in 0..4 {
+        write_checkpoint_object(&root, &format!("prefix/nested/{object:04}"), &[(None, 1)]).await;
+    }
+    let (_, identity) = bound_checkpoint();
+    let mut cache = DataUsageCache::default();
+    cache.prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+    prepare_scanner(&mut scanner, cache);
+    let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel(1);
+    scanner.checkpoint_tx = Some(checkpoint_tx);
+    scanner.checkpoint_objects = SCANNER_CHECKPOINT_OBJECT_INTERVAL - 2;
+    scanner.last_checkpoint_at = Instant::now()
+        .checked_sub(SCANNER_CHECKPOINT_MIN_INTERVAL)
+        .expect("seed compacted checkpoint interval");
+    let mut entry = DataUsageEntry {
+        compacted: true,
+        ..Default::default()
+    };
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut entry,
+        )
+        .await
+        .expect("walk with shared compacted owner accumulator");
+    assert_eq!(entry.objects, 4);
+    assert_eq!(entry.size, 4);
+    assert!(
+        entry.children.is_empty(),
+        "compacted totals must not acquire independently accounted child links"
+    );
+    assert!(
+        matches!(checkpoint_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+        "a shared compacted parent accumulator must not be snapshotted as the current child"
+    );
+}

crates/scanner/src/scanner_io/cache.rs

@@ -28,7 +28,8 @@ pub(crate) fn scanner_cache_lock_timeout() -> Duration {
 pub(crate) enum ScannerCheckpointPersistResult {
     Saved,
     FenceChanged,
-    Failed(StorageError),
+    Skipped,
+    RetryBucket(StorageError),
 }
 
 pub(crate) struct ScannerCheckpointPersistContext<'a> {
@@ -80,7 +81,12 @@ where
     )
     .await;
     if let Err(error) = checkpoint_save {
-        return ScannerCheckpointPersistResult::Failed(error);
+        record_checkpoint_save_failure(cache_name, checkpoint, &error);
+        return if matches!(error, StorageError::PreconditionFailed) {
+            ScannerCheckpointPersistResult::RetryBucket(error)
+        } else {
+            ScannerCheckpointPersistResult::Skipped
+        };
     }
 
     if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, fence_store)
@@ -101,12 +107,48 @@ where
     {
         Ok(next_revisions) => {
             *revisions = next_revisions;
+            if let Some(source) = checkpoint.info.source {
+                let saved = checkpoint.info.scan_checkpoint.as_ref().map(|cursor| {
+                    rustfs_scanner_metrics::metrics::ScannerCheckpointReport {
+                        version: cursor.version,
+                        resume_after: cursor.resume_after.clone(),
+                        reason: cursor.reason.as_str().to_string(),
+                        last_event: "saved".to_string(),
+                    }
+                });
+                global_metrics().record_scanner_checkpoint_saved(
+                    source.pool_index,
+                    source.set_index,
+                    &checkpoint.info.name,
+                    saved,
+                );
+            }
             ScannerCheckpointPersistResult::Saved
         }
-        Err(error) => ScannerCheckpointPersistResult::Failed(error),
+        Err(error) => {
+            // A successful PUT consumed the old CAS revision. Reload the bucket
+            // before another write instead of rebasing this scan over a newer writer.
+            record_checkpoint_save_failure(cache_name, checkpoint, &error);
+            ScannerCheckpointPersistResult::RetryBucket(error)
+        }
     }
 }
 
+fn record_checkpoint_save_failure(cache_name: &str, checkpoint: &DataUsageCache, error: &StorageError) {
+    global_metrics().record_scanner_checkpoint_save_failed();
+    error!(
+        target: "rustfs::scanner::io",
+        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+        component = LOG_COMPONENT_SCANNER,
+        subsystem = LOG_SUBSYSTEM_IO,
+        state = "periodic_checkpoint_save_failed",
+        bucket = %checkpoint.info.name,
+        cache_name = %cache_name,
+        error = %error,
+        "Scanner periodic checkpoint save failed"
+    );
+}
+
 #[derive(Debug)]
 pub(crate) struct ScannerCacheLockGuards {
     scoped: NamespaceLockGuard,

crates/scanner/src/scanner_io/io_cache.rs

@@ -1209,7 +1209,7 @@ impl ScannerIOCache for SetDisks {
                                 )
                                 .await
                                 {
-                                    ScannerCheckpointPersistResult::Saved => {
+                                    ScannerCheckpointPersistResult::Saved | ScannerCheckpointPersistResult::Skipped => {
                                         if cache_guard.is_lock_lost() || ctx_clone.is_cancelled() {
                                             scan_ctx.cancel();
                                             await_scanner_disk_shutdown(scan.as_mut()).await;
@@ -1221,19 +1221,10 @@ impl ScannerIOCache for SetDisks {
                                         await_scanner_disk_shutdown(scan.as_mut()).await;
                                         break Err(Error::other("scanner bucket cache fence changed during checkpoint save"));
                                     }
-                                    ScannerCheckpointPersistResult::Failed(error) => {
-                                        error!(
-                                            target: "rustfs::scanner::io",
-                                            event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                                            component = LOG_COMPONENT_SCANNER,
-                                            subsystem = LOG_SUBSYSTEM_IO,
-                                            bucket = %bucket.name,
-                                            cache_name = %cache_name,
-                                            state = "periodic_checkpoint_save_failed",
-                                            error = %error,
-                                            "Scanner periodic checkpoint save failed"
-                                        );
-                                        checkpoint_channel_closed = true;
+                                    ScannerCheckpointPersistResult::RetryBucket(error) => {
+                                        scan_ctx.cancel();
+                                        await_scanner_disk_shutdown(scan.as_mut()).await;
+                                        break Err(error);
                                     }
                                 }
                             }

crates/scanner/src/scanner_io/tests.rs

@@ -717,7 +717,7 @@ async fn checkpoint_uses_global_cycle_fence_with_set_scoped_cache_revisions() {
                         &mut revisions,
                     )
                     .await,
-                    ScannerCheckpointPersistResult::Failed(StorageError::PreconditionFailed)
+                    ScannerCheckpointPersistResult::RetryBucket(StorageError::PreconditionFailed)
                 ),
                 "matching global fence must not authorize stale cache CAS revisions"
             );
@@ -734,6 +734,175 @@ async fn checkpoint_uses_global_cycle_fence_with_set_scoped_cache_revisions() {
     assert!(mismatched_sets > 0, "multi-pool fixture must expose a set-scoped cycle fence mismatch");
 }
 
+#[derive(Debug)]
+struct CheckpointFaultStore {
+    set: Arc<SetDisks>,
+    fail_put: AtomicBool,
+    fail_revision_read: AtomicBool,
+}
+
+#[async_trait::async_trait]
+impl crate::storage_api::owner::ObjectIO for CheckpointFaultStore {
+    type Error = crate::EcstoreError;
+    type RangeSpec = crate::storage_api::owner::HTTPRangeSpec;
+    type HeaderMap = http::HeaderMap;
+    type ObjectOptions = crate::ScannerObjectOptions;
+    type ObjectInfo = crate::ScannerObjectInfo;
+    type GetObjectReader = crate::ScannerGetObjectReader;
+    type PutObjectReader = crate::ScannerPutObjReader;
+
+    async fn get_object_reader(
+        &self,
+        bucket: &str,
+        object: &str,
+        range: Option<Self::RangeSpec>,
+        headers: Self::HeaderMap,
+        opts: &Self::ObjectOptions,
+    ) -> crate::EcstoreResult<Self::GetObjectReader> {
+        if self.fail_revision_read.swap(false, Ordering::SeqCst) {
+            return Err(StorageError::Io(std::io::Error::new(
+                std::io::ErrorKind::Interrupted,
+                "injected revision read failure",
+            )));
+        }
+        self.set.get_object_reader(bucket, object, range, headers, opts).await
+    }
+
+    async fn put_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut Self::PutObjectReader,
+        opts: &Self::ObjectOptions,
+    ) -> crate::EcstoreResult<Self::ObjectInfo> {
+        if self.fail_put.swap(false, Ordering::SeqCst) {
+            return Err(StorageError::Lock(LockError::Timeout {
+                resource: object.to_string(),
+                timeout: Duration::from_secs(5),
+            }));
+        }
+        self.set.put_object(bucket, object, data, opts).await
+    }
+}
+
+#[async_trait::async_trait]
+impl crate::ScannerConfigObjectDelete for CheckpointFaultStore {
+    async fn delete_config_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: ScannerObjectOptions,
+    ) -> crate::EcstoreResult<crate::ScannerObjectInfo> {
+        crate::ScannerConfigObjectDelete::delete_config_object(self.set.as_ref(), bucket, object, opts).await
+    }
+
+    async fn scanner_data_usage_publication_admission(&self) -> Option<crate::ScannerDataUsagePublicationAdmission> {
+        crate::ScannerConfigObjectDelete::scanner_data_usage_publication_admission(self.set.as_ref()).await
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn checkpoint_retries_next_snapshot_after_lock_failure_and_reloads_after_revision_failure() {
+    let (_temp_dir, fence_store) = setup_local_scanner_store(1, 4).await;
+    let cycle = 8457;
+    let leader_epoch = 19;
+    save_config(
+        fence_store.clone(),
+        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+        crate::scanner::encode_scanner_cycle_fence_for_test(cycle, leader_epoch),
+    )
+    .await
+    .expect("seed global cycle fence");
+    let set = fence_store.pools[0].disk_set[0].clone();
+    let publication_epoch = scanner_publication_epoch(set.clone()).await.expect("read publication epoch");
+    let cache_name = "checkpoint-recovery/.usage-cache.bin";
+    DataUsageCache::default()
+        .save(set.clone(), cache_name)
+        .await
+        .expect("seed cache");
+    let mut revisions = DataUsageCache::read_revisions(set.clone(), cache_name)
+        .await
+        .expect("read initial revisions");
+    let initial_revisions = revisions.clone();
+    let store = Arc::new(CheckpointFaultStore {
+        set: set.clone(),
+        fail_put: AtomicBool::new(true),
+        fail_revision_read: AtomicBool::new(false),
+    });
+    let ctx = CancellationToken::new();
+    let context = || ScannerCheckpointPersistContext {
+        ctx: &ctx,
+        expected_publication_epoch: publication_epoch,
+        cycle,
+        leader_epoch,
+    };
+    let mut checkpoint = DataUsageCache::default();
+    checkpoint.info.name = "checkpoint-recovery".to_string();
+    checkpoint.info.source = Some(DataUsageCacheSource::new(0, 0));
+    checkpoint.info.scan_checkpoint = Some(crate::DataUsageScanCheckpoint::new(
+        "checkpoint-recovery/first".to_string(),
+        crate::DataUsageScanCheckpointReason::Objects,
+    ));
+    assert!(matches!(
+        persist_scanner_checkpoint(store.clone(), fence_store.clone(), context(), cache_name, &checkpoint, &mut revisions).await,
+        ScannerCheckpointPersistResult::Skipped
+    ));
+    assert_eq!(revisions, initial_revisions);
+    assert_eq!(
+        DataUsageCache::read_revisions(set.clone(), cache_name)
+            .await
+            .expect("read after failed save"),
+        initial_revisions
+    );
+
+    checkpoint
+        .info
+        .scan_checkpoint
+        .as_mut()
+        .expect("checkpoint cursor")
+        .resume_after = "checkpoint-recovery/second".to_string();
+    assert!(matches!(
+        persist_scanner_checkpoint(store.clone(), fence_store.clone(), context(), cache_name, &checkpoint, &mut revisions).await,
+        ScannerCheckpointPersistResult::Saved
+    ));
+    assert_ne!(revisions, initial_revisions);
+    let mut loaded = DataUsageCache::default();
+    loaded.load(set.clone(), cache_name).await.expect("load recovered checkpoint");
+    assert_eq!(
+        loaded.info.scan_checkpoint, checkpoint.info.scan_checkpoint,
+        "next snapshot must become durable"
+    );
+
+    let saved_revisions = revisions.clone();
+    checkpoint
+        .info
+        .scan_checkpoint
+        .as_mut()
+        .expect("checkpoint cursor")
+        .resume_after = "checkpoint-recovery/third".to_string();
+    store.fail_revision_read.store(true, Ordering::SeqCst);
+    assert!(matches!(
+        persist_scanner_checkpoint(store, fence_store, context(), cache_name, &checkpoint, &mut revisions).await,
+        ScannerCheckpointPersistResult::RetryBucket(StorageError::Io(_))
+    ));
+    assert_eq!(revisions, saved_revisions, "indeterminate revision refresh must not rebase the writer");
+    loaded
+        .load(set.clone(), cache_name)
+        .await
+        .expect("load committed checkpoint after refresh failure");
+    assert_eq!(
+        loaded.info.scan_checkpoint, checkpoint.info.scan_checkpoint,
+        "PUT completed before revision refresh failed"
+    );
+    assert_ne!(
+        DataUsageCache::read_revisions(set, cache_name)
+            .await
+            .expect("read committed revisions"),
+        saved_revisions
+    );
+}
+
 #[derive(Debug)]
 struct CheckpointFenceStore {
     before: Vec<u8>,

docs/operations/scanner-runtime-controls.md

@@ -172,6 +172,20 @@ catch_up_estimate.discovered_expiry_items
 catch_up_estimate.discovered_transition_items
 ```
 
+### Periodic checkpoint persistence
+
+`metrics.scan_checkpoint` is the in-memory resume hint; it does not confirm a durable save. `metrics.scan_checkpoints_persisted` lists the latest confirmed periodic save for each `(pool_index, set_index)` handled by this process, with its bucket, `saved_unix_secs`, and optional checkpoint cursor. It retains one observation per set, is reset on process restart, and does not include the final bucket save. Query the peer executing the walk for remote-worker observations. `metrics.scan_checkpoint_save_failures` counts failed periodic saves, including failures to refresh CAS revisions after a write; failed attempts do not advance the confirmed save observation.
+
+An ordinary failed save drops that snapshot and leaves the bounded checkpoint receiver active, so the next producer interval tries again. A CAS conflict or an unreadable revision after a successful write cancels the bucket attempt and reloads persisted state through the existing retry path. Leader, cycle, and publication fencing remain mandatory. This preserves future periodic save attempts for a continuing bucket walk.
+
+Periodic snapshots include the suspended ancestor accounting needed to connect a nested active branch to the bucket root. Completed coverage is sealed independently of raw enumeration bookkeeping. A valid coverage frontier can therefore skip completed subtrees after a fenced leader handoff without counting unfinished work as completed. Emission remains bounded by the existing interval/object thresholds and capacity-one channel; invalid snapshot attempts also consume the producer interval so they cannot repeatedly clone the cache on every object.
+
+A resumed walk retains its previous durable coverage while initial enumeration has not recopied the covered subtrees. Unfinished out-of-order heal revisits cannot be certified as completed prefix work, including in cancellation snapshots.
+
+Shared compacted accumulators defer periodic emission until the owning subtree completes; replay inside that compacted subtree remains possible after an interruption. When bucket checkpoint preparation returns `Reset` for existing resume state, it logs `checkpoint_reset` with the validation reason. The `scan_checkpoint_used` counter includes validated coverage adoption by the forward sweep.
+
+A confirmed periodic save does not prove restart resumability. Adoption still validates bucket/set identity and, for non-empty caches, connected cache scope. Skipping completed subtrees additionally requires a validated coverage frontier; raw enumeration state alone does not prove completed subtree coverage.
+
 ## Usage State Reset
 
 The supported break-glass route for rebuilding scanner usage state is `POST /v3/scanner/usage-state/reset` with body `{"mode":"full-rebuild"}`, authenticated as an admin identity holding `ConfigUpdateAdminAction` (route registered in `rustfs/src/admin/route_registration_test.rs`). Use it only after the scanner status shows a usage-floor load failure, a conflicting persisted usage floor, or an operator decision to discard the durable usage baseline and rebuild it from a full scanner pass.