fix(scanner): keep maintenance cycles outside dirty bucket scopes

Force complete bucket scope for deep scans and scheduled maintenance while
preserving the existing planner for verified ordinary dirty work.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/scanner.rs

@@ -1564,7 +1564,7 @@ where
     S: ScannerStorage,
 {
     let cycle_budget = ScannerCycleBudget::new(ctx, scanner_cycle_budget_config());
-    run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget).await
+    run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget, true).await
 }
 
 #[instrument(skip_all)]
@@ -1576,6 +1576,7 @@ async fn run_data_scanner_cycle_with_budget<S>(
     cycle_revision: &mut DataUsageCacheRevision,
     leader_epoch: u64,
     cycle_budget: Arc<ScannerCycleBudget>,
+    requires_full_scan: bool,
 ) -> ScannerCycleOutcome
 where
     S: ScannerStorage,
@@ -1714,6 +1715,9 @@ where
             scan_mode,
             scan_scope: crate::scanner_io::ScannerBucketScanScope::default(),
             persisted_usage_baseline: usage_persist_baseline.data.clone(),
+            requires_full_scan,
+            #[cfg(test)]
+            resolved_scope_observer: None,
         },
     )
     .await;
@@ -2772,6 +2776,7 @@ where
                 &mut cycle_revision,
                 leader_epoch,
                 cycle_budget.clone(),
+                true,
             ),
             guard.lock_lost_notified(),
         )
@@ -3062,6 +3067,9 @@ where
                 &mut cycle_revision,
                 leader_epoch,
                 cycle_budget.clone(),
+                maintenance_features.needs_regular_cycle()
+                    || maintenance_generation_seen != Some(scanner_maintenance_generation())
+                    || !matches!(wake_reason, ScannerCycleWakeReason::DirtyUsage | ScannerCycleWakeReason::ClusterActivity),
             ),
             guard.lock_lost_notified(),
         )

crates/scanner/src/scanner_io/io_cycle.rs

@@ -72,6 +72,9 @@ where
         scan_mode,
         scan_scope: ScannerBucketScanScope::default(),
         persisted_usage_baseline: None,
+        requires_full_scan: true,
+        #[cfg(test)]
+        resolved_scope_observer: None,
     };
     nsscanner_with_storage_status_scoped(store, request).await
 }
@@ -85,6 +88,10 @@ pub(crate) struct ScannerCycleRequest {
     pub(crate) scan_mode: HealScanMode,
     pub(crate) scan_scope: ScannerBucketScanScope,
     pub(crate) persisted_usage_baseline: Option<Bytes>,
+    /// Scheduled maintenance must visit clean buckets even with a valid dirty scope.
+    pub(crate) requires_full_scan: bool,
+    #[cfg(test)]
+    pub(crate) resolved_scope_observer: Option<tokio::sync::oneshot::Sender<ScannerBucketScanScope>>,
 }
 
 struct ScannerBucketScopeResolution<'a> {
@@ -93,6 +100,7 @@ struct ScannerBucketScopeResolution<'a> {
     activity_before: &'a crate::scanner::ScannerActivitySnapshot,
     dirty_usage_snapshot: &'a DirtyUsageSnapshot,
     all_buckets: &'a [BucketInfo],
+    requires_full_scan: bool,
 }
 
 async fn resolve_scanner_bucket_scan_scope<S>(
@@ -103,6 +111,9 @@ async fn resolve_scanner_bucket_scan_scope<S>(
 where
     S: ScannerStorage,
 {
+    if resolution.requires_full_scan {
+        return ScannerBucketScanScope::default();
+    }
     if !resolution.requested_scope.is_default()
         || !resolution.dirty_usage_snapshot.covers_all_pending
         || resolution.dirty_usage_snapshot.generation == u64::MAX
@@ -172,6 +183,9 @@ where
         scan_mode,
         scan_scope,
         persisted_usage_baseline,
+        requires_full_scan,
+        #[cfg(test)]
+        resolved_scope_observer,
     } = request;
     let child_token = ctx.child_token();
     let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
@@ -281,9 +295,14 @@ where
             activity_before: &activity_before,
             dirty_usage_snapshot: &dirty_usage_snapshot,
             all_buckets: &all_buckets,
+            requires_full_scan: requires_full_scan || scan_mode == HealScanMode::Deep,
         },
     )
     .await;
