fix(heal): certify metadata health and marker repair receipts (#7932)

* fix(heal): certify metadata health and marker repair receipts

* fix(heal): match nil selectors to absent null metadata

* test(heal): satisfy clippy for nil selector proof

* test(heal): align normal-scan receipt with nil selector proof

A nil or omitted selector now matches only a nil or absent metadata
version, so a normal scan of the latest UUID version leaves
metadata_verified unset and issues no receipt.

Assert metadata health through the exact latest version id instead, and
pin the omitted-selector path to fail closed.

---------

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

crates/ecstore/src/set_disk/ops/heal.rs

@@ -1105,6 +1105,28 @@ impl SetDisks {
                             });
                         }
 
+                        let requested_nil =
+                            version_id.is_empty() || Uuid::parse_str(version_id).is_ok_and(|requested| requested.is_nil());
+                        let selected_nil = latest_meta.version_id.is_none_or(|selected| selected.is_nil());
+                        let selected_version_matches = if requested_nil {
+                            selected_nil
+                        } else {
+                            latest_meta
+                                .version_id
+                                .is_some_and(|selected| selected.to_string().eq_ignore_ascii_case(version_id))
+                        };
+                        result.metadata_verified = !opts.dry_run
+                            && !latest_meta.is_remote()
+                            && !read_repair_uses_shared_lock
+                            && selected_version_matches
+                            && (protected || latest_meta.deleted)
+                            && !result.after.drives.is_empty()
+                            && result
+                                .after
+                                .drives
+                                .iter()
+                                .all(|drive| drive.state == DriveState::Ok.to_string());
+
                         if !latest_meta.deleted && !latest_meta.is_remote() && !protected {
                             result.detail =
                                 "Legacy object uses standard repair; independent object identity remains unverified".to_owned();
@@ -1836,6 +1858,16 @@ impl SetDisks {
                                 .drives
                                 .iter()
                                 .all(|drive| drive.state == DriveState::Ok.to_string());
+                        result.metadata_repair_verified = latest_meta.deleted
+                            && !opts.dry_run
+                            && !latest_meta.is_remote()
+                            && !read_repair_uses_shared_lock
+                            && result.drives_healed().is_some_and(|healed| healed > 0)
+                            && result
+                                .after
+                                .drives
+                                .iter()
+                                .all(|drive| drive.state == DriveState::Ok.to_string());
 
                         // The object is healthy here; sweep any data dirs left behind
                         // by pre-#3510 unversioned overwrites, which the dangling paths
@@ -2800,6 +2832,8 @@ fn finalize_object_heal_result(
     if lock_lost {
         result.integrity_verified = false;
         result.repair_verified = false;
+        result.metadata_verified = false;
+        result.metadata_repair_verified = false;
         *absence = None;
         error = Some(Error::NamespaceLockQuorumUnavailable {
             mode: "write",
@@ -3282,6 +3316,8 @@ mod heal_result_report_tests {
         let result = rustfs_madmin::heal_commands::HealResultItem {
             integrity_verified: true,
             repair_verified: true,
+            metadata_verified: true,
+            metadata_repair_verified: true,
             ..Default::default()
         };
 
@@ -3289,6 +3325,8 @@ mod heal_result_report_tests {
 
         assert!(!result.integrity_verified);
         assert!(!result.repair_verified);
+        assert!(!result.metadata_verified);
+        assert!(!result.metadata_repair_verified);
         assert!(absence.is_none());
         assert!(matches!(
             error,
@@ -3302,6 +3340,20 @@ mod heal_result_report_tests {
         ));
     }
 
+    #[test]
+    fn absent_and_nil_selected_versions_are_the_same_null_identity() {
+        let requested_nil = Uuid::nil().to_string();
+        let requested = Uuid::parse_str(&requested_nil).expect("nil UUID string");
+        assert!(requested.is_nil());
+
+        let absent: Option<Uuid> = None;
+        let selected_nil = absent.is_none_or(|selected| selected.is_nil());
+        assert!(selected_nil, "absent metadata version selects the null identity");
+
+        let selected = Uuid::new_v4();
+        assert!(!selected.is_nil(), "a concrete UUID must not satisfy a requested null selector");
+    }
+
     #[test]
     fn metadata_less_part_file_accepts_positive_part_numbers_only() {
         assert!(super::metadata_less_part_file("part.1"));

crates/heal/src/heal/manager/scheduler.rs

@@ -811,6 +811,9 @@ pub(super) fn mrf_verified_repair_event_for_target(
     let disposition = match outcome.disposition {
         HealObjectDisposition::Repaired => MrfVerifiedRepairDisposition::Repaired,
         HealObjectDisposition::VerifiedHealthy => MrfVerifiedRepairDisposition::VerifiedHealthy,
+        HealObjectDisposition::MetadataHealthy if target.kind == MrfKind::MetadataCorruption => {
+            MrfVerifiedRepairDisposition::VerifiedHealthy
+        }
         HealObjectDisposition::AuthoritativelyAbsent => MrfVerifiedRepairDisposition::AuthoritativelyAbsent,
         _ => return None,
     };

crates/heal/src/heal/outcome.rs

@@ -76,6 +76,9 @@ pub enum HealObjectDisposition {
     Unknown,
     Repaired,
     VerifiedHealthy,
+    /// Authoritative metadata/presence proof only. This does not certify
+    /// payload integrity.
+    MetadataHealthy,
     AuthoritativelyAbsent,
     Deferred {
         reason: HealDeferredReason,
@@ -106,6 +109,7 @@ impl HealObjectReceipt {
             self.disposition,
             HealObjectDisposition::Repaired
                 | HealObjectDisposition::VerifiedHealthy
+                | HealObjectDisposition::MetadataHealthy
                 | HealObjectDisposition::AuthoritativelyAbsent
         ) && self.identity.kind == expected.kind
             && self.identity.bucket == expected.bucket
@@ -367,7 +371,9 @@ impl HealTaskOutcome {
         counters.overflowed |= !increment_counter(&mut counters.processed);
         let counter = match item.disposition {
             HealObjectDisposition::Repaired => &mut counters.healed,
-            HealObjectDisposition::VerifiedHealthy | HealObjectDisposition::AuthoritativelyAbsent => &mut counters.unchanged,
+            HealObjectDisposition::VerifiedHealthy
+            | HealObjectDisposition::MetadataHealthy
+            | HealObjectDisposition::AuthoritativelyAbsent => &mut counters.unchanged,
             HealObjectDisposition::Failed(_) => &mut counters.failed,
             HealObjectDisposition::Unknown => {
                 counters.overflowed |= !increment_counter(&mut counters.unknown);
@@ -531,6 +537,7 @@ mod canonical_outcome_tests {
             HealObjectDisposition::Unknown,
             HealObjectDisposition::Repaired,
             HealObjectDisposition::VerifiedHealthy,
+            HealObjectDisposition::MetadataHealthy,
             HealObjectDisposition::AuthoritativelyAbsent,
             HealObjectDisposition::Deferred {
                 reason: HealDeferredReason::DanglingDeleteGrace,
@@ -543,7 +550,7 @@ mod canonical_outcome_tests {
             outcome.record(item(disposition));
         }
         let c = &outcome.counters;
-        assert_eq!((c.processed, c.healed, c.unchanged, c.skipped, c.failed, c.unknown), (8, 1, 2, 4, 1, 1));
+        assert_eq!((c.processed, c.healed, c.unchanged, c.skipped, c.failed, c.unknown), (9, 1, 3, 4, 1, 1));
         assert_eq!(c.processed, c.healed + c.unchanged + c.skipped + c.failed);
     }
 

crates/heal/src/heal/storage.rs

@@ -109,7 +109,9 @@ fn verified_object_receipt(
     item: &HealResultItem,
     bucket_incarnation_id: Uuid,
 ) -> Option<HealObjectReceipt> {
-    if opts.dry_run || (!item.integrity_verified && !item.repair_verified) {
+    if opts.dry_run
+        || (!item.integrity_verified && !item.repair_verified && !item.metadata_verified && !item.metadata_repair_verified)
+    {
         return None;
     }
     let resolved_version = Uuid::from_bytes(item.resolved_version_id?);
@@ -120,7 +122,7 @@ fn verified_object_receipt(
     }
     item.drives_reported()?;
     let drives_healed = item.drives_healed()?;
-    if item.repair_verified && drives_healed == 0 {
+    if (item.repair_verified || item.metadata_repair_verified) && drives_healed == 0 {
         return None;
     }
     let ok_drive_state = DriveState::Ok.to_string();
@@ -142,10 +144,12 @@ fn verified_object_receipt(
             pool_index: opts.pool,
             set_index: opts.set,
         },
-        disposition: if drives_healed > 0 {
+        disposition: if drives_healed > 0 && (item.integrity_verified || item.repair_verified || item.metadata_repair_verified) {
             HealObjectDisposition::Repaired
-        } else if item.integrity_verified {
+        } else if drives_healed == 0 && item.integrity_verified {
             HealObjectDisposition::VerifiedHealthy
+        } else if drives_healed == 0 && item.metadata_verified {
+            HealObjectDisposition::MetadataHealthy
         } else {
             return None;
         },
@@ -752,7 +756,10 @@ impl ECStoreHealStorage {
             } else {
                 None
             }
-        } else if error.is_none() && !opts.dry_run && (item.integrity_verified || item.repair_verified) {
+        } else if error.is_none()
+            && !opts.dry_run
+            && (item.integrity_verified || item.repair_verified || item.metadata_verified || item.metadata_repair_verified)
+        {
             let bucket_incarnation_id = match expected {
                 Some(expected) => Some(expected),
                 None => self.ecstore.bucket_incarnation_id(bucket).await.ok(),
@@ -2082,6 +2089,51 @@ mod tests {
         assert!(verified_object_receipt("bucket", "object", None, &options, &item, incarnation).is_none());
     }
 
+    #[test]
+    fn metadata_health_and_marker_repair_have_distinct_receipts() {
+        use super::{HealObjectDisposition, HealOpts, HealResultItem, Uuid, verified_object_receipt};
+        use rustfs_madmin::heal_commands::HealDriveInfo;
+
+        let incarnation = Uuid::new_v4();
+        let version = Uuid::new_v4().to_string();
+        let options = HealOpts::default();
+        let healthy = HealDriveInfo {
+            state: "ok".to_string(),
+            ..Default::default()
+        };
+        let mut item = HealResultItem {
+            metadata_verified: true,
+            version_id: version.clone(),
+            resolved_version_id: Some(*Uuid::parse_str(&version).expect("version UUID").as_bytes()),
+            ..Default::default()
+        };
+        item.before.drives.push(healthy.clone());
+        item.after.drives.push(healthy);
+
+        let receipt = verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation)
+            .expect("metadata quorum should certify metadata health");
+        assert_eq!(receipt.disposition, HealObjectDisposition::MetadataHealthy);
+
+        item.integrity_verified = true;
+        let receipt = verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation)
+            .expect("stronger integrity proof should remain available");
+        assert_eq!(receipt.disposition, HealObjectDisposition::VerifiedHealthy);
+
+        item.integrity_verified = false;
+        item.metadata_verified = false;
+        item.metadata_repair_verified = true;
+        item.before.drives[0].state = "missing".to_string();
+        let receipt = verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation)
+            .expect("a committed metadata repair should certify the marker/version");
+        assert_eq!(receipt.disposition, HealObjectDisposition::Repaired);
+
+        item.before.drives[0].state = "ok".to_string();
+        assert!(
+            verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation).is_none(),
+            "repair proof without a repaired drive must fail closed"
+        );
+    }
+
     #[test]
     fn next_heal_listing_token_returns_none_for_complete_page() {
         assert_eq!(

crates/heal/tests/heal_b5_versioned_regression_test.rs

@@ -28,6 +28,7 @@ use http::HeaderMap;
 use rustfs_filemeta::{FileInfo, FileMeta};
 use rustfs_heal::heal::{
     manager::{HealConfig, HealManager},
+    outcome::HealObjectDisposition,
     storage::{
         ECStoreHealStorage, HealListItem, HealObjectOptions as ObjectOptions, HealPutObjReader as PutObjReader, HealStorageAPI,
     },
@@ -365,7 +366,12 @@ mod serial_tests {
                 assert!(healed.error.is_none(), "exact version repair failed: {:?}", healed.error);
                 if item.is_delete_marker {
                     assert!(!healed.item.integrity_verified);
-                    assert!(healed.receipt.is_none(), "delete markers carry no shard-integrity proof");
+                    let receipt = healed.receipt.expect("delete markers require an exact metadata receipt");
+                    assert_eq!(receipt.identity.version_id, item.version_id);
+                    assert!(matches!(
+                        receipt.disposition,
+                        HealObjectDisposition::Repaired | HealObjectDisposition::MetadataHealthy
+                    ));
                 } else {
                     assert!(healed.item.integrity_verified);
                     let receipt = healed
@@ -435,7 +441,7 @@ mod serial_tests {
             let mut latest_data = versioned_test_data(6);
             latest_data.extend_from_slice(b"new-uuid");
             let latest = put_versioned(&ecstore, &bucket, object, &latest_data).await;
-            versions.push((latest, latest_data.clone()));
+            versions.push((latest.clone(), latest_data.clone()));
 
             let target = disk_paths
                 .iter()
@@ -519,7 +525,17 @@ mod serial_tests {
                 .expect("an omitted selector must still heal latest");
             assert!(latest_result.error.is_none());
             assert_eq!(latest_result.item.object_size, latest_data.len());
-            assert!(latest_result.receipt.is_none(), "a normal scan cannot certify payload integrity");
+            assert!(
+                latest_result.receipt.is_none(),
+                "an omitted selector certifies metadata health only for the null identity"
+            );
+            let latest_receipt = storage
+                .heal_object_with_receipt(&bucket, object, Some(latest.as_str()), &HealOpts::default())
+                .await
+                .expect("a normal scan of the exact latest version")
+                .receipt
+                .expect("a normal scan should certify metadata health");
+            assert_eq!(latest_receipt.disposition, HealObjectDisposition::MetadataHealthy);
             let verified_latest = storage
                 .heal_object_with_receipt(
                     &bucket,
@@ -706,6 +722,46 @@ mod serial_tests {
         );
     }
 
+    /// A missing xl.meta must produce an exact marker repair receipt, while a
+    /// healthy replay must prove metadata health without claiming payload integrity.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial]
+    async fn test_delete_marker_receipt_repair_then_metadata_healthy() {
+        let (disk_paths, ecstore, heal_storage) = heal_env().await;
+        let bucket = "b5-dm-receipt";
+        let object = "marker.bin";
+        create_versioned_bucket(&ecstore, bucket).await;
+
+        let _data = put_versioned(&ecstore, bucket, object, &versioned_test_data(41)).await;
+        let marker = put_delete_marker(&ecstore, bucket, object).await;
+        let obj_dir = object_dir(&disk_paths[0], bucket, object);
+        tokio::fs::remove_file(xl_meta_path(&obj_dir))
+            .await
+            .expect("remove one xl.meta replica");
+
+        let repaired = heal_storage
+            .heal_object_with_receipt(bucket, object, Some(&marker), &recreate_heal_opts())
+            .await
+            .expect("marker repair request");
+        assert!(repaired.error.is_none(), "{:?}", repaired.error);
+        assert!(repaired.item.metadata_repair_verified);
+        let receipt = repaired.receipt.expect("committed marker repair receipt");
+        assert_eq!(receipt.disposition, HealObjectDisposition::Repaired);
+        assert_eq!(receipt.identity.version_id.as_deref(), Some(marker.as_str()));
+
+        let healthy = heal_storage
+            .heal_object_with_receipt(bucket, object, Some(&marker), &recreate_heal_opts())
+            .await
+            .expect("healthy marker replay");
+        assert!(healthy.error.is_none(), "{:?}", healthy.error);
+        assert_eq!(healthy.item.drives_healed(), Some(0));
+        assert!(healthy.item.metadata_verified);
+        assert!(!healthy.item.integrity_verified);
+        let receipt = healthy.receipt.expect("metadata health receipt");
+        assert_eq!(receipt.disposition, HealObjectDisposition::MetadataHealthy);
+        assert_eq!(receipt.identity.version_id.as_deref(), Some(marker.as_str()));
+    }
+
     /// Enumerated unversioned objects select the exact null slot once each.
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[serial]

crates/heal/tests/shard_identity_receipt_test.rs

@@ -200,7 +200,12 @@ async fn receipt_requires_independent_deep_verification() {
             )
             .await
             .expect("normal presence scan");
-        assert!(normal.receipt.is_none(), "a presence scan cannot certify payload integrity");
+        let normal_receipt = normal.receipt.expect("a normal presence scan should certify metadata health");
+        assert_eq!(
+            normal_receipt.disposition,
+            HealObjectDisposition::MetadataHealthy,
+            "a presence scan must not be promoted to payload VerifiedHealthy"
+        );
         let result = storage
             .heal_object_with_receipt(
                 bucket,

crates/madmin/src/heal_commands.rs

@@ -45,6 +45,15 @@ pub struct HealResultItem {
     /// healthy or unchanged object.
     #[serde(skip)]
     pub repair_verified: bool,
+    /// Storage-owner proof that the selected version/marker has authoritative
+    /// metadata from read quorum. This is weaker than payload integrity proof
+    /// and is intentionally in-process only.
+    #[serde(skip)]
+    pub metadata_verified: bool,
+    /// Storage-owner proof that metadata or a delete marker was committed by
+    /// this heal. This is intentionally in-process only.
+    #[serde(skip)]
+    pub metadata_repair_verified: bool,
     #[serde(rename = "resultId")]
     pub result_index: usize,
     #[serde(rename = "type")]
@@ -120,6 +129,10 @@ mod tests {
         assert!(wire.get("resolvedVersionId").is_none());
         assert!(wire.get("repair_verified").is_none());
         assert!(wire.get("repairVerified").is_none());
+        assert!(wire.get("metadata_verified").is_none());
+        assert!(wire.get("metadataVerified").is_none());
+        assert!(wire.get("metadata_repair_verified").is_none());
+        assert!(wire.get("metadataRepairVerified").is_none());
         wire["resolved_version_id"] = serde_json::json!(vec![0; 16]);
         let decoded: HealResultItem = serde_json::from_value(wire).expect("legacy wire shape should remain readable");
         assert_eq!(decoded.resolved_version_id, None, "wire input cannot supply owner proof");

docs/architecture/erasure-coding.md

@@ -216,10 +216,13 @@ Legacy objects retain their existing GET and traditional Heal behavior and
 therefore their residual complete-donor substitution risk. Ordinary shard repair
 and the existing explicit-version metadata recovery path remain available, but
 do not create commitments or certify object identity. Actual drive repairs are
-reported separately from strong integrity receipts. Normal
-presence scans do not issue strong integrity receipts even for protected
-objects. Only a completed exclusive Deep scan/repair with authenticated sources
-can do so. See the [upgrade contract](minio-file-format-compat.md#independent-integrity-upgrade-contract)
+reported separately from strong integrity receipts. Normal presence scans do
+not issue strong integrity receipts even for protected objects; an all-healthy
+protected version or delete marker may instead carry `MetadataHealthy`, which
+proves metadata quorum only. Legacy objects receive no positive receipt. Only a
+completed exclusive Deep scan/repair with authenticated sources
+can produce `VerifiedHealthy` or a payload-backed repair receipt. See the
+[upgrade contract](minio-file-format-compat.md#independent-integrity-upgrade-contract)
 for mixed-version and migration constraints and the
 [rollout runbook](../operations/shard-integrity-rollout.md) for activation and rollback.
 

docs/architecture/minio-file-format-compat.md

@@ -183,7 +183,10 @@ through upgraded coordinators. Recomputing a hash from existing suspect shards,
 or agreeing RS parity alone, does not establish their original identity.
 
 Normal scans, tier metadata scans and unproven legacy scans cannot produce `VerifiedHealthy` or
-`Repaired` integrity receipts. Legacy repair still runs and reports actual
+`Repaired` integrity receipts. A protected object or delete marker may produce
+`MetadataHealthy` only when its exact version was selected from authoritative
+metadata quorum; that disposition proves metadata/presence health and never
+payload integrity. Legacy objects remain without a positive receipt. Legacy repair still runs and reports actual
 before/after drive changes; an unknown strong result does not mean the repair
 was never attempted. Authoritative historical-version absence/cleanup proofs
 remain separate from live payload verification. Retain trusted backups for

docs/operations/shard-integrity-rollout.md

@@ -65,9 +65,12 @@ queue does not imply that these persistent obligations have been cleared.
 Partial-write MRF replay uses Deep verification so a protected object's repair
 can discharge its obligation after verifying the payload. This adds full-object
 read work to those background attempts, including healthy replay targets.
-Normal presence scans likewise cannot certify protected payloads. Local Heal
-of transitioned objects checks metadata without reading the tier payload, so it
-does not issue a payload-integrity receipt even if a descriptor remains. An
+Normal presence scans likewise cannot certify protected payloads; an exact,
+all-healthy protected version or delete marker may receive `MetadataHealthy`,
+which proves authoritative metadata rather than payload integrity. Legacy
+objects receive no positive receipt. Local Heal of transitioned objects
+checks metadata without reading the tier payload, so it does not issue a
+payload-integrity receipt even if a descriptor remains. An
 authoritative historical-version cleanup or absence proof has its own identity
 and commit checks and does not depend on a live payload digest.