heal: require exact owner for object receipts

Require object heal receipts to match the expected bucket incarnation before they can be recorded as positive repair evidence. This prevents stale or cross-incarnation receipts from clearing the wrong heal responsibility.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/heal/src/heal/outcome.rs

@@ -110,7 +110,8 @@ impl HealObjectReceipt {
             && self.identity.version_id == expected.version_id
             && self.identity.pool_index == expected.pool_index
             && self.identity.set_index == expected.set_index
-            && self.identity.bucket_incarnation_id.is_some()
+            && self.identity.bucket_incarnation_id == expected.bucket_incarnation_id
+            && expected.bucket_incarnation_id.is_some()
     }
 }
 
@@ -496,21 +497,31 @@ mod canonical_outcome_tests {
 
     #[test]
     fn positive_receipt_requires_exact_identity_and_bucket_incarnation() {
-        let expected = item(HealObjectDisposition::Unknown).identity;
+        let incarnation = Uuid::new_v4();
+        let expected = HealObjectIdentity {
+            bucket_incarnation_id: Some(incarnation),
+            ..item(HealObjectDisposition::Unknown).identity
+        };
         let mut receipt = HealObjectReceipt {
             identity: expected.clone(),
             disposition: HealObjectDisposition::Repaired,
         };
 
+        receipt.identity.bucket_incarnation_id = None;
         assert!(
             !receipt.verified_for(&expected),
             "a positive storage receipt without bucket incarnation must remain untrusted"
         );
 
-        let incarnation = Uuid::new_v4();
         receipt.identity.bucket_incarnation_id = Some(incarnation);
         assert!(receipt.verified_for(&expected));
 
+        receipt.identity.bucket_incarnation_id = Some(Uuid::new_v4());
+        assert!(
+            !receipt.verified_for(&expected),
+            "a storage receipt for a different bucket incarnation must not clear the requested responsibility"
+        );
+
         receipt.identity.version_id = Some("older-version".to_string());
         assert!(
             !receipt.verified_for(&expected),

crates/heal/src/heal/storage.rs

@@ -20,6 +20,7 @@ use rustfs_madmin::heal_commands::HealResultItem;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use tracing::{debug, error, warn};
+use uuid::Uuid;
 
 use super::outcome::{HealObjectDisposition, HealObjectIdentity, HealObjectKind, HealObjectReceipt};
 use super::progress::stable_generation;
@@ -383,6 +384,11 @@ pub trait HealStorageAPI: Send + Sync {
     /// Check object exists
     async fn object_exists(&self, bucket: &str, object: &str) -> Result<bool>;
 
+    /// Stable bucket incarnation observed before an object heal starts.
+    async fn bucket_incarnation_id(&self, _bucket: &str) -> Result<Option<Uuid>> {
+        Ok(None)
+    }
+
     /// Heal object using ecstore
     async fn heal_object(
         &self,
@@ -1028,6 +1034,14 @@ impl HealStorageAPI for ECStoreHealStorage {
         }
     }
 
+    async fn bucket_incarnation_id(&self, bucket: &str) -> Result<Option<Uuid>> {
+        self.ecstore
+            .bucket_incarnation_id(bucket)
+            .await
+            .map(Some)
+            .map_err(Error::Storage)
+    }
+
     async fn heal_object(
         &self,
         bucket: &str,

crates/heal/src/heal/task/heal_object.rs

@@ -266,8 +266,10 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_object_progress(1, 1, 0, 0, object_size);
                 }
-                let expected_identity =
+                let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?;
+                let mut expected_identity =
                     self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
+                expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id;
                 self.record_verified_storage_receipt(expected_identity, storage_result.receipt)
                     .await;
                 self.record_result_item(result).await;

crates/heal/src/heal/task/tests.rs

@@ -1116,6 +1116,7 @@ struct MockStorage {
     heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
     heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
     heal_object_receipts: Mutex<HashMap<String, VecDeque<HealObjectReceipt>>>,
+    bucket_incarnation_id: Mutex<Option<Uuid>>,
     format_no_heal_required: Mutex<bool>,
     format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
@@ -1219,14 +1220,19 @@ async fn execute_emits_heal_trace_task_state() {
     assert_eq!(trace_attr_string(&completed, "state").as_deref(), Some("completed"));
 }
 
-fn object_receipt(object: &str, version_id: Option<&str>, disposition: HealObjectDisposition) -> HealObjectReceipt {
+fn object_receipt(
+    object: &str,
+    version_id: Option<&str>,
+    disposition: HealObjectDisposition,
+    bucket_incarnation_id: Uuid,
+) -> HealObjectReceipt {
     HealObjectReceipt {
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: "bucket-a".to_string(),
             object: object.to_string(),
             version_id: version_id.map(ToOwned::to_owned),
-            bucket_incarnation_id: Some(Uuid::new_v4()),
+            bucket_incarnation_id: Some(bucket_incarnation_id),
             pool_index: None,
             set_index: None,
         },
@@ -1236,11 +1242,18 @@ fn object_receipt(object: &str, version_id: Option<&str>, disposition: HealObjec
 
 #[tokio::test]
 async fn object_heal_records_matching_positive_storage_receipt() {
+    let incarnation = Uuid::new_v4();
     let storage = Arc::new(MockStorage {
         heal_object_receipts: Mutex::new(HashMap::from([(
             "object-a".to_string(),
-            VecDeque::from([object_receipt("object-a", Some("version-a"), HealObjectDisposition::Repaired)]),
+            VecDeque::from([object_receipt(
+                "object-a",
+                Some("version-a"),
+                HealObjectDisposition::Repaired,
+                incarnation,
+            )]),
         )])),
+        bucket_incarnation_id: Mutex::new(Some(incarnation)),
         ..Default::default()
     });
     let task = HealTask::from_request(
@@ -1262,15 +1275,18 @@ async fn object_heal_records_matching_positive_storage_receipt() {
 
 #[tokio::test]
 async fn object_heal_rejects_mismatched_or_legacy_storage_receipts() {
+    let expected_incarnation = Uuid::new_v4();
     let storage = Arc::new(MockStorage {
         heal_object_receipts: Mutex::new(HashMap::from([(
             "object-a".to_string(),
             VecDeque::from([object_receipt(
                 "object-a",
-                Some("old-version"),
+                Some("version-a"),
                 HealObjectDisposition::Repaired,
+                Uuid::new_v4(),
             )]),
         )])),
+        bucket_incarnation_id: Mutex::new(Some(expected_incarnation)),
         ..Default::default()
     });
     let task = HealTask::from_request(
@@ -1456,6 +1472,10 @@ impl HealStorageAPI for MockStorage {
         Ok(self.object_exists.lock().unwrap().unwrap_or(true))
     }
 
+    async fn bucket_incarnation_id(&self, _bucket: &str) -> Result<Option<Uuid>> {
+        Ok(*self.bucket_incarnation_id.lock().unwrap())
+    }
+
     async fn heal_object(
         &self,
         bucket: &str,