From 82030c83fc7cc51353fbb0212ed1ba87040882f5 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 6 Sep 2026 03:48:07 +0800 Subject: [PATCH] fix(heal): preserve control errors at the last object Keep task timeout and cancellation typed instead of counting an interrupted object as a permanent batch failure. Exercise the real scheduler and executor with retained progress, plus explicit retry budget and request identity. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/manager/tests.rs | 153 ++++++++++++++++++++++- crates/heal/src/heal/task/heal_bucket.rs | 4 + 2 files changed, 154 insertions(+), 3 deletions(-) diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index aa1c1293f..69e546a39 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -455,10 +455,15 @@ impl HealStorageAPI for MockStorage { async fn heal_object( &self, bucket: &str, - _object: &str, + object: &str, _version_id: Option<&str>, _opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { + if bucket.starts_with("heal-start-retry-deadline-object-") && object == "blocked" { + let hook = COMPLETED_RETENTION_HOOKS.lock().await[bucket].clone(); + hook.started.notify_one(); + std::future::pending::<()>().await; + } if bucket == "completed-retention-failed" { return Err(Error::TaskExecutionFailed { message: "retention fixture failure".to_string(), @@ -520,11 +525,35 @@ impl HealStorageAPI for MockStorage { async fn list_objects_for_heal_page( &self, - _bucket: &str, + bucket: &str, _prefix: &str, - _continuation_token: Option<&str>, + continuation_token: Option<&str>, _include_lifecycle_object_info: bool, ) -> Result<(Vec, Option, bool)> { + if bucket.starts_with("heal-start-retry-deadline-") { + if continuation_token.is_some() { + let hook = COMPLETED_RETENTION_HOOKS.lock().await[bucket].clone(); + hook.started.notify_one(); + std::future::pending::<()>().await; + } + let listing_timeout = bucket.starts_with("heal-start-retry-deadline-listing-"); + let names = if listing_timeout { + vec!["completed"] + } else { + vec!["completed", "blocked"] + }; + let objects = names + .into_iter() + .map(|name| crate::heal::storage::HealListItem { + name: name.to_string(), + version_id: None, + mod_time_unix_nanos: None, + lifecycle_object_info: None, + is_delete_marker: false, + }) + .collect(); + return Ok((objects, listing_timeout.then(|| "next".to_string()), listing_timeout)); + } Ok((Vec::new(), None, false)) } @@ -547,6 +576,124 @@ impl HealStorageAPI for MockStorage { } } +async fn assert_heal_start_retry_control_preserves_real_executor_progress(cancel: bool) { + for phase in ["listing", "object"] { + let bucket = format!("heal-start-retry-deadline-{phase}-{cancel}"); + let manager = HealManager::new(Arc::new(MockStorage), None); + let mut request = HealRequest::new( + HealType::Prefix { + bucket: bucket.clone(), + prefix: String::new(), + }, + HealOptions { + timeout: Some(if cancel { + Duration::from_secs(60) + } else { + Duration::from_millis(200) + }), + ..Default::default() + }, + HealPriority::High, + ); + request.source = HealRequestSource::Admin; + let task_id = request.id.clone(); + let hook = Arc::new(CompletedRetentionHook::default()); + { + let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await; + hooks.insert(bucket.clone(), Arc::clone(&hook)); + hooks.insert(task_id.clone(), Arc::clone(&hook)); + } + manager.submit_heal_request(request).await.expect("admit deadline task"); + process_manager_queue_once(&manager).await; + tokio::time::timeout(Duration::from_secs(5), hook.started.notified()) + .await + .expect("executor reaches blocked storage"); + let active = manager.get_task_report(&task_id).await.expect("active report"); + assert_eq!(active.progress.expect("real completed object progress").objects_healed, 1); + if cancel { + manager.active_heals.lock().await[&task_id].cancel_token.cancel(); + } + tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified()) + .await + .expect("deadline archives task"); + let report = manager.get_task_report(&task_id).await.expect("terminal report"); + assert_eq!( + report.status, + if cancel { + HealTaskStatus::Cancelled + } else { + HealTaskStatus::Timeout + }, + "blocked {phase}" + ); + let progress = report.progress.expect("terminal progress retained"); + assert_eq!(progress.objects_healed, 1); + assert_eq!(progress.objects_failed, 0, "interrupted object has no terminal storage result"); + assert_eq!(report.result_items.len(), 1, "completed result retained"); + assert!(!manager.active_heals.lock().await.contains_key(&task_id)); + assert!(!manager.retrying_heals.lock().await.contains_key(&task_id)); + assert!(!manager.heal_queue.lock().await.contains_request_id(&task_id)); + hook.finish.notify_one(); + COMPLETED_RETENTION_HOOKS + .lock() + .await + .retain(|key, _| key != &bucket && key != &task_id); + } +} + +#[tokio::test] +async fn heal_start_retry_deadline_preserves_real_executor_progress() { + assert_heal_start_retry_control_preserves_real_executor_progress(false).await; +} + +#[tokio::test] +async fn heal_start_retry_cancellation_preserves_real_executor_progress() { + assert_heal_start_retry_control_preserves_real_executor_progress(true).await; +} + +#[tokio::test] +async fn heal_start_retry_scheduler_carries_explicit_budget_and_identity() { + let manager = HealManager::new( + Arc::new(MockStorage), + Some(HealConfig { + task_timeout: Duration::ZERO, + ..Default::default() + }), + ); + let mut request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None); + request.source = HealRequestSource::Admin; + request.options.timeout = Some(Duration::from_secs(60)); + let task_id = request.id.clone(); + let created_at = request.created_at; + let hook = Arc::new(CompletedRetentionHook::default()); + COMPLETED_RETENTION_HOOKS + .lock() + .await + .insert(task_id.clone(), Arc::clone(&hook)); + manager + .submit_heal_request(request) + .await + .expect("admit explicit-budget task"); + process_manager_queue_once(&manager).await; + tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified()) + .await + .expect("real read-quorum failure prepares retry"); + let retry = manager.retrying_heals.lock().await[&task_id].request.clone(); + assert_eq!(retry.id, task_id); + assert_eq!(retry.created_at, created_at); + assert_eq!(retry.source, HealRequestSource::Admin); + assert_eq!(retry.retry_attempts, 1); + let remaining = retry.options.timeout.expect("retry retains explicit budget"); + assert!(remaining > Duration::ZERO && remaining < Duration::from_secs(60)); + assert!(matches!( + manager.get_task_status(&task_id).await.expect("retry remains queryable"), + HealTaskStatus::Retrying { retry_attempt: 1, .. } + )); + manager.cancel_task(&task_id).await.expect("cancel held retry"); + hook.finish.notify_one(); + COMPLETED_RETENTION_HOOKS.lock().await.remove(&task_id); +} + struct ManagerRecoveryTestHook { replacement_resume_disk: DiskStore, listed: StdMutex, diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index d44df549e..8a3ebe1b9 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -380,6 +380,10 @@ impl HealTask { }; if let Some(err) = error { + // An interrupted await is not a terminal object repair result. + if matches!(err, Error::TaskTimeout | Error::TaskCancelled) { + return Err(err); + } if Self::is_dangling_delete_grace_error(&err) { telemetry_unknown |= !increment_counter(&mut skipped); warn!(