test(heal): reproduce queued scanner receipt mismatch

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,57 @@
-# 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 canonical receipt baseline
 
 on:
-  push:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
-  pull_request:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
 
 permissions:
   contents: read
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
-
-env:
-  CARGO_TERM_COLOR: always
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-  RUST_BACKTRACE: 1
-
 jobs:
-  rename-safety:
-    name: Rename Safety
-    runs-on: windows-latest
-    timeout-minutes: 60
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-          install-test-tools: 'false'
-
-      - name: Check production Windows dependencies
-        shell: pwsh
-        run: cargo check -p rustfs-ecstore --lib
-
-      - name: Test guarded rename publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
-
-      - name: Test multipart parent durability barriers
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
-
-      - name: Test Windows path conversion
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test Windows volume and disk-space probes
-        shell: pwsh
-        run: |
-          # Map a drive letter to the temp drive's administrative share so the
-          # remote-volume probe also covers mapped network drives.
-          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
-          net use R: "\\localhost\$drive`$" /persistent:no
-          if ($LASTEXITCODE -eq 0) {
-            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
-          } else {
-            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
-          }
-          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
-          $status = $LASTEXITCODE
-          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
-          exit $status
-
-      - name: Test Windows handle guards
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
-
-      - name: Test endpoint path parsing
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
-
-      - name: Test startup temporary-directory cleanup
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
-
-      - name: Test object key Windows admission
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
-
-      - name: Test Windows file-name rules
-        shell: pwsh
-        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
-
-      - name: Test fresh format publication
-        shell: pwsh
-        run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
-
-  # The native Windows main thread only gets the 1 MiB PE stack reserve, while
-  # Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
-  # thread that calls `block_on`, so a debug server once overflowed before it
-  # listened on Windows only. Start a real server to keep that covered.
-  startup-smoke:
-    name: Startup Smoke
-    runs-on: windows-latest
+  verify-red:
+    runs-on: ubuntu-latest
     timeout-minutes: 90
     steps:
-      - name: Checkout repository
+      - name: Checkout
         uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
         with:
           persist-credentials: false
 
-      - name: Setup Rust environment
+      - name: Setup Rust
         uses: ./.github/actions/setup
         with:
           rust-version: stable
-          cache-shared-key: build-x86_64-pc-windows-msvc
+          cache-shared-key: ci-e2e-distributed-hosted
           cache-save-if: 'false'
           install-build-packaging-tools: 'false'
           install-test-tools: 'false'
 
-      - name: Build debug server binary
-        shell: pwsh
-        run: cargo build --locked -p rustfs --bin rustfs
-
-      - name: Print version
-        shell: pwsh
+      - name: Verify the two baseline failures
+        shell: bash
         run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
-
-      - name: Start server and wait for readiness
-        shell: pwsh
-        env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
-        run: |
-          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
-          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
-          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
-          New-Item -ItemType Directory -Force -Path $volume | Out-Null
-          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
-            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
-            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
-            -NoNewWindow -PassThru
-          # Cache the process handle now; otherwise ExitCode reads back empty.
-          $null = $server.Handle
-          $ready = $false
-          try {
-            for ($i = 0; $i -lt 180; $i++) {
-              if ($server.HasExited) { break }
-              try {
-                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
-                if ($response.StatusCode -eq 200) { $ready = $true; break }
-              } catch {}
-              Start-Sleep -Seconds 1
-            }
-          } finally {
-            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
-            $server.WaitForExit()
-            Write-Host '--- server stdout (tail) ---'
-            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
-            Write-Host '--- server stderr (tail) ---'
-            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
-          }
-          if (-not $ready) {
-            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
-          }
+          set -euo pipefail
+          export CARGO_TERM_COLOR=never
+          export CARGO_BUILD_JOBS=2
+          export CARGO_PROFILE_TEST_DEBUG=0
+          export CARGO_INCREMENTAL=0
+          export RUST_BACKTRACE=0
+          failed=0
+          for mode in normal fair; do
+            name="heal::manager::tests::queued_representative::scanner_receipt_tracks_queued_owner_after_${mode}_dequeue"
+            log="${RUNNER_TEMP}/scanner-queued-owner-${mode}.log"
+            set +e
+            cargo test --locked -p rustfs-heal --lib "$name" -- --exact --test-threads=1 >"$log" 2>&1
+            status=$?
+            set -e
+            tail -n 160 "$log"
+            if [[ "$status" -ne 101 ]] ||
+              ! grep -Fxq 'running 1 test' "$log" ||
+              ! grep -Fxq "test ${name} ... FAILED" "$log" ||
+              ! grep -Fq 'receipt must name the remaining queued task' "$log" ||
+              ! grep -Fq 'test result: FAILED. 0 passed; 1 failed;' "$log"; then
+              printf 'Expected exactly one matching failure for %s (cargo exit %s)\n' "$name" "$status" >&2
+              failed=1
+            else
+              printf 'Confirmed expected baseline failure for %s\n' "$name"
+            fi
+          done
+          exit "$failed"

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

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

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

@@ -0,0 +1,232 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+
+fn object_request(source: HealRequestSource, priority: HealPriority, force_start: bool) -> HealRequest {
+    let mut request = HealRequest::new(
+        HealType::Object {
+            bucket: "canonical-owner".to_owned(),
+            object: "object".to_owned(),
+            version_id: None,
+        },
+        HealOptions::default(),
+        priority,
+    );
+    request.source = source;
+    request.force_start = force_start;
+    request
+}
+
+async fn scanner_receipt_tracks_queued_owner(use_fairness: bool) {
+    let manager = HealManager::new_without_root_recovery_for_test(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            queue_size: 16,
+            low_priority_merge_enable: true,
+            overlap_policy: HealOverlapPolicy::Merge,
+            ..HealConfig::default()
+        }),
+    );
+
+    if use_fairness {
+        let mut queue = manager.heal_queue.lock().await;
+        for index in 0..4 {
+            let request = bucket_request(&format!("internal-{index}"), HealPriority::Normal, HealRequestSource::Internal);
+            assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+        }
+        for _ in 0..4 {
+            let (request, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
+            assert_eq!(request.expect("internal work should run").source, HealRequestSource::Internal);
+        }
+    }
+
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let key = PriorityHealQueue::make_dedup_key(&first);
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    for request in [first, second] {
+        assert_eq!(
+            manager
+                .submit_heal_request_with_receipt(request)
+                .await
+                .expect("admit task")
+                .result,
+            HealAdmissionResult::Accepted
+        );
+    }
+
+    let first = {
+        let mut queue = manager.heal_queue.lock().await;
+        let (selected, skipped) = if use_fairness {
+            queue.pop_runnable_with_fairness(|_| true, |_| None)
+        } else {
+            queue.pop_runnable_with_skips(|_| true, |_| None)
+        };
+        assert!(skipped.is_empty());
+        let selected = selected.expect("the first same-priority task should run");
+        assert_eq!(selected.id, first_id);
+        assert!(queue.contains_request_id(&second_id));
+        assert_eq!(
+            queue.queued_request_id_for_dedup_key(&key),
+            Some(first_id.as_str()),
+            "the existing representative for other sources must remain unchanged"
+        );
+        selected
+    };
+    manager
+        .active_heals
+        .lock()
+        .await
+        .insert(first_id.clone(), Arc::new(HealTask::from_request(first, Arc::new(MockStorage))));
+
+    let active = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
+        .await
+        .expect("scanner admission while the first task is active");
+    assert_eq!(active.result, HealAdmissionResult::Merged);
+    assert_eq!(active.task_id, first_id, "active ownership keeps precedence");
+
+    let completed_task = manager
+        .active_heals
+        .lock()
+        .await
+        .remove(&first_id)
+        .expect("first task should leave active ownership");
+    let mut retained = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await;
+    Arc::make_mut(retained.outcome.as_mut().expect("retained task outcome")).finish(None);
+    manager
+        .completed_heals
+        .lock()
+        .await
+        .insert(first_id.clone(), Arc::new(retained));
+    let queued = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
+        .await
+        .expect("scanner admission after the first task leaves active state");
+    assert_eq!(queued.result, HealAdmissionResult::Merged);
+    assert_eq!(queued.task_id, second_id, "the receipt must name the remaining queued task");
+}
+
+#[tokio::test]
+async fn scanner_receipt_tracks_queued_owner_after_normal_dequeue() {
+    scanner_receipt_tracks_queued_owner(false).await;
+}
+
+#[tokio::test]
+async fn scanner_receipt_tracks_queued_owner_after_fair_dequeue() {
+    scanner_receipt_tracks_queued_owner(true).await;
+}
+
+#[tokio::test]
+async fn scanner_receipt_keeps_queued_owner_when_a_different_forced_task_runs() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
+    let first_id = first.id.clone();
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
+        let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert_eq!(selected.expect("higher-priority forced task should run").id, second_id);
+    }
+    let receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("scanner admission after a non-representative task leaves the queue");
+    assert_eq!(receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(receipt.task_id, first_id);
+}
+
+#[tokio::test]
+async fn scanner_receipt_reselects_after_the_queued_owner_is_cancelled() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
+    let third_id = third.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        for request in [first, second, third] {
+            assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+        }
+        let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert_eq!(selected.expect("first task should run").id, first_id);
+    }
+    let first_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("first scanner admission");
+    assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
+    assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
+    let remaining = if first_receipt.task_id == second_id {
+        third_id
+    } else {
+        second_id
+    };
+    manager.heal_queue.lock().await.remove_request_id(&first_receipt.task_id);
+
+    let next_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("scanner admission after the queued owner is removed");
+    assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(next_receipt.task_id, remaining);
+}
+
+#[tokio::test]
+async fn scanner_receipt_reselects_after_multiple_forced_dequeues() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
+    let first_id = first.id.clone();
+    let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
+    let second_id = second.id.clone();
+    let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
+    let third_id = third.id.clone();
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        for request in [first, second, third] {
+            assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
+        }
+        let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
+        assert_eq!(selected.expect("first task should run").id, first_id);
+    }
+    let first_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("first scanner admission");
+    assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
+    assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
+    let remaining = if first_receipt.task_id == second_id {
+        third_id
+    } else {
+        second_id
+    };
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_receipt.task_id, |_| None);
+        assert_eq!(selected.expect("selected queued owner should run").id, first_receipt.task_id);
+    }
+    let next_receipt = manager
+        .submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
+        .await
+        .expect("scanner admission after another queued owner starts");
+    assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
+    assert_eq!(next_receipt.task_id, remaining);
+}