ci: scope admitted collector interaction preflight

# Conflicts:
#	crates/scanner/src/scanner_io/tests.rs

.github/workflows/windows-filesystem.yml

@@ -1,224 +1,144 @@
-# 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 admitted collector cycle preflight
 
 on:
-  push:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
-  pull_request:
-    branches: [ main ]
-    paths:
-      - "crates/ecstore/src/disk/**"
-      - "crates/ecstore/src/layout/**"
-      - "crates/ecstore/src/store/init_format.rs"
-      - "crates/ecstore/src/set_disk/ops/locking.rs"
-      - "crates/ecstore/src/object_api/write_commit_context.rs"
-      - "crates/ecstore/src/bucket/utils.rs"
-      - "crates/ecstore/Cargo.toml"
-      - "crates/utils/src/dunce.rs"
-      - "crates/utils/src/os/**"
-      - "crates/utils/Cargo.toml"
-      - "crates/utils/src/path.rs"
-      - "rustfs/build.rs"
-      - "rustfs/Cargo.toml"
-      - "rustfs/src/main.rs"
-      - "rustfs/src/server/runtime.rs"
-      - "rustfs/src/startup_*.rs"
-      - "Cargo.toml"
-      - "Cargo.lock"
-      - ".github/actions/setup/**"
-      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
+    inputs:
+      expected_head:
+        description: Exact workflow commit SHA
+        required: true
+        type: string
 
 permissions:
   contents: read
 
 concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
+  group: scanner-admitted-collector-${{ 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
+  scoped-test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 110
     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
+      - name: Prepare evidence
+        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
+          evidence="$RUNNER_TEMP/scanner-admitted-collector"
+          mkdir -p "$evidence"
+          printf '%s\n' 'not-started' > "$evidence/build.exit"
+          printf '%s\n' 'not-started' > "$evidence/test.exit"
 
-      - 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
+      - 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: Build debug server binary
-        shell: pwsh
-        run: cargo build --locked -p rustfs --bin rustfs
-
-      - name: Print version
-        shell: pwsh
-        run: |
-          & .\target\debug\rustfs.exe --version
-          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
-
-      - name: Start server and wait for readiness
-        shell: pwsh
+      - name: Verify source and workflow
+        shell: bash
         env:
-          RUSTFS_ACCESS_KEY: rustfsadmin
-          RUSTFS_SECRET_KEY: rustfsadmin
+          EXPECTED_HEAD: ${{ inputs.expected_head }}
         run: |
-          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
-          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
-          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
-          New-Item -ItemType Directory -Force -Path $volume | Out-Null
-          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
-            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
-            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
-            -NoNewWindow -PassThru
-          # Cache the process handle now; otherwise ExitCode reads back empty.
-          $null = $server.Handle
-          $ready = $false
-          try {
-            for ($i = 0; $i -lt 180; $i++) {
-              if ($server.HasExited) { break }
-              try {
-                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
-                if ($response.StatusCode -eq 200) { $ready = $true; break }
-              } catch {}
-              Start-Sleep -Seconds 1
-            }
-          } finally {
-            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
-            $server.WaitForExit()
-            Write-Host '--- server stdout (tail) ---'
-            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
-            Write-Host '--- server stderr (tail) ---'
-            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
-          }
-          if (-not $ready) {
-            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
-          }
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/scanner-admitted-collector"
+          workflow_sha=$(git rev-parse HEAD)
+          source_sha=$(git rev-parse HEAD^)
+          test_sha=$(git rev-parse HEAD^^)
+          combination_sha=$(git rev-parse HEAD^^^)
+          main_sha=$(git rev-parse "${combination_sha}^2")
+          source_tree=$(git rev-parse "${source_sha}^{tree}")
+          test_diff=$(git diff --binary --full-index "$combination_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          fix_diff=$(git diff --binary --full-index "$test_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
+          source_files=$(git diff --name-only "$combination_sha" "$source_sha")
+          workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
+          printf 'expected_head=%s\nhead=%s\nsource=%s\ntest_parent=%s\ncombination=%s\nmain=%s\nsource_tree=%s\ntest_diff=%s\nfix_diff=%s\nsource_files=%s\nworkflow_files=%s\n' \
+            "$EXPECTED_HEAD" "$workflow_sha" "$source_sha" "$test_sha" "$combination_sha" "$main_sha" \
+            "$source_tree" "$test_diff" "$fix_diff" "$source_files" "$workflow_files" | tee "$evidence/source-proof.txt"
+          [[ "$GITHUB_EVENT_NAME" == 'workflow_dispatch' ]]
+          [[ "$EXPECTED_HEAD" =~ ^[0-9a-f]{40}$ ]]
+          [[ "$workflow_sha" == "$EXPECTED_HEAD" && "$workflow_sha" == "$GITHUB_SHA" ]]
+          [[ "$source_sha" == '097f7603db1090f15e390e7e2cde316d148e395f' ]]
+          [[ "$test_sha" == '7952608c03bf3d42a28909b0d9f861dd84a3264f' ]]
+          [[ "$combination_sha" == '5ad283e28ffa0d16b82007c64df75901aaea5b21' ]]
+          [[ "$main_sha" == '8e88419f3a0c4741073c5ea46dd43c434be7f138' ]]
+          [[ "$source_tree" == '67db91a7d0556b8beaa5b903ca602c0981bf7434' ]]
+          [[ "$test_diff" == '2687c156f166ee2dc38861140210b3063c7f4fc7df656b3e4e2a733b619e11a3' ]]
+          [[ "$fix_diff" == 'a50801c7f457af9d3468d421dc1d14efec87be45f67d5af43abb59bc5707b6d6' ]]
+          [[ "$source_files" == 'crates/scanner/src/scanner_io/tests/cycle_collector_diagnostics.rs' ]]
+          [[ "$workflow_files" == '.github/workflows/windows-filesystem.yml' ]]
+          [[ -z "$(git status --porcelain)" ]]
+          printf '%s\n' \
+            'One exact parent unit test starts one isolated child that checks complete, cancelled and competing-revision local cycles.' \
+            'The successful child output is checked inside the parent and is not emitted as a separate log.' \
+            'No rejected old coordinator RPC or MRF responsibility-release test is selected; normal storage background activity can occur.' \
+            'This scoped preflight does not replace complete CI or distributed acceptance.' | tee "$evidence/scope.txt"
+          df -h . | tee "$evidence/disk.txt"
+
+      - name: Setup Rust
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: ci-e2e-distributed-hosted
+          cache-save-if: 'false'
+          install-build-packaging-tools: 'false'
+          install-test-tools: 'false'
+
+      - name: Compile scanner library tests
+        shell: bash
+        timeout-minutes: 80
+        run: |
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/scanner-admitted-collector"
+          set +e
+          timeout --signal=TERM --kill-after=30s 75m \
+            cargo test --locked -p rustfs-scanner --lib --no-run > "$evidence/build.log" 2>&1
+          status=$?
+          set -e
+          printf '%s\n' "$status" > "$evidence/build.exit"
+          tail -n 100 "$evidence/build.log"
+          exit "$status"
+
+      - name: Run the exact collector test
+        shell: bash
+        timeout-minutes: 10
+        run: |
+          set -euo pipefail
+          evidence="$RUNNER_TEMP/scanner-admitted-collector"
+          test_name='scanner_io::tests::cycle_collector_diagnostics::collector_cycle_with_node_admission_preserves_cache_and_reports_outcomes'
+          printf '%s\n' "$test_name" > "$evidence/selection.txt"
+          if [[ -v RUSTFS_TEST_ADMITTED_COLLECTOR_CHILD ]]; then
+            printf '%s\n' 'Unexpected parent child marker' | tee "$evidence/environment.txt"
+            exit 1
+          fi
+          printf '%s\n' 'parent_child_marker=unset' | tee "$evidence/environment.txt"
+          set +e
+          timeout --signal=TERM --kill-after=15s 8m \
+            cargo test --locked -p rustfs-scanner --lib -- --exact "$test_name" --nocapture > "$evidence/test.log" 2>&1
+          status=$?
+          set -e
+          printf '%s\n' "$status" > "$evidence/test.exit"
+          tail -n 120 "$evidence/test.log"
+          selected=$(grep -Fxc 'running 1 test' "$evidence/test.log" || true)
+          passed=$(grep -Fxc "test $test_name ... ok" "$evidence/test.log" || true)
+          summary=$(grep -Fc 'test result: ok. 1 passed; 0 failed; 0 ignored;' "$evidence/test.log" || true)
+          printf 'exit=%s selected=%s passed=%s summary=%s\n' "$status" "$selected" "$passed" "$summary" | tee "$evidence/verification.txt"
+          [[ "$status" -eq 0 && "$selected" -eq 1 && "$passed" -eq 1 && "$summary" -eq 1 ]]
+
+      - name: Preserve evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
+        with:
+          name: scanner-admitted-collector-${{ github.run_id }}
+          path: ${{ runner.temp }}/scanner-admitted-collector/
+          if-no-files-found: error
+          retention-days: 14

crates/scanner/src/scanner_io/io_cycle.rs

@@ -14,6 +14,102 @@
 /// ScannerIO/ScannerIOCycle implementations for ECStore: bucket planning and per-set fan-out.
 use super::*;
 
+#[derive(Clone, Copy)]
+struct CollectorSlotSample<'a> {
+    index: usize,
+    expected: DataUsageCacheSource,
+    observed: bool,
+    snapshot: &'a DataUsageCache,
+}
+
+#[derive(Default)]
+struct CollectorSlotSummary<'a> {
+    observed: usize,
+    unobserved: usize,
+    incomplete: usize,
+    missing_last_update: usize,
+    missing_source: usize,
+    source_mismatch: usize,
+    cycle_mismatch: usize,
+    epoch_mismatch: usize,
+    missing_plan: usize,
+    plan_mismatch: usize,
+    anomalous: usize,
+    sample: Option<CollectorSlotSample<'a>>,
+    source_difference_sample: Option<CollectorSlotSample<'a>>,
+}
+
+fn summarize_collector_slots<'a>(
+    slots: impl IntoIterator<Item = (usize, DataUsageCacheSource, bool, &'a DataUsageCache)>,
+    cycle: u64,
+    leader_epoch: u64,
+    plan: DataUsageScanPlanDigest,
+) -> CollectorSlotSummary<'a> {
+    let mut summary = CollectorSlotSummary::default();
+    for (index, expected, observed, snapshot) in slots {
+        let mut anomalous = !observed;
+        if observed {
+            summary.observed += 1;
+            let info = &snapshot.info;
+            let incomplete = !info.snapshot_complete;
+            let missing_last_update = info.last_update.is_none();
+            let missing_source = info.source.is_none();
+            let source_mismatch = info.source.is_some_and(|source| source != expected);
+            let cycle_mismatch = info.next_cycle != cycle;
+            let epoch_mismatch = info.leader_epoch != leader_epoch;
+            let missing_plan = info.scan_plan_digest.is_none();
+            let plan_mismatch = info.scan_plan_digest.is_some_and(|actual| actual != plan);
+            summary.incomplete += usize::from(incomplete);
+            summary.missing_last_update += usize::from(missing_last_update);
+            summary.missing_source += usize::from(missing_source);
+            summary.source_mismatch += usize::from(source_mismatch);
+            summary.cycle_mismatch += usize::from(cycle_mismatch);
+            summary.epoch_mismatch += usize::from(epoch_mismatch);
+            summary.missing_plan += usize::from(missing_plan);
+            summary.plan_mismatch += usize::from(plan_mismatch);
+            if source_mismatch {
+                let sample_key = (expected.pool_index, expected.set_index, index);
+                if summary
+                    .source_difference_sample
+                    .is_none_or(|sample| sample_key < (sample.expected.pool_index, sample.expected.set_index, sample.index))
+                {
+                    summary.source_difference_sample = Some(CollectorSlotSample {
+                        index,
+                        expected,
+                        observed,
+                        snapshot,
+                    });
+                }
+            }
+            anomalous = incomplete
+                || missing_last_update
+                || missing_source
+                || cycle_mismatch
+                || epoch_mismatch
+                || missing_plan
+                || plan_mismatch;
+        } else {
+            summary.unobserved += 1;
+        }
+        if anomalous {
+            summary.anomalous += 1;
+            let sample_key = (expected.pool_index, expected.set_index, index);
+            if summary
+                .sample
+                .is_none_or(|sample| sample_key < (sample.expected.pool_index, sample.expected.set_index, sample.index))
+            {
+                summary.sample = Some(CollectorSlotSample {
+                    index,
+                    expected,
+                    observed,
+                    snapshot,
+                });
+            }
+        }
+    }
+    summary
+}
+
 #[async_trait::async_trait]
 impl ScannerIO for ECStore {
     async fn nsscanner(
@@ -605,7 +701,8 @@ where
     record_set_scans_active(0);
 
     let results = vec![DataUsageCache::default(); total_results];
-    let results_mutex: Arc<Mutex<Vec<DataUsageCache>>> = Arc::new(Mutex::new(results));
+    let observed_slots = vec![false; total_results];
+    let results_mutex = Arc::new(Mutex::new((results, observed_slots)));
     let first_err_mutex: Arc<Mutex<Option<Error>>> = Arc::new(Mutex::new(None));
     let mut wait_futs = Vec::new();
 
@@ -659,7 +756,8 @@ where
         let receiver_fut = tokio::spawn(async move {
             while let Some(result) = rx.recv().await {
                 let mut results = results_mutex_clone.lock().await;
-                results[results_index_clone] = result;
+                results.0[results_index_clone] = result;
+                results.1[results_index_clone] = true;
             }
         });
         wait_futs.push(AbortOnDropHandle::new(receiver_fut));
@@ -780,7 +878,7 @@ where
     let first_err = first_err_mutex.lock().await.take();
     let has_set_failure = first_err.is_some();
     // Every collector has joined; transfer the caches without duplicating their trees.
-    let results = std::mem::take(&mut *results_mutex.lock().await);
+    let (results, observed_slots) = std::mem::take(&mut *results_mutex.lock().await);
     let completed_all_sets = bucket_plan_complete && scanner_results_form_complete_snapshot(&results, &expected_sources);
     let result = finalize_nsscanner_result(&results, first_err);
     let failed_buckets = bucket_failures.hard.lock().await.clone();
@@ -871,6 +969,72 @@ where
         dirty_usage_status,
         activity_status,
     );
+    if cycle_status == ScannerCycleStatus::Incomplete && result.is_ok() && !budget_elapsed && !ctx.is_cancelled() {
+        let slots =
+            set_disks
+                .iter()
+                .zip(&results)
+                .zip(&observed_slots)
+                .enumerate()
+                .map(|(index, ((set, snapshot), observed))| {
+                    (index, DataUsageCacheSource::new(set.pool_index, set.set_index), *observed, snapshot)
+                });
+        let summary = summarize_collector_slots(slots, want_cycle, leader_epoch, scan_plan_digest);
+        let sample = summary.sample;
+        let source_difference = summary.source_difference_sample;
+        let sample_snapshot = sample.filter(|sample| sample.observed).map(|sample| sample.snapshot);
+        warn!(
+            target: "rustfs::scanner::io",
+            event = EVENT_SCANNER_SET_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_IO,
+            state = "incomplete_cycle_collector_summary",
+            cycle = want_cycle,
+            leader_epoch,
+            configured_sets = set_disks.len(),
+            expected_sources = expected_sources.len(),
+            collector_slots = results.len(),
+            bucket_plan_complete,
+            scan_scope_matches,
+            completed_candidate_present = completed_usage.is_some(),
+            structurally_complete_snapshot,
+            bucket_scan_status = ?bucket_scan_status,
+            dirty_usage_status = ?dirty_usage_status,
+            activity_status = ?activity_status,
+            has_set_failure,
+            observed_slots = summary.observed,
+            unobserved_slots = summary.unobserved,
+            observed_incomplete_slots = summary.incomplete,
+            observed_missing_last_update_slots = summary.missing_last_update,
+            observed_missing_source_slots = summary.missing_source,
+            observed_slot_source_differs = summary.source_mismatch,
+            slot_source_difference_omitted = summary.source_mismatch.saturating_sub(usize::from(source_difference.is_some())),
+            slot_source_difference_index = source_difference.map(|sample| sample.index),
+            slot_source_difference_expected_pool = source_difference.map(|sample| sample.expected.pool_index),
+            slot_source_difference_expected_set = source_difference.map(|sample| sample.expected.set_index),
+            slot_source_difference_snapshot_pool = source_difference.and_then(|sample| sample.snapshot.info.source.map(|source| source.pool_index)),
+            slot_source_difference_snapshot_set = source_difference.and_then(|sample| sample.snapshot.info.source.map(|source| source.set_index)),
+            observed_cycle_mismatch_slots = summary.cycle_mismatch,
+            observed_epoch_mismatch_slots = summary.epoch_mismatch,
+            observed_missing_plan_slots = summary.missing_plan,
+            observed_plan_mismatch_slots = summary.plan_mismatch,
+            anomalous_slots = summary.anomalous,
+            sample_omitted_slots = summary.anomalous.saturating_sub(usize::from(sample.is_some())),
+            sample_slot = sample.map(|sample| sample.index),
+            sample_expected_pool = sample.map(|sample| sample.expected.pool_index),
+            sample_expected_set = sample.map(|sample| sample.expected.set_index),
+            sample_observed = sample.map(|sample| sample.observed),
+            sample_snapshot_source_pool = sample_snapshot.and_then(|snapshot| snapshot.info.source.map(|source| source.pool_index)),
+            sample_snapshot_source_set = sample_snapshot.and_then(|snapshot| snapshot.info.source.map(|source| source.set_index)),
+            sample_snapshot_complete = sample_snapshot.map(|snapshot| snapshot.info.snapshot_complete),
+            sample_snapshot_has_last_update = sample_snapshot.map(|snapshot| snapshot.info.last_update.is_some()),
+            sample_snapshot_cycle = sample_snapshot.map(|snapshot| snapshot.info.next_cycle),
+            sample_snapshot_leader_epoch = sample_snapshot.map(|snapshot| snapshot.info.leader_epoch),
+            sample_snapshot_has_plan = sample_snapshot.map(|snapshot| snapshot.info.scan_plan_digest.is_some()),
+            sample_snapshot_plan_matches = sample_snapshot.and_then(|snapshot| snapshot.info.scan_plan_digest.map(|plan| plan == scan_plan_digest)),
+            "Scanner recorded collector observations for an incomplete cycle"
+        );
+    }
     // Certification is per bucket: the pending set is read after the
     // guardrail verdict so a bucket written during the walk is reported in
     // the same publication that certifies the others. A bucket whose walk
@@ -940,3 +1104,141 @@ where
         .with_scope_fallback(scope_fallback)
         .with_publication_expectation(publication_expectation))
 }
+
+#[cfg(test)]
+mod collector_summary_tests {
+    use super::*;
+
+    const PLAN: DataUsageScanPlanDigest = DataUsageScanPlanDigest([4; 32]);
+
+    fn complete(source: DataUsageCacheSource) -> DataUsageCache {
+        DataUsageCache {
+            info: DataUsageCacheInfo {
+                source: Some(source),
+                next_cycle: 7,
+                leader_epoch: 11,
+                scan_plan_digest: Some(PLAN),
+                last_update: Some(SystemTime::now()),
+                snapshot_complete: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        }
+    }
+
+    #[test]
+    fn collector_distinguishes_an_unobserved_slot_from_an_observed_default() {
+        let default = DataUsageCache::default();
+        let first = DataUsageCacheSource::new(0, 0);
+        let second = DataUsageCacheSource::new(0, 1);
+        let summary = summarize_collector_slots([(0, first, true, &default), (1, second, false, &default)], 7, 11, PLAN);
+
+        assert_eq!((summary.observed, summary.unobserved), (1, 1));
+        assert_eq!((summary.incomplete, summary.missing_source), (1, 1));
+        assert_eq!(summary.anomalous, 2);
+        let sample = summary.sample.expect("a sample should be selected");
+        assert_eq!(sample.expected, first);
+        assert!(sample.observed, "an observed default is not an unobserved slot");
+        assert!(sample.snapshot.info.source.is_none());
+    }
+
+    #[test]
+    fn collector_reports_source_positions_without_inventing_a_slot_failure() {
+        let first = DataUsageCacheSource::new(0, 0);
+        let second = DataUsageCacheSource::new(0, 1);
+        let first_cache = complete(first);
+        let second_cache = complete(second);
+        let expected_sources = HashSet::from([first, second]);
+
+        let swapped = [second_cache, first_cache.clone()];
+        assert!(scanner_results_match_scan_scope(&swapped, &expected_sources));
+        let summary = summarize_collector_slots([(0, first, true, &swapped[0]), (1, second, true, &swapped[1])], 7, 11, PLAN);
+        assert_eq!(summary.source_mismatch, 2);
+        assert!(summary.sample.is_none(), "different positions alone do not identify a failed slot");
+        let difference = summary.source_difference_sample.expect("one bounded position difference");
+        assert_eq!(difference.expected, first);
+        assert_eq!(difference.snapshot.info.source, Some(second));
+
+        let duplicate = [first_cache.clone(), first_cache];
+        assert!(!scanner_results_match_scan_scope(&duplicate, &expected_sources));
+        let summary = summarize_collector_slots([(0, first, true, &duplicate[0]), (1, second, true, &duplicate[1])], 7, 11, PLAN);
+        assert_eq!(summary.source_mismatch, 1);
+        assert!(summary.sample.is_none(), "a duplicate source does not identify which slot caused it");
+        let difference = summary.source_difference_sample.expect("one bounded position difference");
+        assert_eq!(difference.expected, second);
+        assert_eq!(difference.snapshot.info.source, Some(first));
+    }
+
+    #[test]
+    fn collector_counts_generation_and_plan_differences_without_assigning_a_cause() {
+        let first = DataUsageCacheSource::new(0, 0);
+        let second = DataUsageCacheSource::new(0, 1);
+        let mut older = complete(first);
+        older.info.next_cycle = 6;
+        older.info.leader_epoch = 10;
+        older.info.scan_plan_digest = Some(DataUsageScanPlanDigest([9; 32]));
+        let mut missing_plan = complete(second);
+        missing_plan.info.scan_plan_digest = None;
+
+        let slots = [(1, second, true, &missing_plan), (0, first, true, &older)];
+        let summary = summarize_collector_slots(slots, 7, 11, PLAN);
+        assert_eq!((summary.cycle_mismatch, summary.epoch_mismatch), (1, 1));
+        assert_eq!((summary.plan_mismatch, summary.missing_plan), (1, 1));
+        assert_eq!(summary.anomalous, 2);
+        let sample = summary.sample.expect("one deterministic sample");
+        assert_eq!(sample.expected, first);
+        assert_eq!(sample.snapshot.info.next_cycle, 6);
+        assert_eq!(sample.snapshot.info.leader_epoch, 10);
+        assert_eq!(sample.snapshot.info.scan_plan_digest, Some(DataUsageScanPlanDigest([9; 32])));
+
+        let missing = summarize_collector_slots([(1, second, true, &missing_plan)], 7, 11, PLAN);
+        assert!(
+            missing
+                .sample
+                .expect("missing plan sample")
+                .snapshot
+                .info
+                .scan_plan_digest
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn collector_sample_is_bounded_and_complete_slots_do_not_explain_other_incomplete_states() {
+        let first = DataUsageCacheSource::new(0, 0);
+        let second = DataUsageCacheSource::new(0, 1);
+        let first_cache = complete(first);
+        let second_cache = complete(second);
+        let complete_slots = [(0, first, true, &first_cache), (1, second, true, &second_cache)];
+
+        for (sets_complete, bucket_status, dirty_status) in [
+            (false, ScannerBucketScanStatus::PartialCarried, DirtyUsageSnapshotStatus::Current),
+            (true, ScannerBucketScanStatus::Partial, DirtyUsageSnapshotStatus::Current),
+            (true, ScannerBucketScanStatus::Complete, DirtyUsageSnapshotStatus::Unverified),
+        ] {
+            assert_eq!(
+                classify_nsscanner_cycle(
+                    sets_complete,
+                    false,
+                    false,
+                    bucket_status,
+                    dirty_status,
+                    ScannerCycleActivityStatus::Unchanged,
+                ),
+                ScannerCycleStatus::Incomplete
+            );
+            let summary = summarize_collector_slots(complete_slots, 7, 11, PLAN);
+            assert_eq!(summary.anomalous, 0);
+            assert!(summary.sample.is_none(), "complete slot observations do not explain another gate");
+        }
+
+        let default = DataUsageCache::default();
+        let slots = (0..12)
+            .rev()
+            .map(|index| (index, DataUsageCacheSource::new(0, index), false, &default));
+        let summary = summarize_collector_slots(slots, 7, 11, PLAN);
+        assert_eq!(summary.unobserved, 12);
+        assert_eq!(summary.anomalous, 12);
+        assert_eq!(summary.sample.expect("one bounded sample").expected, first);
+    }
+}

crates/scanner/src/scanner_io/tests.rs

@@ -46,6 +46,7 @@ use uuid::Uuid;
 
 mod checkpoint_coverage_identity;
 mod collector_lifecycle;
+mod cycle_collector_diagnostics;
 mod expiry_lock;
 mod listing_window_scope;
 mod node_admission;

crates/scanner/src/scanner_io/tests/cycle_collector_diagnostics.rs

@@ -0,0 +1,606 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
+use std::num::NonZeroU16;
+
+const LEADER_EPOCH: u64 = 11;
+
+#[derive(Clone)]
+struct LogWriter(Arc<StdMutex<Vec<u8>>>);
+
+impl std::io::Write for LogWriter {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        self.0.lock().expect("collector log buffer").extend_from_slice(bytes);
+        Ok(bytes.len())
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+async fn capture_cycle<F: std::future::Future>(future: F) -> (F::Output, Vec<serde_json::Value>) {
+    use tracing::instrument::WithSubscriber as _;
+
+    let buffer = Arc::new(StdMutex::new(Vec::new()));
+    let writer = Arc::clone(&buffer);
+    let subscriber = tracing_subscriber::fmt()
+        .json()
+        .without_time()
+        .with_ansi(false)
+        .with_max_level(tracing::Level::WARN)
+        .with_writer(move || LogWriter(Arc::clone(&writer)))
+        .finish();
+    let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
+    let result = future.with_subscriber(subscriber).await;
+    let logs = buffer.lock().expect("collector log buffer");
+    let events = std::str::from_utf8(&logs)
+        .expect("collector logs should be UTF-8")
+        .lines()
+        .map(|line| serde_json::from_str(line).expect("collector event should be JSON"))
+        .collect();
+    (result, events)
+}
+
+async fn run_cycle(
+    store: &Arc<ECStore>,
+    cycle: u64,
+    max_objects: Option<u64>,
+    requires_full_scan: bool,
+) -> (ScannerCycleResult, Option<DataUsageInfo>, bool) {
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &ctx,
+        ScannerCycleBudgetConfig {
+            max_objects,
+            ..Default::default()
+        },
+    );
+    let (updates, mut receiver) = mpsc::channel(1);
+    let result = tokio::time::timeout(
+        Duration::from_secs(60),
+        nsscanner_with_storage_status_scoped(
+            store.as_ref(),
+            ScannerCycleRequest {
+                ctx: budget.token(),
+                budget: Arc::clone(&budget),
+                updates,
+                want_cycle: cycle,
+                leader_epoch: LEADER_EPOCH,
+                scan_mode: HealScanMode::Normal,
+                scan_scope: ScannerBucketScanScope::default(),
+                persisted_usage_baseline: None,
+                observed_usage_candidate: None,
+                requires_full_scan,
+                service_cohort: None,
+                resolved_scope_observer: None,
+                post_walk_gate: None,
+            },
+        ),
+    )
+    .await
+    .expect("cycle should finish within the test deadline")
+    .expect("cycle should return a status");
+    let snapshot = receiver.recv().await;
+    assert!(receiver.recv().await.is_none(), "a cycle emits at most one snapshot");
+    (result, snapshot, budget.budget_elapsed())
+}
+
+fn collector_events(events: &[serde_json::Value]) -> Vec<&serde_json::Value> {
+    events
+        .iter()
+        .filter(|event| event["fields"]["state"] == "incomplete_cycle_collector_summary")
+        .map(|event| &event["fields"])
+        .collect()
+}
+
+#[tokio::test]
+#[serial]
+async fn collector_reports_an_unobserved_slot_after_a_stale_set_snapshot() {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
+    clear_dirty_usage_buckets_for_tests();
+    store
+        .make_bucket("diagnostic-bucket", &MakeBucketOptions::default())
+        .await
+        .expect("fixture bucket should be created");
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+
+    let set = Arc::clone(&store.pools[0].disk_set[0]);
+    let mut newer = complete_set_usage_cache(&[("diagnostic-bucket", 1)], DataUsageScanPlanDigest([1; 32]));
+    newer.info.source = Some(DataUsageCacheSource::new(0, 0));
+    newer.info.next_cycle = 7;
+    newer.info.leader_epoch = LEADER_EPOCH + 1;
+    newer
+        .save(set, DATA_USAGE_CACHE_NAME)
+        .await
+        .expect("newer set cache should persist");
+
+    let ((result, snapshot, budget_elapsed), events) = capture_cycle(run_cycle(&store, 7, None, true)).await;
+    assert_eq!(result.status, ScannerCycleStatus::Incomplete);
+    assert!(!budget_elapsed);
+    assert!(snapshot.is_none_or(|snapshot| !snapshot.usage_snapshot_complete));
+    let events = collector_events(&events);
+    assert_eq!(events.len(), 1, "a non-budget incomplete cycle needs one collector observation");
+    let fields = events[0];
+    assert_eq!(fields["cycle"], 7);
+    assert_eq!(fields["leader_epoch"], LEADER_EPOCH);
+    assert_eq!(fields["expected_sources"], 1);
+    assert_eq!(fields["observed_slots"], 0);
+    assert_eq!(fields["unobserved_slots"], 1);
+    assert_eq!(fields["sample_expected_pool"], 0);
+    assert_eq!(fields["sample_expected_set"], 0);
+    assert_eq!(fields["sample_observed"], false);
+    assert!(fields.get("sample_snapshot_source_pool").is_none());
+    assert!(fields.get("sample_snapshot_complete").is_none());
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn collector_reports_received_incomplete_cache_only_for_non_budget_partial() {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
+    clear_dirty_usage_buckets_for_tests();
+    store
+        .make_bucket("fresh-bucket", &MakeBucketOptions::default())
+        .await
+        .expect("fixture bucket should be created");
+    for object in ["p0/a", "p1/b", "p2/c", "p3/d"] {
+        let mut reader = ScannerPutObjReader::from_vec(vec![7; 16]);
+        store
+            .put_object("fresh-bucket", object, &mut reader, &ScannerObjectOptions::default())
+            .await
+            .expect("fixture object should persist");
+    }
+    record_dirty_usage_bucket("fresh-bucket");
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+
+    let ((first, _, budget_elapsed), events) = capture_cycle(run_cycle(&store, 1, Some(2), true)).await;
+    assert_eq!(first.status, ScannerCycleStatus::Incomplete);
+    assert!(budget_elapsed);
+    assert!(collector_events(&events).is_empty(), "budget partials do not use this event");
+
+    let ((second, snapshot, budget_elapsed), events) = capture_cycle(run_cycle(&store, 2, None, false)).await;
+    assert_eq!(second.status, ScannerCycleStatus::Incomplete);
+    assert!(!budget_elapsed);
+    assert!(snapshot.is_none_or(|snapshot| !snapshot.usage_snapshot_complete));
+    let events = collector_events(&events);
+    assert_eq!(events.len(), 1, "a non-budget incomplete cycle needs one collector observation");
+    let fields = events[0];
+    assert_eq!(fields["observed_slots"], 1);
+    assert_eq!(fields["unobserved_slots"], 0);
+    assert_eq!(fields["sample_observed"], true);
+    assert_eq!(fields["sample_expected_pool"], 0);
+    assert_eq!(fields["sample_expected_set"], 0);
+    assert_eq!(fields["sample_snapshot_source_pool"], 0);
+    assert_eq!(fields["sample_snapshot_source_set"], 0);
+    assert_eq!(fields["sample_snapshot_complete"], false);
+    assert_eq!(fields["sample_snapshot_has_last_update"], false);
+    assert_eq!(fields["sample_snapshot_cycle"], 2);
+    assert_eq!(fields["sample_snapshot_leader_epoch"], LEADER_EPOCH);
+    assert_eq!(fields["sample_snapshot_has_plan"], true);
+    assert_eq!(fields["sample_snapshot_plan_matches"], true);
+
+    let ((third, snapshot, _), events) = capture_cycle(run_cycle(&store, 3, None, false)).await;
+    assert_eq!(third.status, ScannerCycleStatus::Complete);
+    assert!(snapshot.is_some_and(|snapshot| snapshot.usage_snapshot_complete));
+    assert!(
+        collector_events(&events).is_empty(),
+        "a complete cycle must not report incomplete collector slots"
+    );
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum AdmittedCycle {
+    Complete,
+    Cancelled,
+    CacheChanged,
+}
+
+fn recorded_events(buffer: &StdMutex<Vec<u8>>) -> Vec<serde_json::Value> {
+    let buffer = buffer.lock().expect("collector log buffer");
+    buffer
+        .split_inclusive(|byte| *byte == b'\n')
+        .filter(|line| line.ends_with(b"\n"))
+        .map(|line| serde_json::from_slice(line).expect("complete collector events should be JSON"))
+        .collect()
+}
+
+async fn admitted_collector_cycle(
+    store: &Arc<ECStore>,
+    owner: &Arc<NodeAdmission>,
+    logs: &Arc<StdMutex<Vec<u8>>>,
+    cycle: u64,
+    case: AdmittedCycle,
+) {
+    const BUCKET: &str = "admitted-collector";
+    let set = Arc::clone(&store.pools[0].disk_set[0]);
+    let source = DataUsageCacheSource::new(0, 0);
+    let cache_name = format!("{BUCKET}/{DATA_USAGE_CACHE_NAME}");
+    record_dirty_usage_bucket(BUCKET);
+    let dirty_generation = dirty_usage_buckets_for_tests().get(BUCKET).copied();
+    assert!(dirty_generation.is_some(), "the fixture must start with dirty usage");
+    let original_bucket_revision = DataUsageCache::read_revisions(set.clone(), &cache_name)
+        .await
+        .expect("read bucket cache revisions before the cycle");
+    let original_set_revision = DataUsageCache::read_revisions(set.clone(), DATA_USAGE_CACHE_NAME)
+        .await
+        .expect("read set cache revisions before the cycle");
+    logs.lock().expect("collector log buffer").clear();
+
+    // The worker can acquire its cache lease but cannot start a node operation.
+    let mut blocker = Some(
+        tokio::time::timeout(Duration::from_secs(10), owner.acquire())
+            .await
+            .expect("fixture should acquire the node unit"),
+    );
+    assert_eq!(owner.snapshot().active, 1);
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
+    let (updates, mut receiver) = mpsc::channel(1);
+    let scan = nsscanner_with_storage_status_scoped(
+        store.as_ref(),
+        ScannerCycleRequest {
+            ctx: budget.token(),
+            budget: Arc::clone(&budget),
+            updates,
+            want_cycle: cycle,
+            leader_epoch: LEADER_EPOCH,
+            scan_mode: HealScanMode::Normal,
+            scan_scope: ScannerBucketScanScope::default(),
+            persisted_usage_baseline: None,
+            observed_usage_candidate: None,
+            requires_full_scan: true,
+            service_cohort: None,
+            resolved_scope_observer: None,
+            post_walk_gate: None,
+        },
+    );
+    tokio::pin!(scan);
+    let cache_ready = async {
+        loop {
+            if recorded_events(logs)
+                .iter()
+                .any(|event| event["fields"]["state"] == "cache_ready" && event["fields"]["bucket"] == BUCKET)
+            {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    };
+    tokio::select! {
+        biased;
+        result = &mut scan => panic!("cycle ended before bucket cache admission: {result:?}"),
+        result = tokio::time::timeout(Duration::from_secs(60), cache_ready) => {
+            result.expect("worker should load its cache while the node is occupied");
+        }
+    }
+
+    let mut competing_lease = Some(tokio::select! {
+        biased;
+        result = &mut scan => panic!("cycle ended before yielding its cache lease: {result:?}"),
+        result = tokio::time::timeout(
+            Duration::from_secs(60), acquire_scanner_cache_locks(set.as_ref(), &cache_name, source),
+        ) => result.expect("scanner should yield its cache lease").expect("competing cache lease"),
+    });
+    let expected_bucket_revision = if case == AdmittedCycle::CacheChanged {
+        let mut newer = DataUsageCache::default();
+        newer
+            .load(set.clone(), &cache_name)
+            .await
+            .expect("load the protected bucket cache");
+        assert!(newer.info.snapshot_complete, "the competing writer starts from the certified cache");
+        newer.info.next_cycle = cycle - 1;
+        newer
+            .save(set.clone(), &cache_name)
+            .await
+            .expect("persist the competing cache revision");
+        let updated = DataUsageCache::read_revisions(set.clone(), &cache_name)
+            .await
+            .expect("read the competing cache revision");
+        assert_ne!(updated, original_bucket_revision, "the competing write must change the revision");
+        Some(updated)
+    } else {
+        None
+    };
+
+    if case == AdmittedCycle::Cancelled {
+        ctx.cancel();
+    } else {
+        blocker.take().expect("held node unit").complete();
+        let admitted = async {
+            while owner.snapshot().active != 1 {
+                tokio::task::yield_now().await;
+            }
+        };
+        tokio::select! {
+            biased;
+            result = &mut scan => panic!("cycle ended before reacquiring node admission: {result:?}"),
+            result = tokio::time::timeout(Duration::from_secs(60), admitted) => {
+                result.expect("scanner should acquire the released node unit");
+            }
+        }
+        assert!(
+            !recorded_events(logs)
+                .iter()
+                .any(|event| event["fields"]["state"] == "scan_completed" && event["fields"]["bucket"] == BUCKET),
+            "the worker cannot complete the walk while the other lease is held"
+        );
+    }
+
+    if case != AdmittedCycle::Cancelled {
+        drop(competing_lease.take());
+    }
+    let result = tokio::time::timeout(Duration::from_secs(60), &mut scan)
+        .await
+        .expect("scanner cycle should finish")
+        .expect("scanner cycle should return a status");
+    let snapshot = tokio::time::timeout(Duration::from_secs(10), receiver.recv())
+        .await
+        .expect("cycle update channel should finish");
+    assert!(
+        tokio::time::timeout(Duration::from_secs(10), receiver.recv())
+            .await
+            .expect("cycle update channel should close")
+            .is_none(),
+        "a cycle emits at most one snapshot"
+    );
+    if case == AdmittedCycle::Cancelled {
+        assert_eq!(owner.snapshot().active, 1, "cancellation must not take the occupied node unit");
+        drop(blocker.take());
+        drop(competing_lease.take());
+    }
+    assert_eq!(owner.snapshot().active, 0, "the cycle must release its node capacity");
+    assert!(owner.try_acquire().is_some(), "the node queue must not retain the finished request");
+    let events = recorded_events(logs);
+    let root_summaries: Vec<_> = events
+        .iter()
+        .filter(|event| event["fields"]["state"] == "bucket_root_collector_drained")
+        .map(|event| &event["fields"])
+        .collect();
+    assert_eq!(root_summaries.len(), 1, "each real cycle drains one set collector");
+    let roots = root_summaries[0];
+    assert_eq!(roots["collector_cache_cycle"], cycle);
+    assert_eq!(roots["rejected_roots"], 0);
+    assert_eq!(dirty_usage_buckets_for_tests().get(BUCKET).copied(), dirty_generation);
+
+    match case {
+        AdmittedCycle::Complete => {
+            assert_eq!(result.status, ScannerCycleStatus::Complete);
+            let snapshot = snapshot.expect("complete cycle must produce a usage snapshot");
+            assert!(snapshot.usage_snapshot_complete);
+            assert_eq!(snapshot.objects_total_count, 4);
+            assert_eq!(roots["received_roots"], 1);
+            assert_eq!(roots["distinct_applied_roots"], 1);
+            assert_eq!(roots["cancelled_at_drain"], false);
+            assert!(collector_events(&events).is_empty());
+        }
+        AdmittedCycle::Cancelled | AdmittedCycle::CacheChanged => {
+            assert_eq!(result.status, ScannerCycleStatus::Incomplete);
+            assert!(!result.has_dirty_usage_to_acknowledge());
+            assert!(snapshot.is_none_or(|snapshot| !snapshot.usage_snapshot_complete));
+            assert_eq!(roots["received_roots"], 0);
+            assert_eq!(roots["distinct_applied_roots"], 0);
+            assert_eq!(roots["cancelled_at_drain"], case == AdmittedCycle::Cancelled);
+            assert_eq!(
+                DataUsageCache::read_revisions(set.clone(), DATA_USAGE_CACHE_NAME)
+                    .await
+                    .expect("read the retained set cache revisions"),
+                original_set_revision
+            );
+            let expected = expected_bucket_revision.unwrap_or(original_bucket_revision);
+            assert_eq!(
+                DataUsageCache::read_revisions(set.clone(), &cache_name)
+                    .await
+                    .expect("read the retained bucket cache revisions"),
+                expected
+            );
+            let summaries = collector_events(&events);
+            if case == AdmittedCycle::Cancelled {
+                assert!(summaries.is_empty(), "a cancelled cycle must not report an ordinary incomplete summary");
+            } else {
+                assert!(events.iter().any(|event| {
+                    event["fields"]["state"] == "scan_failed"
+                        && event["fields"]["error"]
+                            .as_str()
+                            .is_some_and(|error| error.contains("scanner cache changed while waiting for node admission"))
+                }));
+                assert_eq!(summaries.len(), 1);
+                let fields = summaries[0];
+                assert_eq!(fields["cycle"], cycle);
+                assert_eq!(fields["leader_epoch"], LEADER_EPOCH);
+                assert_eq!(fields["expected_sources"], 1);
+                assert_eq!(fields["observed_slots"], 1);
+                assert_eq!(fields["unobserved_slots"], 0);
+                assert_eq!(fields["sample_observed"], true);
+                assert_eq!(fields["sample_snapshot_complete"], false);
+                assert_eq!(fields["sample_snapshot_cycle"], cycle);
+            }
+        }
+    }
+    let mut retained = DataUsageCache::default();
+    retained
+        .load(set, &cache_name)
+        .await
+        .expect("load the retained certified bucket cache");
+    assert!(retained.info.snapshot_complete);
+    assert_eq!(
+        retained.info.next_cycle,
+        match case {
+            AdmittedCycle::Complete | AdmittedCycle::Cancelled => 1,
+            AdmittedCycle::CacheChanged => 2,
+        }
+    );
+    assert_eq!(retained.checked_flatten(BUCKET).expect("retained bucket tree").objects, 4);
+}
+
+#[test]
+fn collector_cycle_with_node_admission_preserves_cache_and_reports_outcomes() {
+    const TEST: &str = "scanner_io::tests::cycle_collector_diagnostics::collector_cycle_with_node_admission_preserves_cache_and_reports_outcomes";
+    const CHILD: &str = "RUSTFS_TEST_ADMITTED_COLLECTOR_CHILD";
+    if std::env::var(CHILD).ok().as_deref() != Some(TEST) {
+        struct ChildReaper(std::process::Child);
+
+        impl Drop for ChildReaper {
+            fn drop(&mut self) {
+                if !matches!(self.0.try_wait(), Ok(Some(_))) && self.0.kill().is_ok() {
+                    let _ = self.0.wait();
+                }
+            }
+        }
+
+        // The production admission owner is process-global and cannot be reset.
+        // Snapshot only non-RustFS settings so concurrent environment tests cannot
+        // change the child fixture's scanner or storage configuration.
+        let child_env = std::env::vars_os().filter(|(name, _)| {
+            let name = name.to_string_lossy().to_ascii_uppercase();
+            !name.starts_with("RUSTFS_") && !name.starts_with("MINIO_")
+        });
+        let child = std::process::Command::new(std::env::current_exe().expect("test executable"))
+            .args(["--exact", TEST, "--nocapture", "--color", "never"])
+            .env_clear()
+            .envs(child_env)
+            .env(CHILD, TEST)
+            .env("RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES", "1")
+            .env("RUSTFS_SCANNER_IDLE_MODE", "false")
+            .env("RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS", "off")
+            .env("RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC", "off")
+            .env("RUSTFS_LOCK_ACQUIRE_TIMEOUT", "30")
+            .env("NO_PROXY", "127.0.0.1,localhost")
+            .env("no_proxy", "127.0.0.1,localhost")
+            .stdout(std::process::Stdio::piped())
+            .stderr(std::process::Stdio::piped())
+            .spawn()
+            .expect("start the isolated collector cycle");
+        let mut child = ChildReaper(child);
+        let mut child_stdout = child.0.stdout.take().expect("child stdout pipe");
+        let mut child_stderr = child.0.stderr.take().expect("child stderr pipe");
+        // Drain both pipes while the child runs so neither can block its exit.
+        let stdout_reader = std::thread::spawn(move || {
+            let mut bytes = Vec::new();
+            let error = std::io::Read::read_to_end(&mut child_stdout, &mut bytes).err();
+            (bytes, error)
+        });
+        let stderr_reader = std::thread::spawn(move || {
+            let mut bytes = Vec::new();
+            let error = std::io::Read::read_to_end(&mut child_stderr, &mut bytes).err();
+            (bytes, error)
+        });
+        let started = std::time::Instant::now();
+        let mut timed_out = false;
+        let mut wait_error = None;
+        let status = loop {
+            match child.0.try_wait() {
+                Ok(Some(status)) => break status,
+                Ok(None) if started.elapsed() < Duration::from_secs(300) => {
+                    std::thread::sleep(Duration::from_millis(50));
+                }
+                Ok(None) => {
+                    timed_out = true;
+                    if let Err(error) = child.0.kill() {
+                        // The child can exit between the status check and kill.
+                        if let Some(status) = child.0.try_wait().expect("check child after failed kill") {
+                            break status;
+                        }
+                        panic!("failed to terminate the timed-out collector child: {error}");
+                    }
+                    break child.0.wait().expect("reap the timed-out collector child");
+                }
+                Err(error) => {
+                    wait_error = Some(error);
+                    if let Err(error) = child.0.kill() {
+                        if let Some(status) = child.0.try_wait().expect("check child after failed kill") {
+                            break status;
+                        }
+                        panic!("failed to terminate the collector child after a wait error: {error}");
+                    }
+                    break child.0.wait().expect("reap the collector child after a wait error");
+                }
+            }
+        };
+        let (stdout, stdout_error) = stdout_reader.join().expect("join child stdout reader");
+        let (stderr, stderr_error) = stderr_reader.join().expect("join child stderr reader");
+        let stdout = String::from_utf8_lossy(&stdout);
+        let stderr = String::from_utf8_lossy(&stderr);
+        assert!(
+            !timed_out
+                && wait_error.is_none()
+                && stdout_error.is_none()
+                && stderr_error.is_none()
+                && status.success()
+                && stdout
+                    .lines()
+                    .any(|line| line.starts_with("test result: ok. 1 passed; 0 failed;")),
+            "the isolated collector cycle must finish within 300 seconds and run exactly once \
+             (status={status}, timed_out={timed_out}, wait_error={wait_error:?}, \
+             stdout_error={stdout_error:?}, stderr_error={stderr_error:?}): \
+             {stdout}\n{stderr}"
+        );
+        return;
+    }
+
+    let logs = Arc::new(StdMutex::new(Vec::new()));
+    let writer = Arc::clone(&logs);
+    tracing::subscriber::set_global_default(
+        tracing_subscriber::fmt()
+            .json()
+            .without_time()
+            .with_ansi(false)
+            .with_max_level(tracing::Level::DEBUG)
+            .with_writer(move || LogWriter(Arc::clone(&writer)))
+            .finish(),
+    )
+    .expect("install the isolated collector observer");
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .expect("create the isolated scanner runtime");
+    runtime.block_on(async {
+        let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
+        clear_dirty_usage_buckets_for_tests();
+        store
+            .make_bucket("admitted-collector", &MakeBucketOptions::default())
+            .await
+            .expect("create the fixture bucket");
+        for object in ["p0/a", "p1/b", "p2/c", "p3/d"] {
+            let mut reader = ScannerPutObjReader::from_vec(vec![7; 16]);
+            store
+                .put_object("admitted-collector", object, &mut reader, &ScannerObjectOptions::default())
+                .await
+                .expect("persist the fixture object");
+        }
+        wait_for_namespace_commit_tails(store.as_ref()).await;
+        assert!(!crate::scanner_is_erasure().await, "this fixture must not start scanner healing");
+
+        let owner = NodeAdmission::new(NonZeroU16::new(1).expect("one node unit"));
+        let observation = FeedbackObservation {
+            generation: 1,
+            max_concurrency: 1,
+            pressure: ResourcePressure::Unknown,
+            completed_units: None,
+        };
+        let mut controller = FeedbackController::new();
+        owner.update(&mut controller, Some(observation));
+        let sampler_stop = CancellationToken::new();
+        let sampler_cancel = sampler_stop.clone();
+        let sampler_owner = owner.clone();
+        let sampler = tokio::spawn(async move {
+            let mut tick = tokio::time::interval(Duration::from_secs(1));
+            loop {
+                tokio::select! {
+                    _ = sampler_cancel.cancelled() => break,
+                    _ = tick.tick() => { sampler_owner.update(&mut controller, Some(observation)); }
+                }
+            }
+        });
+        assert!(crate::set_scanner_node_admission(owner.clone()), "install the real node owner once");
+        admitted_collector_cycle(&store, &owner, &logs, 1, AdmittedCycle::Complete).await;
+        admitted_collector_cycle(&store, &owner, &logs, 2, AdmittedCycle::Cancelled).await;
+        admitted_collector_cycle(&store, &owner, &logs, 3, AdmittedCycle::CacheChanged).await;
+        sampler_stop.cancel();
+        sampler.await.expect("stop the admission feedback sampler");
+    });
+}