ci: require the named stack diagnostic test

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,135 @@
-# 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 heal ID stack diagnostic
 
 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: scanner-heal-id-reuse-stack-${{ 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
+  stack-diagnostic:
+    name: Diagnose one scanner test
+    runs-on: ubuntu-latest
+    timeout-minutes: 110
     steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+      - name: Checkout
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
         with:
+          fetch-depth: 5
           persist-credentials: false
+          clean: 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
+      - name: Verify diagnostic source
+        shell: bash
         run: |
-          # Map a drive letter to the temp drive's administrative share so the
-          # remote-volume probe also covers mapped network drives.
-          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
-          net use R: "\\localhost\$drive`$" /persistent:no
-          if ($LASTEXITCODE -eq 0) {
-            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
-          } else {
-            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
-          }
-          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
-          $status = $LASTEXITCODE
-          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
-          exit $status
+          set -euo pipefail
+          output_dir="$RUNNER_TEMP/heal-id-reuse-stack"
+          mkdir -p "$output_dir"
+          workflow_sha=$(git rev-parse HEAD)
+          previous_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}")
+          marker_diff=$(git diff --full-index --binary "$base_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          marker_files=$(git diff --name-only "$base_sha" "$source_sha")
+          workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
+          printf 'workflow=%s\nprevious_workflow=%s\nsource=%s\nbase=%s\nsource_tree=%s\nmarker_diff=%s\nmarker_files=%s\nworkflow_files=%s\n' \
+            "$workflow_sha" "$previous_workflow_sha" "$source_sha" "$base_sha" "$source_tree" "$marker_diff" "$marker_files" "$workflow_files" \
+            | tee "$output_dir/source.txt"
+          [[ "$workflow_sha" == "$GITHUB_SHA" ]]
+          [[ "$previous_workflow_sha" == 'dce2e710b5a6e7b6385749f898d0cb008b25b067' ]]
+          [[ "$source_sha" == '12b3ee9f948be35e54a0cd008aab50009f446ad3' ]]
+          [[ "$base_sha" == 'd31b9fe1b8777482a5ad90c46dedb27170c46d74' ]]
+          [[ "$source_tree" == '55c91e3a126628e309092353ceff9766c3ea06f7' ]]
+          [[ "$marker_diff" == '640e98d000acb90d09f15872423160cac05900f2bae6daf7f6f5e9a012954033' ]]
+          [[ "$marker_files" == 'crates/scanner/tests/heal_generation_observation.rs' ]]
+          [[ "$workflow_files" == '.github/workflows/windows-filesystem.yml' ]]
+          [[ -z "$(git status --porcelain)" ]]
 
-      - 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
+      - name: Install system dependencies
         run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
+          sudo apt-get update
+          sudo apt-get install -y build-essential pkg-config libssl-dev ripgrep
 
-      - name: Start server and wait for readiness
-        shell: pwsh
-        env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
+      - name: Install protoc
+        uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6
+        with:
+          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 single diagnostic test
+        shell: bash
         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
+          output_dir="$RUNNER_TEMP/heal-id-reuse-stack"
+          log="$output_dir/test.log"
+          test_name='overwritten_unversioned_generation_is_not_attributed_to_the_scanned_candidate'
+          if [[ -v RUST_MIN_STACK ]]; then
+            printf 'RUST_MIN_STACK is set; refusing a non-default-stack diagnostic\n' | tee "$output_dir/environment.txt"
+            exit 1
+          fi
+          printf 'RUST_MIN_STACK is unset\nshell_stack_limit_kib=%s\n' "$(ulimit -s)" | tee "$output_dir/environment.txt"
+          printf '%s\n' 'Diagnostic only: the test result is not a repair or full CI verdict.' | tee "$output_dir/scope.txt"
+          set +e
+          cargo test --locked -p rustfs-scanner --features test-util --test heal_generation_observation -- --exact "$test_name" --nocapture >"$log" 2>&1
+          test_status=$?
+          set -e
+          printf '%s\n' "$test_status" | tee "$output_dir/exit-code.txt"
+          tail -n 120 "$log" || printf '%s\n' 'Could not display the diagnostic log' >&2
+          if [[ "$test_status" -ne 0 ]]; then
+            exit "$test_status"
+          fi
+          set +e
+          python3 - "$log" "$test_name" >"$output_dir/selection.txt" 2>&1 <<'PY'
+          import re
+          import sys
+          from pathlib import Path
+
+          def valid_selection(output, expected_name):
+              output = re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", output)
+              runs = re.findall(r"(?m)^running (\d+) tests?[ \t]*\r?$", output)
+              names = re.findall(r"\btest ([^\s]+) \.\.\.", output)
+              summaries = re.findall(r"(?m)^test result: (ok|FAILED)\. (\d+) passed; (\d+) failed; (\d+) ignored;", output)
+              return runs == ["1"] and names == [expected_name] and summaries == [("ok", "1", "0", "0")]
+
+          if __name__ == "__main__":
+              output = Path(sys.argv[1]).read_text(encoding="utf-8", errors="replace")
+              if not valid_selection(output, sys.argv[2]):
+                  print("Test selection or passing summary did not match the named test")
+                  sys.exit(1)
+              print("The named test ran once and passed")
+          PY
+          selection_status=$?
+          set -e
+          cat "$output_dir/selection.txt"
+          exit "$selection_status"
+
+      - name: Preserve diagnostic output
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
+        with:
+          name: scanner-heal-id-reuse-stack-${{ github.run_id }}
+          path: ${{ runner.temp }}/heal-id-reuse-stack/
+          if-no-files-found: warn

crates/ecstore/src/set_disk/ops/heal.rs

@@ -1228,6 +1228,7 @@ impl SetDisks {
                         result.object_size =
                             ObjectInfo::from_file_info(&latest_meta, bucket, object, true).get_actual_size()? as usize;
                         result.resolved_version_id = Some(*latest_meta.version_id.unwrap_or_default().as_bytes());
+                        result.resolved_data_dir = selected_heal_data_dir(&latest_meta, opts, namespace_owner.is_some());
                         // Loop to find number of disks with valid data, per-drive
                         // data state and a list of outdated disks on which data needs
                         // to be healed.
@@ -3225,6 +3226,13 @@ impl SetDisks {
     }
 }
 
+fn selected_heal_data_dir(metadata: &FileInfo, opts: &HealOpts, has_namespace_owner: bool) -> Option<[u8; 16]> {
+    if !has_namespace_owner || opts.read_repair || opts.dry_run || metadata.deleted || metadata.is_remote() {
+        return None;
+    }
+    metadata.data_dir.filter(|id| !id.is_nil()).map(|id| *id.as_bytes())
+}
+
 fn finalize_object_heal_result(
     mut result: HealResultItem,
     mut error: Option<Error>,
@@ -3248,6 +3256,9 @@ fn finalize_object_heal_result(
             achieved: 0,
         });
     }
+    if error.is_some() {
+        result.resolved_data_dir = None;
+    }
     (result, error)
 }
 
@@ -3903,6 +3914,7 @@ mod heal_result_report_tests {
             metadata_verified: true,
             metadata_repair_verified: true,
             legacy_identity_unverified: true,
+            resolved_data_dir: Some([1; 16]),
             ..Default::default()
         };
 
@@ -3913,6 +3925,7 @@ mod heal_result_report_tests {
         assert!(!result.metadata_verified);
         assert!(!result.metadata_repair_verified);
         assert!(!result.legacy_identity_unverified);
+        assert_eq!(result.resolved_data_dir, None);
         assert!(absence.is_none());
         assert!(matches!(
             error,
@@ -3926,6 +3939,121 @@ mod heal_result_report_tests {
         ));
     }
 
+    #[test]
+    fn selected_heal_generation_requires_a_locked_local_object() {
+        let generation = Uuid::new_v4();
+        let mut metadata = FileInfo {
+            data_dir: Some(generation),
+            ..Default::default()
+        };
+        let opts = HealOpts::default();
+        assert_eq!(super::selected_heal_data_dir(&metadata, &opts, true), Some(*generation.as_bytes()));
+        assert_eq!(super::selected_heal_data_dir(&metadata, &opts, false), None);
+        assert_eq!(super::selected_heal_data_dir(&metadata, &HealOpts { dry_run: true, ..opts }, true), None);
+        assert_eq!(
+            super::selected_heal_data_dir(
+                &metadata,
+                &HealOpts {
+                    read_repair: true,
+                    ..opts
+                },
+                true
+            ),
+            None
+        );
+        metadata.data_dir = Some(Uuid::nil());
+        assert_eq!(super::selected_heal_data_dir(&metadata, &opts, true), None);
+        metadata.data_dir = None;
+        assert_eq!(super::selected_heal_data_dir(&metadata, &opts, true), None);
+        metadata.data_dir = Some(generation);
+        metadata.deleted = true;
+        assert_eq!(super::selected_heal_data_dir(&metadata, &opts, true), None);
+        metadata.deleted = false;
+        metadata.transition_status = TRANSITION_COMPLETE.to_owned();
+        assert_eq!(super::selected_heal_data_dir(&metadata, &opts, true), None);
+    }
+
+    #[test]
+    fn failed_heal_does_not_publish_selected_generation() {
+        let result = rustfs_madmin::heal_commands::HealResultItem {
+            resolved_data_dir: Some([1; 16]),
+            ..Default::default()
+        };
+        let mut absence = None;
+        let (result, error) =
+            super::finalize_object_heal_result(result, Some(Error::FileNotFound), &mut absence, false, "b", "o");
+        assert!(error.is_some());
+        assert_eq!(result.resolved_data_dir, None);
+    }
+
+    #[tokio::test]
+    async fn heal_result_observes_current_generation_after_unversioned_overwrite() {
+        let (_temp_dirs, _disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "heal-result-generation";
+        let object = "object.bin";
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create test bucket");
+
+        let put_opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let mut first = PutObjReader::from_vec(vec![0x11; 1024 * 1024]);
+        let first_info = set
+            .put_object(bucket, object, &mut first, &put_opts)
+            .await
+            .expect("write first generation");
+        let first_generation = first_info.data_dir.expect("first object data directory");
+        let heal_opts = HealOpts {
+            scan_mode: HealScanMode::Deep,
+            ..Default::default()
+        };
+        let (first_result, first_error) =
+            crate::storage_api_contracts::heal::HealOperations::heal_object(set.as_ref(), bucket, object, "", &heal_opts)
+                .await
+                .expect("heal first generation");
+        assert!(first_error.is_none());
+        assert_eq!(first_result.resolved_data_dir, Some(*first_generation.as_bytes()));
+
+        let mut second = PutObjReader::from_vec(vec![0x22; 1024 * 1024]);
+        let second_info = set
+            .put_object(bucket, object, &mut second, &put_opts)
+            .await
+            .expect("overwrite unversioned object");
+        let second_generation = second_info.data_dir.expect("second object data directory");
+        assert_ne!(first_generation, second_generation);
+        let (second_result, second_error) =
+            crate::storage_api_contracts::heal::HealOperations::heal_object(set.as_ref(), bucket, object, "", &heal_opts)
+                .await
+                .expect("heal second generation");
+        assert!(second_error.is_none());
+        assert_eq!(second_result.resolved_data_dir, Some(*second_generation.as_bytes()));
+        assert_ne!(second_result.resolved_data_dir, first_result.resolved_data_dir);
+
+        for opts in [
+            HealOpts {
+                no_lock: true,
+                ..heal_opts
+            },
+            HealOpts {
+                read_repair: true,
+                ..heal_opts
+            },
+            HealOpts {
+                dry_run: true,
+                ..heal_opts
+            },
+        ] {
+            let (result, error) =
+                crate::storage_api_contracts::heal::HealOperations::heal_object(set.as_ref(), bucket, object, "", &opts)
+                    .await
+                    .expect("observe heal without an eligible lock");
+            assert!(error.is_none());
+            assert_eq!(result.resolved_data_dir, None);
+        }
+    }
+
     #[test]
     fn absent_and_nil_selected_versions_are_the_same_null_identity() {
         let requested_nil = Uuid::nil().to_string();

crates/heal-contracts/src/outcome.rs

@@ -88,6 +88,12 @@ pub enum HealObjectDisposition {
 pub struct HealObjectReceipt {
     pub identity: HealObjectIdentity,
     pub disposition: HealObjectDisposition,
+    /// Storage-selected data-directory generation, when observed locally.
+    /// This identifies a version of the object, not verified payload contents.
+    pub resolved_data_dir: Option<Uuid>,
+    /// Storage reported authenticated Deep verification of the selected object's shards.
+    /// False means no such proof was retained, not that payload verification failed.
+    pub payload_integrity_verified: bool,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]

crates/heal/src/heal/channel.rs

@@ -990,6 +990,8 @@ mod tests {
                     HealObjectDisposition::Failed(HealFailureClass::RetryExhausted)
                 };
                 outcome.record(HealObjectOutcome {
+                    payload_integrity_verified: false,
+                    resolved_data_dir: None,
                     identity: HealObjectIdentity {
                         kind: HealObjectKind::Object,
                         bucket: "bucket".into(),
@@ -1074,6 +1076,8 @@ mod tests {
         outcome.start();
         for index in 0..256 {
             outcome.record(HealObjectOutcome {
+                payload_integrity_verified: false,
+                resolved_data_dir: None,
                 identity: HealObjectIdentity {
                     kind: HealObjectKind::Object,
                     bucket: "bucket".into(),

crates/heal/src/heal/erasure_healer.rs

@@ -1349,6 +1349,8 @@ impl ErasureSetHealer {
             };
             (
                 HealObjectOutcome {
+                    payload_integrity_verified: false,
+                    resolved_data_dir: None,
                     identity: HealObjectIdentity {
                         kind: HealObjectKind::Metadata,
                         bucket: RUSTFS_META_BUCKET.to_string(),
@@ -2758,6 +2760,8 @@ mod resume_loop_tests {
                 .unwrap_or(Some(HealObjectDisposition::Repaired));
             let receipt = if error.is_none() {
                 disposition.map(|disposition| HealObjectReceipt {
+                    payload_integrity_verified: false,
+                    resolved_data_dir: None,
                     identity: HealObjectIdentity {
                         kind: HealObjectKind::Object,
                         bucket: bucket.to_owned(),

crates/heal/src/heal/erasure_healer/admin.rs

@@ -145,9 +145,11 @@ pub(super) fn skipped(identity: Option<HealObjectIdentity>, reason: &str) -> Opt
     identity.map(|identity| {
         (
             CanonicalObjectOutcome {
+                payload_integrity_verified: false,
                 identity,
                 disposition: HealObjectDisposition::Unknown,
                 detail: Some(reason.to_owned()),
+                resolved_data_dir: None,
             },
             0,
         )
@@ -249,9 +251,11 @@ pub(super) async fn heal_object(
             (size, legacy),
             Some((
                 CanonicalObjectOutcome {
+                    payload_integrity_verified: false,
                     identity,
                     disposition,
                     detail,
+                    resolved_data_dir: None,
                 },
                 failures,
             )),

crates/heal/src/heal/manager.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::heal::{
-    outcome::HealTaskOutcome,
+    outcome::{HealExecutionOutcome, HealTaskOutcome},
     progress::{HealProgress, HealStatistics},
     resume::{
         ReplacementPhase, ReplacementRecoveryBlock, ReplacementRecoveryCandidate, ResumeGc, ResumeManager, ResumeState,
@@ -324,9 +324,18 @@ async fn publish_completed_heal(
         let object = outcome
             .filter(|outcome| completed.heal_type.is_per_object() && !outcome.objects_truncated && outcome.objects.len() == 1)
             .and_then(|outcome| outcome.objects.front())
-            .map(|object| crate::heal::outcome::HealObjectReceipt {
-                identity: object.identity.clone(),
-                disposition: object.disposition.clone(),
+            .map(|object| {
+                let resolved_data_dir = object.resolved_data_dir.filter(|generation| !generation.is_nil());
+                crate::heal::outcome::HealObjectReceipt {
+                    identity: object.identity.clone(),
+                    disposition: object.disposition.clone(),
+                    resolved_data_dir,
+                    payload_integrity_verified: object.payload_integrity_verified
+                        && resolved_data_dir.is_some()
+                        && !completed.options.dry_run
+                        && matches!(completed.status, HealTaskStatus::Completed)
+                        && outcome.is_some_and(|outcome| outcome.execution == HealExecutionOutcome::Completed),
+                }
             });
         rustfs_heal_contracts::completion::publish_heal_completion(rustfs_heal_contracts::completion::HealCompletion {
             task_id: task_id.to_owned(),
@@ -1743,6 +1752,14 @@ impl HealManager {
         Self::new_with_root_recovery(storage, config, None, Arc::new(root_recovery::RootHealRecovery::disabled_for_tests()))
     }
 
+    /// Change the scheduler limit only in an explicitly enabled test build.
+    #[cfg(feature = "test-util")]
+    #[doc(hidden)]
+    pub async fn set_max_concurrent_heals_for_test(&self, limit: usize) {
+        self.config.write().await.max_concurrent_heals = limit;
+        self.notify.notify_one();
+    }
+
     #[cfg(any(test, feature = "test-util"))]
     #[doc(hidden)]
     pub fn new_with_workload_provider_without_root_recovery_for_test(

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

@@ -1384,6 +1384,8 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() {
         durable_anchor: None,
     };
     let matching = HealObjectOutcome {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: "bucket".to_string(),
@@ -1535,6 +1537,8 @@ fn unverified_legacy_fixture(bucket: &str) -> (MrfRepairNoticeTarget, crate::hea
         durable_anchor: Some(anchor.clone()),
     };
     let outcome = HealObjectOutcome {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: bucket.to_string(),
@@ -1729,6 +1733,8 @@ fn completed_mrf_notice_publishes_only_verified_positive_events() {
     outcome.execution = HealExecutionOutcome::Completed;
     outcome.coverage = crate::heal::outcome::HealTraversalCoverage::Complete;
     outcome.record(HealObjectOutcome {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: bucket.to_string(),
@@ -1742,6 +1748,8 @@ fn completed_mrf_notice_publishes_only_verified_positive_events() {
         detail: None,
     });
     outcome.record(HealObjectOutcome {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: bucket.to_string(),

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

@@ -1,5 +1,5 @@
 use super::*;
-use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome};
+use crate::heal::outcome::{HealAbortReason, HealObjectIdentity, HealObjectKind, HealObjectOutcome};
 use rustfs_heal_contracts::{completion::subscribe_heal_completions, heal_channel::HealScanMode};
 
 #[tokio::test]
@@ -22,6 +22,8 @@ async fn canonical_completion_publishes_typed_objects_only_for_terminal_untrunca
     completed.options.scan_mode = HealScanMode::Normal;
     let mut outcome = HealTaskOutcome::default();
     outcome.record(HealObjectOutcome {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: "bucket".into(),
@@ -86,3 +88,135 @@ async fn canonical_completion_publishes_typed_objects_only_for_terminal_untrunca
     .expect("truncated terminal must still be observable");
     assert!(truncated.object.is_none(), "a retained tail cannot certify a whole result");
 }
+
+#[tokio::test]
+async fn completion_carries_local_generation_but_wire_roundtrip_does_not() {
+    let mut receiver = subscribe_heal_completions();
+    let completed_heals = Mutex::new(HashMap::new());
+    let aliases = Mutex::new(HashMap::new());
+    let generation = Uuid::new_v4();
+    let mut outcome = HealTaskOutcome::default();
+    outcome.record(HealObjectOutcome {
+        payload_integrity_verified: true,
+        identity: HealObjectIdentity {
+            kind: HealObjectKind::Object,
+            bucket: "bucket".into(),
+            object: "object".into(),
+            version_id: None,
+            bucket_incarnation_id: Some(Uuid::new_v4()),
+            pool_index: Some(0),
+            set_index: Some(0),
+        },
+        disposition: HealObjectDisposition::VerifiedHealthy,
+        detail: None,
+        resolved_data_dir: Some(generation),
+    });
+    outcome.finish(None);
+    let wire = serde_json::to_value(&outcome).expect("serialize outcome");
+    assert!(wire["objects"][0].get("resolvedDataDir").is_none());
+    assert!(wire["objects"][0].get("payloadIntegrityVerified").is_none());
+    let restored: HealTaskOutcome = serde_json::from_value(wire).expect("deserialize outcome");
+
+    for (observed_outcome, expected_generation, expected_proof) in [(outcome, Some(generation), true), (restored, None, false)] {
+        let id = Uuid::new_v4().to_string();
+        let mut completed = completed_retention_fixture(SystemTime::now());
+        completed.heal_type = HealType::Object {
+            bucket: "bucket".into(),
+            object: "object".into(),
+            version_id: None,
+        };
+        completed.status = HealTaskStatus::Completed;
+        completed.outcome = Some(Arc::new(observed_outcome));
+        publish_completed_heal(&completed_heals, &aliases, &id, completed, true).await;
+        let observed = tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let event = receiver.recv().await.expect("terminal broadcaster stays open");
+                if event.task_id == id {
+                    break event;
+                }
+            }
+        })
+        .await
+        .expect("terminal should be published");
+        let receipt = observed.object.as_ref().expect("single typed outcome");
+        assert_eq!(receipt.resolved_data_dir, expected_generation);
+        assert_eq!(receipt.payload_integrity_verified, expected_proof);
+    }
+}
+
+#[tokio::test]
+async fn completion_requires_success_and_a_non_nil_generation_for_payload_proof() {
+    let mut receiver = subscribe_heal_completions();
+    let completed_heals = Mutex::new(HashMap::new());
+    let aliases = Mutex::new(HashMap::new());
+    let generation = Uuid::new_v4();
+    for (status, execution, dry_run, observed_generation) in [
+        (
+            HealTaskStatus::Cancelled,
+            HealExecutionOutcome::Aborted(HealAbortReason::Cancelled),
+            false,
+            Some(generation),
+        ),
+        (
+            HealTaskStatus::Failed { error: "failure".into() },
+            HealExecutionOutcome::Completed,
+            false,
+            Some(generation),
+        ),
+        (
+            HealTaskStatus::Completed,
+            HealExecutionOutcome::Aborted(HealAbortReason::Deadline),
+            false,
+            Some(generation),
+        ),
+        (HealTaskStatus::Completed, HealExecutionOutcome::Completed, true, Some(generation)),
+        (HealTaskStatus::Completed, HealExecutionOutcome::Completed, false, None),
+        (HealTaskStatus::Completed, HealExecutionOutcome::Completed, false, Some(Uuid::nil())),
+    ] {
+        let id = Uuid::new_v4().to_string();
+        let mut outcome = HealTaskOutcome::default();
+        outcome.record(HealObjectOutcome {
+            identity: HealObjectIdentity {
+                kind: HealObjectKind::Object,
+                bucket: "bucket".into(),
+                object: "object".into(),
+                version_id: None,
+                bucket_incarnation_id: Some(Uuid::new_v4()),
+                pool_index: Some(0),
+                set_index: Some(0),
+            },
+            disposition: HealObjectDisposition::VerifiedHealthy,
+            detail: None,
+            resolved_data_dir: observed_generation,
+            payload_integrity_verified: true,
+        });
+        outcome.execution = execution;
+        let mut completed = completed_retention_fixture(SystemTime::now());
+        completed.heal_type = HealType::Object {
+            bucket: "bucket".into(),
+            object: "object".into(),
+            version_id: None,
+        };
+        completed.options.dry_run = dry_run;
+        completed.status = status;
+        completed.outcome = Some(Arc::new(outcome));
+        publish_completed_heal(&completed_heals, &aliases, &id, completed, true).await;
+        let observed = tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                let event = receiver.recv().await.expect("terminal broadcaster stays open");
+                if event.task_id == id {
+                    break event;
+                }
+            }
+        })
+        .await
+        .expect("terminal should be published");
+        assert!(
+            !observed
+                .object
+                .as_ref()
+                .expect("single typed outcome")
+                .payload_integrity_verified
+        );
+    }
+}

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

@@ -190,6 +190,8 @@ fn terminal_with_outcome(heal_type: &HealType, status: HealTaskStatus) -> Comple
             HealObjectDisposition::VerifiedHealthy
         };
         outcome.record(HealObjectOutcome {
+            payload_integrity_verified: false,
+            resolved_data_dir: None,
             identity: HealObjectIdentity {
                 kind: HealObjectKind::Object,
                 bucket: "bucket".to_string(),
@@ -2045,6 +2047,8 @@ async fn root_recovery_cluster_replay_rejects_successor_and_preserves_completed_
         .insert(bucket.clone(), Some(expected));
     scope.next_bucket_index = 1;
     scope.outcome.record(HealObjectOutcome {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: bucket.clone(),

crates/heal/src/heal/outcome.rs

@@ -18,7 +18,6 @@
 use rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL;
 use serde::{Deserialize, Serialize};
 use std::collections::VecDeque;
-#[cfg(test)]
 use uuid::Uuid;
 
 const MAX_OUTCOME_ITEMS: usize = 128;
@@ -64,6 +63,12 @@ pub struct HealObjectOutcome {
     pub identity: HealObjectIdentity,
     pub disposition: HealObjectDisposition,
     pub detail: Option<String>,
+    /// Local storage generation observation; absent from wire snapshots.
+    #[serde(skip)]
+    pub resolved_data_dir: Option<Uuid>,
+    /// Local payload verification proof; absent from wire snapshots.
+    #[serde(skip)]
+    pub payload_integrity_verified: bool,
 }
 
 impl HealObjectOutcome {
@@ -442,6 +447,8 @@ mod canonical_outcome_tests {
 
     fn item(disposition: HealObjectDisposition) -> HealObjectOutcome {
         HealObjectOutcome {
+            payload_integrity_verified: false,
+            resolved_data_dir: None,
             identity: HealObjectIdentity {
                 kind: HealObjectKind::Object,
                 bucket: "bucket".to_string(),
@@ -496,6 +503,8 @@ mod canonical_outcome_tests {
             ..item(HealObjectDisposition::Unknown).identity
         };
         let mut receipt = HealObjectReceipt {
+            payload_integrity_verified: false,
+            resolved_data_dir: None,
             identity: expected.clone(),
             disposition: HealObjectDisposition::MetadataHealthy,
         };
@@ -740,6 +749,8 @@ mod canonical_outcome_tests {
             ..item(HealObjectDisposition::Unknown).identity
         };
         let mut receipt = HealObjectReceipt {
+            payload_integrity_verified: false,
+            resolved_data_dir: None,
             identity: expected.clone(),
             disposition: HealObjectDisposition::Repaired,
         };

crates/heal/src/heal/storage.rs

@@ -135,6 +135,10 @@ fn verified_object_receipt(
     } else {
         Some(resolved_version.to_string())
     };
+    let resolved_data_dir = item
+        .resolved_data_dir
+        .map(Uuid::from_bytes)
+        .filter(|generation| !generation.is_nil());
     Some(HealObjectReceipt {
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
@@ -154,6 +158,8 @@ fn verified_object_receipt(
         } else {
             return None;
         },
+        resolved_data_dir,
+        payload_integrity_verified: item.integrity_verified && resolved_data_dir.is_some(),
     })
 }
 
@@ -880,6 +886,8 @@ impl ECStoreHealStorage {
                 })
             {
                 Some(HealObjectReceipt {
+                    payload_integrity_verified: false,
+                    resolved_data_dir: None,
                     identity: HealObjectIdentity {
                         kind: HealObjectKind::Object,
                         bucket: proof.bucket,
@@ -1033,6 +1041,8 @@ impl HealStorageAPI for ECStoreHealStorage {
             item: HealResultItem::default(),
             error: None,
             receipt: Some(HealObjectReceipt {
+                payload_integrity_verified: false,
+                resolved_data_dir: None,
                 identity: HealObjectIdentity {
                     kind: HealObjectKind::DeleteMarkerPurge,
                     bucket: bucket.to_owned(),
@@ -2507,6 +2517,42 @@ mod tests {
         assert!(verified_object_receipt("bucket", "object", None, &options, &item, incarnation).is_none());
     }
 
+    #[test]
+    fn receipt_retains_only_the_storage_selected_generation() {
+        use super::{HealOpts, HealResultItem, Uuid, verified_object_receipt};
+        use rustfs_madmin::heal_commands::HealDriveInfo;
+
+        let generation = Uuid::new_v4();
+        let mut item = HealResultItem {
+            integrity_verified: true,
+            resolved_version_id: Some([0; 16]),
+            resolved_data_dir: Some(*generation.as_bytes()),
+            ..Default::default()
+        };
+        let healthy = HealDriveInfo {
+            state: "ok".to_string(),
+            ..Default::default()
+        };
+        item.before.drives.push(healthy.clone());
+        item.after.drives.push(healthy);
+        let incarnation = Uuid::new_v4();
+        let options = HealOpts::default();
+        let receipt =
+            verified_object_receipt("bucket", "object", None, &options, &item, incarnation).expect("healthy storage result");
+        assert_eq!(receipt.resolved_data_dir, Some(generation));
+
+        for unresolved in [None, Some([0; 16])] {
+            item.resolved_data_dir = unresolved;
+            let receipt = verified_object_receipt("bucket", "object", None, &options, &item, incarnation)
+                .expect("missing generation must not alter repair authorization");
+            assert_eq!(receipt.resolved_data_dir, None);
+            assert!(
+                !receipt.payload_integrity_verified,
+                "an unowned or nil generation cannot carry payload proof"
+            );
+        }
+    }
+
     #[test]
     fn metadata_health_and_marker_repair_have_distinct_receipts() {
         use super::{HealObjectDisposition, HealOpts, HealResultItem, Uuid, verified_object_receipt};
@@ -2523,6 +2569,7 @@ mod tests {
             metadata_verified: true,
             version_id: version.clone(),
             resolved_version_id: Some(*Uuid::parse_str(&version).expect("version UUID").as_bytes()),
+            resolved_data_dir: Some(*Uuid::new_v4().as_bytes()),
             ..Default::default()
         };
         item.before.drives.push(healthy.clone());
@@ -2531,11 +2578,13 @@ mod tests {
         let receipt = verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation)
             .expect("metadata quorum should certify metadata health");
         assert_eq!(receipt.disposition, HealObjectDisposition::MetadataHealthy);
+        assert!(!receipt.payload_integrity_verified);
 
         item.integrity_verified = true;
         let receipt = verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation)
             .expect("stronger integrity proof should remain available");
         assert_eq!(receipt.disposition, HealObjectDisposition::VerifiedHealthy);
+        assert!(receipt.payload_integrity_verified);
 
         item.integrity_verified = false;
         item.metadata_verified = false;
@@ -2544,6 +2593,7 @@ mod tests {
         let receipt = verified_object_receipt("bucket", "object", Some(&version), &options, &item, incarnation)
             .expect("a committed metadata repair should certify the marker/version");
         assert_eq!(receipt.disposition, HealObjectDisposition::Repaired);
+        assert!(!receipt.payload_integrity_verified);
 
         item.before.drives[0].state = "ok".to_string();
         assert!(
@@ -2552,6 +2602,52 @@ mod tests {
         );
     }
 
+    #[test]
+    fn repaired_disposition_does_not_imply_deep_payload_verification() {
+        use super::{HealObjectDisposition, HealOpts, HealResultItem, Uuid, verified_object_receipt};
+        use rustfs_madmin::heal_commands::HealDriveInfo;
+
+        let mut item = HealResultItem {
+            repair_verified: true,
+            resolved_version_id: Some([0; 16]),
+            resolved_data_dir: Some(*Uuid::new_v4().as_bytes()),
+            ..Default::default()
+        };
+        item.before.drives.push(HealDriveInfo {
+            state: "missing".to_string(),
+            ..Default::default()
+        });
+        item.after.drives.push(HealDriveInfo {
+            state: "ok".to_string(),
+            ..Default::default()
+        });
+        let options = HealOpts::default();
+        let incarnation = Uuid::new_v4();
+        let normal = verified_object_receipt("bucket", "object", None, &options, &item, incarnation)
+            .expect("repair without a deep proof still has a receipt");
+        assert_eq!(normal.disposition, HealObjectDisposition::Repaired);
+        assert!(!normal.payload_integrity_verified);
+
+        item.integrity_verified = true;
+        let deep = verified_object_receipt("bucket", "object", None, &options, &item, incarnation).expect("storage deep proof");
+        assert_eq!(deep.disposition, HealObjectDisposition::Repaired);
+        assert!(deep.payload_integrity_verified);
+        assert!(
+            verified_object_receipt(
+                "bucket",
+                "object",
+                None,
+                &HealOpts {
+                    dry_run: true,
+                    ..options
+                },
+                &item,
+                incarnation,
+            )
+            .is_none()
+        );
+    }
+
     #[test]
     fn next_heal_listing_token_returns_none_for_complete_page() {
         assert_eq!(

crates/heal/src/heal/task.rs

@@ -686,11 +686,20 @@ impl HealTask {
         if !receipt.verified_for(expected) {
             return false;
         }
+        let resolved_data_dir = receipt.resolved_data_dir.filter(|generation| !generation.is_nil());
+        let payload_integrity_verified = receipt.payload_integrity_verified
+            && resolved_data_dir.is_some()
+            && matches!(
+                receipt.disposition,
+                HealObjectDisposition::Repaired | HealObjectDisposition::VerifiedHealthy
+            );
         let mut outcome = self.outcome.write().await;
         outcome.record(HealObjectOutcome {
             identity: receipt.identity,
             disposition: receipt.disposition,
             detail: None,
+            resolved_data_dir,
+            payload_integrity_verified,
         });
         true
     }
@@ -700,6 +709,8 @@ impl HealTask {
             let mut outcome = self.outcome.write().await;
             outcome.attempt_failed();
             outcome.record(HealObjectOutcome {
+                payload_integrity_verified: false,
+                resolved_data_dir: None,
                 identity,
                 disposition: HealObjectDisposition::Deferred {
                     reason,
@@ -985,6 +996,8 @@ impl HealTask {
             let mut outcome = self.outcome.write().await;
             outcome.attempt_failed();
             outcome.record(HealObjectOutcome {
+                payload_integrity_verified: false,
+                resolved_data_dir: None,
                 identity,
                 disposition: HealObjectDisposition::Deferred {
                     reason: HealDeferredReason::RetiredMarkerProof,
@@ -1146,6 +1159,8 @@ impl HealTask {
                     }
                 };
                 outcome.record(HealObjectOutcome {
+                    payload_integrity_verified: false,
+                    resolved_data_dir: None,
                     identity,
                     disposition,
                     detail: result.as_ref().err().map(ToString::to_string),

crates/heal/src/heal/task/heal_bucket.rs

@@ -852,6 +852,8 @@ impl HealTask {
                                     }
                                 };
                                 self.outcome.write().await.record(HealObjectOutcome {
+                                    payload_integrity_verified: false,
+                                    resolved_data_dir: None,
                                     identity,
                                     disposition,
                                     detail: None,
@@ -970,6 +972,8 @@ impl HealTask {
 
                     if !recorded_authoritative_outcome {
                         self.outcome.write().await.record(HealObjectOutcome {
+                            payload_integrity_verified: false,
+                            resolved_data_dir: None,
                             identity,
                             disposition,
                             detail,
@@ -1005,6 +1009,8 @@ impl HealTask {
                 // No detached sleepers survive abort.
                 for item in deferred.objects.into_iter().chain(inline_retry) {
                     self.outcome.write().await.record(HealObjectOutcome {
+                        payload_integrity_verified: false,
+                        resolved_data_dir: None,
                         identity: self.outcome_identity(
                             bucket,
                             &item.name,

crates/heal/src/heal/task/heal_object.rs

@@ -335,6 +335,8 @@ impl HealTask {
                     && !self.options.dry_run
                 {
                     self.outcome.write().await.record(HealObjectOutcome {
+                        payload_integrity_verified: false,
+                        resolved_data_dir: None,
                         identity: expected_identity,
                         disposition: HealObjectDisposition::Unknown,
                         detail: (!result.detail.is_empty()).then(|| result.detail.clone()),
@@ -518,6 +520,8 @@ impl HealTask {
                     .all(|drive| drive.state == ok_drive_state);
             if healthy_legacy_without_repair {
                 self.outcome.write().await.record(HealObjectOutcome {
+                    payload_integrity_verified: false,
+                    resolved_data_dir: None,
                     identity: expected,
                     disposition: HealObjectDisposition::Unknown,
                     detail: Some(storage_result.item.detail.clone()),

crates/heal/src/heal/task/tests.rs

@@ -1667,6 +1667,8 @@ fn object_receipt(
     bucket_incarnation_id: Uuid,
 ) -> HealObjectReceipt {
     HealObjectReceipt {
+        payload_integrity_verified: false,
+        resolved_data_dir: None,
         identity: HealObjectIdentity {
             kind: HealObjectKind::Object,
             bucket: "bucket-a".to_string(),
@@ -1683,16 +1685,12 @@ fn object_receipt(
 #[tokio::test]
 async fn object_heal_records_matching_positive_storage_receipt() {
     let incarnation = Uuid::new_v4();
+    let generation = Uuid::new_v4();
+    let mut receipt = object_receipt("object-a", Some("version-a"), HealObjectDisposition::Repaired, incarnation);
+    receipt.resolved_data_dir = Some(generation);
+    receipt.payload_integrity_verified = true;
     let storage = Arc::new(MockStorage {
-        heal_object_receipts: Mutex::new(HashMap::from([(
-            "object-a".to_string(),
-            VecDeque::from([object_receipt(
-                "object-a",
-                Some("version-a"),
-                HealObjectDisposition::Repaired,
-                incarnation,
-            )]),
-        )])),
+        heal_object_receipts: Mutex::new(HashMap::from([("object-a".to_string(), VecDeque::from([receipt]))])),
         bucket_incarnation_id: Mutex::new(Some(incarnation)),
         ..Default::default()
     });
@@ -1711,6 +1709,52 @@ async fn object_heal_records_matching_positive_storage_receipt() {
     assert_eq!(object.identity.version_id.as_deref(), Some("version-a"));
     assert!(object.identity.bucket_incarnation_id.is_some());
     assert_eq!(object.disposition, HealObjectDisposition::Repaired);
+    assert_eq!(object.resolved_data_dir, Some(generation));
+    assert!(object.payload_integrity_verified);
+}
+
+#[tokio::test]
+async fn metadata_or_absence_receipts_do_not_carry_payload_proof() {
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None),
+        Arc::new(MockStorage::default()),
+    );
+    for disposition in [
+        HealObjectDisposition::MetadataHealthy,
+        HealObjectDisposition::AuthoritativelyAbsent,
+    ] {
+        let mut receipt = object_receipt("object-a", None, disposition, Uuid::new_v4());
+        receipt.payload_integrity_verified = true;
+        let expected = receipt.identity.clone();
+        assert!(task.record_verified_storage_receipt(&expected, Some(receipt)).await);
+        let outcome = task.get_outcome().await;
+        assert!(
+            !outcome
+                .objects
+                .back()
+                .expect("receipt should be recorded")
+                .payload_integrity_verified
+        );
+    }
+}
+
+#[tokio::test]
+async fn missing_or_nil_storage_generation_stays_unknown_in_recorded_outcome() {
+    let task = HealTask::from_request(
+        HealRequest::object("bucket-a".to_string(), "object-a".to_string(), None),
+        Arc::new(MockStorage::default()),
+    );
+    for generation in [None, Some(Uuid::nil())] {
+        let mut receipt = object_receipt("object-a", None, HealObjectDisposition::VerifiedHealthy, Uuid::new_v4());
+        receipt.resolved_data_dir = generation;
+        receipt.payload_integrity_verified = true;
+        let expected = receipt.identity.clone();
+        assert!(task.record_verified_storage_receipt(&expected, Some(receipt)).await);
+        let outcome = task.get_outcome().await;
+        let object = outcome.objects.back().expect("receipt should be recorded");
+        assert_eq!(object.resolved_data_dir, None);
+        assert!(!object.payload_integrity_verified);
+    }
 }
 
 #[tokio::test]
@@ -1761,17 +1805,9 @@ async fn cancelled_object_heal_rejects_matching_positive_storage_receipt() {
         pool_index: None,
         set_index: None,
     };
-    let accepted = task
-        .record_verified_storage_receipt(
-            &expected,
-            Some(object_receipt(
-                "object-a",
-                Some("version-a"),
-                HealObjectDisposition::Repaired,
-                incarnation,
-            )),
-        )
-        .await;
+    let mut receipt = object_receipt("object-a", Some("version-a"), HealObjectDisposition::Repaired, incarnation);
+    receipt.payload_integrity_verified = true;
+    let accepted = task.record_verified_storage_receipt(&expected, Some(receipt)).await;
 
     let outcome = task.get_outcome().await;
     assert!(!accepted);
@@ -4318,6 +4354,8 @@ async fn mrf_recreate_missing_object_records_exact_absence_receipt_with_scope()
         heal_object_receipts: Mutex::new(HashMap::from([(
             "deleted.bin".to_string(),
             VecDeque::from([HealObjectReceipt {
+                payload_integrity_verified: false,
+                resolved_data_dir: None,
                 identity: HealObjectIdentity {
                     kind: HealObjectKind::Object,
                     bucket: "bucket-a".to_string(),

crates/heal/tests/shard_identity_receipt_test.rs

@@ -196,7 +196,6 @@ async fn receipt_requires_independent_deep_verification() {
                 &object,
                 None,
                 &HealOpts {
-                    no_lock: true,
                     scan_mode: HealScanMode::Normal,
                     ..Default::default()
                 },
@@ -209,13 +208,19 @@ async fn receipt_requires_independent_deep_verification() {
             HealObjectDisposition::MetadataHealthy,
             "a presence scan must not be promoted to payload VerifiedHealthy"
         );
+        assert!(
+            normal_receipt
+                .resolved_data_dir
+                .is_some_and(|generation| !generation.is_nil()),
+            "the Normal scan must have a locked, selected generation"
+        );
+        assert!(!normal_receipt.payload_integrity_verified);
         let result = storage
             .heal_object_with_receipt(
                 bucket,
                 &object,
                 None,
                 &HealOpts {
-                    no_lock: true,
                     scan_mode: HealScanMode::Deep,
                     ..Default::default()
                 },
@@ -224,11 +229,32 @@ async fn receipt_requires_independent_deep_verification() {
         if corrupt_count == 1 {
             let result = result.expect("recoverable repair");
             assert!(result.error.is_none());
+            let receipt = result.receipt.expect("verified repair receipt");
             assert_eq!(
-                result.receipt.expect("verified repair receipt").disposition,
+                receipt.disposition,
                 HealObjectDisposition::Repaired,
                 "presence alone must not produce VerifiedHealthy"
             );
+            assert!(receipt.payload_integrity_verified, "Deep repair verified the protected payload");
+            let unowned = storage
+                .heal_object_with_receipt(
+                    bucket,
+                    &object,
+                    None,
+                    &HealOpts {
+                        no_lock: true,
+                        scan_mode: HealScanMode::Deep,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("unowned Deep inspection");
+            assert!(unowned.item.integrity_verified, "the storage scan did read the payload");
+            assert!(unowned.item.resolved_data_dir.is_none(), "no namespace owner was retained");
+            assert!(
+                !unowned.receipt.expect("unowned result receipt").payload_integrity_verified,
+                "an unowned scan cannot be published as payload proof"
+            );
         } else if let Ok(result) = result {
             assert!(result.error.is_some(), "below quorum cannot be a success");
             assert!(result.receipt.is_none(), "no completion receipt without authoritative sources");
@@ -357,6 +383,10 @@ fn admin_pool_set_repairs_five_shards_and_retains_exact_terminal_after_restart()
                             .find(|receipt| receipt.identity.bucket == *bucket && receipt.identity.object == *object)
                             .expect("one receipt per exact fault");
                         assert_eq!(receipt.disposition, HealObjectDisposition::Repaired);
+                        assert!(
+                            !receipt.payload_integrity_verified,
+                            "erasure-set checkpoint outcomes do not retain local payload proof"
+                        );
                         assert_eq!((receipt.identity.pool_index, receipt.identity.set_index), (Some(0), Some(0)));
                         assert!(receipt.identity.version_id.is_some());
                         assert_eq!(
@@ -528,6 +558,7 @@ fn admin_normal_protected_repairs_report_repaired_for_data_and_parity() {
                             .find(|receipt| receipt.identity.bucket == bucket && receipt.identity.object == *object)
                             .expect("one receipt per repaired object");
                         assert_eq!(receipt.disposition, HealObjectDisposition::Repaired);
+                        assert!(!receipt.payload_integrity_verified, "Normal repair has no Deep payload proof");
                         assert_eq!((receipt.identity.pool_index, receipt.identity.set_index), (Some(0), Some(0)));
                         assert!(receipt.identity.version_id.is_some());
                         assert_eq!(

crates/madmin/src/heal_commands.rs

@@ -86,6 +86,10 @@ pub struct HealResultItem {
     /// Wire-decoded and legacy results cannot supply this in-process proof.
     #[serde(skip)]
     pub resolved_version_id: Option<[u8; 16]>,
+    /// Data directory selected under the storage owner's object lock. This is
+    /// an in-process generation observation, not an integrity proof.
+    #[serde(skip)]
+    pub resolved_data_dir: Option<[u8; 16]>,
     #[serde(rename = "detail")]
     pub detail: String,
     #[serde(default, rename = "deleteRecovery", skip_serializing_if = "Option::is_none")]
@@ -174,12 +178,15 @@ mod tests {
     fn resolved_heal_version_is_not_wire_proof() {
         let item = HealResultItem {
             resolved_version_id: Some([0; 16]),
+            resolved_data_dir: Some([1; 16]),
             legacy_identity_unverified: true,
             ..Default::default()
         };
         let mut wire = serde_json::to_value(&item).expect("heal result should serialize");
         assert!(wire.get("resolved_version_id").is_none());
         assert!(wire.get("resolvedVersionId").is_none());
+        assert!(wire.get("resolved_data_dir").is_none());
+        assert!(wire.get("resolvedDataDir").is_none());
         assert!(wire.get("repair_verified").is_none());
         assert!(wire.get("repairVerified").is_none());
         assert!(wire.get("metadata_verified").is_none());
@@ -191,8 +198,11 @@ mod tests {
         wire["legacy_identity_unverified"] = serde_json::json!(true);
         wire["legacyIdentityUnverified"] = serde_json::json!(true);
         wire["resolved_version_id"] = serde_json::json!(vec![0; 16]);
+        wire["resolved_data_dir"] = serde_json::json!(vec![1; 16]);
+        wire["resolvedDataDir"] = serde_json::json!(vec![1; 16]);
         let decoded: HealResultItem = serde_json::from_value(wire).expect("legacy wire shape should remain readable");
         assert_eq!(decoded.resolved_version_id, None, "wire input cannot supply owner proof");
+        assert_eq!(decoded.resolved_data_dir, None, "wire input cannot supply owner generation");
         assert!(!decoded.legacy_identity_unverified, "wire input cannot supply owner classification");
     }
 

crates/scanner/Cargo.toml

@@ -31,6 +31,7 @@ workspace = true
 
 [features]
 default = []
+test-util = []
 diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller"]
 hotpath = [
     "hotpath/hotpath",
@@ -107,7 +108,7 @@ bytes.workspace = true
 hex-simd.workspace = true
 
 [dev-dependencies]
-rustfs-heal.workspace = true
+rustfs-heal = { workspace = true, features = ["test-util"] }
 rustfs-test-utils.workspace = true
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
 serial_test = { workspace = true }

crates/scanner/src/heal_completion.rs

@@ -28,12 +28,53 @@ use std::{
     time::{Duration, Instant, SystemTime},
 };
 use tokio::sync::{broadcast, mpsc};
+use uuid::Uuid;
 
 const MAX_PENDING: usize = 1024;
+const MAX_PENDING_GROUPS: usize = 1024;
 const MAX_RECENT: usize = 1024;
 const PENDING_TTL: Duration = Duration::from_secs(60 * 60);
 const RECENT_TTL: Duration = Duration::from_secs(10 * 60);
 
+#[cfg(feature = "test-util")]
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ScannerHealTestAdmission {
+    Pending,
+    Unknown,
+    CachedTerminal { data_dir: Option<Uuid> },
+    Unaccounted,
+}
+
+#[cfg(feature = "test-util")]
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ScannerHealTestEvent {
+    Admission {
+        task_id: String,
+        candidate_data_dir: Option<Uuid>,
+        outcome: ScannerHealTestAdmission,
+    },
+    Terminal {
+        task_id: String,
+        terminal_data_dir: Option<Uuid>,
+        duplicate: bool,
+        settled_candidate_data_dirs: Vec<Option<Uuid>>,
+        remaining_candidate_data_dirs: Vec<Option<Uuid>>,
+    },
+}
+
+#[cfg(feature = "test-util")]
+fn test_events() -> &'static broadcast::Sender<ScannerHealTestEvent> {
+    static EVENTS: std::sync::OnceLock<broadcast::Sender<ScannerHealTestEvent>> = std::sync::OnceLock::new();
+    EVENTS.get_or_init(|| broadcast::channel(64).0)
+}
+
+/// Subscribe to bounded, process-local observer events in test builds.
+#[cfg(feature = "test-util")]
+#[doc(hidden)]
+pub fn subscribe_scanner_heal_test_events() -> broadcast::Receiver<ScannerHealTestEvent> {
+    test_events().subscribe()
+}
+
 #[derive(Clone, Debug, Serialize)]
 pub struct ScannerHealTerminalObservation {
     pub execution: &'static str,
@@ -90,6 +131,7 @@ impl Default for ScannerHealObservationStatus {
 #[derive(Debug)]
 pub(crate) struct Candidate {
     identity: Option<HealObjectIdentity>,
+    data_dir: Option<Uuid>,
     mode: HealScanMode,
     dry_run: bool,
     first_seen: Instant,
@@ -97,7 +139,7 @@ pub(crate) struct Candidate {
 }
 
 impl Candidate {
-    fn from_request(request: &HealChannelRequest) -> Option<Self> {
+    fn from_request(request: &HealChannelRequest, data_dir: Option<Uuid>) -> Option<Self> {
         let bytes = request
             .bucket
             .len()
@@ -116,12 +158,36 @@ impl Candidate {
                 pool_index: request.pool_index,
                 set_index: request.set_index,
             }),
+            data_dir: data_dir.filter(|id| !id.is_nil()),
             mode: request.scan_mode.unwrap_or_default(),
             dry_run: request.dry_run.unwrap_or(false),
             first_seen: Instant::now(),
             count: 1,
         })
     }
+
+    fn matches(&self, other: &Self) -> bool {
+        self.identity == other.identity
+            && self.data_dir == other.data_dir
+            && self.mode == other.mode
+            && self.dry_run == other.dry_run
+    }
+}
+
+struct Pending {
+    first: Candidate,
+    // The common single-candidate case does not allocate a vector.
+    others: Vec<Candidate>,
+}
+
+impl Pending {
+    fn iter(&self) -> impl Iterator<Item = &Candidate> {
+        std::iter::once(&self.first).chain(self.others.iter())
+    }
+
+    fn into_iter(self) -> impl Iterator<Item = Candidate> {
+        std::iter::once(self.first).chain(self.others)
+    }
 }
 
 struct Recent {
@@ -131,7 +197,8 @@ struct Recent {
 
 #[derive(Default)]
 struct Observations {
-    pending: HashMap<String, Candidate>,
+    pending: HashMap<String, Pending>,
+    pending_groups: usize,
     recent: HashMap<String, Recent>,
     recent_order: VecDeque<(Instant, String)>,
     // All keys come from fixed execution/disposition/scope/mode labels, never task or object identities.
@@ -145,6 +212,8 @@ struct Observations {
 struct Settlement {
     candidates: u64,
     first_task_observation: bool,
+    generation_mismatch: bool,
+    deep_payload_verified_task: bool,
     execution: &'static str,
     disposition: &'static str,
     scope: &'static str,
@@ -175,33 +244,53 @@ impl Settlement {
             Some(HealObjectDisposition::DryRunObserved) => "dry_run",
             Some(HealObjectDisposition::Unknown) | None => "unknown",
         };
+        let generation_comparison = match (candidate.identity.as_ref(), completion.object.as_ref()) {
+            (Some(expected), Some(actual))
+                if expected.kind == actual.identity.kind
+                    && expected.bucket == actual.identity.bucket
+                    && expected.object == actual.identity.object =>
+            {
+                candidate
+                    .data_dir
+                    .filter(|id| !id.is_nil())
+                    .zip(actual.resolved_data_dir.filter(|id| !id.is_nil()))
+                    .map(|(left, right)| left == right)
+            }
+            _ => None,
+        };
+        let generation_mismatch = generation_comparison == Some(false);
+        let payload_verified = completion
+            .object
+            .as_ref()
+            .is_some_and(|object| object.payload_integrity_verified);
         let scope = if candidate.dry_run || completion.dry_run {
             "dry_run"
         } else if completion.execution != Some(HealExecutionOutcome::Completed) {
             "execution_incomplete"
-        } else if candidate.mode == HealScanMode::Deep && completion.scan_mode != HealScanMode::Deep {
+        } else if candidate.mode == HealScanMode::Deep
+            && completion.scan_mode != HealScanMode::Deep
+            && !(payload_verified && generation_comparison == Some(true))
+        {
             "insufficient_scan"
         } else {
             match (candidate.identity.as_ref(), completion.object.as_ref()) {
-                (Some(expected), Some(actual)) if expected != &actual.identity => "identity_mismatch",
-                (Some(expected), Some(_))
-                    if expected.bucket_incarnation_id.is_some_and(|id| !id.is_nil())
-                        && expected.pool_index.is_some()
-                        && expected.set_index.is_some()
-                        && expected.version_id.as_ref().is_some_and(|version| !version.is_empty()) =>
-                {
-                    // Even an exact version and VerifiedHealthy do not bind the
-                    // candidate's content and leader epoch with today's contract.
-                    "content_and_leader_unknown"
+                (Some(expected), Some(actual)) => {
+                    Self::identity_scope(expected, &actual.identity, candidate.data_dir, actual.resolved_data_dir)
                 }
                 _ => "identity_unknown",
             }
         };
         let first_task_observation = !recent.counted;
+        let deep_payload_verified_task = first_task_observation
+            && completion.execution == Some(HealExecutionOutcome::Completed)
+            && !completion.dry_run
+            && payload_verified;
         recent.counted = true;
         Self {
             candidates: candidate.count,
             first_task_observation,
+            generation_mismatch,
+            deep_payload_verified_task,
             execution,
             disposition,
             scope,
@@ -209,6 +298,58 @@ impl Settlement {
         }
     }
 
+    fn identity_scope(
+        expected: &HealObjectIdentity,
+        actual: &HealObjectIdentity,
+        expected_data_dir: Option<Uuid>,
+        resolved_data_dir: Option<Uuid>,
+    ) -> &'static str {
+        let expected_version = expected.version_id.as_deref().filter(|version| !version.is_empty());
+        let actual_version = actual.version_id.as_deref().filter(|version| !version.is_empty());
+        let expected_incarnation = expected.bucket_incarnation_id.filter(|id| !id.is_nil());
+        let actual_incarnation = actual.bucket_incarnation_id.filter(|id| !id.is_nil());
+        let expected_data_dir = expected_data_dir.filter(|id| !id.is_nil());
+        let resolved_data_dir = resolved_data_dir.filter(|id| !id.is_nil());
+        if expected.kind != actual.kind
+            || expected.bucket != actual.bucket
+            || expected.object != actual.object
+            || expected_version
+                .zip(actual_version)
+                .is_some_and(|(left, right)| left != right)
+            || expected_incarnation
+                .zip(actual_incarnation)
+                .is_some_and(|(left, right)| left != right)
+            || expected
+                .pool_index
+                .zip(actual.pool_index)
+                .is_some_and(|(left, right)| left != right)
+            || expected
+                .set_index
+                .zip(actual.set_index)
+                .is_some_and(|(left, right)| left != right)
+            || expected_data_dir
+                .zip(resolved_data_dir)
+                .is_some_and(|(left, right)| left != right)
+        {
+            "identity_mismatch"
+        } else if expected_version.is_none()
+            || actual_version.is_none()
+            || expected_incarnation.is_none()
+            || actual_incarnation.is_none()
+            || expected.pool_index.is_none()
+            || actual.pool_index.is_none()
+            || expected.set_index.is_none()
+            || actual.set_index.is_none()
+            || expected_data_dir.is_none()
+            || resolved_data_dir.is_none()
+        {
+            "identity_unknown"
+        } else {
+            // Matching identities still do not bind content and leader epoch.
+            "content_and_leader_unknown"
+        }
+    }
+
     fn record(&self) {
         counter!("rustfs_scanner_heal_candidate_terminal_total",
             "disposition" => self.disposition, "scope" => self.scope, "scan_mode" => self.mode.as_str())
@@ -216,6 +357,14 @@ impl Settlement {
         if self.first_task_observation {
             counter!("rustfs_scanner_heal_task_terminal_observations_total").increment(1);
         }
+        // This is independent of scope: an insufficient scan can also have
+        // resolved a different object generation.
+        if self.generation_mismatch {
+            counter!("rustfs_scanner_heal_candidate_generation_mismatch_total").increment(self.candidates);
+        }
+        if self.deep_payload_verified_task {
+            counter!("rustfs_scanner_heal_task_deep_payload_verified_observations_total").increment(1);
+        }
     }
 }
 
@@ -247,18 +396,30 @@ impl Observations {
             return Some(settlement);
         }
         if let Some(pending) = self.pending.get_mut(&task_id) {
-            pending.count = pending.count.saturating_add(candidate.count);
-            if pending.identity != candidate.identity {
-                pending.identity = None;
+            if pending.first.matches(&candidate) {
+                pending.first.count = pending.first.count.saturating_add(candidate.count);
+            } else if let Some(index) = pending.others.iter().position(|group| group.matches(&candidate)) {
+                let group = &mut pending.others[index];
+                group.count = group.count.saturating_add(candidate.count);
+            } else if self.pending_groups < MAX_PENDING_GROUPS {
+                pending.others.push(candidate);
+                self.pending_groups += 1;
+            } else {
+                self.record_unknown("pending_groups_full", candidate.count);
             }
-            if candidate.mode == HealScanMode::Deep {
-                pending.mode = HealScanMode::Deep;
-            }
-            pending.dry_run |= candidate.dry_run;
-        } else if self.pending.len() < MAX_PENDING {
-            self.pending.insert(task_id, candidate);
-        } else {
+        } else if self.pending.len() >= MAX_PENDING {
             self.record_unknown("pending_full", candidate.count);
+        } else if self.pending_groups >= MAX_PENDING_GROUPS {
+            self.record_unknown("pending_groups_full", candidate.count);
+        } else {
+            self.pending.insert(
+                task_id,
+                Pending {
+                    first: candidate,
+                    others: Vec::new(),
+                },
+            );
+            self.pending_groups += 1;
         }
         None
     }
@@ -272,18 +433,22 @@ impl Observations {
             completion,
             counted: false,
         };
-        let settled = self
-            .pending
-            .remove(&task_id)
-            .map(|candidate| Settlement::new(candidate, &mut recent));
+        let mut settled = None;
+        if let Some(pending) = self.pending.remove(&task_id) {
+            self.pending_groups -= 1 + pending.others.len();
+            for candidate in pending.into_iter() {
+                let settlement = Settlement::new(candidate, &mut recent);
+                self.record_settlement(&settlement);
+                if settled.is_none() {
+                    settled = Some(settlement);
+                }
+            }
+        }
         if self.recent.len() == MAX_RECENT {
             self.evict_recent();
         }
         self.recent_order.push_back((now, task_id.clone()));
         self.recent.insert(task_id, recent);
-        if let Some(settlement) = &settled {
-            self.record_settlement(settlement);
-        }
         settled
     }
 
@@ -297,7 +462,9 @@ impl Observations {
         let candidates = self
             .pending
             .drain()
-            .fold(0u64, |count, (_, candidate)| count.saturating_add(candidate.count));
+            .flat_map(|(_, pending)| pending.into_iter())
+            .fold(0u64, |count, candidate| count.saturating_add(candidate.count));
+        self.pending_groups = 0;
         if candidates > 0 {
             self.record_unknown("completion_lost", candidates);
         }
@@ -313,14 +480,32 @@ impl Observations {
 
     fn maintain(&mut self, now: Instant) {
         let mut expired = 0u64;
-        self.pending.retain(|_, candidate| {
-            if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
-                expired = expired.saturating_add(candidate.count);
-                false
-            } else {
-                true
+        let mut expired_groups = 0usize;
+        self.pending.retain(|_, pending| {
+            let previous_len = pending.others.len();
+            pending.others.retain(|candidate| {
+                if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
+                    expired = expired.saturating_add(candidate.count);
+                    expired_groups += 1;
+                    false
+                } else {
+                    true
+                }
+            });
+            if now.saturating_duration_since(pending.first.first_seen) >= PENDING_TTL {
+                expired = expired.saturating_add(pending.first.count);
+                expired_groups += 1;
+                if pending.others.is_empty() {
+                    return false;
+                }
+                pending.first = pending.others.remove(0);
             }
+            if pending.others.len() != previous_len {
+                pending.others.shrink_to_fit();
+            }
+            true
         });
+        self.pending_groups -= expired_groups;
         if expired > 0 {
             self.record_unknown("expired", expired);
         }
@@ -334,6 +519,7 @@ impl Observations {
         let oldest = self
             .pending
             .values()
+            .flat_map(Pending::iter)
             .map(|candidate| now.saturating_duration_since(candidate.first_seen))
             .max();
         gauge!("rustfs_scanner_heal_pending_tasks").set(self.pending.len() as f64);
@@ -370,7 +556,7 @@ fn observer_from(registry: &Mutex<Option<ObserverHandle>>) -> ObserverHandle {
     }
     let state = Arc::new(Mutex::new(Observations::default()));
     let handle = {
-        let (sender, mut admissions) = mpsc::channel(MAX_PENDING);
+        let (sender, mut admissions) = mpsc::channel::<(String, Candidate)>(MAX_PENDING);
         let mut completions = subscribe_heal_completions();
         let observations = state.clone();
         tokio::spawn(async move {
@@ -380,14 +566,79 @@ fn observer_from(registry: &Mutex<Option<ObserverHandle>>) -> ObserverHandle {
                 tokio::select! {
                     candidate = admissions.recv() => match candidate {
                         Some((id, candidate)) => {
-                            let _ = observations.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).admit(id, candidate);
+                            let mut state = observations.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                            #[cfg(feature = "test-util")]
+                            let probe_id = id.clone();
+                            #[cfg(feature = "test-util")]
+                            let probe_identity = candidate.identity.clone();
+                            #[cfg(feature = "test-util")]
+                            let probe_data_dir = candidate.data_dir;
+                            #[cfg(feature = "test-util")]
+                            let probe_mode = candidate.mode;
+                            #[cfg(feature = "test-util")]
+                            let probe_dry_run = candidate.dry_run;
+                            #[cfg(feature = "test-util")]
+                            let unknown_before: u64 = state.unknown_counts.values().copied().sum();
+                            let _settled = state.admit(id, candidate);
+                            #[cfg(feature = "test-util")]
+                            if test_events().receiver_count() != 0 {
+                                let pending = state.pending.get(&probe_id).is_some_and(|pending| {
+                                    pending.iter().any(|candidate| {
+                                        candidate.identity == probe_identity
+                                            && candidate.data_dir == probe_data_dir
+                                            && candidate.mode == probe_mode
+                                            && candidate.dry_run == probe_dry_run
+                                    })
+                                });
+                                let outcome = if _settled.is_some() {
+                                    ScannerHealTestAdmission::CachedTerminal {
+                                        data_dir: state.recent.get(&probe_id)
+                                            .and_then(|recent| recent.completion.object.as_ref())
+                                            .and_then(|object| object.resolved_data_dir),
+                                    }
+                                } else if pending {
+                                    ScannerHealTestAdmission::Pending
+                                } else if state.unknown_counts.values().copied().sum::<u64>() > unknown_before {
+                                    ScannerHealTestAdmission::Unknown
+                                } else {
+                                    ScannerHealTestAdmission::Unaccounted
+                                };
+                                let _ = test_events().send(ScannerHealTestEvent::Admission {
+                                    task_id: probe_id,
+                                    candidate_data_dir: probe_data_dir,
+                                    outcome,
+                                });
+                            }
                         },
                         None => break,
                     },
                     completion = completions.recv() => match completion {
                         Ok(completion) => {
-                            let _ = observations.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
-                                .complete(completion, Instant::now());
+                            let mut state = observations.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                            #[cfg(feature = "test-util")]
+                            let probe_id = completion.task_id.clone();
+                            #[cfg(feature = "test-util")]
+                            let probe_data_dir = completion.object.as_ref().and_then(|object| object.resolved_data_dir);
+                            #[cfg(feature = "test-util")]
+                            let duplicate = state.recent.contains_key(&probe_id);
+                            #[cfg(feature = "test-util")]
+                            let pending_before: Vec<_> = state.pending.get(&probe_id)
+                                .map(|pending| pending.iter().map(|candidate| candidate.data_dir).collect())
+                                .unwrap_or_default();
+                            let _settled = state.complete(completion, Instant::now());
+                            #[cfg(feature = "test-util")]
+                            if test_events().receiver_count() != 0 {
+                                let remaining_candidate_data_dirs = state.pending.get(&probe_id)
+                                    .map(|pending| pending.iter().map(|candidate| candidate.data_dir).collect())
+                                    .unwrap_or_default();
+                                let _ = test_events().send(ScannerHealTestEvent::Terminal {
+                                    task_id: probe_id,
+                                    terminal_data_dir: probe_data_dir,
+                                    duplicate,
+                                    settled_candidate_data_dirs: if _settled.is_some() { pending_before } else { Vec::new() },
+                                    remaining_candidate_data_dirs,
+                                });
+                            }
                         },
                         Err(broadcast::error::RecvError::Lagged(skipped)) => {
                             observations.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).lagged(skipped);
@@ -445,12 +696,14 @@ fn observation_status_from(registry: &Mutex<Option<ObserverHandle>>) -> ScannerH
             observations
                 .pending
                 .values()
+                .flat_map(Pending::iter)
                 .fold(0u64, |count, candidate| count.saturating_add(candidate.count)),
         );
         let now = Instant::now();
         status.oldest_pending_seconds = observations
             .pending
             .values()
+            .flat_map(Pending::iter)
             .map(|candidate| now.saturating_duration_since(candidate.first_seen))
             .max()
             .map(|age| age.as_secs_f64());
@@ -484,9 +737,9 @@ pub fn scanner_heal_observation_status() -> ScannerHealObservationStatus {
 }
 
 /// Capture only the identity actually available at discovery, before dispatch.
-pub(crate) fn prepare(request: &HealChannelRequest) -> Option<Candidate> {
+pub(crate) fn prepare(request: &HealChannelRequest, data_dir: Option<Uuid>) -> Option<Candidate> {
     let _ = observer();
-    Candidate::from_request(request)
+    Candidate::from_request(request, data_dir)
 }
 
 pub(crate) fn observe(candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {

crates/scanner/src/heal_completion/tests.rs

@@ -16,7 +16,7 @@ fn request() -> HealChannelRequest {
 }
 
 fn candidate() -> Candidate {
-    Candidate::from_request(&request()).expect("bounded candidate")
+    Candidate::from_request(&request(), Some(Uuid::from_u128(33))).expect("bounded candidate")
 }
 
 fn completion(task: &str, disposition: HealObjectDisposition) -> Arc<HealCompletion> {
@@ -28,6 +28,8 @@ fn completion(task: &str, disposition: HealObjectDisposition) -> Arc<HealComplet
         object: Some(HealObjectReceipt {
             identity: candidate().identity.expect("exact object identity"),
             disposition,
+            resolved_data_dir: Some(Uuid::from_u128(33)),
+            payload_integrity_verified: false,
         }),
     })
 }
@@ -79,6 +81,50 @@ fn terminal_before_admission_is_observed_without_claiming_coverage() {
     assert!(settled.first_task_observation);
 }
 
+#[test]
+fn reused_task_id_does_not_attach_an_old_terminal_to_a_new_execution() {
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let mut observations = Observations::default();
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    let old_terminal = completion("owner", HealObjectDisposition::VerifiedHealthy);
+    assert!(observations.complete(old_terminal, Instant::now()).is_some());
+
+    let mut new_candidate = candidate();
+    new_candidate.data_dir = Some(Uuid::from_u128(34));
+    let old_terminal_was_reused = observations.admit("owner".into(), new_candidate).is_some();
+    let mut new_terminal = (*completion("owner", HealObjectDisposition::VerifiedHealthy)).clone();
+    new_terminal.object.as_mut().expect("object receipt").resolved_data_dir = Some(Uuid::from_u128(34));
+    let _ = observations.complete(Arc::new(new_terminal), Instant::now());
+
+    assert!(
+        !old_terminal_was_reused,
+        "HEAL_ID_REUSE_UNIT_STALE_TERMINAL: a later execution cannot settle from an earlier terminal"
+    );
+    assert_eq!(
+        recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"),
+        0,
+        "both executions verified the generation discovered for their own candidate"
+    );
+}
+
+#[test]
+fn reused_task_id_with_unchanged_data_dir_does_not_reuse_the_old_terminal() {
+    let mut observations = Observations::default();
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    assert!(
+        observations
+            .complete(completion("owner", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_some()
+    );
+    let old_terminal_was_reused = observations.admit("owner".into(), candidate()).is_some();
+    let _ = observations.complete(completion("owner", HealObjectDisposition::VerifiedHealthy), Instant::now());
+    assert!(
+        !old_terminal_was_reused,
+        "HEAL_ID_REUSE_UNIT_SAME_DATA_STALE_TERMINAL: equal data directories do not identify an execution"
+    );
+}
+
 #[test]
 fn replacement_bucket_version_or_location_cannot_match_an_old_result() {
     for change in 0..4 {
@@ -98,6 +144,445 @@ fn replacement_bucket_version_or_location_cannot_match_an_old_result() {
     }
 }
 
+#[test]
+fn candidate_captures_only_non_nil_object_data_dir() {
+    let generation = Uuid::from_u128(33);
+    for (data_dir, expected) in [(Some(generation), Some(generation)), (Some(Uuid::nil()), None), (None, None)] {
+        let object = crate::ScannerObjectInfo {
+            data_dir,
+            ..Default::default()
+        };
+        let observed = Candidate::from_request(&request(), object.data_dir).expect("bounded object candidate");
+        assert_eq!(observed.data_dir, expected);
+    }
+}
+
+#[test]
+fn generation_scope_requires_known_matching_data_dirs() {
+    let generation = Uuid::from_u128(33);
+    let replacement = Uuid::from_u128(34);
+    for (expected, resolved, scope) in [
+        (Some(generation), Some(generation), "content_and_leader_unknown"),
+        (Some(generation), Some(replacement), "identity_mismatch"),
+        (None, Some(generation), "identity_unknown"),
+        (Some(generation), None, "identity_unknown"),
+        (None, None, "identity_unknown"),
+        (Some(Uuid::nil()), Some(generation), "identity_unknown"),
+        (Some(generation), Some(Uuid::nil()), "identity_unknown"),
+    ] {
+        let mut observed = candidate();
+        observed.data_dir = expected;
+        let mut terminal = (*completion("generation", HealObjectDisposition::VerifiedHealthy)).clone();
+        terminal.object.as_mut().expect("object receipt").resolved_data_dir = resolved;
+        let mut recent = Recent {
+            completion: Arc::new(terminal),
+            counted: false,
+        };
+        assert_eq!(Settlement::new(observed, &mut recent).scope, scope, "{expected:?}, {resolved:?}");
+    }
+}
+
+#[test]
+fn known_generation_conflict_is_not_hidden_by_missing_location() {
+    let mut observed = candidate();
+    observed.identity.as_mut().expect("candidate identity").pool_index = None;
+    let mut terminal = (*completion("generation", HealObjectDisposition::VerifiedHealthy)).clone();
+    let receipt = terminal.object.as_mut().expect("object receipt");
+    receipt.identity.set_index = None;
+    receipt.resolved_data_dir = Some(Uuid::from_u128(34));
+    let mut recent = Recent {
+        completion: Arc::new(terminal),
+        counted: false,
+    };
+    assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_mismatch");
+}
+
+#[test]
+fn merged_generations_settle_independently_in_any_arrival_order() {
+    for terminal_first in [false, true] {
+        for reversed in [false, true] {
+            let recorder = CompletionRecorder::default();
+            let _recorder = metrics::set_default_local_recorder(&recorder);
+            let mut observations = Observations::default();
+            let terminal = completion("owner", HealObjectDisposition::VerifiedHealthy);
+            if terminal_first {
+                assert!(observations.complete(terminal.clone(), Instant::now()).is_none());
+            }
+            let mut generations = [Some(Uuid::from_u128(33)), Some(Uuid::from_u128(34)), None];
+            if reversed {
+                generations.reverse();
+            }
+            for generation in generations {
+                let mut observed = candidate();
+                observed.data_dir = generation;
+                assert_eq!(observations.admit("owner".into(), observed).is_some(), terminal_first);
+            }
+            if !terminal_first {
+                assert!(observations.complete(terminal.clone(), Instant::now()).is_some());
+            }
+            assert!(observations.complete(terminal, Instant::now()).is_none(), "duplicate terminal ignored");
+            assert_eq!(observations.pending_groups, 0);
+            assert_eq!(observations.terminal_tasks, 1);
+            assert_eq!(observations.unknown_counts.get("content_and_leader_unknown"), Some(&1));
+            assert_eq!(observations.unknown_counts.get("identity_mismatch"), Some(&1));
+            assert_eq!(observations.unknown_counts.get("identity_unknown"), Some(&1));
+            assert_eq!(observations.terminal_counts.values().sum::<u64>(), 3);
+            assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 1);
+        }
+    }
+}
+
+#[test]
+fn pending_group_limit_is_global_and_overflow_stays_unknown() {
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let mut observations = Observations::default();
+    for index in 1..=MAX_PENDING_GROUPS {
+        let mut observed = candidate();
+        observed.data_dir = Some(Uuid::from_u128(u128::try_from(index).expect("group index")));
+        assert!(observations.admit("owner".into(), observed).is_none());
+    }
+    assert_eq!(observations.pending_groups, MAX_PENDING_GROUPS);
+    assert_eq!(observations.pending.len(), 1);
+    let mut overflow = candidate();
+    overflow.data_dir = Some(Uuid::from_u128(2000));
+    assert!(observations.admit("owner".into(), overflow).is_none());
+    assert!(observations.admit("another-owner".into(), candidate()).is_none());
+    assert_eq!(observations.unknown_counts.get("pending_groups_full"), Some(&2));
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    assert_eq!(observations.pending_groups, MAX_PENDING_GROUPS);
+
+    assert!(
+        observations
+            .complete(completion("owner", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_some()
+    );
+    assert_eq!(observations.pending_groups, 0);
+    assert_eq!(observations.terminal_tasks, 1);
+    assert_eq!(observations.unknown_counts.get("content_and_leader_unknown"), Some(&2));
+    assert_eq!(
+        observations.unknown_counts.get("identity_mismatch"),
+        Some(&(u64::try_from(MAX_PENDING_GROUPS).expect("bounded group count") - 1))
+    );
+    assert_eq!(
+        recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"),
+        u64::try_from(MAX_PENDING_GROUPS).expect("bounded group count") - 1
+    );
+}
+
+#[test]
+fn insufficient_scan_and_generation_mismatch_are_both_counted() {
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let mut observations = Observations::default();
+    observations.admit("owner".into(), candidate());
+    observations.admit("owner".into(), candidate());
+    let mut terminal = (*completion("owner", HealObjectDisposition::MetadataHealthy)).clone();
+    terminal.scan_mode = HealScanMode::Normal;
+    terminal.object.as_mut().expect("object receipt").resolved_data_dir = Some(Uuid::from_u128(34));
+    let terminal = Arc::new(terminal);
+    let settled = observations
+        .complete(terminal.clone(), Instant::now())
+        .expect("canonical terminal observed");
+    assert_eq!(settled.scope, "insufficient_scan");
+    assert!(settled.generation_mismatch);
+    assert_eq!(observations.unknown_counts.get("insufficient_scan"), Some(&2));
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 2);
+    assert!(observations.complete(terminal, Instant::now()).is_none());
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 2);
+
+    let mut matching = candidate();
+    matching.data_dir = Some(Uuid::from_u128(34));
+    let settled = observations.admit("owner".into(), matching).expect("late merged admission");
+    assert_eq!(settled.scope, "insufficient_scan");
+    assert!(!settled.generation_mismatch);
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 2);
+    assert_eq!(observations.terminal_tasks, 1);
+}
+
+#[test]
+fn proved_payload_task_is_counted_once_for_either_terminal_order() {
+    for terminal_first in [false, true] {
+        for requested_mode in [HealScanMode::Normal, HealScanMode::Deep] {
+            let recorder = CompletionRecorder::default();
+            let _recorder = metrics::set_default_local_recorder(&recorder);
+            let mut observations = Observations::default();
+            let mut terminal = (*completion("owner", HealObjectDisposition::VerifiedHealthy)).clone();
+            terminal.scan_mode = requested_mode;
+            terminal.object.as_mut().expect("object receipt").payload_integrity_verified = true;
+            let terminal = Arc::new(terminal);
+            if terminal_first {
+                assert!(observations.complete(terminal.clone(), Instant::now()).is_none());
+            }
+            observations.admit("owner".into(), candidate());
+            observations.admit("owner".into(), candidate());
+            if !terminal_first {
+                assert!(observations.complete(terminal.clone(), Instant::now()).is_some());
+            }
+            assert!(observations.complete(terminal, Instant::now()).is_none(), "duplicate terminal ignored");
+            assert_eq!(observations.terminal_tasks, 1);
+            assert_eq!(observations.unknown_counts.get("content_and_leader_unknown"), Some(&2));
+            assert_eq!(recorder.total("rustfs_scanner_heal_task_deep_payload_verified_observations_total"), 1);
+        }
+    }
+}
+
+#[test]
+fn task_payload_counter_requires_completed_nondry_proved_receipt() {
+    for change in 0..6 {
+        let recorder = CompletionRecorder::default();
+        let _recorder = metrics::set_default_local_recorder(&recorder);
+        let mut observations = Observations::default();
+        observations.admit("owner".into(), candidate());
+        let mut terminal = (*completion("owner", HealObjectDisposition::VerifiedHealthy)).clone();
+        terminal.object.as_mut().expect("object receipt").payload_integrity_verified = true;
+        match change {
+            0 => terminal.object.as_mut().expect("object receipt").payload_integrity_verified = false,
+            1 => terminal.object = None,
+            2 => terminal.dry_run = true,
+            3 => terminal.execution = None,
+            4 => terminal.execution = Some(HealExecutionOutcome::CompletedWithErrors),
+            _ => terminal.execution = Some(HealExecutionOutcome::Aborted(HealAbortReason::Cancelled)),
+        }
+        let settled = observations
+            .complete(Arc::new(terminal), Instant::now())
+            .expect("terminal observed");
+        assert!(!settled.deep_payload_verified_task, "case {change}");
+        assert_eq!(
+            recorder.total("rustfs_scanner_heal_task_deep_payload_verified_observations_total"),
+            0,
+            "case {change}"
+        );
+    }
+}
+
+#[test]
+fn normal_request_proof_only_satisfies_deep_mode_for_the_same_generation() {
+    for change in 0..7 {
+        let mut observed = candidate();
+        let mut terminal = (*completion("owner", HealObjectDisposition::VerifiedHealthy)).clone();
+        terminal.scan_mode = HealScanMode::Normal;
+        let receipt = terminal.object.as_mut().expect("object receipt");
+        receipt.payload_integrity_verified = true;
+        match change {
+            0 => {}
+            1 => observed.identity.as_mut().expect("candidate identity").version_id = None,
+            2 => observed.data_dir = None,
+            3 => receipt.resolved_data_dir = None,
+            4 => receipt.resolved_data_dir = Some(Uuid::from_u128(34)),
+            5 => receipt.identity.object = "another-object".into(),
+            _ => receipt.payload_integrity_verified = false,
+        }
+        let mut recent = Recent {
+            completion: Arc::new(terminal),
+            counted: false,
+        };
+        let settled = Settlement::new(observed, &mut recent);
+        let expected = match change {
+            0 => "content_and_leader_unknown",
+            1 => "identity_unknown",
+            _ => "insufficient_scan",
+        };
+        assert_eq!(settled.scope, expected, "case {change}");
+        assert_eq!(settled.generation_mismatch, change == 4, "case {change}");
+        assert_eq!(settled.deep_payload_verified_task, change != 6, "case {change}");
+    }
+}
+
+#[test]
+fn merged_candidates_keep_scan_mode_and_dry_run_scopes_separate() {
+    let recorder = CompletionRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let mut observations = Observations::default();
+    let mut normal = candidate();
+    normal.mode = HealScanMode::Normal;
+    let mut dry = candidate();
+    dry.dry_run = true;
+    observations.admit("owner".into(), candidate());
+    observations.admit("owner".into(), normal);
+    observations.admit("owner".into(), dry);
+    assert_eq!(observations.pending_groups, 3);
+    let mut terminal = (*completion("owner", HealObjectDisposition::MetadataHealthy)).clone();
+    terminal.scan_mode = HealScanMode::Normal;
+    terminal.object.as_mut().expect("object receipt").resolved_data_dir = Some(Uuid::from_u128(34));
+    observations.complete(Arc::new(terminal), Instant::now());
+    assert_eq!(observations.unknown_counts.get("insufficient_scan"), Some(&1));
+    assert_eq!(observations.unknown_counts.get("identity_mismatch"), Some(&1));
+    assert_eq!(observations.unknown_counts.get("dry_run"), Some(&1));
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 3);
+    assert_eq!(observations.terminal_tasks, 1);
+}
+
+#[test]
+fn generation_mismatch_counter_requires_matching_object_identity_and_known_data_dirs() {
+    for change in 0..6 {
+        let mut observed = candidate();
+        let mut terminal = (*completion("unrelated", HealObjectDisposition::VerifiedHealthy)).clone();
+        let receipt = terminal.object.as_mut().expect("object receipt");
+        receipt.resolved_data_dir = Some(Uuid::from_u128(34));
+        match change {
+            0 => receipt.identity.bucket = "another-bucket".into(),
+            1 => receipt.identity.object = "another-object".into(),
+            2 => receipt.identity.kind = HealObjectKind::Metadata,
+            3 => observed.data_dir = None,
+            4 => observed.data_dir = Some(Uuid::nil()),
+            _ => receipt.resolved_data_dir = Some(Uuid::nil()),
+        }
+        let mut recent = Recent {
+            completion: Arc::new(terminal),
+            counted: false,
+        };
+        assert!(!Settlement::new(observed, &mut recent).generation_mismatch, "case {change}");
+    }
+}
+
+#[test]
+fn group_expiry_and_lag_release_capacity_without_settling_candidates() {
+    let mut observations = Observations::default();
+    let now = Instant::now();
+    let mut old = candidate();
+    old.first_seen = now;
+    let mut newer = candidate();
+    newer.data_dir = Some(Uuid::from_u128(34));
+    newer.first_seen = now + Duration::from_secs(1);
+    observations.admit("owner".into(), old);
+    observations.admit("owner".into(), newer);
+    observations.maintain(now + PENDING_TTL);
+    assert_eq!(observations.pending_groups, 1);
+    assert_eq!(observations.pending["owner"].first.data_dir, Some(Uuid::from_u128(34)));
+    assert_eq!(observations.unknown_counts.get("expired"), Some(&1));
+
+    observations.lagged(1);
+    assert_eq!(observations.pending_groups, 0);
+    assert_eq!(observations.unknown_counts.get("completion_lost"), Some(&1));
+    assert!(
+        observations
+            .complete(completion("owner", HealObjectDisposition::VerifiedHealthy), Instant::now())
+            .is_none()
+    );
+    assert_eq!(observations.terminal_tasks, 0);
+    assert!(observations.admit("new-owner".into(), candidate()).is_none());
+    assert_eq!(observations.pending_groups, 1);
+}
+
+#[test]
+fn matching_merged_generations_still_reject_a_replacement_terminal() {
+    let mut observations = Observations::default();
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    assert!(observations.admit("owner".into(), candidate()).is_none());
+    let mut terminal = (*completion("owner", HealObjectDisposition::VerifiedHealthy)).clone();
+    terminal.object.as_mut().expect("object receipt").resolved_data_dir = Some(Uuid::from_u128(34));
+    let settled = observations
+        .complete(Arc::new(terminal), Instant::now())
+        .expect("canonical terminal observed");
+    assert_eq!(settled.candidates, 2);
+    assert_eq!(settled.scope, "identity_mismatch");
+    assert_eq!(observations.unknown_counts.get("identity_mismatch"), Some(&2));
+}
+
+#[test]
+fn early_terminal_compares_each_late_admission_generation() {
+    let mut observations = Observations::default();
+    let terminal = completion("fast", HealObjectDisposition::VerifiedHealthy);
+    assert!(observations.complete(terminal.clone(), Instant::now()).is_none());
+    let mut replaced = candidate();
+    replaced.data_dir = Some(Uuid::from_u128(34));
+    let first = observations.admit("fast".into(), replaced).expect("early terminal retained");
+    assert_eq!(first.scope, "identity_mismatch");
+    assert!(first.first_task_observation);
+    assert!(observations.complete(terminal, Instant::now()).is_none(), "duplicate terminal ignored");
+    let second = observations
+        .admit("fast".into(), candidate())
+        .expect("merged admission observed");
+    assert_eq!(second.scope, "content_and_leader_unknown");
+    assert!(!second.first_task_observation);
+    assert_eq!(observations.terminal_tasks, 1);
+}
+
+#[test]
+fn missing_identity_fields_are_unknown_on_either_side() {
+    let changes: [(&str, fn(&mut HealObjectIdentity)); 6] = [
+        ("version", |identity| identity.version_id = None),
+        ("empty version", |identity| identity.version_id = Some(String::new())),
+        ("incarnation", |identity| identity.bucket_incarnation_id = None),
+        ("nil incarnation", |identity| identity.bucket_incarnation_id = Some(Uuid::nil())),
+        ("pool", |identity| identity.pool_index = None),
+        ("set", |identity| identity.set_index = None),
+    ];
+    for (label, change) in changes {
+        for candidate_missing in [true, false] {
+            let mut observed = candidate();
+            let mut terminal = (*completion("partial", HealObjectDisposition::VerifiedHealthy)).clone();
+            if candidate_missing {
+                change(observed.identity.as_mut().expect("candidate identity"));
+            } else {
+                change(&mut terminal.object.as_mut().expect("receipt").identity);
+            }
+            let mut recent = Recent {
+                completion: Arc::new(terminal),
+                counted: false,
+            };
+            assert_eq!(
+                Settlement::new(observed, &mut recent).scope,
+                "identity_unknown",
+                "{label}, candidate missing: {candidate_missing}"
+            );
+        }
+    }
+}
+
+#[test]
+fn known_identity_conflicts_are_not_hidden_by_other_missing_fields() {
+    for change in 0..3 {
+        let mut observed = candidate();
+        let mut terminal = (*completion("conflict", HealObjectDisposition::VerifiedHealthy)).clone();
+        let expected = observed.identity.as_mut().expect("candidate identity");
+        let actual = &mut terminal.object.as_mut().expect("receipt").identity;
+        match change {
+            0 => {
+                expected.version_id = None;
+                actual.bucket = "replacement".into();
+            }
+            1 => {
+                expected.pool_index = None;
+                actual.version_id = Some(Uuid::from_u128(12).to_string());
+            }
+            _ => {
+                expected.bucket_incarnation_id = None;
+                actual.set_index = Some(2);
+            }
+        }
+        let mut recent = Recent {
+            completion: Arc::new(terminal),
+            counted: false,
+        };
+        assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_mismatch", "case {change}");
+    }
+}
+
+#[test]
+fn scanner_candidate_without_location_records_identity_unknown() {
+    let mut scanner_request = request();
+    scanner_request.expected_bucket_incarnation_id = None;
+    scanner_request.pool_index = None;
+    scanner_request.set_index = None;
+    let mut observations = Observations::default();
+    assert!(
+        observations
+            .admit(
+                "scanner".into(),
+                Candidate::from_request(&scanner_request, Some(Uuid::from_u128(33))).expect("bounded scanner candidate"),
+            )
+            .is_none()
+    );
+    let settlement = observations
+        .complete(completion("scanner", HealObjectDisposition::VerifiedHealthy), Instant::now())
+        .expect("terminal candidate observed");
+    assert_eq!(settlement.scope, "identity_unknown");
+    assert_eq!(observations.unknown_counts.get("identity_unknown"), Some(&1));
+    assert_eq!(observations.unknown_counts.get("identity_mismatch"), None);
+}
+
 #[test]
 fn unversioned_overwrite_and_unknown_leader_never_certify_payload() {
     let mut observed = candidate();
@@ -228,7 +713,7 @@ fn terminal_history_eviction_loses_evidence_without_inventing_a_completion() {
 fn identity_size_is_bounded_before_cloning() {
     let mut large = request();
     large.object_prefix = Some("x".repeat(MAX_COMPLETION_IDENTITY_BYTES));
-    assert!(Candidate::from_request(&large).is_none());
+    assert!(Candidate::from_request(&large, Some(Uuid::from_u128(33))).is_none());
 }
 
 #[test]
@@ -346,6 +831,57 @@ fn observer_status_preserves_loss_expiry_and_rejected_admissions() {
     assert_eq!(unknown["observer_full_or_closed"], 2);
 }
 
+#[tokio::test]
+async fn broadcast_terminal_rejects_overwrite_for_merged_candidates() {
+    use rustfs_heal_contracts::{completion::publish_heal_completion, heal_channel::HealAdmissionResult};
+
+    let registry = Mutex::new(None);
+    let observer = observer_from(&registry);
+    let canonical = Uuid::new_v4().to_string();
+    let object = crate::ScannerObjectInfo {
+        data_dir: Some(Uuid::from_u128(33)),
+        ..Default::default()
+    };
+    for result in [HealAdmissionResult::Accepted, HealAdmissionResult::Merged] {
+        observe_on(
+            &observer,
+            Candidate::from_request(&request(), object.data_dir),
+            Ok(&HealAdmissionReceipt {
+                result,
+                task_id: canonical.clone(),
+            }),
+        );
+    }
+    tokio::time::timeout(Duration::from_secs(2), async {
+        while observation_status_from(&registry).retained_pending_candidates != Some(2) {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("both admissions must reach the observer before the terminal");
+
+    let mut terminal = (*completion(&canonical, HealObjectDisposition::VerifiedHealthy)).clone();
+    terminal.object.as_mut().expect("object receipt").resolved_data_dir = Some(Uuid::from_u128(34));
+    publish_heal_completion(terminal);
+    tokio::time::timeout(Duration::from_secs(2), async {
+        while observation_status_from(&registry).terminal_task_observations_total != Some(1) {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("broadcast terminal must reach the observer");
+
+    let status = observation_status_from(&registry);
+    assert_eq!(status.retained_pending_candidates, Some(0));
+    assert_eq!(status.integrity_state, "unknown");
+    assert_eq!(status.terminal_candidates.len(), 1);
+    assert_eq!(status.terminal_candidates[0].scope, "identity_mismatch");
+    assert_eq!(status.terminal_candidates[0].count, 2);
+    assert_eq!(status.unknown_candidates.len(), 1);
+    assert_eq!(status.unknown_candidates[0].reason, "identity_mismatch");
+    assert_eq!(status.unknown_candidates[0].count, 2);
+}
+
 #[tokio::test]
 async fn observer_status_tracks_canonical_terminals_without_a_metrics_recorder() {
     use rustfs_heal_contracts::{completion::publish_heal_completion, heal_channel::HealAdmissionResult};
@@ -491,9 +1027,9 @@ async fn live_observer_associates_early_completion_with_canonical_merged_receipt
         result: HealAdmissionResult::Merged,
         task_id: canonical_id,
     };
-    observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged));
+    observe_on(&sender, Candidate::from_request(&source_request, Some(Uuid::from_u128(33))), Ok(&merged));
     recorder.wait_for_terminals(2).await;
-    observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged));
+    observe_on(&sender, Candidate::from_request(&source_request, Some(Uuid::from_u128(33))), Ok(&merged));
     recorder.wait_for_terminals(3).await;
 
     assert_eq!(recorder.total("rustfs_scanner_heal_candidate_terminal_total"), 3);

crates/scanner/src/lib.rs

@@ -81,6 +81,9 @@ pub use error::ScannerError;
 pub use heal_completion::{
     ScannerHealObservationStatus, ScannerHealTerminalObservation, ScannerHealUnknownObservation, scanner_heal_observation_status,
 };
+#[cfg(feature = "test-util")]
+#[doc(hidden)]
+pub use heal_completion::{ScannerHealTestAdmission, ScannerHealTestEvent, subscribe_scanner_heal_test_events};
 pub use prefix_usage::{BucketPrefixUsageResponse, bucket_prefix_usage, invalidate_prefix_usage_cache};
 pub use remote_scanner::{
     NS_SCANNER_MAX_REQUEST_BODY_SIZE, RemoteScannerAdmission, RemoteScannerRequest, admit_remote_scanner_request,

crates/scanner/src/scanner_folder.rs

@@ -1603,7 +1603,7 @@ impl FolderScanner {
         let candidate_type = pending_scanner_heal_candidate_type(kind);
         let priority = request.priority;
         let scan_mode = request.scan_mode.unwrap_or(self.scan_mode);
-        let result = match send_scanner_heal_request(candidate_type, request).await {
+        let result = match send_scanner_heal_request(candidate_type, request, None).await {
             Ok(result) => result,
             Err(err) => {
                 self.update_pending_scanner_heal_after_admission(

crates/scanner/src/scanner_folder/item_actions.rs

@@ -477,10 +477,11 @@ pub(super) fn record_heal_candidate_admission(
 pub(super) async fn send_scanner_heal_request(
     candidate_type: &'static str,
     request: HealChannelRequest,
+    data_dir: Option<uuid::Uuid>,
 ) -> Result<HealAdmissionResult, ScannerError> {
     let priority = request.priority;
     let trace_context = scanner_heal_candidate_trace_context(&request);
-    let candidate = crate::heal_completion::prepare(&request);
+    let candidate = crate::heal_completion::prepare(&request, data_dir);
     match send_heal_request_with_receipt(request).await {
         Ok(receipt) => {
             crate::heal_completion::observe(candidate, Ok(&receipt));
@@ -1161,6 +1162,7 @@ impl ScannerItem {
                 scan_mode,
                 HealChannelPriority::Low,
             ),
+            oi.data_dir,
         )
         .await;
 

crates/scanner/tests/heal_generation_observation.rs

@@ -0,0 +1,452 @@
+// 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.
+
+//! One local process exercises the scanner, heal manager, and real ECStore.
+
+use rustfs_ecstore::api::object::{ShardIntegrityWriteMode, WriteCommitGuard, WriteCompletion};
+#[cfg(feature = "test-util")]
+use rustfs_heal::heal::task::{HealRequest, HealTaskStatus};
+use rustfs_heal::{
+    HealChannelProcessor,
+    heal::{
+        manager::{HealConfig, HealManager},
+        storage::ECStoreHealStorage,
+    },
+};
+#[cfg(feature = "test-util")]
+use rustfs_heal_contracts::{
+    completion::HealCompletion,
+    heal_channel::{HealAdmissionResult, HealChannelResponse, HealRequestSource, subscribe_heal_responses},
+};
+use rustfs_heal_contracts::{
+    completion::subscribe_heal_completions,
+    heal_channel::{HealScanMode, init_heal_channels},
+    outcome::HealExecutionOutcome,
+};
+#[cfg(feature = "test-util")]
+use rustfs_scanner::{ScannerHealTestAdmission, ScannerHealTestEvent, ScannerObjectInfo, subscribe_scanner_heal_test_events};
+use rustfs_scanner::{
+    ScannerObjectOptions, ScannerPutObjReader, SizeSummary, scanner_folder::ScannerItem, scanner_heal_observation_status,
+};
+use rustfs_storage_api::{NamespaceLocking, ObjectIO, ObjectOperations};
+use rustfs_test_utils::TestECStoreEnv;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex, atomic::AtomicU64, atomic::Ordering},
+    time::Duration,
+};
+use time::OffsetDateTime;
+#[cfg(feature = "test-util")]
+use tokio::sync::broadcast;
+#[cfg(feature = "test-util")]
+use uuid::Uuid;
+
+#[derive(Default)]
+struct CounterRecorder(Mutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
+
+impl metrics::Recorder for CounterRecorder {
+    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
+    fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        metrics::Counter::from_arc(
+            self.0
+                .lock()
+                .expect("counter recorder")
+                .entry(key.clone())
+                .or_default()
+                .clone(),
+        )
+    }
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+    fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
+        metrics::Histogram::noop()
+    }
+}
+
+impl CounterRecorder {
+    fn total(&self, name: &str) -> u64 {
+        self.0
+            .lock()
+            .expect("counter recorder")
+            .iter()
+            .filter(|(key, _)| key.name() == name)
+            .map(|(_, value)| value.load(Ordering::Relaxed))
+            .sum()
+    }
+}
+
+fn stack_probe_stage(stage: &str) {
+    use std::io::Write;
+
+    let mut stderr = std::io::stderr().lock();
+    let _ = writeln!(stderr, "HEAL_ID_REUSE_STACK_STAGE: {stage}");
+    let _ = stderr.flush();
+}
+
+async fn put_protected(env: &TestECStoreEnv, bucket: &str, object: &str, fill: u8, mod_time: OffsetDateTime) {
+    let lock = env.ecstore.new_ns_lock(bucket, object).await.expect("fixture namespace lock");
+    let guard = WriteCommitGuard::acquire(&lock, Duration::from_secs(30))
+        .await
+        .expect("fixture write ownership");
+    let mut options = ScannerObjectOptions {
+        mod_time: Some(mod_time),
+        shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected),
+        write_completion: WriteCompletion::TailDrained,
+        ..Default::default()
+    };
+    options.add_write_commit_guard(&guard);
+    env.ecstore
+        .put_object(bucket, object, &mut ScannerPutObjReader::from_vec(vec![fill; 256 * 1024 + 137]), &options)
+        .await
+        .expect("write protected object to the real erasure set");
+}
+
+#[cfg(feature = "test-util")]
+async fn wait_for_observer_admission(
+    events: &mut broadcast::Receiver<ScannerHealTestEvent>,
+    owner: &str,
+    data_dir: Uuid,
+) -> ScannerHealTestAdmission {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            match events.recv().await.expect("observer test event stream") {
+                ScannerHealTestEvent::Admission {
+                    task_id,
+                    candidate_data_dir: Some(observed),
+                    outcome,
+                } if task_id == owner && observed == data_dir => break outcome,
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("observer must process the candidate admission")
+}
+
+#[cfg(feature = "test-util")]
+async fn wait_for_observer_terminal(
+    events: &mut broadcast::Receiver<ScannerHealTestEvent>,
+    owner: &str,
+    data_dir: Uuid,
+) -> (bool, Vec<Option<Uuid>>, Vec<Option<Uuid>>) {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            match events.recv().await.expect("observer test event stream") {
+                ScannerHealTestEvent::Terminal {
+                    task_id,
+                    terminal_data_dir: Some(observed),
+                    duplicate,
+                    settled_candidate_data_dirs,
+                    remaining_candidate_data_dirs,
+                } if task_id == owner && observed == data_dir => {
+                    break (duplicate, settled_candidate_data_dirs, remaining_candidate_data_dirs);
+                }
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("observer must process the requested terminal")
+}
+
+#[cfg(feature = "test-util")]
+async fn wait_for_idle(manager: &HealManager) {
+    tokio::time::timeout(Duration::from_secs(30), async {
+        while manager.get_active_tasks_count().await != 0 || manager.get_queue_length().await != 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("the previous execution must leave the active queue");
+}
+
+#[cfg(feature = "test-util")]
+async fn run_reused_owner_round(
+    manager: &HealManager,
+    scanner: &mut ScannerItem,
+    admissions: &mut broadcast::Receiver<HealChannelResponse>,
+    completions: &mut broadcast::Receiver<Arc<HealCompletion>>,
+    observer_events: &mut broadcast::Receiver<ScannerHealTestEvent>,
+    owner: &str,
+    candidate: ScannerObjectInfo,
+    data_dir: Uuid,
+) -> (ScannerHealTestAdmission, (bool, Vec<Option<Uuid>>, Vec<Option<Uuid>>)) {
+    manager.set_max_concurrent_heals_for_test(0).await;
+    let mut request = HealRequest::object(scanner.bucket.clone(), scanner.object_path(), None);
+    request.id = owner.to_owned();
+    request.source = HealRequestSource::Scanner;
+    request.options.scan_mode = HealScanMode::Deep;
+    request.options.recreate_missing = false;
+    let receipt = manager
+        .submit_heal_request_with_receipt(request)
+        .await
+        .expect("submit new execution with reused task ID");
+    assert_eq!(receipt.result, HealAdmissionResult::Accepted);
+    assert_eq!(receipt.task_id, owner);
+
+    temp_env::async_with_vars([("RUSTFS_SCANNER_DEEP_VERIFY_COOLDOWN_SECS", Some("0"))], async {
+        scanner
+            .apply_actions(vec![candidate], None, Default::default(), &[], &mut SizeSummary::default())
+            .await;
+    })
+    .await;
+    let response = tokio::time::timeout(Duration::from_secs(30), admissions.recv())
+        .await
+        .expect("scanner admission response deadline")
+        .expect("scanner admission response delivered");
+    assert_eq!(response.request_id, owner, "scanner must merge into the new owner");
+    assert!(response.success);
+    assert_eq!(response.data.as_deref(), Some(&b"admission=merged,reason=none"[..]));
+    assert_eq!(
+        manager.get_task_status(owner).await.expect("new owner status"),
+        HealTaskStatus::Pending,
+        "new owner must remain queued while the observer handles admission"
+    );
+    let admission = wait_for_observer_admission(observer_events, owner, data_dir).await;
+    assert_ne!(admission, ScannerHealTestAdmission::Unaccounted, "observer did not account for admission");
+
+    manager.set_max_concurrent_heals_for_test(1).await;
+    let terminal = tokio::time::timeout(Duration::from_secs(30), completions.recv())
+        .await
+        .expect("new execution terminal deadline")
+        .expect("new terminal delivered");
+    assert_eq!(terminal.task_id, owner);
+    assert_eq!(terminal.execution, Some(HealExecutionOutcome::Completed));
+    assert_eq!(terminal.object.as_ref().expect("new storage receipt").resolved_data_dir, Some(data_dir));
+    let observed_terminal = wait_for_observer_terminal(observer_events, owner, data_dir).await;
+    wait_for_idle(manager).await;
+    (admission, observed_terminal)
+}
+
+#[cfg(feature = "test-util")]
+fn assert_observer_resolution(
+    admission: &ScannerHealTestAdmission,
+    terminal: &(bool, Vec<Option<Uuid>>, Vec<Option<Uuid>>),
+    data_dir: Uuid,
+) {
+    let (duplicate, settled, remaining) = terminal;
+    assert!(
+        remaining.is_empty(),
+        "the owner's candidate must not remain pending after terminal processing"
+    );
+    match admission {
+        ScannerHealTestAdmission::Pending => {
+            assert!(!duplicate, "the new terminal must not be discarded as a duplicate");
+            assert_eq!(settled.as_slice(), &[Some(data_dir)], "only the expected candidate must be settled");
+        }
+        ScannerHealTestAdmission::Unknown => {}
+        ScannerHealTestAdmission::CachedTerminal { .. } | ScannerHealTestAdmission::Unaccounted => {
+            panic!("unexpected admission state after stale-terminal assertion")
+        }
+    }
+}
+
+#[tokio::test]
+async fn overwritten_unversioned_generation_is_not_attributed_to_the_scanned_candidate() {
+    stack_probe_stage("test body entered");
+    let root = tempfile::tempdir().expect("local erasure-set fixture");
+    stack_probe_stage("ECStore build started");
+    let env = TestECStoreEnv::builder().base_dir(root.path()).build().await;
+    stack_probe_stage("ECStore build finished");
+    let bucket = "scanner-generation";
+    let object = "overwritten-object";
+    stack_probe_stage("bucket creation started");
+    env.make_bucket(bucket, false).await;
+
+    // The old timestamp exercises the scanner's real Deep selection without
+    // altering the ObjectInfo returned by storage.
+    let now = OffsetDateTime::now_utc();
+    stack_probe_stage("generation A write started");
+    put_protected(&env, bucket, object, 0x41, now - time::Duration::minutes(4)).await;
+    stack_probe_stage("generation A write finished");
+    let discovered = env
+        .ecstore
+        .get_object_info(bucket, object, &ScannerObjectOptions::default())
+        .await
+        .expect("read generation A from storage");
+    assert!(discovered.version_id.is_none_or(|id| id.is_nil()), "fixture must be unversioned");
+    let generation_a = discovered.data_dir.filter(|id| !id.is_nil()).expect("generation A");
+
+    stack_probe_stage("generation B write started");
+    put_protected(&env, bucket, object, 0x42, now - time::Duration::minutes(3)).await;
+    stack_probe_stage("generation B write finished");
+    let current = env
+        .ecstore
+        .get_object_info(bucket, object, &ScannerObjectOptions::default())
+        .await
+        .expect("read generation B from storage");
+    assert!(current.version_id.is_none_or(|id| id.is_nil()), "replacement must remain unversioned");
+    let generation_b = current.data_dir.filter(|id| !id.is_nil()).expect("generation B");
+    assert_ne!(generation_a, generation_b, "the overwrite must change data_dir");
+
+    let manager = Arc::new(HealManager::new(
+        Arc::new(ECStoreHealStorage::new(env.ecstore.clone())),
+        Some(HealConfig {
+            enable_auto_heal: false,
+            mainline_throttle_enable: false,
+            max_concurrent_heals: 1,
+            low_priority_merge_enable: true,
+            event_driven_scheduler_enable: true,
+            heal_interval: Duration::from_millis(10),
+            ..Default::default()
+        }),
+    ));
+    stack_probe_stage("heal manager start started");
+    manager.start().await.expect("start production heal manager");
+    stack_probe_stage("heal manager start finished");
+    let (commands, receipts) = init_heal_channels().expect("initialize process-local heal channels");
+    let processor_manager = manager.clone();
+    let processor = tokio::spawn(async move {
+        HealChannelProcessor::new(processor_manager)
+            .start_with_receipts(commands, receipts)
+            .await
+    });
+    let mut completions = subscribe_heal_completions();
+    #[cfg(feature = "test-util")]
+    let mut observer_events = subscribe_scanner_heal_test_events();
+    let recorder = CounterRecorder::default();
+    let _recorder = metrics::set_default_local_recorder(&recorder);
+    let mut scanner = ScannerItem {
+        path: String::new(),
+        bucket: bucket.into(),
+        prefix: String::new(),
+        object_name: object.into(),
+        file_type: std::fs::metadata(root.path()).expect("fixture root metadata").file_type(),
+        lifecycle: None,
+        object_lock: None,
+        replication: None,
+        heal_enabled: true,
+        heal_bitrot: true,
+        debug: false,
+    };
+    stack_probe_stage("scanner apply started");
+    temp_env::async_with_vars([("RUSTFS_SCANNER_DEEP_VERIFY_COOLDOWN_SECS", Some("0"))], async {
+        scanner
+            .apply_actions(vec![discovered], None, Default::default(), &[], &mut SizeSummary::default())
+            .await;
+    })
+    .await;
+    stack_probe_stage("scanner apply finished");
+
+    let terminal = tokio::time::timeout(Duration::from_secs(30), completions.recv())
+        .await
+        .expect("manager terminal deadline")
+        .expect("manager terminal delivered");
+    stack_probe_stage("initial heal terminal received");
+    assert_eq!(terminal.execution, Some(HealExecutionOutcome::Completed));
+    assert_eq!(terminal.scan_mode, HealScanMode::Deep);
+    assert!(!terminal.dry_run);
+    let receipt = terminal.object.as_ref().expect("production object receipt");
+    assert_eq!((&*receipt.identity.bucket, &*receipt.identity.object), (bucket, object));
+    assert_eq!(receipt.resolved_data_dir, Some(generation_b), "terminal must describe generation B");
+    assert!(
+        receipt.payload_integrity_verified,
+        "real protected Deep scan must prove the executed payload"
+    );
+
+    let status = tokio::time::timeout(Duration::from_secs(30), async {
+        loop {
+            let status = scanner_heal_observation_status();
+            if status.terminal_task_observations_total == Some(1) {
+                break status;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("scanner must observe the production terminal");
+    assert_eq!(status.integrity_state, "unknown");
+    assert_eq!(status.retained_pending_candidates, Some(0));
+    assert_eq!(status.terminal_candidates.len(), 1);
+    assert_eq!(status.terminal_candidates[0].scope, "identity_mismatch");
+    assert_eq!(status.terminal_candidates[0].scan_mode, HealScanMode::Deep.as_str());
+    assert_eq!(status.terminal_candidates[0].count, 1);
+    assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 1);
+    assert_eq!(recorder.total("rustfs_scanner_heal_task_deep_payload_verified_observations_total"), 1);
+
+    #[cfg(feature = "test-util")]
+    {
+        let owner = terminal.task_id.clone();
+        let initial = scanner_heal_observation_status();
+        assert_eq!(initial.state, "active");
+        let observer_generation = initial.generation.expect("observer generation");
+        let lost_notifications = initial.lost_completion_notifications.expect("observer loss counter");
+        let _ = wait_for_observer_terminal(&mut observer_events, &owner, generation_b).await;
+        wait_for_idle(&manager).await;
+        let mut admissions = subscribe_heal_responses();
+
+        // The same ID first runs again without an overwrite: equal data_dir
+        // still does not mean that this is the earlier execution.
+        stack_probe_stage("same-data owner round started");
+        let (same_admission, same_terminal) = run_reused_owner_round(
+            &manager,
+            &mut scanner,
+            &mut admissions,
+            &mut completions,
+            &mut observer_events,
+            &owner,
+            current,
+            generation_b,
+        )
+        .await;
+        stack_probe_stage("same-data owner round finished");
+
+        stack_probe_stage("generation C write started");
+        put_protected(&env, bucket, object, 0x43, now - time::Duration::minutes(2)).await;
+        stack_probe_stage("generation C write finished");
+        let changed = env
+            .ecstore
+            .get_object_info(bucket, object, &ScannerObjectOptions::default())
+            .await
+            .expect("read generation C from storage");
+        let generation_c = changed.data_dir.filter(|id| !id.is_nil()).expect("generation C");
+        assert_ne!(generation_b, generation_c);
+        stack_probe_stage("changed-data owner round started");
+        let (changed_admission, changed_terminal) = run_reused_owner_round(
+            &manager,
+            &mut scanner,
+            &mut admissions,
+            &mut completions,
+            &mut observer_events,
+            &owner,
+            changed,
+            generation_c,
+        )
+        .await;
+        stack_probe_stage("changed-data owner round finished");
+
+        let final_status = scanner_heal_observation_status();
+        assert_eq!(final_status.state, "active");
+        assert_eq!(final_status.generation, Some(observer_generation));
+        assert_eq!(final_status.lost_completion_notifications, Some(lost_notifications));
+
+        let same_stale = matches!(&same_admission, ScannerHealTestAdmission::CachedTerminal { .. });
+        let changed_stale = matches!(&changed_admission, ScannerHealTestAdmission::CachedTerminal { .. });
+        assert!(
+            !same_stale && !changed_stale,
+            "HEAL_ID_REUSE_E2E_STALE_TERMINAL: same_data={same_stale} changed_data={changed_stale}"
+        );
+        assert_observer_resolution(&same_admission, &same_terminal, generation_b);
+        assert_observer_resolution(&changed_admission, &changed_terminal, generation_c);
+        assert_eq!(recorder.total("rustfs_scanner_heal_candidate_generation_mismatch_total"), 1);
+        assert_eq!(final_status.integrity_state, "unknown");
+    }
+
+    manager.stop().await.expect("stop production heal manager");
+    processor.abort();
+    let _ = processor.await;
+}

docs/operations/scanner-runtime-controls.md

@@ -626,7 +626,9 @@ typed terminal result:
 | `rustfs_scanner_heal_pending_tasks` | Canonical tasks awaiting a terminal observation. |
 | `rustfs_scanner_heal_pending_oldest_seconds` | Age of the oldest retained pending candidate; refreshed every five seconds. |
 | `rustfs_scanner_heal_candidate_terminal_total` | Associated candidates, labeled by object `disposition`, expected `scan_mode`, and unresolved `scope`. |
+| `rustfs_scanner_heal_candidate_generation_mismatch_total` | Associated candidates whose non-nil data-directory IDs differ for the same kind, bucket, and object name. It has no object labels and can overlap `scope` values such as `insufficient_scan`. It proves neither payload integrity nor leader identity. |
 | `rustfs_scanner_heal_task_terminal_observations_total` | Associated task observations, deduplicated within the bounded recent history. This is not an execution count. |
+| `rustfs_scanner_heal_task_deep_payload_verified_observations_total` | The first associated observation of a canonical task that is `Completed`, is not a dry run, and carries storage's payload-integrity proof. A `Normal` request can be upgraded to a deep scan internally. Restart or recent-history eviction can allow recounts; this does not prove candidate coverage, leader identity, or full-bucket verification. |
 | `rustfs_scanner_heal_candidate_unknown_total` | Candidates whose integrity remains unknown, labeled by reason. |
 | `rustfs_scanner_heal_completion_lost_total` | Terminal notifications skipped by a lagging observer. |