fix(scanner): skip child prefixes removed before they are opened (#8479)

A child prefix can be deleted after its parent listed it and before scan_folder opens it. read_dir then returned NotFound, which propagated through the compacted subtree and dropped the whole top-level child (child_scan_failed), so the cycle never produced a durable usage snapshot under continuous deletes.

Treat NotFound on open like a directory that vanished during iteration for non-root folders. A missing bucket root still reports NotFound so namespace removal handling is unchanged.

Refs #8478

Co-authored-by: Anton <5527364+basic-system@users.noreply.github.com>

crates/scanner/src/scanner_folder.rs

@@ -1657,8 +1657,26 @@ impl FolderScanner {
                 "Scanner folder state updated"
             );
 
+            // A child prefix can be removed after its parent listed it and before
+            // it is opened here (e.g. concurrent object deletes). Treat it like a
+            // 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 tokio::fs::read_dir(&dir_path).await {
-                Ok(dir_reader) => dir_reader,
+                Ok(dir_reader) => Some(dir_reader),
+                Err(e) if e.kind() == ErrorKind::NotFound && folder.parent.is_some() => {
+                    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_before_open",
+                        error = %e,
+                        "Scanner folder state updated"
+                    );
+                    None
+                }
                 Err(e) => return Err(ScannerError::Io(e)),
             };
             let mut pending_entry_progress = 0_u64;
@@ -1667,7 +1685,11 @@ impl FolderScanner {
             let raw_enumeration_committed_entries = self.raw_enumeration_committed_entry_oracle(&folder.name);
 
             loop {
-                let entry = match dir_reader.next_entry().await {
+                let Some(reader) = dir_reader.as_mut() else {
+                    raw_enumeration_complete = true;
+                    break;
+                };
+                let entry = match reader.next_entry().await {
                     Ok(Some(entry)) => entry,
                     Ok(None) => {
                         raw_enumeration_complete = true;

crates/scanner/src/scanner_folder/tests.rs

@@ -2172,6 +2172,59 @@ async fn test_scan_folder_skips_unreadable_child_directory() {
     assert!(result.is_ok(), "expected unreadable child directory to be skipped");
 }
 
+#[tokio::test]
+#[serial]
+async fn test_scan_folder_treats_child_removed_before_open_as_empty() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone());
+
+    // The parent listed `gone/`, then a concurrent delete removed it before the
+    // child scan opened it.
+    std::fs::create_dir_all(temp_dir.join("bucket")).expect("failed to create bucket dir");
+
+    scanner.old_cache.info.name = "bucket".to_string();
+    scanner.new_cache.info.name = "bucket".to_string();
+    scanner.update_cache.info.name = "bucket".to_string();
+
+    let folder = CachedFolder {
+        name: "bucket/gone".to_string(),
+        parent: Some(hash_path("bucket")),
+        object_heal_prob_div: 1,
+    };
+
+    let mut into = DataUsageEntry::default();
+    let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await;
+
+    assert!(result.is_ok(), "expected a child removed before open to be skipped, got {result:?}");
+    assert_eq!(into.objects, 0);
+    assert_eq!(into.size, 0);
+}
+
+#[tokio::test]
+#[serial]
+async fn test_scan_folder_missing_bucket_root_still_reports_not_found() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 0, &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();
+
+    let folder = CachedFolder {
+        name: "bucket".to_string(),
+        parent: None,
+        object_heal_prob_div: 1,
+    };
+
+    let mut into = DataUsageEntry::default();
+    let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await;
+
+    assert!(
+        matches!(&result, Err(ScannerError::Io(e)) if e.kind() == ErrorKind::NotFound),
+        "expected a missing bucket root to keep reporting NotFound, got {result:?}"
+    );
+}
+
 #[tokio::test]
 #[serial]
 async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {