test(heal): scope force alias regression workflow

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,103 @@
-# 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 force alias regression
 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
-
+  group: heal-force-alias-red-${{ github.ref }}
+  cancel-in-progress: false
 env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  RUST_BACKTRACE: 1
-
+  CARGO_TERM_COLOR: never
+  CARGO_INCREMENTAL: '0'
+  CARGO_BUILD_JOBS: '1'
+  RUST_BACKTRACE: '0'
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
+  NO_PROXY: 127.0.0.1,localhost
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
-    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: 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
-        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
+  scoped-red-test:
+    runs-on: ubuntu-latest
     timeout-minutes: 90
     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
-        uses: ./.github/actions/setup
+          clean: false
+      - name: Verify source snapshot
+        shell: bash
+        run: |
+          set -euo pipefail
+          workflow_sha=$(git rev-parse HEAD)
+          source_sha=$(git rev-parse HEAD^)
+          base_sha=$(git rev-parse HEAD^^)
+          source_tree=$(git rev-parse "$source_sha^{tree}")
+          source_diff=$(git diff --full-index --binary "$base_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          source_files=$(git diff --name-only "$base_sha" "$source_sha")
+          workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
+          expected_source_files=$(printf '%s\n' 'crates/heal/src/heal/manager.rs' 'crates/heal/src/heal/manager/tests/admin_overlap.rs')
+          printf 'head=%s source=%s base=%s tree=%s diff=%s\n' "$workflow_sha" "$source_sha" "$base_sha" "$source_tree" "$source_diff"
+          [[ "$workflow_sha" == "$GITHUB_SHA" ]]
+          [[ "$source_sha" == 'c5712213727062e67d1e9e295c6a148e202b887a' ]]
+          [[ "$base_sha" == '157a3a6c7d72be064820bf52a687fe36d8383994' ]]
+          [[ "$source_tree" == 'abca475e165d642da5333e993932b5c58da38ee6' ]]
+          [[ "$source_diff" == 'a97a06b7e82caa392c440ae03438b3fac9ddeea6723063e847b20e150bbdf5ba' ]]
+          [[ "$source_files" == "$expected_source_files" ]]
+          [[ "$workflow_files" == '.github/workflows/windows-filesystem.yml' ]]
+          [[ -z "$(git status --porcelain)" ]]
+      - name: Install system dependencies
+        shell: bash
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y build-essential pkg-config libssl-dev ripgrep
+      - name: Install protoc
+        uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6
         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
+          version: '35.1'
+          repo-token: ${{ github.token }}
+      - name: Install flatc
+        uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b
+        with:
+          version: '25.12.19'
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8
+        with:
+          toolchain: stable
+      - name: Run the named regression
+        shell: bash
         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
+          df -h .
+          test_name=heal::manager::tests::admin_overlap::force_start_and_alias_same_id_do_not_misroute
+          log="$RUNNER_TEMP/heal-force-alias-red.log"
+          set +e
+          cargo test --locked -p rustfs-heal --lib -- --exact "$test_name" >"$log" 2>&1
+          test_status=$?
+          set -e
+          tail -n 120 "$log"
+          if [[ "$test_status" -ne 101 ]]; then
+            printf 'Expected test exit 101, received %s\n' "$test_status"
+            exit 1
+          fi
+          grep -Fxq -- 'running 1 test' "$log"
+          grep -Fxq -- "test $test_name ... FAILED" "$log"
+          for source in scanner internal; do
+            for timing in published_before pending_before during; do
+              if [[ "$timing" == pending_before ]]; then
+                alias_reached=true
+              else
+                alias_reached=false
+              fi
+              grep -Fxq -- "alias_force_observed source=$source timing=$timing alias_publication_reached=$alias_reached reached_force_window=true force=accepted alias=merged forced_queued=true canonical_original=true canonical_force=false force_pending=false force_missing=false force_invalid=true alias_pending=true alias_invalid=false original_pending=true receipt_matches=true valid=false" "$log"
+            done
+          done
+          grep -Fq -- 'each id must have exactly one admitted owner and route to that owner' "$log"
+          grep -Fq -- 'test result: FAILED. 0 passed; 1 failed;' "$log"
+          printf 'Confirmed the expected failure of the one selected test\n'
+      - name: Preserve test output
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
+        with:
+          name: heal-force-alias-red-${{ github.run_id }}
+          path: ${{ runner.temp }}/heal-force-alias-red.log
+          if-no-files-found: warn

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()
@@ -1545,6 +1576,8 @@ impl HealManager {
             return;
         }
 
+        #[cfg(test)]
+        tests::admin_overlap::pause_alias_publication(alias_id).await;
         self.task_aliases.lock().await.insert(
             alias_id.to_string(),
             HealTaskAlias {
@@ -2119,28 +2152,61 @@ 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();
+            #[cfg(test)]
+            tests::admin_overlap::pause_force_cancellation(&request.id).await;
+            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 +2255,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()
@@ -2282,13 +2300,22 @@ impl HealManager {
             None
         };
 
+        let scanner_queued_owner_required = request.source == HealRequestSource::Scanner
+            && mrf_notice_target.is_none()
+            && request.expected_mrf_bucket_incarnation_id.is_none();
         let duplicate = (!request.force_start && rejection.is_none()).then(|| {
             active_heal_for_dedup_key(&active_heals, &dedup_key)
                 .map(|(task_id, _)| (task_id, "active"))
                 .or_else(|| {
-                    queue
-                        .queued_request_id_for_dedup_key(&dedup_key)
-                        .map(|queued_id| (queued_id.to_string(), "queued"))
+                    let queued_id = if scanner_queued_owner_required {
+                        queue.scanner_queued_request_id_for_dedup_key(&dedup_key)
+                    } else {
+                        queue.queued_request_id_for_dedup_key(&dedup_key)
+                    };
+                    queued_id.map(|queued_id| (queued_id.to_string(), "queued")).or_else(|| {
+                        (scanner_queued_owner_required && queue.dedup_keys.contains_key(&dedup_key))
+                            .then(|| (request.id.clone(), "queued_owner_unknown"))
+                    })
                 })
                 .or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying")))
         });
@@ -2296,12 +2323,12 @@ impl HealManager {
             // HS-06: under the minio_error overlap policy an exact duplicate
             // admin start reports the typed AlreadyRunning rejection instead
             // of the silent merge (MinIO's ErrHealAlreadyRunning).
-            let mut admission = if (request.source == HealRequestSource::Admin
-                && config.overlap_policy == HealOverlapPolicy::MinioError)
+            let mut admission = if duplicate_state == "queued_owner_unknown"
+                || (request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError)
                 || (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable()))
             {
-                // A running/retrying task may have observed the object before
-                // this write committed. Its receipt cannot prove the new lease.
+                // An unknown queued owner cannot certify a receipt. A running or
+                // retrying task may have observed the object before this write.
                 HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
             } else {
                 Self::duplicate_admission_for_request(&request, &config)
@@ -2514,10 +2541,17 @@ impl HealManager {
                 .remove(&request_id, &request_heal_type, request_source)
                 .await?;
         }
-        if admission == HealAdmissionResult::Merged
-            && let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key)
-        {
-            task_id = queued_id.to_owned();
+        if admission == HealAdmissionResult::Merged {
+            let queued_id = if scanner_queued_owner_required {
+                queue.scanner_queued_request_id_for_dedup_key(&dedup_key)
+            } else {
+                queue.queued_request_id_for_dedup_key(&dedup_key)
+            };
+            if let Some(queued_id) = queued_id {
+                task_id = queued_id.to_owned();
+            } else if scanner_queued_owner_required {
+                return Err(Error::Other("Scanner queued owner is unavailable".to_owned()));
+            }
         }
         let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
         let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);

crates/heal/src/heal/manager/queue.rs

@@ -13,15 +13,23 @@
 // limitations under the License.
 /// The priority heal queue and its per-key dedup index.
 use super::*;
+use std::collections::BTreeMap;
 
-/// Per-key bookkeeping for the queued-request dedup index: how many queued
-/// requests hold the key, and the id of the first request that opened it —
-/// the O(1) stand-in for the former heap scan when a merge receipt needs to
-/// name a queued representative.
+/// Per-key bookkeeping for the queued-request dedup index. The legacy
+/// representative is retained for existing non-scanner callers; scanner
+/// receipts use the separately tracked queued holders.
 #[derive(Debug)]
 pub(super) struct DedupKeyEntry {
     pub(super) refcount: usize,
     pub(super) representative_request_id: String,
+    pub(super) scanner_queued_owners: ScannerQueuedOwners,
+}
+
+#[derive(Debug)]
+pub(super) enum ScannerQueuedOwners {
+    Single(u64),
+    // Once a key has forced duplicates, keep every live holder until it is released.
+    Indexed(BTreeMap<u64, String>),
 }
 
 /// Priority queue wrapper for heal requests
@@ -306,11 +314,7 @@ impl PriorityHealQueue {
 
     #[cfg(test)]
     pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
-        self.heap.pop().map(|item| {
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
-            self.remove_best_effort_count(item.request.source);
-            item.request
-        })
+        self.heap.pop().map(|item| self.remove_queued_item(item))
     }
 
     pub(super) fn is_empty(&self) -> bool {
@@ -332,14 +336,32 @@ impl PriorityHealQueue {
         // also reserves the dedup key for later non-forced duplicates. The first
         // request that opens the key becomes the named representative for merge
         // receipts (taken before `request` moves into the heap).
-        self.dedup_keys
-            .entry(key.clone())
-            .or_insert_with(|| DedupKeyEntry {
-                refcount: 0,
-                representative_request_id: request.id.clone(),
-            })
-            .refcount += 1;
         self.sequence += 1;
+        if let Some(entry) = self.dedup_keys.get_mut(&key) {
+            match &mut entry.scanner_queued_owners {
+                ScannerQueuedOwners::Single(first_sequence) => {
+                    let mut owners = BTreeMap::new();
+                    if entry.refcount == 1 {
+                        owners.insert(*first_sequence, entry.representative_request_id.clone());
+                    }
+                    owners.insert(self.sequence, request.id.clone());
+                    entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(owners);
+                }
+                ScannerQueuedOwners::Indexed(owners) => {
+                    owners.insert(self.sequence, request.id.clone());
+                }
+            }
+            entry.refcount += 1;
+        } else {
+            self.dedup_keys.insert(
+                key.clone(),
+                DedupKeyEntry {
+                    refcount: 1,
+                    representative_request_id: request.id.clone(),
+                    scanner_queued_owners: ScannerQueuedOwners::Single(self.sequence),
+                },
+            );
+        }
         let source = request.source;
         self.heap.push(PriorityQueueItem {
             priority: request.priority,
@@ -402,7 +424,7 @@ impl PriorityHealQueue {
         self.heap = retained;
 
         let displaced = displaced.map(|item| {
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
             self.remove_best_effort_count(item.request.source);
             self.refresh_dedup_representative(&item.dedup_key);
             item.request
@@ -446,11 +468,7 @@ impl PriorityHealQueue {
 
     #[cfg(test)]
     pub(super) fn pop(&mut self) -> Option<HealRequest> {
-        self.heap.pop().map(|item| {
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
-            self.remove_best_effort_count(item.request.source);
-            item.request
-        })
+        self.heap.pop().map(|item| self.remove_queued_item(item))
     }
 
     pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
@@ -475,14 +493,7 @@ impl PriorityHealQueue {
 
         self.restore_deferred_items(deferred);
 
-        (
-            selected.map(|item| {
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
-                self.remove_best_effort_count(item.request.source);
-                item.request
-            }),
-            skipped,
-        )
+        (selected.map(|item| self.remove_queued_item(item)), skipped)
     }
 
     pub(super) fn pop_runnable_with_fairness<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
@@ -554,15 +565,19 @@ impl PriorityHealQueue {
         self.heap = BinaryHeap::from(items);
 
         let chosen = chosen.map(|item| {
-            let request = item.request;
-            Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
-            self.remove_best_effort_count(request.source);
+            let request = self.remove_queued_item(item);
             self.record_dispatch(&request);
             request
         });
         (chosen, skipped)
     }
 
+    fn remove_queued_item(&mut self, item: PriorityQueueItem) -> HealRequest {
+        Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
+        self.remove_best_effort_count(item.request.source);
+        item.request
+    }
+
     fn record_dispatch(&mut self, request: &HealRequest) {
         if is_best_effort_source(request.source) {
             self.mrf_normal_dispatches_since_best_effort = 0;
@@ -667,10 +682,19 @@ impl PriorityHealQueue {
         }
     }
 
-    pub(super) fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, key: &str) {
-        if let Some(entry) = dedup_keys.get_mut(key) {
+    fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, item: &PriorityQueueItem) {
+        if let Some(entry) = dedup_keys.get_mut(&item.dedup_key) {
+            match &mut entry.scanner_queued_owners {
+                ScannerQueuedOwners::Single(sequence) if entry.refcount != 1 || *sequence != item.sequence => {
+                    entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(BTreeMap::new());
+                }
+                ScannerQueuedOwners::Single(_) => {}
+                ScannerQueuedOwners::Indexed(owners) => {
+                    owners.remove(&item.sequence);
+                }
+            }
             if entry.refcount <= 1 {
-                dedup_keys.remove(key);
+                dedup_keys.remove(&item.dedup_key);
             } else {
                 entry.refcount -= 1;
             }
@@ -747,15 +771,19 @@ impl PriorityHealQueue {
         self.dedup_keys.get(key).map(|entry| entry.representative_request_id.as_str())
     }
 
-    /// Re-elect the representative for `key` from the queue entries holding
-    /// it. Needed after a holder leaves the queue *without* becoming active
-    /// (canceled by id, or displaced): the former opener may be the request
-    /// that just left, and a merge receipt must never name an id that
-    /// resolves nowhere. The scheduler pop path does not need this — the
-    /// popped request surfaces in `active_heals` under the same id and the
-    /// duplicate pre-check consults active heals before the queue. No-op for
-    /// released keys; the survivor scan only runs when a key still has
-    /// holders, which under forced duplicates is the rare admin path.
+    pub(super) fn scanner_queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> {
+        let entry = self.dedup_keys.get(key)?;
+        match &entry.scanner_queued_owners {
+            ScannerQueuedOwners::Single(_) if entry.refcount == 1 => Some(entry.representative_request_id.as_str()),
+            ScannerQueuedOwners::Indexed(owners) if owners.len() == entry.refcount => {
+                owners.first_key_value().map(|(_, id)| id.as_str())
+            }
+            _ => None,
+        }
+    }
+
+    /// Re-elect the legacy representative from the heap after cancellation or
+    /// displacement. Dequeue leaves it unchanged for existing callers.
     pub(super) fn refresh_dedup_representative(&mut self, key: &str) {
         if !self.dedup_keys.contains_key(key) {
             return;
@@ -786,7 +814,7 @@ impl PriorityHealQueue {
         while let Some(item) = self.heap.pop() {
             if removed.is_none() && item.request.id == request_id {
                 let key = item.dedup_key.clone();
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
                 self.remove_best_effort_count(item.request.source);
                 affected_key = Some(key);
                 removed = Some(item.request);
@@ -812,7 +840,7 @@ impl PriorityHealQueue {
 
         while let Some(item) = self.heap.pop() {
             if should_remove(&item.request) {
-                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
+                Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
                 self.remove_best_effort_count(item.request.source);
                 affected_keys.push(item.dedup_key);
                 removed.push(item.request);

crates/heal/src/heal/manager/tests.rs

@@ -34,6 +34,7 @@ mod auto_scan;
 mod completion;
 mod disposition;
 mod legacy_ambiguity_logging;
+mod queued_representative;
 mod running_mainline;
 
 use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo};

crates/heal/src/heal/manager/tests/admin_overlap.rs

@@ -17,6 +17,41 @@ use rustfs_heal_contracts::heal_channel::HealScanMode;
 use std::future::Future;
 use std::task::Poll;
 
+struct AdmissionInterleavingHook {
+    reached: Notify,
+    release: Notify,
+}
+
+static FORCE_CANCELLATION_HOOKS: LazyLock<StdMutex<HashMap<String, Arc<AdmissionInterleavingHook>>>> =
+    LazyLock::new(|| StdMutex::new(HashMap::new()));
+
+static ALIAS_PUBLICATION_HOOKS: LazyLock<StdMutex<HashMap<String, Arc<AdmissionInterleavingHook>>>> =
+    LazyLock::new(|| StdMutex::new(HashMap::new()));
+
+pub(in crate::heal::manager) async fn pause_force_cancellation(request_id: &str) {
+    let hook = FORCE_CANCELLATION_HOOKS
+        .lock()
+        .expect("force cancellation hooks")
+        .get(request_id)
+        .cloned();
+    if let Some(hook) = hook {
+        hook.reached.notify_one();
+        hook.release.notified().await;
+    }
+}
+
+pub(in crate::heal::manager) async fn pause_alias_publication(request_id: &str) {
+    let hook = ALIAS_PUBLICATION_HOOKS
+        .lock()
+        .expect("alias publication hooks")
+        .get(request_id)
+        .cloned();
+    if let Some(hook) = hook {
+        hook.reached.notify_one();
+        hook.release.notified().await;
+    }
+}
+
 #[derive(Clone, Copy, Debug)]
 enum OwnerState {
     Active,
@@ -450,6 +485,373 @@ 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 force_start_and_alias_same_id_do_not_misroute() {
+    let mut observed = Vec::new();
+    for source in [HealRequestSource::Scanner, HealRequestSource::Internal] {
+        for timing in ["published_before", "pending_before", "during"] {
+            let manager = Arc::new(manager_with_policy(HealOverlapPolicy::Merge));
+            install_owner(&manager, admin_prefix_request("bucket", "scope/child/"), OwnerState::Active).await;
+            let mut original = HealRequest::object("other-bucket".to_string(), "object".to_string(), None);
+            original.id = format!("alias-original-{}-{timing}", source.as_str());
+            original.source = source;
+            original.options.recreate_missing = false;
+            let original_id = original.id.clone();
+            assert_eq!(
+                manager
+                    .submit_heal_request(original.clone())
+                    .await
+                    .expect("queue original owner"),
+                HealAdmissionResult::Accepted
+            );
+
+            let mut replacement = admin_prefix_request("bucket", "scope/");
+            replacement.id = format!("alias-forced-{}-{timing}", source.as_str());
+            replacement.force_start = true;
+            let request_id = replacement.id.clone();
+            let mut duplicate = original;
+            duplicate.id = request_id.clone();
+            let mut alias_submission = Box::pin(manager.submit_heal_request(duplicate));
+            let alias_hook = (timing == "pending_before").then(|| {
+                let hook = Arc::new(AdmissionInterleavingHook {
+                    reached: Notify::new(),
+                    release: Notify::new(),
+                });
+                ALIAS_PUBLICATION_HOOKS
+                    .lock()
+                    .expect("install alias publication hook")
+                    .insert(request_id.clone(), Arc::clone(&hook));
+                hook
+            });
+            let mut alias_result = None;
+            if timing == "published_before" {
+                alias_result = Some(alias_submission.as_mut().await);
+                assert_eq!(
+                    alias_result
+                        .as_ref()
+                        .expect("pre-published alias result")
+                        .as_ref()
+                        .expect("merge the alias"),
+                    &HealAdmissionResult::Merged
+                );
+                assert_eq!(manager.canonical_task_id(&request_id).await, original_id);
+            } else if timing == "pending_before" {
+                let hook = alias_hook.as_ref().expect("pending alias hook");
+                let mut reached = std::pin::pin!(hook.reached.notified());
+                let mut drive_alias = std::pin::pin!(tokio::task::unconstrained(alias_submission.as_mut()));
+                tokio::time::timeout(
+                    Duration::from_secs(5),
+                    std::future::poll_fn(|cx| {
+                        if reached.as_mut().poll(cx).is_ready() {
+                            return Poll::Ready(());
+                        }
+                        assert!(drive_alias.as_mut().poll(cx).is_pending(), "alias admission ended before publication");
+                        reached.as_mut().poll(cx)
+                    }),
+                )
+                .await
+                .expect("alias admission reached publication before forceStart begins");
+            }
+
+            let hook = Arc::new(AdmissionInterleavingHook {
+                reached: Notify::new(),
+                release: Notify::new(),
+            });
+            FORCE_CANCELLATION_HOOKS
+                .lock()
+                .expect("install force cancellation hook")
+                .insert(request_id.clone(), Arc::clone(&hook));
+            let mut force_submission = Box::pin(manager.submit_heal_request_with_receipt(replacement));
+            let (force_result, reached_force_window) = if timing == "during" {
+                let mut reached = std::pin::pin!(hook.reached.notified());
+                let mut drive_force = std::pin::pin!(tokio::task::unconstrained(force_submission.as_mut()));
+                tokio::time::timeout(
+                    Duration::from_secs(5),
+                    std::future::poll_fn(|cx| {
+                        if reached.as_mut().poll(cx).is_ready() {
+                            return Poll::Ready(());
+                        }
+                        assert!(drive_force.as_mut().poll(cx).is_pending(), "force admission ended before the test window");
+                        reached.as_mut().poll(cx)
+                    }),
+                )
+                .await
+                .expect("forceStart reached the cancellation window before the alias begins");
+                (None, true)
+            } else {
+                let result = match futures::poll!(tokio::task::unconstrained(force_submission.as_mut())) {
+                    Poll::Ready(result) => Some(result),
+                    Poll::Pending => None,
+                };
+                let reached = futures::poll!(std::pin::pin!(hook.reached.notified())).is_ready();
+                (result, reached)
+            };
+
+            // A correct force admission may wait for alias publication or
+            // reserve its id. Do not wait for either admission while a test
+            // hook is blocking the other.
+            if let Some(hook) = &alias_hook {
+                hook.release.notify_one();
+            }
+            if alias_result.is_none()
+                && let Poll::Ready(result) = futures::poll!(tokio::task::unconstrained(alias_submission.as_mut()))
+            {
+                alias_result = Some(result);
+            }
+            hook.release.notify_one();
+            let (alias_result, force_result) = tokio::time::timeout(Duration::from_secs(5), async {
+                tokio::join!(
+                    async {
+                        match alias_result {
+                            Some(result) => result,
+                            None => alias_submission.as_mut().await,
+                        }
+                    },
+                    async {
+                        match force_result {
+                            Some(result) => result,
+                            None => force_submission.as_mut().await,
+                        }
+                    },
+                )
+            })
+            .await
+            .expect("both admissions finish after releasing the test gates");
+            FORCE_CANCELLATION_HOOKS
+                .lock()
+                .expect("remove force hook")
+                .remove(&request_id);
+            ALIAS_PUBLICATION_HOOKS.lock().expect("remove alias hook").remove(&request_id);
+            let alias_result = alias_result.expect("alias admission returns a decision");
+            let force_receipt = force_result.expect("force admission returns a decision");
+            let forced_queued = manager
+                .heal_queue
+                .lock()
+                .await
+                .requests()
+                .any(|queued| queued.id == request_id);
+            let canonical = manager.canonical_task_id(&request_id).await;
+            let force_status = manager.get_task_status_for_path("bucket/scope", &request_id).await;
+            let alias_status = manager.get_task_status_for_path("other-bucket/object", &request_id).await;
+            let original_status = manager.get_task_status_for_path("other-bucket/object", &original_id).await;
+            let force_pending = matches!(&force_status, Ok(HealTaskStatus::Pending));
+            let force_missing = matches!(&force_status, Err(Error::TaskNotFound { task_id }) if task_id == &request_id);
+            let force_invalid = matches!(&force_status, Err(Error::InvalidClientToken));
+            let alias_pending = matches!(&alias_status, Ok(HealTaskStatus::Pending));
+            let alias_invalid = matches!(&alias_status, Err(Error::InvalidClientToken));
+            let original_pending = matches!(&original_status, Ok(HealTaskStatus::Pending));
+            let alias_won = alias_result == HealAdmissionResult::Merged
+                && force_receipt.result == HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+                && !forced_queued
+                && canonical == original_id
+                && alias_pending
+                && force_missing;
+            let force_won = force_receipt.result == HealAdmissionResult::Accepted
+                && alias_result == HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+                && forced_queued
+                && canonical == request_id
+                && force_pending
+                && alias_invalid;
+            let valid = force_receipt.task_id == request_id && original_pending && (alias_won || force_won);
+            println!(
+                "alias_force_observed source={} timing={} alias_publication_reached={} reached_force_window={} force={} alias={} forced_queued={} canonical_original={} canonical_force={} force_pending={} force_missing={} force_invalid={} alias_pending={} alias_invalid={} original_pending={} receipt_matches={} valid={}",
+                source.as_str(),
+                timing,
+                alias_hook.is_some(),
+                reached_force_window,
+                force_receipt.result.result_label(),
+                alias_result.result_label(),
+                forced_queued,
+                canonical == original_id,
+                canonical == request_id,
+                force_pending,
+                force_missing,
+                force_invalid,
+                alias_pending,
+                alias_invalid,
+                original_pending,
+                force_receipt.task_id == request_id,
+                valid
+            );
+            observed.push((source, timing, valid));
+        }
+    }
+    assert!(
+        observed.iter().all(|(_, _, valid)| *valid),
+        "each id must have exactly one admitted owner and route to that owner: {observed:?}"
+    );
+}
+
+#[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);

crates/heal/src/heal/manager/tests/queued_representative.rs

@@ -0,0 +1,398 @@
+// Copyright 2026 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.
+
+use super::*;
+
+fn object_request(source: HealRequestSource, priority: HealPriority, force_start: bool) -> HealRequest {
+    let mut request = HealRequest::new(
+        HealType::Object {
+            bucket: "canonical-owner".to_owned(),
+            object: "object".to_owned(),
+            version_id: None,
+        },
+        HealOptions::default(),
+        priority,
+    );
+    request.source = source;
+    request.force_start = force_start;
+    request
+}
+
+async fn scanner_receipt_tracks_queued_owner(use_fairness: bool) {
+    let manager = HealManager::new_without_root_recovery_for_test(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 16,
+            low_priority_merge_enable: true,
+            overlap_policy: HealOverlapPolicy::Merge,
+            ..HealConfig::default()
+        }),
+    );
+
+    if use_fairness {
+        let mut queue = manager.heal_queue.lock().await;
+        for index in 0..4 {
+            let request = bucket_request(&format!("internal-{index}"), HealPriority::Normal, HealRequestSource::Internal);
+            assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+        }
+        for _ in 0..4 {
+            let (request, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
+            assert_eq!(request.expect("internal work should run").source, HealRequestSource::Internal);
+        }
+    }
+
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    for request in [first, second] {
+        assert_eq!(
+            manager
+                .submit_heal_request_with_receipt(request)
+                .await
+                .expect("admit task")
+                .result,
+            HealAdmissionResult::Accepted
+        );
+    }
+
+    let first = {
+        let mut queue = manager.heal_queue.lock().await;
+        let (selected, skipped) = if use_fairness {
+            queue.pop_runnable_with_fairness(|_| true, |_| None)
+        } else {
+            queue.pop_runnable_with_skips(|_| true, |_| None)
+        };
+        assert!(skipped.is_empty());
+        let selected = selected.expect("the first same-priority task should run");
+        assert_eq!(selected.id, first_id);
+        assert!(queue.contains_request_id(&second_id));
+        assert_eq!(
+            queue.queued_request_id_for_dedup_key(&key),
+            Some(first_id.as_str()),
+            "the existing representative for other sources must remain unchanged"
+        );
+        selected
+    };
+    manager
+        .active_heals
+        .lock()
+        .await
+        .insert(first_id.clone(), Arc::new(HealTask::from_request(first, Arc::new(MockStorage))));
+
+    let active = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
+        .await
+        .expect("scanner admission while the first task is active");
+    assert_eq!(active.result, HealAdmissionResult::Merged);
+    assert_eq!(active.task_id, first_id, "active ownership keeps precedence");
+
+    let completed_task = manager
+        .active_heals
+        .lock()
+        .await
+        .remove(&first_id)
+        .expect("first task should leave active ownership");
+    let mut retained = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await;
+    Arc::make_mut(retained.outcome.as_mut().expect("retained task outcome")).finish(None);
+    manager
+        .completed_heals
+        .lock()
+        .await
+        .insert(first_id.clone(), Arc::new(retained));
+    let queued = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
+        .await
+        .expect("scanner admission after the first task leaves active state");
+    assert_eq!(queued.result, HealAdmissionResult::Merged);
+    assert_eq!(queued.task_id, second_id, "the receipt must name the remaining queued task");
+}
+
+#[tokio::test]
+async fn scanner_receipt_tracks_queued_owner_after_normal_dequeue() {
+    scanner_receipt_tracks_queued_owner(false).await;
+}
+
+#[tokio::test]
+async fn scanner_receipt_tracks_queued_owner_after_fair_dequeue() {
+    scanner_receipt_tracks_queued_owner(true).await;
+}
+
+#[tokio::test]
+async fn scanner_receipt_keeps_queued_owner_when_a_different_forced_task_runs() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
+    let first_id = first.id.clone();
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
+        let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert_eq!(selected.expect("higher-priority forced task should run").id, second_id);
+    }
+    let receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("scanner admission after a non-representative task leaves the queue");
+    assert_eq!(receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(receipt.task_id, first_id);
+}
+
+#[tokio::test]
+async fn scanner_receipt_reselects_after_the_queued_owner_is_cancelled() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
+    let third_id = third.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        for request in [first, second, third] {
+            assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+        }
+        let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert_eq!(selected.expect("first task should run").id, first_id);
+    }
+    let first_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("first scanner admission");
+    assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
+    assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
+    let remaining = if first_receipt.task_id == second_id {
+        third_id
+    } else {
+        second_id
+    };
+    manager.heal_queue.lock().await.remove_request_id(&first_receipt.task_id);
+
+    let next_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("scanner admission after the queued owner is removed");
+    assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(next_receipt.task_id, remaining);
+}
+
+#[tokio::test]
+async fn scanner_receipt_reselects_after_multiple_forced_dequeues() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
+    let third_id = third.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        for request in [first, second, third] {
+            assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+        }
+        let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert_eq!(selected.expect("first task should run").id, first_id);
+    }
+    let first_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("first scanner admission");
+    assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
+    assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
+    let remaining = if first_receipt.task_id == second_id {
+        third_id
+    } else {
+        second_id
+    };
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_receipt.task_id, |_| None);
+        assert_eq!(selected.expect("selected queued owner should run").id, first_receipt.task_id);
+    }
+    let next_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("scanner admission after another queued owner starts");
+    assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(next_receipt.task_id, remaining);
+}
+
+#[test]
+fn scanner_owner_index_survives_a_single_remaining_holder_and_new_forced_work() {
+    let mut queue = PriorityHealQueue::new();
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
+    let third_id = third.id.clone();
+
+    assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+    assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_)));
+    assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
+    assert_eq!(queue.pop().expect("first queued task").id, first_id);
+    assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(first_id.as_str()));
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
+
+    assert_eq!(queue.push(third), QueuePushOutcome::Accepted);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
+    assert_eq!(queue.pop().expect("second queued task").id, second_id);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
+}
+
+#[test]
+fn scanner_owner_index_distinguishes_holders_with_the_same_request_id() {
+    let mut queue = PriorityHealQueue::new();
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let shared_id = first.id.clone();
+    let mut second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    second.id.clone_from(&shared_id);
+    let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
+    let third_id = third.id.clone();
+    for request in [first, second, third] {
+        assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+    }
+
+    assert_eq!(queue.pop().expect("first holder").id, shared_id);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(shared_id.as_str()));
+    assert_eq!(queue.remove_request_id(&shared_id).expect("second holder").id, shared_id);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
+    assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
+}
+
+#[test]
+fn scanner_owner_index_tracks_bulk_removal_and_key_release() {
+    let mut queue = PriorityHealQueue::new();
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::Low, true);
+    let third_id = third.id.clone();
+    for request in [first, second, third] {
+        assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+    }
+
+    let removed = queue.remove_matching(|request| request.id == first_id || request.id == second_id);
+    assert_eq!(removed.len(), 2);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
+    assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
+    assert_eq!(queue.remove_matching(|request| request.id == third_id).len(), 1);
+    assert!(!queue.dedup_keys.contains_key(&key));
+
+    let next = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let next_id = next.id.clone();
+    assert_eq!(queue.push(next), QueuePushOutcome::Accepted);
+    assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_)));
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(next_id.as_str()));
+}
+
+#[test]
+fn scanner_owner_index_tracks_displacement_without_changing_legacy_reselection() {
+    let mut queue = PriorityHealQueue::new();
+    let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
+    let first_id = first.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true);
+    let second_id = second.id.clone();
+    assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+    assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
+
+    let incoming = bucket_request("displacement", HealPriority::Urgent, HealRequestSource::Internal);
+    let displaced = queue
+        .push_displacing_lower_priority(incoming)
+        .expect("lower priority task should be displaced");
+    assert_eq!(displaced.id, first_id);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
+    assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
+}
+
+#[tokio::test]
+async fn scanner_receipt_rejects_a_missing_or_empty_holder_index() {
+    for missing in [true, false] {
+        let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+        let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+        let key = PriorityHealQueue::make_dedup_key(&first);
+        let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+        {
+            let mut queue = manager.heal_queue.lock().await;
+            assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+            assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
+            let entry = queue.dedup_keys.get_mut(&key).expect("dedup key should exist");
+            entry.scanner_queued_owners = if missing {
+                ScannerQueuedOwners::Single(0)
+            } else {
+                ScannerQueuedOwners::Indexed(std::collections::BTreeMap::new())
+            };
+        }
+
+        for leave_one in [false, true] {
+            if leave_one {
+                manager.heal_queue.lock().await.pop().expect("one holder should leave");
+            }
+            let request = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+            let request_id = request.id.clone();
+            let receipt = manager
+                .submit_heal_request_with_receipt(request)
+                .await
+                .expect("admission should fail closed");
+            assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
+            assert_eq!(receipt.task_id, request_id);
+        }
+    }
+}
+
+#[test]
+fn scanner_and_legacy_owners_keep_their_distinct_reselection_orders() {
+    let mut queue = PriorityHealQueue::new();
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let second = object_request(HealRequestSource::Admin, HealPriority::Low, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::Normal, true);
+    let third_id = third.id.clone();
+    for request in [first, second, third] {
+        assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+    }
+
+    assert_eq!(queue.remove_request_id(&first_id).expect("first holder should be removed").id, first_id);
+    assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
+    assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
+}
+
+#[test]
+fn scanner_owner_index_tracks_a_series_of_forced_holders() {
+    let mut queue = PriorityHealQueue::new();
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let mut ids = vec![first.id.clone()];
+    assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+    for _ in 0..128 {
+        let request = object_request(HealRequestSource::Internal, HealPriority::High, true);
+        ids.push(request.id.clone());
+        assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+    }
+    for id in &ids {
+        assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(id.as_str()));
+        assert_eq!(&queue.pop().expect("indexed holder should remain queued").id, id);
+    }
+    assert!(!queue.dedup_keys.contains_key(&key));
+}