heal: latch object receipt owner before repair

Capture the expected bucket incarnation before invoking object repair so a post-repair owner change cannot rewrite the responsibility that a storage receipt is allowed to prove.

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

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

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

@@ -162,6 +162,10 @@ impl HealTask {
             pool: self.options.pool_index,
             set: self.options.set_index,
         };
+        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;
 
         let heal_fut = self.storage.heal_object_with_receipt(bucket, object, version_id, &heal_opts);
         let heal_result = if self.source == HealRequestSource::ReadRepair {
@@ -266,10 +270,6 @@ impl HealTask {
                     let mut progress = self.progress.write().await;
                     progress.update_object_progress(1, 1, 0, 0, object_size);
                 }
-                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

@@ -1117,6 +1117,7 @@ struct MockStorage {
     heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
     heal_object_receipts: Mutex<HashMap<String, VecDeque<HealObjectReceipt>>>,
     bucket_incarnation_id: Mutex<Option<Uuid>>,
+    bucket_incarnation_after_object_heal: Mutex<Option<Uuid>>,
     format_no_heal_required: Mutex<bool>,
     format_error: Mutex<Option<Error>>,
     global_format_calls: Mutex<u32>,
@@ -1273,6 +1274,39 @@ async fn object_heal_records_matching_positive_storage_receipt() {
     assert_eq!(object.disposition, HealObjectDisposition::Repaired);
 }
 
+#[tokio::test]
+async fn object_heal_latches_expected_incarnation_before_repair() {
+    let original_incarnation = Uuid::new_v4();
+    let successor_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::VerifiedHealthy,
+                original_incarnation,
+            )]),
+        )])),
+        bucket_incarnation_id: Mutex::new(Some(original_incarnation)),
+        bucket_incarnation_after_object_heal: Mutex::new(Some(successor_incarnation)),
+        ..Default::default()
+    });
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), Some("version-a".to_string())),
+        storage,
+    );
+
+    task.execute().await.expect("mock object heal should complete");
+
+    let outcome = task.get_outcome().await;
+    assert_eq!(outcome.counters.unchanged, 1);
+    assert_eq!(outcome.counters.unknown, 0);
+    let object = outcome.objects.front().expect("latched receipt should be recorded");
+    assert_eq!(object.identity.bucket_incarnation_id, Some(original_incarnation));
+    assert_eq!(object.disposition, HealObjectDisposition::VerifiedHealthy);
+}
+
 #[tokio::test]
 async fn object_heal_rejects_mismatched_or_legacy_storage_receipts() {
     let expected_incarnation = Uuid::new_v4();
@@ -1586,6 +1620,9 @@ impl HealStorageAPI for MockStorage {
             return Ok((HealResultItem::default(), Some(Error::Disk(DiskError::FileNotFound))));
         }
         self.healed_objects.lock().unwrap().push(object.to_string());
+        if let Some(bucket_incarnation_id) = self.bucket_incarnation_after_object_heal.lock().unwrap().take() {
+            *self.bucket_incarnation_id.lock().unwrap() = Some(bucket_incarnation_id);
+        }
         Ok((
             HealResultItem {
                 object_size: 1,