diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 20416da6a..65ebfd598 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -59,6 +59,19 @@ pub(crate) struct OutcomeFinishTestHook { pub(crate) static OUTCOME_FINISH_TEST_HOOK: std::sync::LazyLock>>> = std::sync::LazyLock::new(|| tokio::sync::Mutex::new(None)); +#[cfg(test)] +static STORAGE_RECEIPT_TEST_HOOK: std::sync::LazyLock>>> = + 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(); diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index 1862bd152..2f7eadfc5 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/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 } } } diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index c67515c7d..989ad9f7a 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/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();