diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs
index 0bb265220..14c5453f0 100644
--- a/crates/scanner/src/scanner_folder.rs
+++ b/crates/scanner/src/scanner_folder.rs
@@ -96,8 +96,8 @@ const SCANNER_CHECKPOINT_OBJECT_INTERVAL: u64 = 1024;
 const SCANNER_CHECKPOINT_MIN_INTERVAL: Duration = Duration::from_secs(5);
 const SCANNER_CHECKPOINT_INTERVAL: Duration = Duration::from_secs(60);
 const SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT: usize = 128;
-// Erasure data directories contain direct part.N files; keep namespace probes bounded.
-const ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT: usize = 64;
+// Bound cooperative directory work without treating a batch boundary as EOF.
+const SCANNER_DIRECTORY_BATCH_ENTRIES: usize = 64;
 const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
 const ENV_DATA_USAGE_UPDATE_DIR_CYCLES: &str = "RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES";
 const ENV_HEAL_OBJECT_SELECT_PROB: &str = "RUSTFS_HEAL_OBJECT_SELECT_PROB";
@@ -419,6 +419,25 @@ enum FolderScanSource {
     Existing,
 }
 
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum FolderScanMode {
+    Namespace,
+    ErasureCandidate { reuse_namespace: bool },
+}
+
+#[derive(Debug, PartialEq, Eq)]
+enum FolderScanOutcome {
+    Namespace,
+    ErasureData,
+    Vanished,
+}
+
+fn is_erasure_candidate(name: &str) -> bool {
+    name.rsplit('/')
+        .next()
+        .is_some_and(|name| uuid::Uuid::parse_str(name).is_ok_and(|id| !id.is_nil()))
+}
+
 #[derive(Clone, Debug)]
 struct QueuedFolder {
     folder: CachedFolder,
@@ -920,16 +939,54 @@ async fn wait_for_scanner_io<T>(ctx: &CancellationToken, wait: impl std::future:
     }
 }
 
+struct ScannerDirectory {
+    reader: tokio::fs::ReadDir,
+    batch_entries: usize,
+    #[cfg(test)]
+    ordered: Option<tests::erasure_probe::OrderedEntries>,
+}
+
+impl ScannerDirectory {
+    async fn open(path: &str) -> std::io::Result<Self> {
+        let reader = tokio::fs::read_dir(path).await?;
+        #[cfg(test)]
+        let (reader, ordered) = {
+            let mut reader = reader;
+            let ordered = tests::erasure_probe::ordered_entries(path, &mut reader).await?;
+            (reader, ordered)
+        };
+        Ok(Self {
+            reader,
+            batch_entries: 0,
+            #[cfg(test)]
+            ordered,
+        })
+    }
+
+    async fn next_entry(&mut self) -> std::io::Result<Option<tokio::fs::DirEntry>> {
+        #[cfg(test)]
+        if let Some(entries) = &mut self.ordered {
+            return entries.pop_front().transpose();
+        }
+        self.reader.next_entry().await
+    }
+}
+
 async fn next_scanner_dir_entry(
     ctx: &CancellationToken,
     sleeper: &DynamicSleeper,
-    entries: &mut tokio::fs::ReadDir,
+    entries: &mut ScannerDirectory,
 ) -> Result<Option<tokio::fs::DirEntry>, ScannerError> {
+    if entries.batch_entries == SCANNER_DIRECTORY_BATCH_ENTRIES {
+        wait_for_scanner_io(ctx, tokio::task::yield_now()).await?;
+        entries.batch_entries = 0;
+    }
     let entry = wait_for_scanner_io(ctx, entries.next_entry()).await??;
     if entry.is_some() {
         // Tokio prefetches bounded chunks. Pacing consumption also paces the
         // next refill, without charging an extra entry for EOF.
         wait_for_scanner_io(ctx, sleeper.admit_disk_io(1, 0)).await?;
+        entries.batch_entries += 1;
     }
     Ok(entry)
 }
@@ -1166,6 +1223,17 @@ impl FolderScanner {
         self.record_scan_resume_hint(folder);
     }
 
+    fn can_reuse_covered_namespace(&self, name: &str, compacted: bool) -> bool {
+        self.old_cache.info.scan_progress.is_some()
+            && !compacted
+            && self
+                .resume_frontier
+                .as_deref()
+                .is_some_and(|frontier| scan_frontier_covers_path(name, frontier))
+            && self.old_cache.find(name).is_some()
+            && !self.resume_subtree_has_failures(name)
+    }
+
     fn resume_subtree_has_failures(&self, folder: &str) -> bool {
         if self.resume_failed_paths.is_empty() {
             return false;
@@ -1294,9 +1362,10 @@ impl FolderScanner {
         parent_hash: &mut DataUsageHash,
         parent_entry: &mut DataUsageEntry,
         child_entry: &mut DataUsageEntry,
-    ) -> Result<(), ScannerError> {
+        mode: FolderScanMode,
+    ) -> Result<FolderScanOutcome, 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;
+            return Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode)).await;
         }
         let depth = self.checkpoint_depth;
         if depth >= MAX_DATA_USAGE_CACHE_DEPTH {
@@ -1310,7 +1379,7 @@ impl FolderScanner {
         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;
+        let result = Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode)).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);
@@ -1632,23 +1701,79 @@ impl FolderScanner {
         self.send_update().await;
     }
 
+    async fn record_missing_erasure_metadata(
+        &mut self,
+        folder: &CachedFolder,
+        into: &mut DataUsageEntry,
+    ) -> Result<(), ScannerError> {
+        let metadata_path = path_join_buf(&[&self.root, &folder.name, STORAGE_FORMAT_FILE]);
+        self.mark_failed_path_seen(&metadata_path);
+        let retry_suppressed = self.failed_retry_suppressed(&metadata_path);
+
+        into.failed_objects = into.failed_objects.saturating_add(1);
+        if !retry_suppressed {
+            self.record_failed(&metadata_path);
+
+            let failed_cache_entries = self.new_cache.info.failed_objects.len();
+            if failed_cache_entries > 0 && should_log_failed_object(failed_cache_entries) {
+                warn!(
+                    target: "rustfs::scanner::folder",
+                    event = EVENT_SCANNER_FOLDER_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_FOLDER,
+                    path = %metadata_path,
+                    failed_objects = failed_cache_entries,
+                    state = "object_metadata_missing",
+                    "Scanner found erasure object data without metadata"
+                );
+            }
+        }
+
+        if !retry_suppressed {
+            let (bucket, object) = path2_bucket_object_with_base_path(&self.root, &folder.name);
+            if !bucket.is_empty() && !object.is_empty() {
+                self.send_required_scanner_heal_request(
+                    PendingScannerHealKind::Object,
+                    bucket.clone(),
+                    Some(object.clone()),
+                    None,
+                    build_non_destructive_object_heal_request(bucket, object, self.scan_mode, HealChannelPriority::High),
+                )
+                .await?;
+            }
+        }
+        Ok(())
+    }
+
     /// Scan a folder recursively
     /// Files found in the folders will be added to new_cache.
