ci: run ownership interaction red-owner

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,58 @@
-# Copyright 2024 RustFS Team
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-name: Windows Filesystem Tests
-
+name: Heal ownership interaction red-owner
 on:
-  push:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
-  pull_request:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
-
 permissions:
   contents: read
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
-
-env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  RUST_BACKTRACE: 1
-
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
+  focused:
+    runs-on: ubuntu-latest
+    timeout-minutes: 120
     steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+      - name: Checkout
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
         with:
+          fetch-depth: 3
           persist-credentials: false
-
-      - name: Setup Rust environment
+      - name: Verify source snapshot
+        shell: bash
+        run: |
+          set -euo pipefail
+          actual_head=$(git rev-parse HEAD)
+          actual_source=$(git rev-parse HEAD^)
+          actual_base=$(git rev-parse HEAD^^)
+          actual_tree=$(git rev-parse 'HEAD^^{tree}')
+          actual_diff=$(git diff --full-index --binary "$actual_base" "$actual_source" | sha256sum | cut -d' ' -f1)
+          actual_files=$(git diff --name-only "$actual_source" "$actual_head")
+          printf 'head %s source %s base %s tree %s diff %s files %s\n' "$actual_head" "$actual_source" "$actual_base" "$actual_tree" "$actual_diff" "$actual_files"
+          [[ "$actual_head" == "$GITHUB_SHA" ]]
+          [[ "$actual_source" == "733bf1ce7cd5817a2c53bdd8ac26f4968518cc57" ]]
+          [[ "$actual_base" == "ebcd8429ded7dc7d2361b73e48070f35dd46725f" ]]
+          [[ "$actual_tree" == "73d4f909e7ccc7d84ed16e62f549d12feff9560c" ]]
+          [[ "$actual_diff" == "481706c361582a9cd51d97cc5b5e1aa4e4f83c076586e36717ba19a2d34d2176" ]]
+          [[ "$actual_files" == ".github/workflows/windows-filesystem.yml" ]]
+      - name: Setup Rust
         uses: ./.github/actions/setup
         with:
           rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
+          cache-shared-key: ci-e2e-distributed-hosted
           cache-save-if: 'false'
           install-build-packaging-tools: 'false'
           install-test-tools: 'false'
-
-      - name: Check production Windows dependencies
-        shell: pwsh
-        run: cargo check -p rustfs-ecstore --lib
-
-      - name: Test guarded rename publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
-
-      - name: Test multipart parent durability barriers
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
-
-      - name: Test Windows path conversion
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test Windows volume and disk-space probes
-        shell: pwsh
+      - name: Run the selected test
+        shell: bash
         run: |
-          # Map a drive letter to the temp drive's administrative share so the
-          # remote-volume probe also covers mapped network drives.
-          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
-          net use R: "\\localhost\$drive`$" /persistent:no
-          if ($LASTEXITCODE -eq 0) {
-            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
-          } else {
-            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
-          }
-          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
-          $status = $LASTEXITCODE
-          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
-          exit $status
-
-      - name: Test Windows handle guards
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
-
-      - name: Test endpoint path parsing
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
-
-      - name: Test startup temporary-directory cleanup
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
-
-      - name: Test object key Windows admission
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
-
-      - name: Test Windows file-name rules
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test fresh format publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
-
-  # The native Windows main thread only gets the 1 MiB PE stack reserve, while
-  # Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
-  # thread that calls `block_on`, so a debug server once overflowed before it
-  # listened on Windows only. Start a real server to keep that covered.
-  startup-smoke:
-    name: Startup Smoke
-    runs-on: windows-latest
-    timeout-minutes: 90
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Build debug server binary
-        shell: pwsh
-        run: cargo build --locked -p rustfs --bin rustfs
-
-      - name: Print version
-        shell: pwsh
-        run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
-
-      - name: Start server and wait for readiness
-        shell: pwsh
-        env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
-        run: |
-          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
-          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
-          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
-          New-Item -ItemType Directory -Force -Path $volume | Out-Null
-          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
-            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
-            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
-            -NoNewWindow -PassThru
-          # Cache the process handle now; otherwise ExitCode reads back empty.
-          $null = $server.Handle
-          $ready = $false
-          try {
-            for ($i = 0; $i -lt 180; $i++) {
-              if ($server.HasExited) { break }
-              try {
-                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
-                if ($response.StatusCode -eq 200) { $ready = $true; break }
-              } catch {}
-              Start-Sleep -Seconds 1
-            }
-          } finally {
-            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
-            $server.WaitForExit()
-            Write-Host '--- server stdout (tail) ---'
-            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
-            Write-Host '--- server stderr (tail) ---'
-            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
-          }
-          if (-not $ready) {
-            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
-          }
+          set -euo pipefail
+          export CARGO_TERM_COLOR=never CARGO_BUILD_JOBS=2 CARGO_PROFILE_TEST_DEBUG=0 CARGO_INCREMENTAL=0 RUST_BACKTRACE=0
+          name="heal::manager::tests::admin_overlap::forced_id_conflict_preserves_scanner_queued_owner"
+          log="$RUNNER_TEMP/interaction.log"
+          set +e
+          cargo test --locked -p rustfs-heal --lib "$name" -- --exact --test-threads=1 >"$log" 2>&1
+          status=$?
+          set -e
+          tail -n 160 "$log"
+          [[ "$status" -eq 101 ]]
+          grep -Fxq 'running 1 test' "$log"
+          grep -Fxq "test ${name} ... FAILED" "$log"
+          grep -Fq 'the completed holder cannot own a queued receipt' "$log"
+          grep -Fq 'test result: FAILED. 0 passed; 1 failed;' "$log"
+          printf 'Confirmed expected stale queued owner failure\n'

crates/heal/src/heal/manager.rs

@@ -625,6 +625,37 @@ fn request_matches_request(request: &HealRequest, existing: &HealRequest) -> boo
         && request.heal_endpoints == existing.heal_endpoints
 }
 
+fn lookup_request_id_admission(
+    request: &HealRequest,
+    active_heals: &HashMap<String, Arc<HealTask>>,
+    queue: &PriorityHealQueue,
+    retrying_heals: &HashMap<String, RetryingHeal>,
+    durable_owners: &[HealRequest],
+) -> Option<(bool, &'static str)> {
+    active_heals
+        .get(&request.id)
+        .map(|task| (request_matches_task(request, task), "active"))
+        .or_else(|| {
+            queue
+                .requests()
+                .find(|queued| queued.id == request.id)
+                .map(|queued| (request_matches_request(request, queued), "queued"))
+        })
+        .or_else(|| {
+            retrying_heals
+                .get(&request.id)
+                .map(|retrying| (request_matches_request(request, &retrying.request), "retrying"))
+        })
+        // A durable-only owner must be resumed by recovery, not overwritten
+        // with a new execution budget by a replay before recovery completes.
+        .or_else(|| {
+            durable_owners
+                .iter()
+                .find(|owner| owner.id == request.id)
+                .map(|_| (false, "durable"))
+        })
+}
+
 fn retrying_heal_for_dedup_key(retrying_heals: &HashMap<String, RetryingHeal>, key: &str) -> Option<(String, HealType)> {
     retrying_heals
         .iter()
@@ -2119,28 +2150,59 @@ impl HealManager {
         let mut queue = self.heal_queue.lock().await;
         let mut retrying_heals = self.retrying_heals.lock().await;
 
-        let request_id_admission = active_heals
-            .get(&request.id)
-            .map(|task| (request_matches_task(&request, task), "active"))
-            .or_else(|| {
-                queue
-                    .requests()
-                    .find(|queued| queued.id == request.id)
-                    .map(|queued| (request_matches_request(&request, queued), "queued"))
-            })
-            .or_else(|| {
-                retrying_heals
-                    .get(&request.id)
-                    .map(|retrying| (request_matches_request(&request, &retrying.request), "retrying"))
-            })
-            // A durable-only owner must be resumed by recovery, not overwritten
-            // with a new execution budget by a replay before recovery completes.
-            .or_else(|| {
-                durable_owners
-                    .iter()
-                    .find(|owner| owner.id == request.id)
-                    .map(|_| (false, "durable"))
-            });
+        let mut request_id_admission =
+            lookup_request_id_admission(&request, &active_heals, &queue, &retrying_heals, &durable_owners);
+
+        if request_id_admission.is_none() && source == HealRequestSource::Admin && force_start {
+            let overlapping = active_heals
+                .values()
+                .map(|task| (&task.id, &task.heal_type, &task.options, task.source))
+                .chain(
+                    queue
+                        .requests()
+                        .chain(retrying_heals.values().map(|retrying| &retrying.request))
+                        .chain(durable_owners.iter())
+                        .map(|owner| (&owner.id, &owner.heal_type, &owner.options, owner.source)),
+                )
+                .filter(|(id, heal_type, options, source)| {
+                    **id != request.id
+                        && *source == HealRequestSource::Admin
+                        && heal_scopes_overlap(&request.heal_type, &request.options, heal_type, options)
+                            != OverlapVerdict::Disjoint
+                })
+                .map(|(id, _, _, _)| id.clone())
+                .collect::<HashSet<_>>();
+            drop(retrying_heals);
+            drop(queue);
+            drop(active_heals);
+            active_lock_hold += active_lock_acquired.elapsed();
+            for task_id in &overlapping {
+                match self.cancel_task(task_id).await {
+                    Ok(_) => info!(
+                        target: "rustfs::heal::manager",
+                        event = EVENT_HEAL_QUEUE_ADMISSION,
+                        component = LOG_COMPONENT_HEAL,
+                        subsystem = LOG_SUBSYSTEM_MANAGER,
+                        request_id = %request.id,
+                        cancelled_task_id = %task_id,
+                        result = "force_start_cancelled_overlap",
+                        "Admin forceStart cancelled an overlapping heal task"
+                    ),
+                    Err(Error::TaskNotFound { .. }) => {} // The owner completed before cancellation acquired its state.
+                    Err(err) => return Err(err),
+                }
+            }
+            durable_owners.retain(|owner| !overlapping.contains(&owner.id));
+            let reacquire_start = Instant::now();
+            active_heals = self.active_heals.lock().await;
+            active_lock_acquired = Instant::now();
+            active_lock_wait += active_lock_acquired.duration_since(reacquire_start);
+            queue = self.heal_queue.lock().await;
+            retrying_heals = self.retrying_heals.lock().await;
+            // Non-admin admission can claim this id while cancellation drops the state locks.
+            request_id_admission = lookup_request_id_admission(&request, &active_heals, &queue, &retrying_heals, &durable_owners);
+        }
+
         if let Some((matches_existing, duplicate_state)) = request_id_admission {
             let admission = if matches_existing {
                 if accept_same_request_id_replay {
@@ -2189,54 +2251,6 @@ impl HealManager {
             });
         }
 
-        if source == HealRequestSource::Admin && force_start {
-            let overlapping = active_heals
-                .values()
-                .map(|task| (&task.id, &task.heal_type, &task.options, task.source))
-                .chain(
-                    queue
-                        .requests()
-                        .chain(retrying_heals.values().map(|retrying| &retrying.request))
-                        .chain(durable_owners.iter())
-                        .map(|owner| (&owner.id, &owner.heal_type, &owner.options, owner.source)),
-                )
-                .filter(|(id, heal_type, options, source)| {
-                    **id != request.id
-                        && *source == HealRequestSource::Admin
-                        && heal_scopes_overlap(&request.heal_type, &request.options, heal_type, options)
-                            != OverlapVerdict::Disjoint
-                })
-                .map(|(id, _, _, _)| id.clone())
-                .collect::<HashSet<_>>();
-            drop(retrying_heals);
-            drop(queue);
-            drop(active_heals);
-            active_lock_hold += active_lock_acquired.elapsed();
-            for task_id in &overlapping {
-                match self.cancel_task(task_id).await {
-                    Ok(_) => info!(
-                        target: "rustfs::heal::manager",
-                        event = EVENT_HEAL_QUEUE_ADMISSION,
-                        component = LOG_COMPONENT_HEAL,
-                        subsystem = LOG_SUBSYSTEM_MANAGER,
-                        request_id = %request.id,
-                        cancelled_task_id = %task_id,
-                        result = "force_start_cancelled_overlap",
-                        "Admin forceStart cancelled an overlapping heal task"
-                    ),
-                    Err(Error::TaskNotFound { .. }) => {} // The owner completed before cancellation acquired its state.
-                    Err(err) => return Err(err),
-                }
-            }
-            durable_owners.retain(|owner| !overlapping.contains(&owner.id));
-            let reacquire_start = Instant::now();
-            active_heals = self.active_heals.lock().await;
-            active_lock_acquired = Instant::now();
-            active_lock_wait += active_lock_acquired.duration_since(reacquire_start);
-            queue = self.heal_queue.lock().await;
-            retrying_heals = self.retrying_heals.lock().await;
-        }
-
         let rejection = if source == HealRequestSource::Admin {
             let live_owners = active_heals
                 .values()

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