+    #[cfg(test)]
+    if let Some(observer) = resolved_scope_observer {
+        let _ = observer.send(scan_scope.clone());
+    }
     let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle));
     let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await;
     let tier_registry_generation = tier_registry.generation;

crates/scanner/src/scanner_io/tests.rs

@@ -317,6 +317,104 @@ async fn scanner_cycle_is_deferred_while_terminal_decommission_is_blocked() {
     }
 }
 
+#[tokio::test]
+#[serial]
+async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    clear_dirty_usage_buckets_for_tests();
+    for bucket in ["hot-bucket", "cold-bucket"] {
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let mut reader = ScannerPutObjReader::from_vec(b"initial".to_vec());
+        store.pools[0].disk_set[0]
+            .put_object(bucket, "initial", &mut reader, &ScannerObjectOptions::default())
+            .await
+            .expect("initial object should persist");
+    }
+    let mut baseline = None;
+    for (index, (scan_mode, requires_full_scan, explicit_scope)) in [
+        (HealScanMode::Normal, true, false),
+        (HealScanMode::Normal, false, false),
+        (HealScanMode::Deep, false, false),
+        (HealScanMode::Normal, true, false),
+        (HealScanMode::Deep, false, true),
+        (HealScanMode::Normal, true, true),
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        if index > 0 {
+            let mut reader = ScannerPutObjReader::from_vec(b"maintenance".to_vec());
+            store.pools[0].disk_set[0]
+                .put_object("cold-bucket", &format!("added-{index}"), &mut reader, &ScannerObjectOptions::default())
+                .await
+                .expect("cold bucket mutation should persist");
+            // Only the hot bucket is in the usage hint. The cold result must
+            // come from this cycle's storage walk, not its previous baseline.
+            record_dirty_usage_bucket("hot-bucket");
+        }
+        let requested_scope = if explicit_scope {
+            ScannerBucketScanScope::from_dirty_buckets(
+                HashSet::from(["hot-bucket".to_string()]),
+                DataUsageScanPlanDigest([7; 32]),
+            )
+        } else {
+            ScannerBucketScanScope::default()
+        };
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let (updates, mut receiver) = mpsc::channel(1);
+        let (observer, observed_scope) = tokio::sync::oneshot::channel();
+        let cycle = u64::try_from(index + 1).expect("test cycle should fit");
+        let result = tokio::time::timeout(
+            Duration::from_secs(30),
+            nsscanner_with_storage_status_scoped(
+                store.as_ref(),
+                ScannerCycleRequest {
+                    ctx,
+                    budget,
+                    updates,
+                    want_cycle: cycle,
+                    leader_epoch: 11,
+                    scan_mode,
+                    scan_scope: requested_scope,
+                    persisted_usage_baseline: baseline,
+                    requires_full_scan,
+                    resolved_scope_observer: Some(observer),
+                },
+            ),
+        )
+        .await
+        .expect("cycle should finish within the test deadline")
+        .expect("cycle should succeed");
+        assert_eq!(result.status, ScannerCycleStatus::Complete, "cycle {cycle}");
+        let resolved = observed_scope.await.expect("production resolver should report its scope");
+        if index == 1 {
+            assert_eq!(
+                resolved.selected_buckets.as_deref(),
+                Some(&HashSet::from(["hot-bucket".to_string()])),
+                "ordinary dirty work must retain the existing planner"
+            );
+        } else {
+            assert!(resolved.is_default(), "cycle {cycle} must visit the full maintenance scope");
+        }
+        let mut snapshot = receiver.recv().await.expect("cycle should publish a snapshot");
+        assert!(snapshot.usage_snapshot_complete, "cycle {cycle}");
+        assert_eq!(
+            snapshot.buckets_usage["cold-bucket"].objects_count,
+            u64::try_from(index + 1).expect("count should fit")
+        );
+        assert_eq!(snapshot.buckets_usage["hot-bucket"].objects_count, 1);
+        assert_eq!(snapshot.scanner_cycle, Some(cycle));
+        assert_eq!(snapshot.scanner_epoch, Some(11));
+        snapshot.usage_snapshot_converged = Some(true);
+        baseline = Some(Bytes::from(serde_json::to_vec(&snapshot).expect("complete baseline should encode")));
+    }
+    clear_dirty_usage_buckets_for_tests();
+}
+
 #[tokio::test]
 async fn data_usage_publish_fails_when_receiver_is_closed() {
     let (updates, receiver) = mpsc::channel(1);