From 8679570c2a328f3095f371be25a9e68997298bf7 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 22 Aug 2026 16:23:40 +0800 Subject: [PATCH] fix(heal): retain displaced task status (#6370) --- crates/heal/src/heal/channel.rs | 54 +++++ crates/heal/src/heal/manager.rs | 115 +++++++++- crates/heal/src/heal/manager/auto_scan.rs | 17 +- crates/heal/src/heal/manager/scheduler.rs | 28 ++- crates/heal/src/heal/manager/tests.rs | 263 +++++++++++++++++++++- 5 files changed, 461 insertions(+), 16 deletions(-) diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index edfcf2613..e00435100 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -1640,6 +1640,60 @@ mod tests { assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0); } + #[tokio::test] + async fn test_process_query_request_reports_displaced_terminal_detail() { + let heal_manager = Arc::new(HealManager::new( + Arc::new(MockStorage), + Some(HealConfig { + queue_size: 1, + ..HealConfig::default() + }), + )); + let mut displaced = HealRequest::new( + HealType::Bucket { + bucket: "displaced-channel".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + displaced.id = "displaced-channel-task".to_string(); + let displaced_id = displaced.id.clone(); + heal_manager + .submit_heal_request(displaced) + .await + .expect("initial channel task should queue"); + heal_manager + .submit_heal_request(HealRequest::new( + HealType::Bucket { + bucket: "successor-channel".to_string(), + }, + HealOptions::default(), + HealPriority::High, + )) + .await + .expect("successor channel task should displace the initial task"); + + let processor = HealChannelProcessor::new(heal_manager); + let (tx, rx) = oneshot::channel(); + processor + .process_query_request("displaced-channel".to_string(), displaced_id, None, tx) + .await + .expect("displaced query should process"); + let response = rx + .await + .expect("query response should be returned") + .expect("displaced query should remain successful"); + let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should exist")) + .expect("status payload should be json"); + assert_eq!(payload["summary"], "stopped"); + assert!( + response + .error + .as_deref() + .is_some_and(|detail| detail.contains("reason=displaced")) + ); + } + #[tokio::test] async fn test_process_query_request_reports_running_for_queued_task() { let heal_manager = create_test_heal_manager(); diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index bad6d4b5b..fc8255e0b 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -40,6 +40,7 @@ use tracing::{debug, error, info, warn}; use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read}; const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60); +const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again"; const LOG_COMPONENT_HEAL: &str = "heal"; const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner"; const LOG_SUBSYSTEM_MANAGER: &str = "manager"; @@ -120,26 +121,30 @@ struct MrfRepairNoticeTarget { version_id: Option<[u8; 16]>, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] struct HealAdmissionDecision { result: HealAdmissionResult, - displaced_task_id: Option, + displaced_request: Option, } impl HealAdmissionDecision { const fn new(result: HealAdmissionResult) -> Self { Self { result, - displaced_task_id: None, + displaced_request: None, } } - fn accepted_with_displacement(displaced_task_id: String) -> Self { + fn accepted_with_displacement(displaced_request: HealRequest) -> Self { Self { result: HealAdmissionResult::Accepted, - displaced_task_id: Some(displaced_task_id), + displaced_request: Some(displaced_request), } } + + fn displaced_task_id(&self) -> Option<&str> { + self.displaced_request.as_ref().map(|request| request.id.as_str()) + } } fn lock_mrf_repair_notice_targets( @@ -151,6 +156,55 @@ fn lock_mrf_repair_notice_targets( } } +fn lock_displaced_terminals( + registry: &StdMutex>>, +) -> StdMutexGuard<'_, HashMap>> { + match registry.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +fn record_displaced_terminal( + registry: &StdMutex>>, + request: &HealRequest, +) -> Arc { + let terminal = Arc::new(CompletedHealStatus { + heal_type: request.heal_type.clone(), + status: HealTaskStatus::Failed { + error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"), + }, + result_items_truncated: false, + completed_at: SystemTime::now(), + seqed_items: Vec::new(), + next_seq: 0, + min_seq: 0, + }); + let mut terminals = lock_displaced_terminals(registry); + prune_completed_heal_statuses(&mut terminals); + terminals.insert(request.id.clone(), Arc::clone(&terminal)); + terminal +} + +async fn remove_displaced_task_aliases( + aliases: &Arc>>, + terminals: &StdMutex>>, + task_id: &str, + terminal: &Arc, +) { + let mut aliases = aliases.lock().await; + let alias_ids = aliases + .iter() + .filter_map(|(alias_id, alias)| (alias.task_id == task_id).then_some(alias_id.clone())) + .collect::>(); + let mut displaced_terminals = lock_displaced_terminals(terminals); + prune_completed_heal_statuses(&mut displaced_terminals); + for alias_id in alias_ids { + displaced_terminals.insert(alias_id, Arc::clone(terminal)); + } + aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id); +} + async fn remove_task_aliases_for_task(registry: &Arc>>, task_id: &str) { registry .lock() @@ -618,6 +672,14 @@ pub struct HealManager { /// are shared so the lookup helper can hand a completed entry to a /// caller without cloning the retained result window. completed_heals: Arc>>>, + /// Terminals for requests removed by priority displacement. An Accepted + /// task ID remains queryable for the same process lifetime and the normal + /// ten-minute status TTL; clients should treat `reason=displaced` as a + /// terminal result and submit a fresh request. This sidecar is synchronous + /// so admission can publish the terminal while the queue transition is + /// still under its lock, without awaiting another tokio lock. Queue state + /// is process-local, so this guarantee does not extend across restart. + displaced_terminals: Arc>>>, /// Client tokens merged into an existing task id. task_aliases: Arc>>, /// Heal tasks waiting for a retry backoff to expire. @@ -659,6 +721,7 @@ struct HealQueueContext<'a> { heal_queue: &'a Arc>, active_heals: &'a Arc>>>, completed_heals: &'a Arc>>>, + displaced_terminals: &'a Arc>>>, task_aliases: &'a Arc>>, retrying_heals: &'a Arc>>, mrf_repair_notice_targets: &'a Arc>>>, @@ -874,7 +937,7 @@ impl HealManager { result = "accepted_by_displacement", "Heal queue request accepted by displacement" }); - return HealAdmissionDecision::accepted_with_displacement(displaced.id); + return HealAdmissionDecision::accepted_with_displacement(displaced); } demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", { @@ -1105,6 +1168,7 @@ impl HealManager { active_heals: Arc::new(Mutex::new(HashMap::new())), heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())), completed_heals: Arc::new(Mutex::new(HashMap::new())), + displaced_terminals: Arc::new(StdMutex::new(HashMap::new())), task_aliases: Arc::new(Mutex::new(HashMap::new())), retrying_heals: Arc::new(Mutex::new(HashMap::new())), mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())), @@ -1209,6 +1273,10 @@ impl HealManager { active_heals.clear(); publish_active_heal_count(&active_heals); self.completed_heals.lock().await.clear(); + // Do not let the synchronous guard live across the following async lock. + { + lock_displaced_terminals(&self.displaced_terminals).clear(); + } self.task_aliases.lock().await.clear(); self.retrying_heals.lock().await.clear(); lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear(); @@ -1459,7 +1527,11 @@ impl HealManager { task_id = queued_id.to_owned(); } let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable; - let displaced_task_id = admission_decision.displaced_task_id; + let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned); + let displaced_terminal = admission_decision + .displaced_request + .as_ref() + .map(|request| record_displaced_terminal(&self.displaced_terminals, request)); if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged) && let Some(target) = mrf_notice_target { @@ -1473,8 +1545,12 @@ impl HealManager { drop(queue); drop(active_heals); - if let Some(displaced_task_id) = displaced_task_id { - self.remove_aliases_for_task(&displaced_task_id).await; + if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) { + // The queue has already removed the displaced request, so the + // synchronous terminal sidecar was published before aliases and + // MRF ownership are cleaned up. + remove_displaced_task_aliases(&self.task_aliases, &self.displaced_terminals, &displaced_task_id, &displaced_terminal) + .await; } if should_notify { @@ -1549,6 +1625,15 @@ impl HealManager { } } + if terminal_completed.is_none() { + let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals); + prune_completed_heal_statuses(&mut displaced_terminals); + terminal_completed = displaced_terminals + .get(canonical_task_id) + .filter(|terminal| matches_path(&terminal.heal_type)) + .cloned(); + } + match terminal_completed { Some(completed) => TaskStateLookup::Completed(completed), None => TaskStateLookup::NotFound, @@ -1669,9 +1754,19 @@ impl HealManager { let mut completed_heals = self.completed_heals.lock().await; prune_completed_heal_statuses(&mut completed_heals); - completed_heals + if completed_heals .values() .any(|completed| heal_type_matches_path(&completed.heal_type, heal_path)) + { + return true; + } + drop(completed_heals); + + let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals); + prune_completed_heal_statuses(&mut displaced_terminals); + displaced_terminals + .values() + .any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path)) } /// Get task progress diff --git a/crates/heal/src/heal/manager/auto_scan.rs b/crates/heal/src/heal/manager/auto_scan.rs index 10ac136a4..f7eb4482d 100644 --- a/crates/heal/src/heal/manager/auto_scan.rs +++ b/crates/heal/src/heal/manager/auto_scan.rs @@ -21,6 +21,7 @@ impl HealManager { let heal_queue = self.heal_queue.clone(); let active_heals = self.active_heals.clone(); let task_aliases = self.task_aliases.clone(); + let displaced_terminals = self.displaced_terminals.clone(); let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone(); let storage = self.storage.clone(); let replacement_recovery_anchors = self.replacement_recovery_anchors.clone(); @@ -481,6 +482,10 @@ impl HealManager { let admission = admission_decision.result; let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable; + let displaced_terminal = admission_decision + .displaced_request + .as_ref() + .map(|request| record_displaced_terminal(&displaced_terminals, request)); if matches!(admission, HealAdmissionResult::Accepted) && let Some(anchor) = recovery_anchor { @@ -491,8 +496,16 @@ impl HealManager { } drop(queue); drop(config); - if let Some(displaced_task_id) = admission_decision.displaced_task_id { - remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await; + if let (Some(displaced_task_id), Some(displaced_terminal)) = + (admission_decision.displaced_task_id().map(ToOwned::to_owned), displaced_terminal) + { + remove_displaced_task_aliases( + &task_aliases, + &displaced_terminals, + &displaced_task_id, + &displaced_terminal, + ) + .await; lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id); } if matches!(admission, HealAdmissionResult::Accepted) { diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index fbee9f212..c76a037ae 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -21,6 +21,7 @@ impl HealManager { let heal_queue = self.heal_queue.clone(); let active_heals = self.active_heals.clone(); let completed_heals = self.completed_heals.clone(); + let displaced_terminals = self.displaced_terminals.clone(); let task_aliases = self.task_aliases.clone(); let retrying_heals = self.retrying_heals.clone(); let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone(); @@ -53,6 +54,7 @@ impl HealManager { heal_queue: &heal_queue, active_heals: &active_heals, completed_heals: &completed_heals, + displaced_terminals: &displaced_terminals, task_aliases: &task_aliases, retrying_heals: &retrying_heals, mrf_repair_notice_targets: &mrf_repair_notice_targets, @@ -71,6 +73,7 @@ impl HealManager { heal_queue: &heal_queue, active_heals: &active_heals, completed_heals: &completed_heals, + displaced_terminals: &displaced_terminals, task_aliases: &task_aliases, retrying_heals: &retrying_heals, mrf_repair_notice_targets: &mrf_repair_notice_targets, @@ -98,6 +101,7 @@ impl HealManager { heal_queue, active_heals, completed_heals, + displaced_terminals, task_aliases, retrying_heals, mrf_repair_notice_targets, @@ -183,6 +187,7 @@ impl HealManager { let active_heals_clone = active_heals.clone(); let heal_queue_clone = heal_queue.clone(); let completed_heals_clone = completed_heals.clone(); + let displaced_terminals_clone = displaced_terminals.clone(); let task_aliases_clone = task_aliases.clone(); let retrying_heals_clone = retrying_heals.clone(); let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone(); @@ -363,6 +368,7 @@ impl HealManager { let retry_heal_queue = heal_queue_clone.clone(); let retrying_heals_for_spawn = retrying_heals_clone.clone(); let retry_task_aliases = task_aliases_clone.clone(); + let retry_displaced_terminals = displaced_terminals_clone.clone(); let retry_mrf_repair_notice_targets = mrf_repair_notice_targets_clone.clone(); let retry_completed_heals = completed_heals_clone.clone(); let retry_notify = notify_clone.clone(); @@ -430,6 +436,14 @@ impl HealManager { let admission = admission_decision.result; let should_notify = matches!(admission, HealAdmissionResult::Accepted) && retry_config.event_driven_scheduler_enable; + // Publish the terminal synchronously while the + // queue transition is protected. The subsequent + // queue -> retrying handoff retains the lock order + // used by operations_snapshot. + let displaced_terminal = admission_decision + .displaced_request + .as_ref() + .map(|request| record_displaced_terminal(&retry_displaced_terminals, request)); match admission { HealAdmissionResult::Accepted => { // Transfer ownership while holding queue -> retrying, @@ -437,10 +451,18 @@ impl HealManager { #[cfg(test)] pause_retry_ownership_transition(&retry_request_id, true).await; retrying_heals_for_spawn.lock().await.remove(&retry_request_id); - let displaced_task_id = admission_decision.displaced_task_id; + let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned); drop(queue); - if let Some(displaced_task_id) = displaced_task_id { - remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await; + if let (Some(displaced_task_id), Some(displaced_terminal)) = + (displaced_task_id, displaced_terminal) + { + remove_displaced_task_aliases( + &retry_task_aliases, + &retry_displaced_terminals, + &displaced_task_id, + &displaced_terminal, + ) + .await; remove_mrf_repair_notice_targets( &retry_mrf_repair_notice_targets, &displaced_task_id, diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index ad50c4bf7..585b7a4e7 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -84,6 +84,7 @@ async fn process_manager_queue_once(manager: &HealManager) { heal_queue: &manager.heal_queue, active_heals: &manager.active_heals, completed_heals: &manager.completed_heals, + displaced_terminals: &manager.displaced_terminals, task_aliases: &manager.task_aliases, retrying_heals: &manager.retrying_heals, mrf_repair_notice_targets: &manager.mrf_repair_notice_targets, @@ -2778,7 +2779,10 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() { HealAdmissionResult::Accepted ); assert_eq!(manager.get_queue_length().await, 1); - assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. }))); + assert!(matches!( + manager.get_task_status(&low_id).await, + Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced") + )); assert_eq!( manager .get_task_status(&high_id) @@ -2788,6 +2792,263 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() { ); } +#[tokio::test] +async fn displaced_task_remains_queryable() { + let manager = HealManager::new( + Arc::new(MockStorage), + Some(HealConfig { + queue_size: 1, + ..HealConfig::default() + }), + ); + let mut displaced = HealRequest::new( + HealType::Bucket { + bucket: "displaced-bucket".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + displaced.id = "displaced-task".to_string(); + let displaced_id = displaced.id.clone(); + manager + .submit_heal_request(displaced) + .await + .expect("displaced request should queue"); + + let successor = HealRequest::new( + HealType::Bucket { + bucket: "successor-bucket".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + manager + .submit_heal_request(successor) + .await + .expect("successor should displace low work"); + + let report = manager + .get_task_report(&displaced_id) + .await + .expect("displaced report should remain queryable"); + assert!(matches!(report.status, HealTaskStatus::Failed { ref error } if error.contains("reason=displaced"))); +} + +#[tokio::test] +async fn displaced_archive_failure_keeps_queryable_terminal() { + let manager = HealManager::new(Arc::new(MockStorage), None); + let mut request = HealRequest::new( + HealType::Bucket { + bucket: "archive-failure".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + request.id = "archive-failure-task".to_string(); + let request_id = request.id.clone(); + // The synchronous sidecar is the authoritative fallback when the normal + // completed-task archive has no entry (the failure window that must not + // turn an Accepted ID into NotFound). + record_displaced_terminal(&manager.displaced_terminals, &request); + assert!(manager.completed_heals.lock().await.is_empty()); + assert!(matches!( + manager.get_task_status(&request_id).await, + Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced") + )); +} + +#[tokio::test] +async fn scheduler_retry_displacement_keeps_evicted_task_queryable() { + let manager = Arc::new(HealManager::new( + Arc::new(MockStorage), + Some(HealConfig { + queue_size: 1, + event_driven_scheduler_enable: false, + ..HealConfig::default() + }), + )); + let mut retry_request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None); + retry_request.priority = HealPriority::High; + let retry_id = retry_request.id.clone(); + manager + .submit_heal_request(retry_request) + .await + .expect("retry request should queue"); + + // Process exactly one queue cycle so the retry task is spawned without a + // background scheduler consuming the filler request before the retry wakes. + process_manager_queue_once(&manager).await; + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if manager.retrying_heals.lock().await.contains_key(&retry_id) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("retry request should enter backoff"); + + let filler = HealRequest::new( + HealType::Bucket { + bucket: "retry-displaced-filler".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + let filler_id = filler.id.clone(); + manager + .submit_heal_request(filler) + .await + .expect("filler request should occupy the queue"); + + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if matches!( + manager.get_task_status(&filler_id).await, + Ok(HealTaskStatus::Failed { ref error }) if error.contains("reason=displaced") + ) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("retry admission should displace the filler request"); + assert_eq!(manager.get_queue_length().await, 1); + assert_eq!( + manager.get_task_status(&retry_id).await.expect("retry should be queued"), + HealTaskStatus::Pending + ); +} + +#[tokio::test] +async fn concurrent_displacers_produce_one_terminal_generation() { + let manager = Arc::new(HealManager::new( + Arc::new(MockStorage), + Some(HealConfig { + queue_size: 1, + ..HealConfig::default() + }), + )); + let mut displaced = HealRequest::new( + HealType::Bucket { + bucket: "concurrent-displaced".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + displaced.id = "concurrent-displaced-task".to_string(); + let displaced_id = displaced.id.clone(); + manager + .submit_heal_request(displaced) + .await + .expect("initial request should queue"); + + let first = HealRequest::new( + HealType::Bucket { + bucket: "concurrent-successor-a".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + let second = HealRequest::new( + HealType::Bucket { + bucket: "concurrent-successor-b".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + let (first_result, second_result) = tokio::join!(manager.submit_heal_request(first), manager.submit_heal_request(second)); + let accepted = [&first_result, &second_result] + .into_iter() + .filter(|result| matches!(result, Ok(HealAdmissionResult::Accepted))) + .count(); + assert_eq!(accepted, 1, "exactly one concurrent displacer should win the full queue"); + assert!( + first_result.is_ok() && second_result.is_ok(), + "the losing request should receive a typed Full result" + ); + let terminals = lock_displaced_terminals(&manager.displaced_terminals); + assert_eq!(terminals.len(), 1); + assert!(terminals.contains_key(&displaced_id)); +} + +#[tokio::test] +async fn successor_chain_is_bounded_and_authorized() { + let manager = HealManager::new( + Arc::new(MockStorage), + Some(HealConfig { + queue_size: 1, + ..HealConfig::default() + }), + ); + let mut original = HealRequest::new( + HealType::Bucket { + bucket: "authorized-original".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + original.id = "authorized-original-task".to_string(); + let original_id = original.id.clone(); + manager.submit_heal_request(original).await.expect("original should queue"); + let mut duplicate = HealRequest::new( + HealType::Bucket { + bucket: "authorized-original".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + duplicate.id = "authorized-duplicate-task".to_string(); + let duplicate_id = duplicate.id.clone(); + manager + .submit_heal_request(duplicate) + .await + .expect("same-target duplicate should merge"); + let successor = HealRequest::new( + HealType::Bucket { + bucket: "authorized-successor".to_string(), + }, + HealOptions::default(), + HealPriority::High, + ); + let successor_id = successor.id.clone(); + manager.submit_heal_request(successor).await.expect("successor should queue"); + assert!(manager.task_aliases.lock().await.is_empty()); + assert!(matches!(manager.get_task_status(&original_id).await, Ok(HealTaskStatus::Failed { .. }))); + assert!(matches!(manager.get_task_status(&duplicate_id).await, Ok(HealTaskStatus::Failed { .. }))); + assert_eq!( + manager + .get_task_status(&successor_id) + .await + .expect("successor should remain queued"), + HealTaskStatus::Pending + ); +} + +#[tokio::test] +async fn displaced_terminal_expires_after_bounded_ttl() { + let manager = HealManager::new(Arc::new(MockStorage), None); + let mut request = HealRequest::new( + HealType::Bucket { + bucket: "expires".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + request.id = "expires-task".to_string(); + let request_id = request.id.clone(); + record_displaced_terminal(&manager.displaced_terminals, &request); + { + let mut terminals = lock_displaced_terminals(&manager.displaced_terminals); + let entry = + Arc::get_mut(terminals.get_mut(&request_id).expect("terminal should be retained")).expect("test owns terminal entry"); + entry.completed_at = SystemTime::now() - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_secs(1); + } + assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. }))); +} + #[tokio::test] async fn test_displacing_registered_mrf_task_drops_notice_ownership() { let storage: Arc = Arc::new(MockStorage);