fix(scanner): retain child objects beneath object prefixes

crates/scanner/src/scanner_folder.rs

@@ -425,7 +425,10 @@ enum FolderScanSource {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum FolderScanMode {
     Namespace,
-    ErasureCandidate { reuse_namespace: bool },
+    ErasureCandidate {
+        reuse_namespace: bool,
+        parent_has_metadata: bool,
+    },
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -2065,7 +2068,9 @@ impl FolderScanner {
                     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
+                        if let FolderScanMode::ErasureCandidate {
+                            reuse_namespace: true, ..
+                        } = mode
                             && !into.compacted
                             && self.old_cache.find(&folder.name).is_some()
                         {
@@ -2079,10 +2084,23 @@ impl FolderScanner {
                             return Ok(FolderScanOutcome::Namespace);
                         }
                         candidate_unclassified = false;
-                        if !self.budget.try_start_directory() {
-                            return Err(ScannerError::Other("Operation cancelled".to_string()));
+                        // An empty UUID directory beside object metadata has
+                        // no namespace work. Its listing was already charged;
+                        // nonempty namespaces still need directory admission.
+                        let empty_object_child = matches!(
+                            mode,
+                            FolderScanMode::ErasureCandidate {
+                                parent_has_metadata: true,
+                                ..
+                            }
+                        ) && deferred_metadata.is_none()
+                            && queued_folders.is_empty();
+                        if !empty_object_child {
+                            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?;
                         }
-                        wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_preset()).await?;
                     }
                     let Some(entry) = deferred_metadata.take() else { break };
                     (entry, true)
@@ -2552,20 +2570,6 @@ impl FolderScanner {
                 return Err(ScannerError::Other("Operation cancelled".to_string()));
             }
 
-            if found_object_metadata && self.is_erasure_mode {
-                // If we found an object in erasure mode, we skip subdirs (only datadirs)...
-                debug!(
-                    target: "rustfs::scanner::folder",
-                    event = EVENT_SCANNER_FOLDER_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_FOLDER,
-                    folder = %folder.name,
-                    state = "erasure_object_found",
-                    "Scanner folder descent stopped after erasure object"
-                );
-                break;
-            }
-
             // Checkpointed sweeps need child entries to validate their resume
             // frontier (#8292). Outside that, also compact a wide bucket root:
             // excluding it left #8961's ~170k top-level prefixes unbounded until
@@ -2654,6 +2658,7 @@ impl FolderScanner {
                 let mut child_mode = if self.is_erasure_mode && is_erasure_candidate(&folder_item.name) {
                     FolderScanMode::ErasureCandidate {
                         reuse_namespace: reuse_covered,
+                        parent_has_metadata: found_object_metadata,
                     }
                 } else {
                     FolderScanMode::Namespace
@@ -2721,7 +2726,10 @@ impl FolderScanner {
                                 }
                                 // Sampling may skip namespace work, never the
                                 // classification that protects its parent object.
-                                child_mode = FolderScanMode::ErasureCandidate { reuse_namespace: true };
+                                child_mode = FolderScanMode::ErasureCandidate {
+                                    reuse_namespace: true,
+                                    parent_has_metadata: found_object_metadata,
+                                };
                             } else {
                                 folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                             }
@@ -2816,6 +2824,11 @@ impl FolderScanner {
                         updates.cache.delete_recursive(&h);
                     }
                     into.children.remove(&clone_cache_value(&h.0));
+                    // Object keys can also be prefixes. A classified payload
+                    // directory must not hide their other namespace children.
+                    if found_object_metadata {
+                        continue;
+                    }
                     // Earlier namespace siblings may have completed, but none
                     // can certify coverage of this unresolved parent object.
                     self.coverage_gap |= forward_sweep;

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

@@ -183,6 +183,117 @@ fn charged_ops(root: &std::path::Path) -> u64 {
     status.charged_ops
 }
 
+#[tokio::test]
+#[serial]
+async fn erasure_object_metadata_preserves_child_object_namespaces() {
+    for (corrupt_parent, cached_parent) in [(false, false), (true, false), (false, true)] {
+        let (mut scanner, root) = build_test_scanner().await;
+        let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
+        scanner.old_cache.info.name = "bucket".into();
+        scanner.new_cache.info.name = "bucket".into();
+        scanner.is_erasure_mode = true;
+        bind_budget(&mut scanner, &root, 1_000_000);
+
+        let namespace = format!("parent/{}/nested", Uuid::new_v4());
+        for object in ["parent", "parent/child", namespace.as_str()] {
+            write_test_object_metadata_bytes(&root, "bucket", object, &metadata_for_object_version("bucket", object, None)).await;
+        }
+        let payload = root.join("bucket/parent").join(Uuid::new_v4().to_string());
+        tokio::fs::create_dir_all(&payload)
+            .await
+            .expect("create parent payload directory");
+        tokio::fs::write(payload.join("part.1"), b"shard")
+            .await
+            .expect("create actual payload evidence");
+        if corrupt_parent {
+            tokio::fs::write(root.join("bucket/parent/xl.meta"), b"")
+                .await
+                .expect("corrupt only the parent metadata");
+        }
+        if cached_parent {
+            scanner.old_cache.replace("bucket", "", DataUsageEntry::default());
+            scanner.old_cache.replace(
+                "bucket/parent",
+                "bucket",
+                DataUsageEntry {
+                    objects: 1,
+                    size: 1,
+                    compacted: true,
+                    ..Default::default()
+                },
+            );
+            let interval = data_usage_update_dir_cycles();
+            scanner.old_cache.info.next_cycle = u64::from(
+                (0..interval)
+                    .find(|cycle| hash_path("bucket/parent").mod_(*cycle, interval))
+                    .expect("one normal sampling cycle revisits the old parent-only cache"),
+            );
+        }
+
+        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 object and child namespaces independently");
+        let total = scanner.new_cache.size_recursive("bucket").expect("retain bucket accounting");
+        assert_eq!(total.objects, if corrupt_parent { 2 } else { 3 });
+        assert_eq!(total.size, if corrupt_parent { 2 } else { 3 });
+        assert_eq!(total.failed_objects, usize::from(corrupt_parent));
+        assert_eq!(
+            scanner.new_cache.info.failed_objects.len(),
+            usize::from(corrupt_parent),
+            "payload evidence must not invent missing parent metadata"
+        );
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn erasure_object_prefix_children_keep_the_object_budget() {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 100, &mut scanner, root.clone());
+    scanner.old_cache.info.name = "bucket".into();
+    scanner.new_cache.info.name = "bucket".into();
+    scanner.is_erasure_mode = true;
+    bind_budget(&mut scanner, &root, 1_000_000);
+    for object in ["parent", "parent/child", "parent/other"] {
+        write_test_object_metadata(&root, "bucket", object).await;
+    }
+    let visits = metadata_visits::observe(&mut scanner);
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_objects: Some(2),
+            ..Default::default()
+        },
+    );
+    scanner.budget = budget.clone();
+    let result = scanner
+        .scan_folder(
+            budget.token(),
+            CachedFolder {
+                name: "bucket".into(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        )
+        .await;
+    assert!(result.is_err(), "parent metadata cannot hide children or bypass their budget");
+    assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects));
+    assert_eq!(budget.progress().0, 2);
+    assert_eq!(visits.load(Ordering::Relaxed), 2);
+}
+
 #[tokio::test]
 #[serial]
 async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
@@ -204,7 +315,10 @@ async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
                 ctx,
                 folder,
                 &mut DataUsageEntry::default(),
-                FolderScanMode::ErasureCandidate { reuse_namespace: false },
+                FolderScanMode::ErasureCandidate {
+                    reuse_namespace: false,
+                    parent_has_metadata: false,
+                },
                 false,
             )
             .await
@@ -271,7 +385,10 @@ async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await;
@@ -311,7 +428,10 @@ async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
             ctx.clone(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await;
@@ -382,7 +502,16 @@ async fn erasure_candidate_defers_metadata_until_part_classification_finishes()
     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 }, false)
+        .scan_folder_inner(
+            ctx,
+            folder,
+            &mut into,
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
+            false,
+        )
         .await
         .expect("classify candidate");
     assert_eq!(outcome, FolderScanOutcome::ErasureData);
@@ -594,7 +723,10 @@ async fn erasure_candidate_part_symlink_is_not_data_evidence() {
             CancellationToken::new(),
             folder,
             &mut DataUsageEntry::default(),
-            FolderScanMode::ErasureCandidate { reuse_namespace: false },
+            FolderScanMode::ErasureCandidate {
+                reuse_namespace: false,
+                parent_has_metadata: false,
+            },
             false,
         )
         .await