fix(heal): retain cancellation after successful bucket storage I/O

crates/heal/src/heal/task.rs

@@ -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();

crates/heal/src/heal/task/heal_bucket.rs

@@ -796,21 +796,27 @@ 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, storage_result.receipt).await;
-                                        if !recorded_authoritative_outcome
-                                            && !heal_opts.dry_run
-                                            && !storage_result.item.detail.is_empty()
-                                        {
-                                            detail = Some(storage_result.item.detail.clone());
+                                        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),
+                                            );
+                                            if !recorded_authoritative_outcome
+                                                && !heal_opts.dry_run
+                                                && !storage_result.item.detail.is_empty()
+                                            {
+                                                detail = Some(storage_result.item.detail.clone());
+                                            }
+                                            self.record_result_item(storage_result.item).await;
+                                            None
                                         }
-                                        self.record_result_item(storage_result.item).await;
-                                        None
                                     }
                                 }
                             }

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

@@ -304,6 +304,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();