test(heal): exercise concurrent queued ownership

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