fix(scanner): discover sub-quorum heal candidates (#6384)

* fix(scanner): preserve unversioned heal retries

* fix(scanner): bound orphan heal discovery fallback

* fix(filemeta): fence unsafe heal key components

* fix(scanner): preserve exact overflow heal versions

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/scanner/src/scanner_folder/item_actions.rs

@@ -13,6 +13,8 @@
 // limitations under the License.
 /// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
 use super::*;
+#[cfg(test)]
+use rustfs_filemeta::MetadataResolutionParams;
 
 /// Cached folder information for scanning
 #[derive(Clone, Debug)]
@@ -89,6 +91,22 @@ pub(super) fn build_object_heal_request(
     }
 }
 
+/// Build the versionless inspection request used when discovery cannot prove
+/// a destructive version identity (for example an unversioned object or a
+/// bounded candidate overflow). The explicit flag is the fail-closed safety
+/// boundary; callers must not reconstruct it with the destructive default.
+pub(super) fn build_non_destructive_object_heal_request(
+    bucket: String,
+    object: String,
+    scan_mode: HealScanMode,
+    priority: HealChannelPriority,
+) -> HealChannelRequest {
+    let mut request = build_object_heal_request(bucket, object, None, scan_mode, priority);
+    request.remove_corrupted = Some(false);
+    request
+}
+
+#[cfg(test)]
 pub(super) fn resolve_object_heal_entry(
     entries: &MetaCacheEntries,
     resolver: MetadataResolutionParams,

crates/scanner/src/scanner_folder/ledger.rs

@@ -105,6 +105,29 @@ impl FolderScanner {
         }
     }
 
+    /// Preserve the discovery reason when a candidate could not be admitted
+    /// immediately. The existing string field is intentionally reused so the
+    /// scanner's map-encoded cache schema stays backward compatible.
+    pub(super) fn mark_pending_scanner_heal_reason(
+        &mut self,
+        kind: PendingScannerHealKind,
+        bucket: &str,
+        object: Option<&str>,
+        version_id: Option<&str>,
+        reason: &str,
+    ) {
+        if let Some(entry) = self
+            .new_cache
+            .info
+            .pending_heals
+            .iter_mut()
+            .find(|entry| pending_scanner_heal_matches(entry, kind, bucket, object, version_id))
+        {
+            entry.last_admission_reason = reason.to_string();
+            self.sync_pending_heals();
+        }
+    }
+
     pub(super) fn prune_pending_scanner_heals(&mut self) {
         let now = Self::now_secs();
         let before_expiry = self.new_cache.info.pending_heals.len();
@@ -305,13 +328,22 @@ pub(super) fn build_pending_scanner_heal_request(entry: &PendingScannerHeal) ->
     match entry.kind {
         PendingScannerHealKind::Bucket => Some(build_bucket_heal_request(entry.bucket.clone(), HealChannelPriority::High)),
         PendingScannerHealKind::Object => entry.object.as_ref().map(|object| {
-            build_object_heal_request(
-                entry.bucket.clone(),
-                object.clone(),
-                entry.version_id.clone(),
-                entry.scan_mode,
-                HealChannelPriority::High,
-            )
+            if entry.version_id.is_none() {
+                build_non_destructive_object_heal_request(
+                    entry.bucket.clone(),
+                    object.clone(),
+                    entry.scan_mode,
+                    HealChannelPriority::High,
+                )
+            } else {
+                build_object_heal_request(
+                    entry.bucket.clone(),
+                    object.clone(),
+                    entry.version_id.clone(),
+                    entry.scan_mode,
+                    HealChannelPriority::High,
+                )
+            }
         }),
     }
 }

crates/scanner/src/scanner_folder/tests.rs

@@ -17,7 +17,7 @@ use crate::SCANNER_SLEEPER;
 use super::*;
 use crate::storage_api::VersionPurgeStatusType;
 use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
-use rustfs_filemeta::{FileInfo, FileMeta};
+use rustfs_filemeta::{FileInfo, FileMeta, MetadataResolutionParams};
 use std::io::Write;
 #[cfg(unix)]
 use std::os::unix::fs::{PermissionsExt, symlink};
@@ -994,6 +994,21 @@ fn test_build_object_heal_request_omits_nil_version_id() {
     assert_eq!(request.recreate_missing, Some(false));
 }
 