-    #[allow(clippy::never_loop)]
-    #[allow(unused_assignments)]
     pub async fn scan_folder(
         &mut self,
         ctx: CancellationToken,
         folder: CachedFolder,
         into: &mut DataUsageEntry,
     ) -> Result<(), ScannerError> {
+        self.scan_folder_inner(ctx, folder, into, FolderScanMode::Namespace)
+            .await
+            .map(|_| ())
+    }
+
+    #[allow(clippy::never_loop)]
+    #[allow(unused_assignments)]
+    async fn scan_folder_inner(
+        &mut self,
+        ctx: CancellationToken,
+        folder: CachedFolder,
+        into: &mut DataUsageEntry,
+        mode: FolderScanMode,
+    ) -> Result<FolderScanOutcome, ScannerError> {
         let done_folder = Metrics::time(Metric::ScanFolder);
         let trace_started_at = trace_start_instant();
 
         if ctx.is_cancelled() {
             return Err(ScannerError::Other("Operation cancelled".to_string()));
         }
-        if !self.budget.try_start_directory() {
+        if mode == FolderScanMode::Namespace && !self.budget.try_start_directory() {
             return Err(ScannerError::Other("Operation cancelled".to_string()));
         }
 
@@ -1656,7 +1781,7 @@ impl FolderScanner {
         // Store initial compaction state.
         let was_compacted = into.compacted;
 
-        loop {
+        'scan: loop {
             if ctx.is_cancelled() {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
@@ -1705,12 +1830,16 @@ impl FolderScanner {
             };
             let active_object_lock = self.old_cache.info.object_lock.clone();
 
-            wait_for_scanner_io(&ctx, self.sleeper.sleep_folder()).await?;
+            if mode != FolderScanMode::Namespace {
+                // A candidate may be object payload rather than a namespace.
+                wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
+            } else {
+                wait_for_scanner_io(&ctx, self.sleeper.sleep_folder()).await?;
+            }
 
             let mut existing_folders: Vec<CachedFolder> = Vec::new();
             let mut new_folders: Vec<CachedFolder> = Vec::new();
             let mut found_object_metadata = false;
-            let mut erasure_data_directory_candidates: Vec<(CachedFolder, bool, String)> = Vec::new();
             let mut object_count: u64 = 0;
             let yield_every_objects = scanner_yield_every_n_objects();
 
@@ -1731,9 +1860,13 @@ impl FolderScanner {
             // directory that vanished during iteration instead of failing the
             // whole parent subtree. The bucket root (no parent) still propagates
             // NotFound so namespace removal is detected by the caller.
-            let mut dir_reader = match wait_for_scanner_io(&ctx, tokio::fs::read_dir(&dir_path)).await? {
+            let mut dir_reader = match wait_for_scanner_io(&ctx, ScannerDirectory::open(&dir_path)).await? {
                 Ok(dir_reader) => Some(dir_reader),
-                Err(e) if e.kind() == ErrorKind::NotFound && folder.parent.is_some() => {
+                Err(e)
+                    if folder.parent.is_some()
+                        && (e.kind() == ErrorKind::NotFound
+                            || (mode != FolderScanMode::Namespace && e.kind() == ErrorKind::NotADirectory)) =>
+                {
                     debug!(
                         target: "rustfs::scanner::folder",
                         event = EVENT_SCANNER_FOLDER_STATE,
@@ -1744,6 +1877,11 @@ impl FolderScanner {
                         error = %e,
                         "Scanner folder state updated"
                     );
+                    if mode != FolderScanMode::Namespace {
+                        self.coverage_gap |= e.kind() == ErrorKind::NotADirectory && self.old_cache.info.scan_progress.is_some();
+                        done_folder();
+                        return Ok(FolderScanOutcome::Vanished);
+                    }
                     None
                 }
                 Err(e) => return Err(ScannerError::Io(e)),
@@ -1753,67 +1891,111 @@ impl FolderScanner {
             let mut raw_enumeration_complete = false;
             let raw_enumeration_committed_entries = self.raw_enumeration_committed_entry_oracle(&folder.name);
 
+            let mut candidate_unclassified = mode != FolderScanMode::Namespace;
+            let mut deferred_metadata = None;
+
             loop {
-                let Some(reader) = dir_reader.as_mut() else {
+                // Hold at most one metadata entry until a candidate reaches
+                // EOF: a later part must not leave object accounting or ILM work.
+                let (entry, replayed_metadata) = if let Some(reader) = dir_reader.as_mut() {
+                    let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
+                        Ok(Some(entry)) => entry,
+                        Ok(None) => {
+                            raw_enumeration_complete = true;
+                            dir_reader = None;
+                            continue;
+                        }
+                        Err(ScannerError::Io(e))
+                            if candidate_unclassified && matches!(e.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) =>
+                        {
+                            return Err(ScannerError::Io(e));
+                        }
+                        Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotFound => {
+                            debug!(
+                                target: "rustfs::scanner::folder",
+                                event = EVENT_SCANNER_FOLDER_STATE,
+                                component = LOG_COMPONENT_SCANNER,
+                                subsystem = LOG_SUBSYSTEM_FOLDER,
+                                dir_path = %dir_path,
+                                state = "dir_missing_during_iteration",
+                                error = %e,
+                                "Scanner folder state updated"
+                            );
+                            raw_enumeration_complete = true;
+                            dir_reader = None;
+                            continue;
+                        }
+                        Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotADirectory => {
+                            warn!(
+                                target: "rustfs::scanner::folder",
+                                event = EVENT_SCANNER_FOLDER_STATE,
+                                component = LOG_COMPONENT_SCANNER,
+                                subsystem = LOG_SUBSYSTEM_FOLDER,
+                                dir_path = %dir_path,
+                                state = "dir_became_non_directory",
+                                error = %e,
+                                "Scanner folder state updated"
+                            );
+                            raw_enumeration_complete = true;
+                            dir_reader = None;
+                            continue;
+                        }
+                        Err(e) => return Err(e),
+                    };
+                    (entry, false)
+                } else {
                     raw_enumeration_complete = true;
-                    break;
-                };
-                let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
-                    Ok(Some(entry)) => entry,
-                    Ok(None) => {
-                        raw_enumeration_complete = true;
-                        break;
+                    if candidate_unclassified {
+                        // Apply namespace reuse only after the current listing
+                        // proves there is no direct erasure part.
+                        if let FolderScanMode::ErasureCandidate { reuse_namespace: true } = mode
+                            && !into.compacted
+                            && self.old_cache.find(&folder.name).is_some()
+                        {
+                            // The caller stitches this cached child into its
+                            // parent; do not clone a growing child set into the
+                            // unused provisional accumulator as well.
+                            self.new_cache.copy_with_children(&self.old_cache, &this_hash, &folder.parent);
+                            self.budget.record_entries_visited(pending_entry_progress);
+                            self.finish_raw_enumeration_parent(&folder.name);
+                            done_folder();
+                            return Ok(FolderScanOutcome::Namespace);
+                        }
+                        candidate_unclassified = false;
+                        if !self.budget.try_start_directory() {
+                            return Err(ScannerError::Other("Operation cancelled".to_string()));
+                        }
+                        wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                     }
-                    Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotFound => {
-                        debug!(
-                            target: "rustfs::scanner::folder",
-                            event = EVENT_SCANNER_FOLDER_STATE,
-                            component = LOG_COMPONENT_SCANNER,
-                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                            dir_path = %dir_path,
-                            state = "dir_missing_during_iteration",
-                            error = %e,
-                            "Scanner folder state updated"
-                        );
-                        raw_enumeration_complete = true;
-                        break;
-                    }
-                    Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotADirectory => {
-                        warn!(
-                            target: "rustfs::scanner::folder",
-                            event = EVENT_SCANNER_FOLDER_STATE,
-                            component = LOG_COMPONENT_SCANNER,
-                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                            dir_path = %dir_path,
-                            state = "dir_became_non_directory",
-                            error = %e,
-                            "Scanner folder state updated"
-                        );
-                        raw_enumeration_complete = true;
-                        break;
-                    }
-                    Err(e) => return Err(e),
+                    let Some(entry) = deferred_metadata.take() else { break };
+                    (entry, true)
                 };
                 let file_name = entry.file_name().to_string_lossy().to_string();
                 if file_name.is_empty() || file_name == "." || file_name == ".." {
                     continue;
                 }
-                let raw_entry_consumed_by_owner_index = raw_enumeration_committed_entries.contains(&file_name);
-                if !raw_entry_consumed_by_owner_index {
-                    #[cfg(test)]
-                    tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
-                    pending_entry_progress = pending_entry_progress.saturating_add(1);
-                    if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
-                        || last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
-                    {
-                        self.budget.record_entries_visited(pending_entry_progress);
-                        pending_entry_progress = 0;
-                        last_entry_progress = Instant::now();
+                if !replayed_metadata {
+                    let raw_entry_consumed_by_owner_index = raw_enumeration_committed_entries.contains(&file_name);
+                    if !raw_entry_consumed_by_owner_index {
+                        #[cfg(test)]
+                        tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
+                        pending_entry_progress = pending_entry_progress.saturating_add(1);
+                        if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
+                            || last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
+                        {
+                            self.budget.record_entries_visited(pending_entry_progress);
+                            pending_entry_progress = 0;
+                            last_entry_progress = Instant::now();
+                        }
                     }
+                    self.record_raw_enumeration_entry(&folder.name, &file_name);
+                    self.maybe_send_checkpoint(&folder, &this_hash, into);
                 }
-                self.record_raw_enumeration_entry(&folder.name, &file_name);
-                self.maybe_send_checkpoint(&folder, &this_hash, into);
                 let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE;
+                if candidate_unclassified && is_storage_format_entry {
+                    deferred_metadata = Some(entry);
+                    continue;
+                }
 
                 let file_path = entry.path().to_string_lossy().to_string();
 
@@ -1858,6 +2040,15 @@ impl FolderScanner {
                     Err(e) => return Err(ScannerError::Io(e)),
                 };
 
+                if candidate_unclassified && entry_type.is_file() && is_erasure_part_name(&file_name) {
+                    self.budget.record_entries_visited(pending_entry_progress);
+                    // Positive data evidence completes classification, not raw
+                    // namespace coverage. Discard its provisional resume owner.
+                    self.finish_raw_enumeration_parent(&folder.name);
+                    done_folder();
+                    return Ok(FolderScanOutcome::ErasureData);
+                }
+
                 // Metadata presence establishes an erasure object boundary;
                 // parsing failures still belong to accounting and healing. A
                 // directory named `xl.meta` remains a valid namespace prefix,
@@ -1950,11 +2141,6 @@ impl FolderScanner {
                         object_heal_prob_div: folder.object_heal_prob_div,
                     };
 
-                    if self.is_erasure_mode && uuid::Uuid::parse_str(&file_name).is_ok_and(|data_dir_id| !data_dir_id.is_nil()) {
-                        erasure_data_directory_candidates.push((this, exists, file_path));
-                        continue;
-                    }
-
                     if exists && self.should_reuse_clean_root_child(&folder, into, &this, &h, &abandoned_children) {
                         abandoned_children.remove(&h.key());
                         self.reuse_clean_root_child(&this, &h, into);
@@ -2164,73 +2350,6 @@ impl FolderScanner {
                 self.finish_raw_enumeration_parent(&folder.name);
             }
 
-            let mut found_erasure_data_directory = false;
-            if self.is_erasure_mode && !found_object_metadata {
-                for (_, _, path) in &erasure_data_directory_candidates {
-                    if contains_erasure_part_file(path, &ctx, &self.sleeper).await? {
-                        found_erasure_data_directory = true;
-                        break;
-                    }
-                }
-            }
-
-            if !found_object_metadata && !found_erasure_data_directory {
-                for (candidate, exists, _) in erasure_data_directory_candidates {
-                    let h = hash_path(&candidate.name);
-                    if exists && self.should_reuse_clean_root_child(&folder, into, &candidate, &h, &abandoned_children) {
-                        abandoned_children.remove(&h.key());
-                        self.reuse_clean_root_child(&candidate, &h, into);
-                    } else if exists {
-                        abandoned_children.remove(&h.key());
-                        self.update_cache.copy_with_children(&self.old_cache, &h, &candidate.parent);
-                        existing_folders.push(candidate);
-                    } else {
-                        abandoned_children.remove(&h.key());
-                        new_folders.push(candidate);
-                    }
-                }
-            }
-
-            if self.is_erasure_mode && found_erasure_data_directory && !found_object_metadata {
-                found_object_metadata = true;
-                let metadata_path = path_join_buf(&[&dir_path, STORAGE_FORMAT_FILE]);
-                self.mark_failed_path_seen(&metadata_path);
-                let retry_suppressed = self.failed_retry_suppressed(&metadata_path);
-
-                into.failed_objects = into.failed_objects.saturating_add(1);
-                if !retry_suppressed {
-                    self.record_failed(&metadata_path);
-
-                    let failed_cache_entries = self.new_cache.info.failed_objects.len();
-                    if failed_cache_entries > 0 && should_log_failed_object(failed_cache_entries) {
-                        warn!(
-                            target: "rustfs::scanner::folder",
-                            event = EVENT_SCANNER_FOLDER_STATE,
-                            component = LOG_COMPONENT_SCANNER,
-                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                            path = %metadata_path,
-                            failed_objects = failed_cache_entries,
-                            state = "object_metadata_missing",
-                            "Scanner found erasure object data without metadata"
-                        );
-                    }
-                }
-
-                if !retry_suppressed {
-                    let (bucket, object) = path2_bucket_object_with_base_path(&self.root, &folder.name);
-                    if !bucket.is_empty() && !object.is_empty() {
-                        self.send_required_scanner_heal_request(
-                            PendingScannerHealKind::Object,
-                            bucket.clone(),
-                            Some(object.clone()),
-                            None,
-                            build_non_destructive_object_heal_request(bucket, object, self.scan_mode, HealChannelPriority::High),
-                        )
-                        .await?;
-                    }
-                }
-            }
-
             if ctx.is_cancelled() {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
@@ -2320,7 +2439,6 @@ impl FolderScanner {
             }));
             let has_queued_folders = !queued_folders.is_empty();
             let forward_sweep = self.old_cache.info.scan_progress.is_some();
-            let forward_resume_after = self.resume_frontier.clone();
             let resume_order = if forward_sweep {
                 order_queued_folders_for_resume(&mut queued_folders, None)
             } else {
@@ -2342,14 +2460,15 @@ impl FolderScanner {
 
                 let mut folder_item = queued_folder.folder;
                 let h = hash_path(&folder_item.name);
-                if forward_sweep
-                    && !into.compacted
-                    && forward_resume_after
-                        .as_deref()
-                        .is_some_and(|resume| scan_frontier_covers_path(&folder_item.name, resume))
-                    && self.old_cache.find(&folder_item.name).is_some()
-                    && !self.resume_subtree_has_failures(&folder_item.name)
-                {
+                let reuse_covered = self.can_reuse_covered_namespace(&folder_item.name, into.compacted);
+                let mut child_mode = if self.is_erasure_mode && is_erasure_candidate(&folder_item.name) {
+                    FolderScanMode::ErasureCandidate {
+                        reuse_namespace: reuse_covered,
+                    }
+                } else {
+                    FolderScanMode::Namespace
+                };
+                if child_mode == FolderScanMode::Namespace && reuse_covered {
                     self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
                     into.add_child(&h);
                     continue;
@@ -2399,60 +2518,90 @@ impl FolderScanner {
                         {
                             let next_cycle = self.old_cache.info.next_cycle as u32;
                             if !h.mod_(next_cycle, data_usage_update_dir_cycles()) {
-                                // Transfer and add as child...
-                                self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
-                                into.add_child(&h);
-                                self.record_scan_resume_hint(&folder_item.name);
-                                continue;
+                                if child_mode == FolderScanMode::Namespace {
+                                    self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
+                                    into.add_child(&h);
+                                    self.record_scan_resume_hint(&folder_item.name);
+                                    continue;
+                                }
+                                // Sampling may skip namespace work, never the
+                                // classification that protects its parent object.
+                                child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
+                            } else {
+                                folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                             }
-
-                            folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                         }
                     }
                 }
 
                 (self.update_current_path)(&folder_item.name).await;
 
+                let outcome = if into.compacted {
+                    Box::pin(self.scan_folder_inner(ctx.clone(), folder_item.clone(), into, child_mode)).await?
+                } else {
+                    let mut dst = DataUsageEntry::default();
+                    match self
+                        .scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst, child_mode)
+                        .await
+                    {
+                        Ok(outcome) => outcome,
+                        Err(e) => {
+                            if ctx.is_cancelled() {
+                                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,
+                                folder = %folder.name,
+                                child_folder = %folder_item.name,
+                                state = "child_scan_failed",
+                                error = %e,
+                                "Scanner child folder scan failed"
+                            );
+                            self.coverage_gap |= forward_sweep;
+                            continue;
+                        }
+                    }
+                };
+                if outcome == FolderScanOutcome::Vanished {
+                    self.new_cache.delete_recursive(&h);
+                    self.update_cache.delete_recursive(&h);
+                    into.children.remove(&h.key());
+                    continue;
+                }
+                if outcome == FolderScanOutcome::ErasureData {
+                    self.new_cache.delete_recursive(&h);
+                    self.update_cache.delete_recursive(&h);
+                    into.children.remove(&h.key());
+                    // Earlier namespace siblings may have completed, but none
+                    // can certify coverage of this unresolved parent object.
+                    self.coverage_gap |= forward_sweep;
+                    if self.coverage_frontier.as_deref().is_some_and(|frontier| {
+                        frontier
+                            .strip_prefix(&folder.name)
+                            .is_some_and(|suffix| suffix.starts_with('/'))
+                    }) {
+                        self.coverage_frontier = None;
+                    }
+                    self.record_missing_erasure_metadata(&folder, into).await?;
+                    break 'scan;
+                }
                 if into.compacted {
-                    // In compacted mode child totals are accumulated directly into the parent entry.
-                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
-                    fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
                     self.record_completed_child(&folder_item.name);
                     self.send_update_for_entry(&this_hash, &folder.parent, into).await;
                     tokio::task::yield_now().await;
                 } else {
-                    let mut dst = DataUsageEntry::default();
-
-                    // Use Box::pin for recursive async call
-                    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;
-                            return Err(e);
-                        }
-                        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;
-                        continue;
-                    }
                     tokio::task::yield_now().await;
-
                     into.add_child(&h);
                     self.record_completed_child(&folder_item.name);
                     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);
                     self.send_update().await;
@@ -2829,7 +2978,14 @@ impl FolderScanner {
 
                         // Use Box::pin for recursive async call
                         if let Err(e) = self
-                            .scan_child_folder(ctx.clone(), folder_item.clone(), &mut this_hash, into, &mut dst)
+                            .scan_child_folder(
+                                ctx.clone(),
+                                folder_item.clone(),
+                                &mut this_hash,
+                                into,
+                                &mut dst,
+                                FolderScanMode::Namespace,
+                            )
                             .await
                         {
                             if ctx.is_cancelled() {
@@ -2924,7 +3080,7 @@ impl FolderScanner {
         let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
         emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
 
-        Ok(())
+        Ok(FolderScanOutcome::Namespace)
     }
 
     pub fn as_mut_new_cache(&mut self) -> &mut DataUsageCache {
diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs
index bcc9bd003..d91400307 100644
--- a/crates/scanner/src/scanner_folder/item_actions.rs
+++ b/crates/scanner/src/scanner_folder/item_actions.rs
@@ -1300,48 +1300,10 @@ pub(super) fn classify_get_size_failure(item: &ScannerItem, err: &StorageError)
     GetSizeFailureAction::RecordFailed
 }
 
-pub(super) async fn contains_erasure_part_file(
-    path: &str,
-    ctx: &CancellationToken,
-    sleeper: &DynamicSleeper,
-) -> Result<bool, ScannerError> {
-    wait_for_scanner_io(ctx, sleeper.admit_disk_io(1, 0)).await?;
-    let mut entries = match wait_for_scanner_io(ctx, tokio::fs::read_dir(path)).await? {
-        Ok(entries) => entries,
-        Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => return Ok(false),
-        Err(err) => return Err(ScannerError::Io(err)),
-    };
-
-    for _ in 0..ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT {
-        let entry = match next_scanner_dir_entry(ctx, sleeper, &mut entries).await {
-            Ok(Some(entry)) => entry,
-            Ok(None) => return Ok(false),
-            Err(ScannerError::Io(err)) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => {
-                return Ok(false);
-            }
-            Err(err) => return Err(err),
-        };
-        let file_name = entry.file_name();
-        let Some(part_number) = file_name
-            .to_str()
-            .and_then(|name| name.strip_prefix("part."))
-            .and_then(|number| number.parse::<u32>().ok())
-        else {
-            continue;
-        };
-        if part_number == 0 {
-            continue;
-        }
-
-        match wait_for_scanner_io(ctx, entry.file_type()).await? {
-            Ok(file_type) if file_type.is_file() => return Ok(true),
-            Ok(_) => {}
-            Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::TooManyLinks) => {}
-            Err(err) => return Err(ScannerError::Io(err)),
-        }
-    }
-
-    Ok(false)
+pub(super) fn is_erasure_part_name(name: &str) -> bool {
+    name.strip_prefix("part.")
+        .and_then(|number| number.parse::<u32>().ok())
+        .is_some_and(|number| number > 0)
 }
 
 #[cfg(test)]
diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs
index d9a9180d5..f0179dab1 100644
--- a/crates/scanner/src/scanner_folder/tests.rs
+++ b/crates/scanner/src/scanner_folder/tests.rs
@@ -16,6 +16,7 @@ use crate::SCANNER_SLEEPER;
 
 use super::*;
 
+pub(super) mod erasure_probe;
 mod mrf_ownership;
 use crate::storage_api::VersionPurgeStatusType;
 use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
@@ -151,6 +152,29 @@ struct CapturedLogWriter {
 }
 
 impl CapturedLogs {
+    fn capture<F: std::future::Future>(&self, level: tracing::Level, future: F) -> impl std::future::Future<Output = F::Output> {
+        use tracing::instrument::WithSubscriber as _;
+
+        let subscriber = tracing_subscriber::fmt()
+            .json()
+            .with_max_level(level)
+            .with_writer(self.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+        // Re-enter the dispatcher on every poll instead of leaving a
+        // thread-local default installed while this future is suspended.
+        let captured = future.with_subscriber(subscriber);
+        // Match ecstore's test_tracing guard: with a single dispatcher,
+        // first registration on an unsubscribed sibling can cache Never.
+        // Construct both dispatchers before polling and retain the inert one.
+        let interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
+        async move {
+            let _interest_guard = interest_guard;
+            captured.await
+        }
+    }
+
     fn contents(&self) -> String {
         let buffer = self
             .buffer
@@ -587,6 +611,7 @@ async fn checkpoint_ancestor_depth_limit_preserves_suspended_accounting() {
             &mut hash,
             &mut parent,
             &mut child,
+            FolderScanMode::Namespace,
         )
         .await
         .expect_err("unsupported depth must stop before recursion");
@@ -2448,113 +2473,6 @@ async fn scan_folder_reuses_symlink_size_without_bypassing_disk_validation() {
     );
 }
 
-#[tokio::test(start_paused = true)]
-async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
-    let path = std::env::temp_dir().join(format!("scanner-cancel-budget-{}", Uuid::new_v4()));
-    let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
-        &path,
-        crate::sleeper::ScannerDiskIoBudgetLimits {
-            iops: Some(0),
-            bytes_per_sec: None,
-        },
-    );
-    sleeper.admit_disk_io(1, 0).await;
-    let ctx = CancellationToken::new();
-    let cancel = ctx.clone();
-    let task = tokio::spawn(async move {
-        tokio::task::yield_now().await;
-        cancel.cancel();
-    });
-    let started = tokio::time::Instant::now();
-    assert!(
-        contains_erasure_part_file(path.to_string_lossy().as_ref(), &ctx, &sleeper)
-            .await
-            .is_err()
-    );
-    task.await.expect("cancellation task");
-    assert_eq!(
-        started.elapsed(),
-        Duration::ZERO,
-        "cancellation must interrupt the one-second budget wait"
-    );
-
-    assert!(wait_for_scanner_io(&ctx, sleeper.admit_disk_io(1, 0)).await.is_err());
-    let status = crate::sleeper::scanner_disk_io_budget_status()
-        .into_iter()
-        .find(|status| status.disk == path.to_string_lossy())
-        .expect("bound disk budget");
-    assert_eq!(status.charged_ops, 2, "an already cancelled wait must not charge another operation");
-}
-
-#[tokio::test]
-async fn erasure_directory_probe_respects_entry_limit_and_disk_budget() {
-    let temp_dir = tempfile::tempdir().expect("create probe directory");
-    for index in 0..=ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT {
-        tokio::fs::write(temp_dir.path().join(format!("namespace-{index}")), b"")
-            .await
-            .expect("create non-part entry");
-    }
-    let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
-        temp_dir.path(),
-        crate::sleeper::ScannerDiskIoBudgetLimits {
-            iops: Some(1_000_000),
-            bytes_per_sec: None,
-        },
-    );
-    assert!(
-        !contains_erasure_part_file(temp_dir.path().to_string_lossy().as_ref(), &CancellationToken::new(), &sleeper)
-            .await
-            .expect("probe directory without erasure parts")
-    );
-    let status = crate::sleeper::scanner_disk_io_budget_status()
-        .into_iter()
-        .find(|status| status.disk == temp_dir.path().to_string_lossy())
-        .expect("bound disk budget");
-    assert_eq!(
-        status.charged_ops,
-        1 + u64::try_from(ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT).expect("entry limit fits u64"),
-        "one directory open plus at most the bounded probe entries"
-    );
-    assert_eq!(status.charged_bytes, 0);
-}
-
-#[tokio::test(start_paused = true)]
-async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
-    let temp_dir = tempfile::tempdir().expect("create probe directory");
-    tokio::fs::write(temp_dir.path().join("part.1"), b"shard")
-        .await
-        .expect("create erasure part");
-    let sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
-        temp_dir.path(),
-        crate::sleeper::ScannerDiskIoBudgetLimits {
-            iops: Some(0),
-            bytes_per_sec: None,
-        },
-    );
-    let ctx = CancellationToken::new();
-    let cancel = ctx.clone();
-    let disk = temp_dir.path().to_string_lossy().into_owned();
-    let task = tokio::spawn(async move {
-        loop {
-            if crate::sleeper::scanner_disk_io_budget_status()
-                .iter()
-                .any(|status| status.disk == disk && status.charged_ops == 2)
-            {
-                cancel.cancel();
-                return;
-            }
-            tokio::task::yield_now().await;
-        }
-    });
-    let started = tokio::time::Instant::now();
-    let result = contains_erasure_part_file(temp_dir.path().to_string_lossy().as_ref(), &ctx, &sleeper).await;
-    task.abort();
-    assert!(ctx.is_cancelled(), "probe must reach the entry admission after opening its directory");
-    task.await.expect("entry-wait cancellation task");
-    assert!(result.is_err(), "cancelled probe must not report an erasure part");
-    assert_eq!(started.elapsed(), Duration::ZERO, "entry admission must not wait out its one-second debt");
-}
-
 #[tokio::test]
 #[serial]
 async fn test_scan_folder_missing_bucket_root_still_reports_not_found() {
@@ -2745,15 +2663,11 @@ async fn test_scan_folder_xl_meta_named_directory_uses_namespace_descent() {
 #[serial]
 async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
     let logs = CapturedLogs::default();
-    let subscriber = tracing_subscriber::fmt()
-        .json()
-        .with_max_level(tracing::Level::ERROR)
-        .with_writer(logs.clone())
-        .with_ansi(false)
-        .without_time()
-        .finish();
-    let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+    logs.capture(tracing::Level::ERROR, scan_corrupt_metadata_with_captured_logs(&logs))
+        .await;
+}
 
+async fn scan_corrupt_metadata_with_captured_logs(logs: &CapturedLogs) {
     let (mut scanner, temp_dir) = build_test_scanner().await;
     // Canonicalize for the "drive" field comparison (scanner resolves symlinks).
     let canonical_temp_dir = std::fs::canonicalize(&temp_dir).unwrap_or_else(|_| temp_dir.clone());
@@ -4215,14 +4129,11 @@ async fn test_scan_data_folder_success_clears_resume_hint() {
 #[serial]
 async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
     let logs = CapturedLogs::default();
-    let subscriber = tracing_subscriber::fmt()
-        .json()
-        .with_max_level(tracing::Level::DEBUG)
-        .with_writer(logs.clone())
-        .with_ansi(false)
-        .without_time()
-        .finish();
-    let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+    logs.capture(tracing::Level::DEBUG, scan_unresolved_objects_with_captured_logs(&logs))
+        .await;
+}
+
+async fn scan_unresolved_objects_with_captured_logs(logs: &CapturedLogs) {
     let (scanner, temp_dir) = build_test_scanner().await;
     let _guard = TestGuard {
         temp_dir: Some(temp_dir.clone()),
diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs
index 912e57983..ff34493ca 100644
--- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs
+++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs
@@ -900,14 +900,11 @@ async fn checkpoint_fixture_hot_bucket_certifies_after_mixed_end_and_leader_hand
 
 async fn hot_bucket_certifies_first_full_sweep_after_mixed_end(advance_epoch: bool) {
     let logs = CapturedLogs::default();
-    let subscriber = tracing_subscriber::fmt()
-        .json()
-        .with_max_level(tracing::Level::DEBUG)
-        .with_writer(logs.clone())
-        .with_ansi(false)
-        .without_time()
-        .finish();
-    let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+    logs.capture(tracing::Level::DEBUG, run_hot_bucket_certification(advance_epoch, &logs))
+        .await;
+}
+
+async fn run_hot_bucket_certification(advance_epoch: bool, logs: &CapturedLogs) {
     let (scanner, root) = build_test_scanner().await;
     let _guard = TestGuard {
         temp_dir: Some(root.clone()),
@@ -1056,6 +1053,49 @@ async fn hot_bucket_certifies_first_full_sweep_after_mixed_end(advance_epoch: bo
     assert_eq!(fields["size_reconciliation_entries"], 0);
 }
 
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn checkpoint_admission_log_capture_survives_another_future_dispatcher() {
+    let logs = CapturedLogs::default();
+    let (mut cache, identity) = bound_checkpoint();
+    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+    let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
+    let (done_tx, done_rx) = tokio::sync::oneshot::channel();
+    let captured = logs.capture(tracing::Level::DEBUG, async move {
+        entered_tx.send(()).expect("announce captured future");
+        ready_rx.await.expect("competing dispatcher is installed");
+        for cycle in 100..104 {
+            cache.prepare_bucket_checkpoint("bucket", cycle, 9, SOURCE, PLAN, identity);
+            assert_eq!(cache.info.next_cycle, cycle);
+        }
+        done_tx.send(()).expect("announce checkpoint completion");
+    });
+    let competing = async {
+        entered_rx.await.expect("captured future has yielded");
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::ERROR)
+            .with_writer(std::io::sink)
+            .finish();
+        // An unrelated future keeps a quieter thread-local default while
+        // the captured future resumes on this same runtime thread.
+        let _guard = tracing::subscriber::set_default(subscriber);
+        ready_tx.send(()).expect("announce competing dispatcher");
+        done_rx.await.expect("checkpoint preparation has completed");
+    };
+    tokio::join!(captured, competing);
+
+    let events = logs
+        .contents()
+        .lines()
+        .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("scanner JSON event"))
+        .filter(|event| event["fields"]["state"] == "bucket_checkpoint_prepared")
+        .collect::<Vec<_>>();
+    assert_eq!(events.len(), 4, "all checkpoint admissions must use the captured dispatcher");
+    for (cycle, event) in (100..104).zip(events) {
+        assert_eq!(event["fields"]["cycle"], cycle);
+    }
+}
+
 async fn write_checkpoint_object(root: &std::path::Path, object: &str, versions: &[(Option<Uuid>, i64)]) {
     let mut metadata = FileMeta::new();
     for (index, (version_id, size)) in versions.iter().enumerate() {
@@ -1407,3 +1447,39 @@ where
     }
     panic!("finite stable fixture must converge using the same four-object budget without an unbounded final sweep");
 }
+
+fn emit_first_registration_capture_event() {
+    tracing::debug!(
+        target: "rustfs::scanner::folder",
+        event = EVENT_SCANNER_FOLDER_STATE,
+        component = LOG_COMPONENT_SCANNER,
+        subsystem = LOG_SUBSYSTEM_FOLDER,
+        state = "first_registration_capture",
+        "Scanner test capture exercised first callsite registration"
+    );
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[serial]
+async fn checkpoint_log_capture_survives_first_registration_on_unsubscribed_thread() {
+    let logs = CapturedLogs::default();
+    // Construct the capture before polling it: its subscriber must be live
+    // when the other thread first registers this otherwise unused callsite.
+    let captured = logs.capture(tracing::Level::DEBUG, async {
+        emit_first_registration_capture_event();
+    });
+    std::thread::spawn(emit_first_registration_capture_event)
+        .join()
+        .expect("unsubscribed first-registration thread");
+    captured.await;
+    let events = logs
+        .contents()
+        .lines()
+        .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("scanner JSON event"))
+        .filter(|event| event["fields"]["state"] == "first_registration_capture")
+        .count();
+    assert_eq!(
+        events, 1,
+        "first registration without a subscriber must not disable the captured callsite"
+    );
+}
diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs
index 93355f7d2..25953fb0b 100644
--- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs
+++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs
@@ -999,14 +999,11 @@ async fn periodic_resume_uses_traversal_order_for_punctuation_siblings() {
 #[serial]
 async fn periodic_failed_nested_entries_cannot_certify_a_bucket_with_diagnostics_disabled() {
     let logs = CapturedLogs::default();
-    let subscriber = tracing_subscriber::fmt()
-        .json()
-        .with_max_level(tracing::Level::DEBUG)
-        .with_writer(logs.clone())
-        .with_ansi(false)
-        .without_time()
-        .finish();
-    let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+    logs.capture(tracing::Level::DEBUG, periodic_failed_entries_with_captured_logs(&logs))
+        .await;
+}
+
+async fn periodic_failed_entries_with_captured_logs(logs: &CapturedLogs) {
     let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await;
     let result = temp_env::async_with_vars([(ENV_FAILED_OBJECT_TTL_SECS, Some("0"))], async {
         let mut cache = loaded;
diff --git a/crates/scanner/src/scanner_folder/tests/erasure_probe.rs b/crates/scanner/src/scanner_folder/tests/erasure_probe.rs
new file mode 100644
index 000000000..43d1c5899
--- /dev/null
+++ b/crates/scanner/src/scanner_folder/tests/erasure_probe.rs
@@ -0,0 +1,827 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+use std::collections::{HashMap, VecDeque};
+use std::sync::LazyLock;
+
+#[derive(Clone, Copy)]
+enum DirectoryOrder {
+    PartLast,
+    FailAfter(usize, std::io::ErrorKind),
+    FailOpen(std::io::ErrorKind),
+}
+
+pub(in crate::scanner_folder) type OrderedEntries = VecDeque<std::io::Result<tokio::fs::DirEntry>>;
+static ORDERED_DIRECTORIES: LazyLock<Mutex<HashMap<String, DirectoryOrder>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+
+struct OrderedDirectory(String);
+
+impl OrderedDirectory {
+    fn new(path: &std::path::Path) -> Self {
+        let path = path.to_string_lossy().into_owned();
+        ORDERED_DIRECTORIES
+            .lock()
+            .expect("ordered directory registry")
+            .insert(path.clone(), DirectoryOrder::PartLast);
+        Self(path)
+    }
+    fn failing(path: &std::path::Path, entries: usize, kind: std::io::ErrorKind) -> Self {
+        let guard = Self::new(path);
+        ORDERED_DIRECTORIES
+            .lock()
+            .expect("ordered directory registry")
+            .insert(guard.0.clone(), DirectoryOrder::FailAfter(entries, kind));
+        guard
+    }
+    fn fail_open(path: &std::path::Path, kind: std::io::ErrorKind) -> Self {
+        let guard = Self::new(path);
+        ORDERED_DIRECTORIES
+            .lock()
+            .expect("ordered directory registry")
+            .insert(guard.0.clone(), DirectoryOrder::FailOpen(kind));
+        guard
+    }
+}
+
+impl Drop for OrderedDirectory {
+    fn drop(&mut self) {
+        ORDERED_DIRECTORIES.lock().expect("ordered directory cleanup").remove(&self.0);
+    }
+}
+
+pub(in crate::scanner_folder) async fn ordered_entries(
+    path: &str,
+    reader: &mut tokio::fs::ReadDir,
+) -> std::io::Result<Option<OrderedEntries>> {
+    let order = ORDERED_DIRECTORIES
+        .lock()
+        .expect("ordered directory registry")
+        .get(path)
+        .copied();
+    let Some(order) = order else { return Ok(None) };
+    if let DirectoryOrder::FailOpen(kind) = order {
+        return Err(std::io::Error::from(kind));
+    }
+    let mut entries = Vec::new();
+    while let Some(entry) = reader.next_entry().await? {
+        entries.push(entry);
+    }
+    // Real entries and file types, with an explicit adversarial order rather
+    // than an assumption about the filesystem's enumeration order.
+    entries.sort_by_key(|entry| (entry.file_name().to_str() == Some("part.1"), entry.file_name()));
+    let mut entries: OrderedEntries = entries.into_iter().map(Ok).collect();
+    if let DirectoryOrder::FailAfter(count, kind) = order {
+        entries.truncate(count);
+        entries.push_back(Err(std::io::Error::from(kind)));
+    }
+    Ok(Some(entries))
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_finds_late_part_without_scanning_data_children() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let data_dir = root.join("bucket/object").join(Uuid::new_v4().to_string());
+    tokio::fs::create_dir_all(&data_dir).await.expect("create candidate");
+    for index in 0..65 {
+        tokio::fs::write(data_dir.join(format!("part.{index}.integrity.orphan")), b"")
+            .await
+            .expect("write sidecar");
+    }
+    tokio::fs::write(data_dir.join("part.1"), b"shard")
+        .await
+        .expect("write late part");
+    tokio::fs::create_dir(data_dir.join("namespace-child"))
+        .await
+        .expect("stage a child before the part");
+    let _order = OrderedDirectory::new(&data_dir);
+    scanner.is_erasure_mode = true;
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
+        &root,
+        crate::sleeper::ScannerDiskIoBudgetLimits {
+            iops: Some(1_000_000),
+            bytes_per_sec: None,
+        },
+    );
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_directories: Some(1),
+            ..Default::default()
+        },
+    );
+    scanner.budget = budget.clone();
+    let mut into = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            budget.token(),
+            CachedFolder {
+                name: "bucket/object".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut into,
+        )
+        .await
+        .expect("late part must establish the parent object boundary");
+    assert_eq!(into.failed_objects, 1);
+    assert_eq!(into.objects, 0);
+    assert!(
+        scanner
+            .new_cache
+            .info
+            .failed_objects
+            .contains_key(root.join("bucket/object/xl.meta").to_string_lossy().as_ref())
+    );
+    assert_eq!(budget.reason(), None, "payload directories must not consume namespace directory quota");
+    assert!(scanner.new_cache.info.pending_heals.iter().any(|pending| {
+        pending.kind == PendingScannerHealKind::Object
+            && pending.bucket == "bucket"
+            && pending.object.as_deref() == Some("object")
+    }));
+}
+
+async fn candidate_scanner() -> (FolderScanner, std::path::PathBuf, CachedFolder) {
+    let (mut scanner, root) = build_test_scanner().await;
+    for info in [
+        &mut scanner.old_cache.info,
+        &mut scanner.new_cache.info,
+        &mut scanner.update_cache.info,
+    ] {
+        info.name = "bucket".into();
+    }
+    scanner.is_erasure_mode = true;
+    let name = format!("bucket/{}", Uuid::new_v4());
+    tokio::fs::create_dir_all(root.join(&name))
+        .await
+        .expect("create UUID namespace");
+    let folder = CachedFolder {
+        name,
+        parent: Some(hash_path("bucket")),
+        object_heal_prob_div: 1,
+    };
+    (scanner, root, folder)
+}
+
+fn bind_budget(scanner: &mut FolderScanner, root: &std::path::Path, iops: u64) {
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest).for_disk_with_budget_for_tests(
+        root,
+        crate::sleeper::ScannerDiskIoBudgetLimits {
+            iops: Some(iops),
+            bytes_per_sec: None,
+        },
+    );
+}
+
+fn charged_ops(root: &std::path::Path) -> u64 {
+    let status = crate::sleeper::scanner_disk_io_budget_status()
+        .into_iter()
+        .find(|status| status.disk == root.to_string_lossy())
+        .expect("bound candidate budget");
+    assert_eq!(status.charged_bytes, 0);
+    status.charged_ops
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
+    for entry_count in [0, 1, 63, 64, 65, 129] {
+        let (mut scanner, root, folder) = candidate_scanner().await;
+        let _guard = TestGuard {
+            temp_dir: Some(root.clone()),
+        };
+        for index in 0..entry_count {
+            tokio::fs::write(root.join(&folder.name).join(format!("namespace-{index}")), b"")
+                .await
+                .expect("create non-part entry");
+        }
+        bind_budget(&mut scanner, &root, 1_000_000);
+        let ctx = CancellationToken::new();
+        scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
+        let outcome = scanner
+            .scan_folder_inner(
+                ctx,
+                folder,
+                &mut DataUsageEntry::default(),
+                FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            )
+            .await
+            .expect("classify full UUID namespace");
+        assert_eq!(outcome, FolderScanOutcome::Namespace);
+        assert_eq!(charged_ops(&root), 1 + entry_count, "one open and each entry once, never EOF");
+        assert_eq!(scanner.budget.entries_visited(), entry_count);
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_namespace_uses_one_listing_and_no_extra_probe_charge() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for index in 0..129 {
+        tokio::fs::write(root.join(&folder.name).join(format!("namespace-{index}")), b"")
+            .await
+            .expect("write namespace entry");
+    }
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let mut into = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut into,
+        )
+        .await
+        .expect("scan namespace through parent");
+    assert_eq!(
+        charged_ops(&root),
+        132,
+        "parent open/entry + candidate open/129 entries, no second listing"
+    );
+    assert_eq!(into.failed_objects, 0);
+    assert!(scanner.new_cache.info.failed_objects.is_empty());
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    bind_budget(&mut scanner, &root, 0);
+    scanner.sleeper.admit_disk_io(1, 0).await;
+    let ctx = CancellationToken::new();
+    let cancel = ctx.clone();
+    let task = tokio::spawn(async move {
+        tokio::task::yield_now().await;
+        cancel.cancel();
+    });
+    let started = tokio::time::Instant::now();
+    let result = scanner
+        .scan_folder_inner(
+            ctx.clone(),
+            folder,
+            &mut DataUsageEntry::default(),
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+        )
+        .await;
+    task.await.expect("open-wait cancellation");
+    assert!(result.is_err());
+    assert_eq!(started.elapsed(), Duration::ZERO);
+    assert!(wait_for_scanner_io(&ctx, scanner.sleeper.admit_disk_io(1, 0)).await.is_err());
+    assert_eq!(charged_ops(&root), 2, "already cancelled operations must not charge");
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    tokio::fs::write(root.join(&folder.name).join("part.1"), b"shard")
+        .await
+        .expect("write part");
+    bind_budget(&mut scanner, &root, 0);
+    let ctx = CancellationToken::new();
+    let cancel = ctx.clone();
+    let disk = root.clone();
+    let task = tokio::spawn(async move {
+        loop {
+            if charged_ops(&disk) == 2 {
+                cancel.cancel();
+                return;
+            }
+            tokio::task::yield_now().await;
+        }
+    });
+    let started = tokio::time::Instant::now();
+    let result = scanner
+        .scan_folder_inner(
+            ctx.clone(),
+            folder,
+            &mut DataUsageEntry::default(),
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+        )
+        .await;
+    assert!(ctx.is_cancelled());
+    task.await.expect("entry-wait cancellation");
+    assert!(result.is_err(), "cancelled classification cannot become found or EOF");
+    assert_eq!(started.elapsed(), Duration::ZERO);
+    assert_eq!(charged_ops(&root), 2);
+}
+
+#[test]
+fn erasure_part_names_require_a_positive_u32() {
+    for name in ["part.1", "part.10000", "part.4294967295"] {
+        assert!(is_erasure_part_name(name));
+    }
+    for name in [
+        "part.0",
+        "part.",
+        "part.-1",
+        "part.4294967296",
+        "part.1.integrity.orphan",
+        "Part.1",
+    ] {
+        assert!(!is_erasure_part_name(name), "{name}");
+    }
+}
+
+fn enable_checkpoint(scanner: &mut FolderScanner) {
+    let plan = crate::DataUsageScanPlanDigest([11; 32]);
+    let identity = crate::DataUsageScanIdentity {
+        version: 1,
+        bucket_incarnation: Uuid::from_u128(7),
+        set_layout: crate::DataUsageScanPlanDigest([12; 32]),
+        publication_epoch: 3,
+        tier_registry_generation: 0,
+        scan_mode: HealScanMode::Normal,
+    };
+    let mut cache = DataUsageCache::default();
+    cache.prepare_bucket_checkpoint("bucket", 7, 3, crate::DataUsageCacheSource::new(1, 0), plan, identity);
+    cache.replace("bucket", "", DataUsageEntry::default());
+    scanner.old_cache = cache.clone();
+    scanner.new_cache = cache.clone();
+    scanner.update_cache = cache;
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_defers_metadata_until_part_classification_finishes() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let path = root.join(&folder.name);
+    for index in 0..65 {
+        tokio::fs::write(path.join(format!("part.{index}.integrity.orphan")), b"")
+            .await
+            .expect("write sidecar");
+    }
+    let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
+    tokio::fs::write(path.join("xl.meta"), metadata_for_object("bucket", object))
+        .await
+        .expect("write candidate metadata");
+    tokio::fs::write(path.join("part.1"), b"shard")
+        .await
+        .expect("write late part");
+    let _order = OrderedDirectory::new(&path);
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let ctx = CancellationToken::new();
+    scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
+    let mut into = DataUsageEntry::default();
+    let outcome = scanner
+        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false })
+        .await
+        .expect("classify candidate");
+    assert_eq!(outcome, FolderScanOutcome::ErasureData);
+    assert_eq!(into.objects, 0, "candidate xl.meta cannot start accounting or maintenance before EOF");
+    assert_eq!(scanner.budget.progress(), (0, 0));
+    assert_eq!(scanner.budget.entries_visited(), 67);
+    assert_eq!(charged_ops(&root), 68, "metadata entry only: no metadata stat or read reservation");
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_cancelled_batch_keeps_partial_checkpoint_and_reclassifies_on_resume() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let path = root.join(&folder.name);
+    for index in 0..65 {
+        tokio::fs::write(path.join(format!("part.{index}.integrity.orphan")), b"")
+            .await
+            .expect("write sidecar");
+    }
+    tokio::fs::write(path.join("part.1"), b"shard")
+        .await
+        .expect("write late part");
+    let _order = OrderedDirectory::new(&path);
+    enable_checkpoint(&mut scanner);
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
+    scanner.budget = budget.clone();
+    let cancel_budget = budget.clone();
+    let cancel = tokio::spawn(async move {
+        loop {
+            if cancel_budget.entries_visited() >= 65 {
+                cancel_budget.cancel_for_runtime();
+                return;
+            }
+            tokio::task::yield_now().await;
+        }
+    });
+    let parent_folder = CachedFolder {
+        name: "bucket".into(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+    let mut into = DataUsageEntry::default();
+    let result = scanner.scan_folder(budget.token(), parent_folder.clone(), &mut into).await;
+    cancel.await.expect("batch cancellation task");
+    assert!(result.is_err());
+    assert_eq!(
+        budget.entries_visited(),
+        65,
+        "parent entry and first full candidate batch are semantic progress"
+    );
+    assert_eq!(charged_ops(&root), 67, "two opens and 65 entries, cancellation before the next batch");
+    assert!(
+        scanner.new_cache.info.failed_objects.is_empty(),
+        "unfinished classification cannot manufacture missing metadata"
+    );
+    let snapshot = scanner
+        .checkpoint_snapshot(&parent_folder, &hash_path("bucket"), &into)
+        .expect("partial raw checkpoint");
+    assert!(!snapshot.info.snapshot_complete);
+    assert!(snapshot.info.scan_coverage_receipt.is_none());
+    let cursor = snapshot
+        .validated_raw_enumeration_cursor()
+        .expect("validated candidate cursor");
+    assert_eq!(cursor.parent, folder.name);
+    assert_eq!(cursor.entries_seen, 64);
+    let index = snapshot
+        .validated_raw_enumeration_page_index()
+        .expect("validated candidate index");
+    assert_eq!(index.indexed_entries().expect("index entries").len(), 64);
+    let bytes = serde_json::to_vec(&snapshot).expect("serialize partial checkpoint");
+    let reloaded: DataUsageCache = serde_json::from_slice(&bytes).expect("reload partial checkpoint");
+    scanner.old_cache = reloaded.clone();
+    scanner.new_cache = reloaded.clone();
+    scanner.update_cache = reloaded;
+    scanner.raw_enumeration_progress.clear();
+    let next_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
+    scanner.budget = next_budget.clone();
+    scanner
+        .scan_folder(next_budget.token(), parent_folder, &mut DataUsageEntry::default())
+        .await
+        .expect("revalidate resumed candidate");
+    assert!(
+        scanner
+            .new_cache
+            .info
+            .failed_objects
+            .contains_key(root.join("bucket/xl.meta").to_string_lossy().as_ref())
+    );
+    assert_eq!(
+        charged_ops(&root),
+        136,
+        "resume recharges actual reads; the raw index is not a filesystem seek"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_late_data_sibling_prevents_parent_coverage_certification() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    enable_checkpoint(&mut scanner);
+    scanner.is_erasure_mode = true;
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let first = format!("object/{}", Uuid::from_u128(1));
+    write_test_object_metadata(&root, "bucket", &first).await;
+    let data = root.join("bucket/object").join(Uuid::from_u128(2).to_string());
+    tokio::fs::create_dir_all(&data).await.expect("create later data sibling");
+    tokio::fs::write(data.join("part.1"), b"shard")
+        .await
+        .expect("write later part");
+    let mut into = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket/object".into(),
+                parent: Some(hash_path("bucket")),
+                object_heal_prob_div: 1,
+            },
+            &mut into,
+        )
+        .await
+        .expect("scan mixed siblings");
+    assert_eq!(
+        scanner
+            .new_cache
+            .find(&format!("bucket/{first}"))
+            .expect("preceding UUID object retained")
+            .objects,
+        1,
+        "the preceding legitimate UUID object was scanned once"
+    );
+    assert_eq!(into.failed_objects, 1);
+    assert!(scanner.coverage_gap);
+    assert!(
+        scanner.coverage_frontier.is_none(),
+        "completed siblings cannot certify unresolved parent coverage"
+    );
+    assert!(
+        scanner
+            .new_cache
+            .info
+            .failed_objects
+            .contains_key(root.join("bucket/object/xl.meta").to_string_lossy().as_ref())
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_rechecks_data_below_a_restored_namespace_frontier() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    enable_checkpoint(&mut scanner);
+    tokio::fs::write(root.join(&folder.name).join("part.1"), b"shard")
+        .await
+        .expect("write part");
+    let hash = hash_path(&folder.name);
+    scanner.old_cache.replace_hashed(
+        &hash,
+        &Some(hash_path("bucket")),
+        &DataUsageEntry {
+            objects: 1,
+            compacted: true,
+            ..Default::default()
+        },
+    );
+    scanner.resume_frontier = Some(folder.name.clone());
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let mut into = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut into,
+        )
+        .await
+        .expect("reclassify covered UUID directory");
+    assert_eq!(into.failed_objects, 1);
+    assert_eq!(into.objects, 0, "stale namespace totals cannot bypass current part evidence");
+    assert!(scanner.new_cache.find(&folder.name).is_none());
+}
+
+#[cfg(unix)]
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_part_symlink_is_not_data_evidence() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let target = root.join("target");
+    tokio::fs::write(&target, b"shard").await.expect("write symlink target");
+    symlink(&target, root.join(&folder.name).join("part.1")).expect("create part symlink");
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let outcome = scanner
+        .scan_folder_inner(
+            CancellationToken::new(),
+            folder,
+            &mut DataUsageEntry::default(),
+            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+        )
+        .await
+        .expect("classify symlink namespace");
+    assert_eq!(outcome, FolderScanOutcome::Namespace);
+    assert_eq!(charged_ops(&root), 3, "open, entry, symlink target stat each once");
+    assert!(scanner.new_cache.info.failed_objects.is_empty());
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_covered_namespace_reuses_totals_after_classification() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
+    write_test_object_metadata(&root, "bucket", object).await;
+    enable_checkpoint(&mut scanner);
+    scanner.old_cache.replace_hashed(
+        &hash_path(&folder.name),
+        &folder.parent,
+        &DataUsageEntry {
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    scanner.resume_frontier = Some(folder.name.clone());
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &ctx,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_objects: Some(1),
+            ..Default::default()
+        },
+    );
+    scanner.budget = budget.clone();
+    scanner
+        .scan_folder(
+            budget.token(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        )
+        .await
+        .expect("covered candidate must not exhaust object quota again");
+    assert_eq!(budget.reason(), None);
+    assert_eq!(budget.progress(), (0, 1), "only the parent consumes namespace quota");
+    assert_eq!(charged_ops(&root), 4, "list both directories once without re-reading covered metadata");
+    assert_eq!(scanner.new_cache.find(&folder.name).expect("reused candidate").objects, 1);
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_iteration_error_cannot_replay_deferred_metadata_as_eof() {
+    for kind in [std::io::ErrorKind::NotFound, std::io::ErrorKind::NotADirectory] {
+        let (mut scanner, root, folder) = candidate_scanner().await;
+        let _guard = TestGuard {
+            temp_dir: Some(root.clone()),
+        };
+        let path = root.join(&folder.name);
+        let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
+        write_test_object_metadata(&root, "bucket", object).await;
+        tokio::fs::write(path.join("part.1"), b"shard")
+            .await
+            .expect("write unread part");
+        let _error = OrderedDirectory::failing(&path, 1, kind);
+        bind_budget(&mut scanner, &root, 1_000_000);
+        let result = scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: "bucket".into(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut DataUsageEntry::default(),
+            )
+            .await;
+        assert!(
+            matches!(result, Err(ScannerError::Io(err)) if err.kind() == kind),
+            "candidate error must remain typed and cannot become namespace completion"
+        );
+        assert_eq!(
+            charged_ops(&root),
+            4,
+            "two opens, parent entry, deferred metadata entry; no metadata read"
+        );
+        assert!(
+            scanner.new_cache.info.failed_objects.is_empty(),
+            "unknown classification is not missing-metadata evidence"
+        );
+        assert!(scanner.new_cache.find(&folder.name).is_none());
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_preserves_compacted_namespace_sampling_after_classification() {
+    let (mut scanner, root, folder) = candidate_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    let object = folder.name.strip_prefix("bucket/").expect("candidate in bucket");
+    write_test_object_metadata(&root, "bucket", object).await;
+    let hash = hash_path(&folder.name);
+    scanner.old_cache.replace_hashed(
+        &hash,
+        &folder.parent,
+        &DataUsageEntry {
+            objects: 1,
+            compacted: true,
+            ..Default::default()
+        },
+    );
+    let skipped_cycle = (0..data_usage_update_dir_cycles())
+        .find(|cycle| !hash.mod_(*cycle, data_usage_update_dir_cycles()))
+        .expect("at least one cycle samples another namespace");
+    scanner.old_cache.info.next_cycle = u64::from(skipped_cycle);
+    bind_budget(&mut scanner, &root, 1_000_000);
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &ctx,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_objects: Some(1),
+            ..Default::default()
+        },
+    );
+    scanner.budget = budget.clone();
+    scanner
+        .scan_folder(
+            budget.token(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        )
+        .await
+        .expect("sampled candidate must not consume object quota");
+    assert_eq!(budget.reason(), None);
+    assert_eq!(budget.progress(), (0, 1));
+    assert_eq!(
+        charged_ops(&root),
+        4,
+        "sampling still requires full current classification, without metadata re-read"
+    );
+    assert_eq!(scanner.new_cache.find(&folder.name).expect("sampled candidate").objects, 1);
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_candidate_vanished_before_open_preserves_other_namespace_progress() {
+    for kind in [std::io::ErrorKind::NotFound, std::io::ErrorKind::NotADirectory] {
+        let (mut scanner, root, folder) = candidate_scanner().await;
+        let _guard = TestGuard {
+            temp_dir: Some(root.clone()),
+        };
+        enable_checkpoint(&mut scanner);
+        scanner.old_cache.replace_hashed(
+            &hash_path(&folder.name),
+            &folder.parent,
+            &DataUsageEntry {
+                objects: 1,
+                ..Default::default()
+            },
+        );
+        write_test_object_metadata(&root, "bucket", "remaining").await;
+        let _vanished = OrderedDirectory::fail_open(&root.join(&folder.name), kind);
+        bind_budget(&mut scanner, &root, 1_000_000);
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new_with_progress_tracking(
+            &ctx,
+            crate::scanner_budget::ScannerCycleBudgetConfig {
+                max_directories: Some(2),
+                ..Default::default()
+            },
+        );
+        scanner.budget = budget.clone();
+        let mut into = DataUsageEntry::default();
+        scanner
+            .scan_folder(
+                budget.token(),
+                CachedFolder {
+                    name: "bucket".into(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut into,
+            )
+            .await
+            .expect("concurrent candidate deletion or replacement must not fail the remaining namespace scan");
+        assert_eq!(budget.reason(), None);
+        assert_eq!(budget.progress().1, 2, "only parent and remaining namespace consume directory quota");
+        assert!(
+            scanner.new_cache.find(&folder.name).is_none(),
+            "vanished candidate cannot retain stale cached totals"
+        );
+        assert_eq!(
+            scanner
+                .new_cache
+                .find("bucket/remaining")
+                .expect("remaining object scanned")
+                .objects,
+            1
+        );
+        assert_eq!(
+            charged_ops(&root),
+            7,
+            "vanished open is charged but contributes no entry or metadata read"
+        );
+        assert!(scanner.new_cache.info.failed_objects.is_empty());
+        assert!(
+            scanner.new_cache.info.pending_heals.is_empty(),
+            "vanished directory is not missing-xl.meta evidence"
+        );
+        assert_eq!(
+            scanner.coverage_gap,
+            kind == std::io::ErrorKind::NotADirectory,
+            "a replaced directory retains the existing incomplete-coverage behavior"
+        );
+    }
+}
diff --git a/crates/scanner/src/sleeper.rs b/crates/scanner/src/sleeper.rs
index da35fa3e8..1ab03d41b 100644
--- a/crates/scanner/src/sleeper.rs
+++ b/crates/scanner/src/sleeper.rs
@@ -225,7 +225,7 @@ impl DynamicSleeper {
         self.admit_disk_io(1, 0).await;
     }
 
-    async fn sleep_folder_preset(&self) {
+    pub(crate) async fn sleep_folder_preset(&self) {
         if !Self::should_sleep() {
             return;
         }
diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md
index f38ae1768..1b3e3758e 100644
--- a/docs/operations/scanner-runtime-controls.md
+++ b/docs/operations/scanner-runtime-controls.md
@@ -142,7 +142,9 @@ The admission gate orders scanner I/O behind foreground requests but does not bo
 
 Only `xl.meta` files enter object metadata accounting. Other payload files consume entry charges without extra size stats or metadata-read charges. An `xl.meta` symlink reuses its resolved target size for the byte charge. Cancellation interrupts folder pacing, budget waits, and directory/stat waits. Filesystem work already dispatched to a blocking worker can still finish.
 
-Erasure data-directory probes for missing `xl.meta` use the same directory-open and entry budget. They retain their 64-entry limit and remain cancellation-aware without adding a preset folder delay per probe.
+UUID candidates under a directory without `xl.meta` are classified during their normal directory traversal. Each directory open and entry is charged once; traversal yields every 64 entries and continues under the budget until a regular `part.N` is found or EOF is reached. A full batch is not evidence of absence. Candidate metadata accounting is deferred until classification finishes. A data directory retains the existing missing-parent-metadata and non-destructive heal path; a namespace continues from the same listing, without a second probe/open/entry pass. Data directories do not consume namespace-directory quota or a preset folder delay; a candidate proven to be a namespace consumes that quota and delay once.
+
+Candidate entries use the normal raw-enumeration progress and checkpoint path. Cancellation or an iteration error leaves classification incomplete, never an EOF or successful coverage result. A child already absent or replaced by a non-directory at open is handled as vanished, without stale namespace totals or a missing-metadata heal request. A replacement retains the existing coverage gap; mid-iteration errors never become EOF. A clean namespace covered by a restored frontier reuses its prior totals only after current classification reaches EOF; metadata is not read or charged to object quota again. A later data candidate prevents completed namespace siblings from certifying the unresolved parent. Existing raw-page indexes are membership receipts, not portable filesystem seek offsets: a later cycle or process restart reopens and revalidates the directory, with actual reads charged again. This does not guarantee that a directory larger than the configured cycle budget can finish within that budget.
 
 - Media is detected once per disk path from the backing block device's `rotational` flag on Linux, through partitions, `dm-*`, and software RAID; any spindle in the stack counts as rotational. Undetectable media, including non-Linux hosts and overlay filesystems without a bind-mounted volume, uses the rotational defaults because a spindle is the case that needs protection.
 - `auto` resolves to `100` operations/s and `16 MiB/s` on rotational or unknown media and to `4000` operations/s and `256 MiB/s` on flash. The rotational values leave roughly half of a 7200 rpm spindle's random reads to foreground traffic; the flash values are a few percent of a SATA SSD. The `fastest` speed preset resolves `auto` to no cap; explicit numbers always apply.