fix(scanner): report incomplete collector observations

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 listing_window_scope;
 mod partial_bucket_carry;
 mod per_bucket_certification;

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

@@ -0,0 +1,191 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+
+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();
+}