diff --git a/crates/heal/src/heal/manager/tests/admin_overlap.rs b/crates/heal/src/heal/manager/tests/admin_overlap.rs index 703e74f0a..e39c51b52 100644 --- a/crates/heal/src/heal/manager/tests/admin_overlap.rs +++ b/crates/heal/src/heal/manager/tests/admin_overlap.rs @@ -450,6 +450,176 @@ async fn admin_overlap_normal_start_cannot_enter_force_start_cancellation_window assert_eq!(manager.heal_queue.lock().await.len(), 1); } +#[tokio::test] +async fn admin_force_start_rechecks_request_id_after_cancelling_overlaps() { + for source in [HealRequestSource::Scanner, HealRequestSource::Internal] { + let manager = manager_with_policy(HealOverlapPolicy::Merge); + let previous_id = install_owner(&manager, admin_prefix_request("bucket", "scope/child/"), OwnerState::Active).await; + let aliases = manager.task_aliases.lock().await; + let mut replacement = admin_prefix_request("bucket", "scope/"); + replacement.force_start = true; + let request_id = replacement.id.clone(); + let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement)); + + tokio::time::timeout( + Duration::from_secs(5), + std::future::poll_fn(|cx| { + assert!( + std::pin::pin!(tokio::task::unconstrained(forced.as_mut())) + .poll(cx) + .is_pending() + ); + if manager.active_heals.try_lock().is_ok() { + Poll::Ready(()) + } else { + cx.waker().wake_by_ref(); + Poll::Pending + } + }), + ) + .await + .expect("forceStart reaches cancellation with registry locks released"); + + let mut competing = HealRequest::object("other-bucket".to_string(), "object".to_string(), None); + competing.id = request_id.clone(); + competing.source = source; + competing.options.recreate_missing = false; + let receipt = manager + .submit_heal_request_with_receipt(competing) + .await + .expect("non-admin admission during forceStart cancellation"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, request_id); + + drop(aliases); + let receipt = forced.await.expect("forceStart returns a conflict"); + assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)); + assert_eq!(receipt.task_id, request_id); + let queue = manager.heal_queue.lock().await; + let same_id: Vec<_> = queue.requests().filter(|request| request.id == request_id).collect(); + assert_eq!(same_id.len(), 1, "conflicting requests must not share an id"); + assert_eq!(same_id[0].source, source); + drop(queue); + assert!( + matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled)), + "overlapping admin cancellation precedes the conflict recheck" + ); + } +} + +#[tokio::test] +async fn forced_id_conflict_preserves_scanner_queued_owner() { + for competing_source in [HealRequestSource::Scanner, HealRequestSource::Internal] { + let manager = manager_with_policy(HealOverlapPolicy::Merge); + let previous_id = install_owner(&manager, admin_object_request("object"), OwnerState::Active).await; + let mut queued_ids = Vec::new(); + for source in [ + HealRequestSource::Scanner, + HealRequestSource::Internal, + HealRequestSource::Scanner, + ] { + let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None); + request.source = source; + request.force_start = true; + let id = request.id.clone(); + let receipt = manager + .submit_heal_request_with_receipt(request) + .await + .expect("queue forced work"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, id); + queued_ids.push(id); + } + + // The first holder leaves the queue, leaving two forced tasks queued. + let selected = { + let mut queue = manager.heal_queue.lock().await; + let (selected, skipped) = queue.pop_runnable_with_skips(|_| true, |_| None); + assert!(skipped.is_empty()); + let selected = selected.expect("first queued task should run"); + assert_eq!(selected.id, queued_ids[0]); + assert_eq!(queue.len(), 2); + selected + }; + let completed_task = HealTask::from_request(selected, manager.storage.clone()); + let mut completed = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await; + Arc::make_mut(completed.outcome.as_mut().expect("completed task outcome")).finish(None); + manager + .completed_heals + .lock() + .await + .insert(queued_ids[0].clone(), Arc::new(completed)); + + let aliases = manager.task_aliases.lock().await; + let mut replacement = admin_object_request("object"); + replacement.force_start = true; + let competing_id = replacement.id.clone(); + let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement)); + tokio::time::timeout( + Duration::from_secs(5), + std::future::poll_fn(|cx| { + assert!( + std::pin::pin!(tokio::task::unconstrained(forced.as_mut())) + .poll(cx) + .is_pending() + ); + if manager.active_heals.try_lock().is_ok() { + Poll::Ready(()) + } else { + cx.waker().wake_by_ref(); + Poll::Pending + } + }), + ) + .await + .expect("forceStart reaches cancellation with registry locks released"); + + let mut competing = HealRequest::object("bucket".to_string(), "object".to_string(), None); + competing.id = competing_id.clone(); + competing.source = competing_source; + competing.force_start = true; + let receipt = manager + .submit_heal_request_with_receipt(competing) + .await + .expect("competing same-key forced work"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_eq!(receipt.task_id, competing_id); + + drop(aliases); + let receipt = forced.await.expect("forceStart returns an ID conflict"); + assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)); + assert_eq!(receipt.task_id, competing_id); + assert!(!manager.active_heals.lock().await.contains_key(&competing_id)); + assert!(!manager.retrying_heals.lock().await.contains_key(&competing_id)); + assert!(matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled))); + { + let queue = manager.heal_queue.lock().await; + let same_id: Vec<_> = queue.requests().filter(|request| request.id == competing_id).collect(); + assert_eq!(same_id.len(), 1, "the competing ID must have one live owner"); + assert_eq!(same_id[0].source, competing_source); + assert_eq!(queue.len(), 3); + } + + let mut scanner = HealRequest::object("bucket".to_string(), "object".to_string(), None); + scanner.source = HealRequestSource::Scanner; + let receipt = manager + .submit_heal_request_with_receipt(scanner) + .await + .expect("scanner should merge into queued work"); + assert_eq!(receipt.result, HealAdmissionResult::Merged); + assert_ne!(receipt.task_id, queued_ids[0], "the completed holder cannot own a queued receipt"); + assert!( + manager + .heal_queue + .lock() + .await + .requests() + .any(|request| request.id == receipt.task_id), + "the canonical receipt must name a live queued task" + ); + } +} + #[tokio::test] async fn admin_overlap_cancel_cannot_miss_a_queued_to_active_transition() { let manager = manager_with_policy(HealOverlapPolicy::Merge);