ci: accept interleaved test tracing output

ci: scope scanner read repair preflight
test(scanner): cover scoped read repair admission
2026-10-28 23:07:39 +00:00 · 2026-10-28 09:39:48 +08:00 · 2026-10-28 09:27:28 +08:00 · 2026-10-28 09:26:19 +08:00 · 2026-10-28 09:11:45 +08:00 · 2026-10-28 09:08:37 +08:00
7 changed files with 1281 additions and 336 deletions
@@ -1,224 +1,119 @@
-# 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: Scanner Read Repair Queue Preflight

 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:
+    inputs:
+      expected_head:
+        description: Full SHA of this workflow commit
+        required: true
+        type: string

 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-test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 120
+    env:
+      CARGO_TERM_COLOR: never
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
+      NO_PROXY: 127.0.0.1,localhost
    steps:
-      - name: Checkout repository
+      - name: Checkout candidate
        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
        with:
+          fetch-depth: 6
          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
-    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
+      - name: Verify candidate and workflow head
+        shell: bash
        env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
+          EXPECTED_HEAD: ${{ inputs.expected_head }}
        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
+          actual_head=$(git rev-parse HEAD)
+          source=$(git rev-parse 'e8382ebefc8f6feac681fbbd67d30fefdd3bf46b^{commit}')
+          test_base=$(git rev-parse "${source}^")
+          main_base=$(git rev-parse "${source}~3")
+          base_tree=$(git rev-parse "${test_base}^{tree}")
+          candidate_tree=$(git rev-parse "${source}^{tree}")
+          workflow_tree=$(git rev-parse "${actual_head}^{tree}")
+          test_diff=$(git diff --full-index --binary "$test_base" "$source" | sha256sum | cut -d' ' -f1)
+          combined_diff=$(git diff --full-index --binary "$main_base" "$source" | sha256sum | cut -d' ' -f1)
+          source_files=$(git diff --name-only "$test_base" "$source")
+          workflow_files=$(git diff --name-only "$source" "$actual_head")
+          printf 'head=%s source=%s test_base=%s main_base=%s base_tree=%s candidate_tree=%s workflow_tree=%s test_diff=%s combined_diff=%s\n' \
+            "$actual_head" "$source" "$test_base" "$main_base" "$base_tree" "$candidate_tree" "$workflow_tree" "$test_diff" "$combined_diff"
+          printf 'source_files=%s workflow_files=%s\n' "$source_files" "$workflow_files"
+          [[ "$EXPECTED_HEAD" =~ ^[0-9a-f]{40}$ ]]
+          [[ "$actual_head" == "$EXPECTED_HEAD" && "$actual_head" == "$GITHUB_SHA" ]]
+          git merge-base --is-ancestor "$source" "$actual_head"
+          [[ "$source" == e8382ebefc8f6feac681fbbd67d30fefdd3bf46b ]]
+          [[ "$test_base" == 41d06b4d7a1eee1cec4db8df65bc741b834831eb ]]
+          [[ "$main_base" == 7a3bf8cdbd781116a0970585b995ade668c0e662 ]]
+          [[ "$base_tree" == 01b101acf0119bec8243cab1556cc0218170b6bf ]]
+          [[ "$candidate_tree" == 7cb63e65ca31372a85fd06657608c24cc3b8d063 ]]
+          [[ "$test_diff" == d7dd41fa96e9ee83e7a3c64a0b27efbe2265771e2b026e29e12e94393fab9dd2 ]]
+          [[ "$combined_diff" == 8b898f5821031f5e567c0a0ab51c4039c39467c42e32c54d916aca20e59eba2d ]]
+          [[ "$source_files" == crates/scanner/tests/read_repair_scanner_queue.rs ]]
+          [[ "$workflow_files" == .github/workflows/windows-filesystem.yml ]]
+          [[ -z "$(git status --porcelain)" ]]
+
+      - name: Setup Rust
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: ci-e2e-distributed-hosted
+          cache-save-if: 'false'
+          install-build-packaging-tools: 'false'
+          install-test-tools: 'false'
+
+      # The single integration test uses local disks and an in-process channel.
+      - name: Run the named Scanner and ReadRepair test
+        id: scoped_test
+        shell: bash
+        env:
+          CARGO_BUILD_JOBS: '2'
+          CARGO_INCREMENTAL: '0'
+          CARGO_PROFILE_TEST_DEBUG: '0'
+          RUST_BACKTRACE: '1'
+          TEST_LOG: ${{ runner.temp }}/read-repair-scanner-test.log
+          TEST_NAME: dangling_metadata_read_repair_and_scanner_have_distinct_queued_owners
+        run: |
+          set -euo pipefail
+          validate_test_log() {
+            local test_status="$1" log_file="$2" test_name="$3"
+            if (( test_status != 0 )); then
+              printf 'cargo test exited %s\n' "$test_status" >&2
+              return "$test_status"
+            fi
+            awk -v name="$test_name" '
+              $0 == "running 1 test" { runs++ }
+              index($0, "test " name " ...") == 1 {
+                suffix = substr($0, length("test " name " ...") + 1)
+                if (suffix == "" || suffix ~ /^[[:space:]]/) names++
+              }
+              /^test result: ok[.] 1 passed; 0 failed; 0 ignored; [0-9]+ measured; [0-9]+ filtered out; finished in / { passed++ }
+              END {
+                printf "Harness markers: running=%d named=%d passed=%d\n", runs, names, passed
+                if (runs != 1 || names != 1 || passed != 1) exit 1
+              }
+            ' "$log_file"
          }
