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},
@@ -59,6 +59,19 @@ pub(crate) struct OutcomeFinishTestHook {
 pub(crate) static OUTCOME_FINISH_TEST_HOOK: std::sync::LazyLock<tokio::sync::Mutex<Option<Arc<OutcomeFinishTestHook>>>> =
     std::sync::LazyLock::new(|| tokio::sync::Mutex::new(None));
 
+#[cfg(test)]
+static STORAGE_RECEIPT_TEST_HOOK: std::sync::LazyLock<tokio::sync::Mutex<Option<Arc<OutcomeFinishTestHook>>>> =
+    std::sync::LazyLock::new(|| tokio::sync::Mutex::new(None));
+
+#[cfg(test)]
+async fn pause_storage_receipt(task_id: &str) {
+    let hook = STORAGE_RECEIPT_TEST_HOOK.lock().await.clone();
+    if let Some(hook) = hook.filter(|hook| hook.task_id == task_id) {
+        hook.reached.notify_one();
+        hook.release.notified().await;
+    }
+}
+
 #[cfg(test)]
 async fn pause_outcome_finish(task_id: &str) {
     let hook = OUTCOME_FINISH_TEST_HOOK.lock().await.clone();
@@ -436,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
@@ -508,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))),
@@ -669,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_bucket.rs

@@ -796,16 +796,22 @@ impl HealTask {
                                 match unavailable_recreate_error(&storage_result.item, &heal_opts) {
                                     Some(error) => Some(error),
                                     None => {
-                                        telemetry_unknown |= !increment_counter(&mut healed);
-                                        telemetry_unknown |= !add_bytes(
-                                            &mut bytes,
-                                            u64::try_from(storage_result.item.object_size).unwrap_or(u64::MAX),
-                                        );
+                                        #[cfg(test)]
+                                        super::pause_storage_receipt(&self.id).await;
                                         recorded_authoritative_outcome = self
                                             .record_verified_storage_receipt(identity.clone(), storage_result.receipt)
                                             .await;
-                                        self.record_result_item(storage_result.item).await;
-                                        None
+                                        if !recorded_authoritative_outcome && self.cancel_token.is_cancelled() {
+                                            Some(Error::TaskCancelled)
+                                        } else {
+                                            telemetry_unknown |= !increment_counter(&mut healed);
+                                            telemetry_unknown |= !add_bytes(
+                                                &mut bytes,
+                                                u64::try_from(storage_result.item.object_size).unwrap_or(u64::MAX),
+                                            );
+                                            self.record_result_item(storage_result.item).await;
+                                            None
+                                        }
                                     }
                                 }
                             }

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

@@ -278,6 +278,53 @@ mod canonical_outcome {
         assert_eq!(storage.healed_objects.lock().expect("healed objects").len(), 2);
     }
 
+    #[tokio::test]
+    async fn bucket_cancellation_after_storage_success_keeps_typed_cancelled_responsibility() {
+        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", None, HealObjectDisposition::Repaired, incarnation)]),
+            )])),
+            bucket_incarnation_id: Mutex::new(Some(incarnation)),
+            ..Default::default()
+        });
+        let task = Arc::new(bucket_task(storage.clone()));
+        let hook = Arc::new(OutcomeFinishTestHook {
+            task_id: task.id.clone(),
+            reached: tokio::sync::Notify::new(),
+            release: tokio::sync::Notify::new(),
+        });
+        *STORAGE_RECEIPT_TEST_HOOK.lock().await = Some(hook.clone());
+        let runner = {
+            let task = task.clone();
+            tokio::spawn(async move { task.execute().await })
+        };
+        tokio::time::timeout(Duration::from_secs(10), hook.reached.notified())
+            .await
+            .expect("storage success must reach the receipt boundary");
+        assert_eq!(storage.healed_objects.lock().expect("storage mutations").as_slice(), ["object-a"]);
+        task.cancel().await.expect("cancel before receipt recording");
+        hook.release.notify_one();
+        let result = tokio::time::timeout(Duration::from_secs(10), runner)
+            .await
+            .expect("cancelled worker must stop")
+            .expect("worker must not panic");
+        STORAGE_RECEIPT_TEST_HOOK.lock().await.take();
+        assert!(matches!(result, Err(Error::TaskCancelled)));
+        let outcome = task.get_outcome().await;
+        assert_eq!(outcome.execution, HealExecutionOutcome::Aborted(HealAbortReason::Cancelled));
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+        assert_eq!(outcome.counters.processed, 1);
+        assert_eq!(outcome.counters.healed, 0);
+        assert_eq!(outcome.counters.unchanged, 0);
+        assert_eq!(outcome.counters.unknown, 0);
+        assert_eq!(outcome.counters.skipped, 1);
+        assert_eq!(outcome.objects.len(), 1);
+        assert_eq!(outcome.objects[0].identity.object, "object-a");
+        assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Cancelled);
+    }
+
     #[tokio::test]
     async fn bucket_heal_rejects_stale_receipts_without_double_recording() {
         let expected_incarnation = Uuid::new_v4();
@@ -1866,6 +1913,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]