+#[test]
+fn test_build_non_destructive_object_heal_request_disables_removal() {
+    let request = build_non_destructive_object_heal_request(
+        "bucket".to_string(),
+        "path/to/object".to_string(),
+        HealScanMode::Deep,
+        HealChannelPriority::High,
+    );
+
+    assert_eq!(request.object_version_id, None);
+    assert_eq!(request.remove_corrupted, Some(false));
+    assert_eq!(request.recreate_missing, Some(false));
+    assert_eq!(request.source, HealRequestSource::Scanner);
+}
+
 #[test]
 fn test_build_bucket_heal_request_disables_recreate_for_scanner() {
     let request = build_bucket_heal_request("bucket".to_string(), HealChannelPriority::Low);
@@ -1133,6 +1148,42 @@ fn test_pending_heal_reconstructs_object_request_with_version() {
     assert_eq!(request.source, HealRequestSource::Scanner);
 }
 
+#[test]
+fn test_pending_heal_reconstructs_unversioned_request_without_removal() {
+    let pending = pending_heal(PendingScannerHealKind::Object, "bucket", Some("object"), None, 1, 1);
+
+    let request = build_pending_scanner_heal_request(&pending).expect("unversioned object request should rebuild");
+
+    assert!(request.object_version_id.is_none());
+    assert_eq!(request.remove_corrupted, Some(false));
+    assert_eq!(request.recreate_missing, Some(false));
+}
+
+#[tokio::test]
+async fn test_pending_heal_reason_preserves_sub_quorum_discovery() {
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
+
+    scanner.update_pending_scanner_heal_after_admission(
+        PendingScannerHealKind::Object,
+        "bucket",
+        Some("object"),
+        Some("version-a"),
+        HealScanMode::Deep,
+        HealAdmissionResult::Full,
+    );
+    scanner.mark_pending_scanner_heal_reason(
+        PendingScannerHealKind::Object,
+        "bucket",
+        Some("object"),
+        Some("version-a"),
+        "sub_quorum_metadata",
+    );
+
+    assert_eq!(scanner.new_cache.info.pending_heals.len(), 1);
+    assert_eq!(scanner.new_cache.info.pending_heals[0].last_admission_reason, "sub_quorum_metadata");
+}
+
 #[test]
 fn test_pending_heal_retry_candidates_respect_cap_and_order() {
     let pending: Vec<PendingScannerHeal> = (0..(MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET + 2))
@@ -1335,6 +1386,20 @@ fn metadata_for_object(bucket: &str, object: &str) -> Vec<u8> {
     meta.marshal_msg().expect("test metadata should marshal")
 }
 
+fn metadata_for_object_version(bucket: &str, object: &str, version_id: Option<Uuid>) -> Vec<u8> {
+    let mut file_info = FileInfo::new(object, 4, 2);
+    file_info.volume = bucket.to_string();
+    file_info.name = object.to_string();
+    file_info.version_id = version_id;
+    file_info.versioned = version_id.is_some();
+    file_info.mod_time = Some(OffsetDateTime::now_utc());
+    file_info.size = 1;
+
+    let mut meta = FileMeta::new();
+    meta.add_version(file_info).expect("test metadata version should be accepted");
+    meta.marshal_msg().expect("test metadata should marshal")
+}
+
 async fn write_test_object_metadata(root: &std::path::Path, bucket: &str, object: &str) {
     write_test_object_metadata_bytes(root, bucket, object, &metadata_for_object(bucket, object)).await;
 }
@@ -1738,12 +1803,21 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
     let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
     let heal_starts = Arc::new(AtomicUsize::new(0));
     let heal_starts_clone = heal_starts.clone();
+    let healed_versions = Arc::new(Mutex::new(Vec::<Option<String>>::new()));
+    let healed_versions_clone = healed_versions.clone();
     let mut heal_rx =
         rustfs_common::heal_channel::init_heal_channel().expect("heal channel should initialize once for scanner tests");
     let _heal_responder = tokio::spawn(async move {
         while let Some(command) = heal_rx.recv().await {
-            if let rustfs_common::heal_channel::HealChannelCommand::Start { response_tx, .. } = command {
+            if let rustfs_common::heal_channel::HealChannelCommand::Start {
+                request, response_tx, ..
+            } = command
+            {
                 heal_starts_clone.fetch_add(1, Ordering::Relaxed);
+                healed_versions_clone
+                    .lock()
+                    .expect("heal version capture lock should not be poisoned")
+                    .push(request.object_version_id);
                 let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
             }
         }
@@ -1751,13 +1825,18 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
 
     let bucket = "src-archive";
     let object = "snapshots/37b3f20d941e2f5e6d99114d9bb2f3e67a8a2e5c9c4c5a1b0d6e7f8091a2b3c4";
-    let metadata = metadata_for_object(bucket, object);
-    write_test_object_metadata_bytes(&temp_dir, bucket, object, &metadata).await;
+    let orphan_version = Uuid::from_u128(0x1934);
+    let shared_version = Uuid::from_u128(0x1935);
+    let orphan_metadata = metadata_for_object_version(bucket, object, Some(orphan_version));
+    let shared_metadata = metadata_for_object_version(bucket, object, Some(shared_version));
+    write_test_object_metadata_bytes(&temp_dir, bucket, object, &orphan_metadata).await;
+    let mut expected_metadata = vec![(temp_dir.join(bucket).join(object).join("xl.meta"), orphan_metadata.clone())];
 
     let mut disks = vec![scanner.local_disk.clone()];
     for disk_name in ["disk2", "disk3", "disk4"] {
         let disk_root = temp_dir.join(disk_name);
-        write_test_object_metadata_bytes(&disk_root, bucket, object, &metadata).await;
+        write_test_object_metadata_bytes(&disk_root, bucket, object, &shared_metadata).await;
+        expected_metadata.push((disk_root.join(bucket).join(object).join("xl.meta"), shared_metadata.clone()));
         let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("failed to create extra disk endpoint");
         let disk = new_disk(
             &endpoint,
@@ -1806,8 +1885,29 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
         .new_cache
         .checked_flatten(bucket)
         .expect("healed cache must contain canonical child links");
-    assert_eq!(root.objects, 1);
+    // The fixture intentionally exposes two divergent version histories, so
+    // the scanner keeps both logical versions visible while discovering heals.
+    assert_eq!(root.objects, 2);
     assert!(heal_starts.load(Ordering::Relaxed) > 0, "test must execute the heal child-link path");
+    let orphan_version_text = orphan_version.to_string();
+    assert!(
+        healed_versions
+            .lock()
+            .expect("heal version capture lock should not be poisoned")
+            .iter()
+            .any(|version| version.as_deref() == Some(orphan_version_text.as_str())),
+        "sub-quorum orphan version must be submitted as an exact heal candidate"
+    );
+    for (path, expected) in expected_metadata {
+        assert_eq!(
+            tokio::fs::read(&path)
+                .await
+                .expect("scanner discovery must not delete metadata"),
+            expected,
+            "scanner discovery must not modify candidate metadata: {}",
+            path.display()
+        );
+    }
 }
 
 #[tokio::test]