+          [[ ! -e "$TEST_LOG" ]]
+          set +e
+          cargo test --locked -p rustfs-scanner --test read_repair_scanner_queue -- "$TEST_NAME" --exact --test-threads=1 >"$TEST_LOG" 2>&1
+          status=$?
+          set -e
+          cat "$TEST_LOG"
+          validate_test_log "$status" "$TEST_LOG" "$TEST_NAME"
+
+      - name: Upload test log
+        if: ${{ always() && (steps.scoped_test.outcome == 'success' || steps.scoped_test.outcome == 'failure') }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: scanner-read-repair-queue-${{ github.run_id }}
+          path: ${{ runner.temp }}/read-repair-scanner-test.log
+          retention-days: 7
+          if-no-files-found: error
@@ -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()
@@ -2282,13 +2296,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 +2319,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 +2537,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);
@@ -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);
@@ -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};
@@ -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);
@@ -0,0 +1,524 @@
+// 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
+}
+
+fn scoped_object_request(
+    source: HealRequestSource,
+    version_id: &str,
+    pool_index: Option<usize>,
+    set_index: Option<usize>,
+    force_start: bool,
+) -> HealRequest {
+    let mut request = object_request(source, HealPriority::Low, force_start);
+    let HealType::Object { version_id: version, .. } = &mut request.heal_type else {
+        unreachable!("object_request always constructs an object heal")
+    };
+    *version = Some(version_id.to_owned());
+    request.options.pool_index = pool_index;
+    request.options.set_index = set_index;
+    request.options.recreate_missing = source != HealRequestSource::Scanner;
+    request
+}
+
+#[tokio::test]
+async fn scanner_receipt_tracks_scoped_read_repair_forced_holders() {
+    let manager = HealManager::new_without_root_recovery_for_test(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 16,
+            low_priority_merge_enable: true,
+            ..HealConfig::default()
+        }),
+    );
+    let version = "00000000-0000-4000-8000-000000000001";
+    let original = scoped_object_request(HealRequestSource::ReadRepair, version, Some(0), Some(1), false);
+    let original_id = original.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&original);
+    let admitted = manager
+        .submit_heal_request_with_receipt(original)
+        .await
+        .expect("admit read-repair owner");
+    assert_eq!(admitted.result, HealAdmissionResult::Accepted);
+    assert_eq!(admitted.task_id, original_id);
+
+    let receipt = manager
+        .submit_heal_request_with_receipt(scoped_object_request(HealRequestSource::Scanner, version, Some(0), Some(1), false))
+        .await
+        .expect("scanner merges into the scoped read-repair owner");
+    assert_eq!(receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(receipt.task_id, original_id);
+
+    let first_forced = scoped_object_request(HealRequestSource::Internal, version, Some(0), Some(1), true);
+    let first_forced_id = first_forced.id.clone();
+    let second_forced = scoped_object_request(HealRequestSource::Internal, version, Some(0), Some(1), true);
+    let second_forced_id = second_forced.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        assert_eq!(queue.push(first_forced), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(second_forced), QueuePushOutcome::Accepted);
+        let (selected, skipped) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert!(skipped.is_empty());
+        assert_eq!(selected.expect("read-repair owner should leave first").id, original_id);
+        assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(original_id.as_str()));
+    }
+
+    let receipt = manager
+        .submit_heal_request_with_receipt(scoped_object_request(HealRequestSource::Scanner, version, Some(0), Some(1), false))
+        .await
+        .expect("scanner merges into the remaining owner");
+    assert_eq!(receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(receipt.task_id, first_forced_id);
+
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_forced_id, |_| None);
+        assert_eq!(selected.expect("first forced holder should leave").id, first_forced_id);
+        assert!(queue.contains_request_id(&second_forced_id));
+    }
+    let receipt = manager
+        .submit_heal_request_with_receipt(scoped_object_request(HealRequestSource::Scanner, version, Some(0), Some(1), false))
+        .await
+        .expect("scanner merges into the last queued holder");
+    assert_eq!(receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(receipt.task_id, second_forced_id);
+}
+
+#[tokio::test]
+async fn scanner_receipt_does_not_merge_read_repair_with_a_different_scope_or_version() {
+    let version = "00000000-0000-4000-8000-000000000001";
+    let other_version = "00000000-0000-4000-8000-000000000002";
+    for (candidate_version, pool, set) in [
+        (version, None, None),
+        (version, Some(1), Some(1)),
+        (version, Some(0), Some(2)),
+        (other_version, Some(0), Some(1)),
+    ] {
+        let manager = HealManager::new_without_root_recovery_for_test(
+            Arc::new(MockStorage),
+            Some(HealConfig {
+                queue_size: 16,
+                low_priority_merge_enable: true,
+                ..HealConfig::default()
+            }),
+        );
+        let original = scoped_object_request(HealRequestSource::ReadRepair, version, Some(0), Some(1), false);
+        let original_id = original.id.clone();
+        let initial = manager
+            .submit_heal_request_with_receipt(original)
+            .await
+            .expect("admit read-repair owner");
+        assert_eq!(initial.result, HealAdmissionResult::Accepted);
+        assert_eq!(initial.task_id, original_id);
+
+        let candidate = scoped_object_request(HealRequestSource::Scanner, candidate_version, pool, set, false);
+        let candidate_id = candidate.id.clone();
+        let receipt = manager
+            .submit_heal_request_with_receipt(candidate)
+            .await
+            .expect("admit scanner with a distinct scope or version");
+        assert_eq!(
+            receipt.result,
+            HealAdmissionResult::Accepted,
+            "version={candidate_version}, pool={pool:?}, set={set:?}"
+        );
+        assert_eq!(receipt.task_id, candidate_id);
+        let queue = manager.heal_queue.lock().await;
+        assert!(queue.contains_request_id(&original_id));
+        assert!(queue.contains_request_id(&candidate_id));
+    }
+}
+
+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));
+}
@@ -0,0 +1,297 @@
+// 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.
+
+//! One process exercises dangling metadata, scanner disk reads, and heal admission.
+
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use rustfs_ecstore::api::{
+    capacity::path2_bucket_object_with_base_path,
+    config::storageclass::{CLASS_STANDARD, DEFAULT_KVS, STANDARD, lookup_config},
+    disk::{DeleteOptions, DiskAPI, DiskError, ReadOptions, STORAGE_FORMAT_FILE},
+    object::{ObjectOptions, PutObjReader, WriteCompletion},
+};
+use rustfs_filemeta::FileInfo;
+use rustfs_heal::{
+    HealChannelProcessor,
+    heal::{
+        manager::{HealConfig, HealManager},
+        storage::ECStoreHealStorage,
+        task::HealTaskStatus,
+    },
+};
+use rustfs_heal_contracts::heal_channel::{
+    HealChannelCommand, HealChannelResponse, HealRequestSource, init_heal_channels, subscribe_heal_responses,
+};
+use rustfs_scanner::{scanner_folder::ScannerItem, scanner_heal_observation_status, scanner_io::ScannerIODisk};
+use rustfs_storage_api::{ObjectIO, ObjectOperations};
+use rustfs_test_utils::TestECStoreEnv;
+use time::OffsetDateTime;
+use tokio::{sync::broadcast, time::timeout};
+use uuid::Uuid;
+
+const DEADLINE: Duration = Duration::from_secs(60);
+
+async fn accepted_response(responses: &mut broadcast::Receiver<HealChannelResponse>, id: &str) {
+    let response = timeout(DEADLINE, responses.recv())
+        .await
+        .expect("heal admission response timed out")
+        .expect("heal admission response channel closed or lagged");
+    assert_eq!(response.request_id, id, "admission must belong to the forwarded request");
+    assert!(response.success, "heal admission failed: {:?}", response.error);
+    assert_eq!(response.data.as_deref(), Some(&b"admission=accepted,reason=none"[..]));
+}
+
+async fn wait_for_scanner_pending(expected: usize) {
+    let expected_candidates = u64::try_from(expected).expect("bounded scanner candidate count");
+    timeout(DEADLINE, async {
+        loop {
+            let status = scanner_heal_observation_status();
+            assert_ne!(status.state, "stopped", "scanner observer stopped");
+            assert!(status.unknown_candidates.is_empty(), "scanner observer lost a candidate: {status:?}");
+            if status.state == "active" {
+                assert_eq!(status.lost_completion_notifications, Some(0));
+                assert_eq!(status.terminal_task_observations_total, Some(0));
+                if status.retained_pending_tasks == Some(expected)
+                    && status.retained_pending_candidates == Some(expected_candidates)
+                    && status.queued_admissions == Some(0)
+                {
+                    break;
+                }
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("scanner receipt observation timed out: {:?}", scanner_heal_observation_status()));
+}
+
+#[tokio::test]
+async fn dangling_metadata_read_repair_and_scanner_have_distinct_queued_owners() {
+    // This test binary owns the process-wide channels and storage-class config.
+    temp_env::async_with_vars(
+        [
+            ("RUSTFS_STORAGE_CLASS_STANDARD", Some("EC:2")),
+            ("RUSTFS_STORAGE_CLASS_RRS", Some("EC:1")),
+        ],
+        async {
+            let root = tempfile::tempdir().expect("create isolated six-disk fixture");
+            let env = TestECStoreEnv::builder().disk_count(6).base_dir(root.path()).build().await;
+            let mut kvs = (*DEFAULT_KVS).clone();
+            kvs.insert(CLASS_STANDARD.to_owned(), "EC:2".to_owned());
+            let storage_class = lookup_config(&kvs, 6).expect("configure EC 4+2");
+            assert_eq!(storage_class.get_parity_for_sc(STANDARD), Some(2));
+            env.ecstore.set_storage_class(storage_class);
+
+            let bucket = "read-repair-scanner";
+            let object = "torn/latest";
+            env.make_bucket(bucket, true).await;
+            let sets = env.ecstore.all_set_disks();
+            assert_eq!(sets.len(), 1, "fixture must have one erasure set");
+            let set = &sets[0];
+            assert_eq!((set.pool_index, set.set_index), (0, 0));
+            let options = |seconds| ObjectOptions {
+                versioned: true,
+                mod_time: Some(OffsetDateTime::from_unix_timestamp(seconds).expect("fixture timestamp")),
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            };
+            let v1 = set
+                .put_object(
+                    bucket,
+                    object,
+                    &mut PutObjReader::from_vec(b"version one".to_vec()),
+                    &options(1_700_000_000),
+                )
+                .await
+                .expect("write committed first version")
+                .version_id
+                .expect("first version id");
+            let v2 = set
+                .put_object(
+                    bucket,
+                    object,
+                    &mut PutObjReader::from_vec(b"version two".to_vec()),
+                    &options(1_700_000_060),
+                )
+                .await
+                .expect("write second version")
+                .version_id
+                .expect("second version id");
+            assert_ne!(v1, v2);
+
+            let disks = set.disks.read().await.clone();
+            assert_eq!(disks.len(), 6);
+            for disk in &disks[3..] {
+                let disk = disk.as_ref().expect("all fixture disks are online");
+                let mut remove = FileInfo {
+                    volume: bucket.to_owned(),
+                    name: object.to_owned(),
+                    version_id: Some(v2),
+                    ..Default::default()
+                };
+                remove.set_skip_tier_free_version();
+                disk.delete_version(bucket, object, remove, false, DeleteOptions::default())
+                    .await
+                    .expect("remove second version from one fixture disk");
+            }
+            for (index, disk) in disks.iter().enumerate() {
+                let disk = disk.as_ref().expect("all fixture disks are online");
+                disk.read_version("", bucket, object, &v1.to_string(), &ReadOptions::default())
+                    .await
+                    .expect("first version remains on every disk");
+                let second = disk
+                    .read_version("", bucket, object, &v2.to_string(), &ReadOptions::default())
+                    .await;
+                if index < 3 {
+                    let info = second.expect("second version remains on three disks");
+                    assert_eq!((info.erasure.data_blocks, info.erasure.parity_blocks), (4, 2));
+                } else {
+                    let error = second.expect_err("second version must be absent from this disk");
+                    assert!(DiskError::is_err_version_not_found(&error), "unexpected disk error: {error:?}");
+                }
+            }
+
+            let manager = Arc::new(HealManager::new(
+                Arc::new(ECStoreHealStorage::new(env.ecstore.clone())),
+                Some(HealConfig {
+                    enable_auto_heal: false,
+                    low_priority_merge_enable: true,
+                    mainline_throttle_enable: false,
+                    queue_size: 16,
+                    ..Default::default()
+                }),
+            ));
+            // No manager scheduler runs: accepted requests remain in the real queue.
+            let (mut commands, mut receipts) = init_heal_channels().expect("initialize process-local heal channels");
+            let (command_tx, command_rx) = tokio::sync::mpsc::unbounded_channel();
+            let (receipt_tx, receipt_rx) = tokio::sync::mpsc::unbounded_channel();
+            let processor_manager = manager.clone();
+            let processor = tokio::spawn(async move {
+                HealChannelProcessor::new(processor_manager)
+                    .start_with_receipts(command_rx, receipt_rx)
+                    .await
+            });
+            let mut responses = subscribe_heal_responses();
+
+            // The dangling-latest read uses submit_read_repair_heal, whose MRF intent is None.
+            let tagged = set
+                .put_object_tags(bucket, object, "stage=committed", &ObjectOptions::default())
+                .await
+                .expect("latest tagging skips the dangling second version");
+            assert_eq!(tagged.version_id, Some(v1));
+
+            // Read-repair submits asynchronously, so wait for its actual request and admission.
+            let command = timeout(DEADLINE, commands.recv())
+                .await
+                .expect("read-repair request timed out")
+                .expect("legacy heal command channel closed");
+            let read_repair_id = match &command {
+                HealChannelCommand::Start { request, .. } => {
+                    assert_eq!(request.source, HealRequestSource::ReadRepair);
+                    assert_eq!(request.bucket, bucket);
+                    assert_eq!(request.object_prefix.as_deref(), Some(object));
+                    assert_eq!(request.object_version_id.as_deref(), Some(v2.to_string().as_str()));
+                    assert_eq!(request.pool_index, Some(set.pool_index));
+                    assert_eq!(request.set_index, Some(set.set_index));
+                    request.id.clone()
+                }
+                _ => panic!("expected read-repair start command"),
+            };
+            command_tx.send(command).expect("forward original read-repair command once");
+            accepted_response(&mut responses, &read_repair_id).await;
+
+            // This isolated process has made no scanner admission before the disk read.
+            let initial = scanner_heal_observation_status();
+            assert_eq!(initial.state, "not_started");
+            assert_eq!(initial.retained_pending_tasks, None);
+            assert_eq!(initial.retained_pending_candidates, None);
+
+            let scan_disk = disks[0].as_ref().expect("first fixture disk").clone();
+            let disk_path = scan_disk.path();
+            let metadata_path = disk_path.join(bucket).join(object).join(STORAGE_FORMAT_FILE);
+            let (_, scanner_path) = path2_bucket_object_with_base_path(
+                disk_path.to_str().expect("fixture disk path is UTF-8"),
+                metadata_path.to_str().expect("fixture metadata path is UTF-8"),
+            );
+            let scanner = ScannerItem {
+                path: scanner_path,
+                bucket: bucket.to_owned(),
+                prefix: object.to_owned(),
+                object_name: STORAGE_FORMAT_FILE.to_owned(),
+                file_type: tokio::fs::metadata(&metadata_path)
+                    .await
+                    .expect("stat surviving real metadata")
+                    .file_type(),
+                lifecycle: None,
+                object_lock: None,
+                replication: None,
+                heal_enabled: true,
+                heal_bitrot: false,
+                debug: false,
+            };
+            let scan = tokio::spawn(async move { scan_disk.get_size_with_tier_names(scanner, &[]).await });
+            let mut scanner_ids = HashMap::new();
+            for index in 0..2 {
+                let command = timeout(DEADLINE, receipts.recv())
+                    .await
+                    .expect("scanner request timed out")
+                    .expect("scanner heal receipt channel closed");
+                let request = &command.request;
+                assert_eq!(request.source, HealRequestSource::Scanner);
+                assert_eq!(request.bucket, bucket);
+                assert_eq!(request.object_prefix.as_deref(), Some(object));
+                assert_eq!((request.pool_index, request.set_index), (None, None));
+                let version = Uuid::parse_str(request.object_version_id.as_deref().expect("scanner version id"))
+                    .expect("scanner version id is a UUID");
+                assert!(version == v1 || version == v2, "unexpected scanner version {version}");
+                let id = request.id.clone();
+                assert!(scanner_ids.insert(version, id.clone()).is_none(), "duplicate scanner version");
+                receipt_tx.send(command).expect("forward original scanner command once");
+                accepted_response(&mut responses, &id).await;
+                // Broadcast precedes the oneshot; wait until Scanner consumes each receipt.
+                wait_for_scanner_pending(index + 1).await;
+            }
+            timeout(DEADLINE, scan)
+                .await
+                .expect("scanner disk read timed out")
+                .expect("scanner task failed")
+                .expect("scanner read real object metadata");
+            assert!(
+                matches!(receipts.try_recv(), Err(tokio::sync::mpsc::error::TryRecvError::Empty)),
+                "scanner submitted an unexpected extra request"
+            );
+            assert!(
+                matches!(commands.try_recv(), Err(tokio::sync::mpsc::error::TryRecvError::Empty)),
+                "an unexpected legacy request was submitted"
+            );
+            assert_eq!(scanner_ids.len(), 2);
+            let scanner_v2_id = scanner_ids.get(&v2).expect("scanner submitted the dangling version");
+            assert_ne!(scanner_v2_id, &read_repair_id);
+            assert_eq!(manager.get_queue_length().await, 3);
+            for id in scanner_ids.values().chain(std::iter::once(&read_repair_id)) {
+                assert_eq!(manager.get_task_status(id).await.expect("queued owner status"), HealTaskStatus::Pending);
+            }
+
+            drop(command_tx);
+            drop(receipt_tx);
+            timeout(DEADLINE, processor)
+                .await
+                .expect("heal channel processor shutdown timed out")
+                .expect("heal channel processor task failed")
+                .expect("heal channel processor stopped");
+        },
+    )
+    .await;
+}
Author	SHA1	Message	Date
overtrue	8d9122942d	ci: accept interleaved test tracing output	2026-10-28 09:39:48 +08:00
overtrue	16ebff514c	ci: scope scanner read repair preflight	2026-10-28 09:27:28 +08:00
overtrue	e8382ebefc	test(scanner): cover scoped read repair admission	2026-10-28 09:26:19 +08:00
overtrue	41d06b4d7a	test(heal): cover scoped read repair ownership	2026-10-28 09:11:45 +08:00
overtrue	c96fc5cf26	fix(heal): preserve queued ownership under force admission	2026-10-28 09:08:37 +08:00