diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 0f97924a4..ab07ba1eb 100644 --- a/crates/scanner/src/scanner.rs +++ b/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( cycle_revision: &mut DataUsageCacheRevision, leader_epoch: u64, cycle_budget: Arc, + 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(), ) diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index e45c7e7c0..cd9723d73 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/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, + /// 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>, } 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( @@ -103,6 +111,9 @@ async fn resolve_scanner_bucket_scan_scope( 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; diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 2bbbc918f..b93a2e344 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/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);