fix(heal): retain resolved object identity on terminal failures

crates/heal/src/heal/task.rs

@@ -37,7 +37,7 @@ use std::{
     collections::VecDeque,
     future::Future,
     sync::{
-        Arc,
+        Arc, OnceLock,
         atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
     },
     time::{Duration, Instant, SystemTime},
@@ -449,6 +449,9 @@ pub struct HealTask {
     replacement_start_retry_count: Arc<AtomicU32>,
     /// Set when the locked storage read classified the single target as absent.
     object_absent: Arc<AtomicBool>,
+    /// The first resolved incarnation belongs to this task even if storage
+    /// fails or the bucket is recreated before its terminal outcome is recorded.
+    single_object_incarnation: Arc<OnceLock<Option<Uuid>>>,
     /// Task status
     pub status: Arc<RwLock<HealTaskStatus>>,
     /// Progress tracking
@@ -521,6 +524,7 @@ impl HealTask {
             replacement_running: Arc::new(AtomicBool::new(false)),
             replacement_start_retry_count: Arc::new(AtomicU32::new(0)),
             object_absent: Arc::new(AtomicBool::new(false)),
+            single_object_incarnation: Arc::new(OnceLock::new()),
             status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
             progress: Arc::new(RwLock::new(HealProgress::new())),
             result_items: Arc::new(RwLock::new(VecDeque::with_capacity(MAX_RETAINED_HEAL_RESULT_ITEMS))),
@@ -682,7 +686,9 @@ impl HealTask {
             } => (bucket, object, Some(version_id.as_str())),
             _ => return None,
         };
-        Some(self.outcome_identity(bucket, object, version, self.options.pool_index, self.options.set_index))
+        let mut identity = self.outcome_identity(bucket, object, version, self.options.pool_index, self.options.set_index);
+        identity.bucket_incarnation_id = self.single_object_incarnation.get().copied().flatten();
+        Some(identity)
     }
 
     pub(super) async fn record_verified_storage_receipt(

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

@@ -220,6 +220,11 @@ impl HealTask {
         let mut expected_identity =
             self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
         expected_identity.bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, self.options.dry_run).await?;
+        if matches!(self.heal_type, HealType::Object { .. }) {
+            expected_identity.bucket_incarnation_id = *self
+                .single_object_incarnation
+                .get_or_init(|| expected_identity.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 {

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

@@ -1941,6 +1941,127 @@ async fn failed_object_heal_rejects_matching_positive_storage_receipt() {
                 | HealObjectDisposition::AuthoritativelyAbsent
         )
     }));
+    assert_eq!(outcome.objects.len(), 1);
+    assert_eq!(outcome.objects[0].identity.bucket_incarnation_id, Some(incarnation));
+}
+
+#[tokio::test]
+async fn object_failure_retains_incarnation_for_inner_and_outer_storage_errors() {
+    for failure in [
+        MockHealObjectOutcome::PermissionDenied,
+        MockHealObjectOutcome::RetryableReadQuorum,
+        MockHealObjectOutcome::RetryableLockTimeout,
+        MockHealObjectOutcome::ErrOther("backend failed"),
+    ] {
+        let incarnation = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            bucket_incarnation_id: Mutex::new(Some(incarnation)),
+            heal_object_outcome: Mutex::new(Some(failure)),
+            ..Default::default()
+        });
+        let mut request = HealRequest::object("bucket-a".to_owned(), "object-a".to_owned(), Some("version-a".to_owned()));
+        request.options.pool_index = Some(2);
+        request.options.set_index = Some(3);
+        let task = HealTask::from_request(request, storage);
+        task.execute()
+            .await
+            .expect_err("the failed storage operation must remain failed");
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.objects.len(), 1);
+        assert_eq!(
+            (outcome.counters.processed, outcome.counters.failed, outcome.counters.attempt_failures),
+            (1, 1, 1)
+        );
+        assert_eq!((outcome.counters.healed, outcome.counters.unchanged, outcome.counters.unknown), (0, 0, 0));
+        let object = &outcome.objects[0];
+        assert_eq!(object.identity.bucket_incarnation_id, Some(incarnation));
+        assert_eq!(object.identity.bucket, "bucket-a");
+        assert_eq!(object.identity.object, "object-a");
+        assert_eq!(object.identity.version_id.as_deref(), Some("version-a"));
+        assert_eq!((object.identity.pool_index, object.identity.set_index), (Some(2), Some(3)));
+        assert!(matches!(object.disposition, HealObjectDisposition::Failed(_)));
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn object_abort_keeps_pre_storage_incarnation_after_bucket_recreation() {
+    for cancel in [false, true] {
+        let original = Uuid::new_v4();
+        let storage = Arc::new(MockStorage {
+            bucket_incarnation_id: Mutex::new(Some(original)),
+            block_heal_object: Mutex::new(true),
+            ..Default::default()
+        });
+        let mut request = HealRequest::object("bucket-a".to_owned(), "object-a".to_owned(), None);
+        request.options.timeout = Some(Duration::from_secs(10));
+        let task = HealTask::from_request(request, storage.clone());
+        let execution = task.execute();
+        tokio::pin!(execution);
+        assert!(futures::poll!(&mut execution).is_pending());
+        assert_eq!(storage.heal_object_calls.lock().expect("storage boundary").as_slice(), ["object-a"]);
+        *storage.bucket_incarnation_id.lock().expect("recreated bucket") = Some(Uuid::new_v4());
+        if cancel {
+            task.cancel().await.expect("cancel blocked storage");
+        } else {
+            tokio::time::advance(Duration::from_secs(11)).await;
+        }
+        let result = execution.await;
+        assert!(matches!(
+            (&result, cancel),
+            (Err(Error::TaskCancelled), true) | (Err(Error::TaskTimeout), false)
+        ));
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.objects.len(), 1);
+        assert_eq!(outcome.objects[0].identity.bucket_incarnation_id, Some(original));
+        assert_eq!((outcome.counters.processed, outcome.counters.skipped), (1, 1));
+        assert_eq!((outcome.counters.failed, outcome.counters.healed, outcome.counters.unknown), (0, 0, 0));
+        assert_eq!(
+            outcome.objects[0].disposition,
+            if cancel {
+                HealObjectDisposition::Cancelled
+            } else {
+                HealObjectDisposition::Deferred {
+                    reason: HealDeferredReason::Deadline,
+                    retry_not_before: None,
+                }
+            }
+        );
+    }
+}
+
+#[tokio::test]
+async fn object_unverified_outcome_keeps_original_incarnation_without_claiming_repair() {
+    let original = Uuid::new_v4();
+    let successor = Uuid::new_v4();
+    let storage = Arc::new(MockStorage {
+        bucket_incarnation_id: Mutex::new(Some(original)),
+        bucket_incarnation_after_object_heal: Mutex::new(Some(successor)),
+        ..Default::default()
+    });
+    let task = HealTask::from_request(HealRequest::object("bucket-a".to_owned(), "object-a".to_owned(), None), storage.clone());
+    task.execute().await.expect("legacy storage keeps its unverified result");
+    let outcome = task.get_outcome().await;
+    assert_eq!(outcome.objects[0].identity.bucket_incarnation_id, Some(original));
+    assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Unknown);
+    assert_eq!((outcome.counters.healed, outcome.counters.unknown), (0, 1));
+    assert_eq!(*storage.bucket_incarnation_id.lock().expect("successor"), Some(successor));
+}
+
+#[tokio::test]
+async fn object_dry_run_does_not_adopt_bucket_incarnation() {
+    let storage = Arc::new(MockStorage {
+        bucket_incarnation_id: Mutex::new(Some(Uuid::new_v4())),
+        ..Default::default()
+    });
+    let mut request = HealRequest::object("bucket-a".to_owned(), "object-a".to_owned(), None);
+    request.options.dry_run = true;
+    let task = HealTask::from_request(request, storage);
+    task.execute().await.expect("dry-run observation");
+    let outcome = task.get_outcome().await;
+    assert_eq!(outcome.objects.len(), 1);
+    assert_eq!(outcome.objects[0].identity.bucket_incarnation_id, None);
+    assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::DryRunObserved);
+    assert_eq!(outcome.counters.healed, 0);
 }
 
 #[tokio::test]