diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 860ebb2cd..f568050d8 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -105,6 +105,12 @@ struct DirtyUsageSnapshot { covers_all_pending: bool, } +#[derive(Clone, Debug, Default)] +pub(crate) struct ScannerBucketScanScope { + selected_buckets: Option>>, + baseline_scan_plan_digest: Option, +} + pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool { matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR)) } @@ -146,6 +152,7 @@ fn object_lock_config_enabled(config: &ObjectLockConfiguration) -> bool { pub struct ScannerBucketScanPlan { buckets: Vec, all_buckets: Arc>, + scope: ScannerBucketScanScope, digest: DataUsageScanPlanDigest, leader_epoch: u64, tier_registry_generation: u64, @@ -732,6 +739,8 @@ mod dirty_usage; mod guards; mod io_cache; mod io_cycle; +#[cfg(test)] +use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan}; pub(crate) use io_cycle::nsscanner_with_storage_status; mod io_disk; #[cfg(test)] diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index c99ffacd1..a8686a1e4 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -14,6 +14,93 @@ /// ScannerIOCache implementation for SetDisks: bucket ordering, worker fan-out, merge, and publish. use super::*; +#[derive(Clone, Copy)] +pub(super) struct ScannerSetCacheGeneration { + pub(super) want_cycle: u64, + pub(super) leader_epoch: u64, + pub(super) tier_registry_generation: u64, + pub(super) source: DataUsageCacheSource, + pub(super) scan_plan_digest: DataUsageScanPlanDigest, +} + +pub(super) struct PreparedScopedSetScan { + pub(super) buckets: Vec, + pub(super) cache: DataUsageCache, +} + +pub(super) fn prepare_scoped_set_scan( + old_cache: &DataUsageCache, + set_buckets: &[BucketInfo], + all_buckets: &[BucketInfo], + scope: &ScannerBucketScanScope, + generation: ScannerSetCacheGeneration, +) -> Option { + let (Some(selected_buckets), Some(baseline_scan_plan_digest)) = (&scope.selected_buckets, scope.baseline_scan_plan_digest) + else { + return None; + }; + if selected_buckets.is_empty() + || !old_cache.info.snapshot_complete + || old_cache.info.last_update.is_none() + || old_cache.info.name != DATA_USAGE_ROOT + || old_cache.info.next_cycle > generation.want_cycle + || old_cache.info.leader_epoch != generation.leader_epoch + || old_cache.info.tier_registry_generation != Some(generation.tier_registry_generation) + || old_cache.info.source != Some(generation.source) + || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest) + || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT + || old_cache.checked_flatten_complete_scope(DATA_USAGE_ROOT).is_none() + { + return None; + } + + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: generation.want_cycle, + leader_epoch: generation.leader_epoch, + tier_registry_generation: Some(generation.tier_registry_generation), + source: Some(generation.source), + snapshot_complete: false, + scan_plan_digest: Some(generation.scan_plan_digest), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + lkg_snapshot_complete: true, + lkg_next_cycle: Some(old_cache.info.next_cycle), + lkg_last_update: old_cache.info.last_update, + lkg_leader_epoch: Some(old_cache.info.leader_epoch), + lkg_scan_plan_digest: old_cache.info.scan_plan_digest, + ..Default::default() + }, + cache: HashMap::new(), + }; + cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); + let root_hash = crate::hash_path(DATA_USAGE_ROOT); + let mut current_bucket_names = HashSet::with_capacity(all_buckets.len()); + for bucket in all_buckets { + if !current_bucket_names.insert(bucket.name.as_str()) { + return None; + } + if selected_buckets.contains(&bucket.name) { + cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); + continue; + } + + let bucket_hash = crate::hash_path(&bucket.name); + old_cache.find(&bucket.name)?; + cache.copy_with_children(old_cache, &bucket_hash, &Some(root_hash.clone())); + cache.find(&bucket.name)?; + } + + Some(PreparedScopedSetScan { + buckets: set_buckets + .iter() + .filter(|bucket| selected_buckets.contains(&bucket.name)) + .cloned() + .collect(), + cache, + }) +} + #[async_trait::async_trait] impl ScannerIOCache for SetDisks { #[tracing::instrument(skip(self, budget, scan_plan, updates))] @@ -27,8 +114,9 @@ impl ScannerIOCache for SetDisks { scan_mode: HealScanMode, ) -> Result<()> { let ScannerBucketScanPlan { - buckets, + mut buckets, all_buckets, + scope, digest: scan_plan_digest, leader_epoch, tier_registry_generation, @@ -63,26 +151,57 @@ impl ScannerIOCache for SetDisks { "Scanner old data usage cache load failed; rebuilding from bucket caches" ); } + let scoped_scan = prepare_scoped_set_scan( + &old_cache, + &buckets, + &all_buckets, + &scope, + ScannerSetCacheGeneration { + want_cycle, + leader_epoch, + tier_registry_generation, + source, + scan_plan_digest, + }, + ); + let mut scoped_cache = scoped_scan.map(|prepared| { + buckets = prepared.buckets; + prepared.cache + }); if buckets.is_empty() { let now = SystemTime::now(); - let mut cache = DataUsageCache { - info: DataUsageCacheInfo { - name: DATA_USAGE_ROOT.to_string(), - next_cycle: want_cycle, - last_update: Some(now), - leader_epoch, - tier_registry_generation: Some(tier_registry_generation), - source: Some(source), - snapshot_complete: true, - scan_plan_digest: Some(scan_plan_digest), - cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, - ..Default::default() - }, - cache: HashMap::new(), + let mut cache = match scoped_cache.take() { + Some(cache) => cache, + None => { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: want_cycle, + leader_epoch, + tier_registry_generation: Some(tier_registry_generation), + source: Some(source), + scan_plan_digest: Some(scan_plan_digest), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + ..Default::default() + }, + cache: HashMap::new(), + }; + cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); + for bucket in all_buckets.iter() { + cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); + } + cache + } }; - cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); - for bucket in all_buckets.iter() { - cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); + cache.info.last_update = Some(now); + cache.info.snapshot_complete = true; + cache.info.lkg_snapshot_complete = false; + cache.info.lkg_next_cycle = None; + cache.info.lkg_last_update = None; + cache.info.lkg_leader_epoch = None; + cache.info.lkg_scan_plan_digest = None; + if cache.find(DATA_USAGE_ROOT).is_none() { + cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); } reset_disk_bucket_scan_gauges(&pool_label, &set_label); return persist_and_publish_cache_snapshot( @@ -269,92 +388,102 @@ impl ScannerIOCache for SetDisks { record_disk_bucket_scans_active(0, &pool_label, &set_label); let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone()); - // Fence a stale set aggregate before copying entries into per-bucket work caches. - if old_cache.info.next_cycle <= want_cycle - && old_cache.info.leader_epoch <= leader_epoch - && old_cache.info.tier_registry_generation != Some(tier_registry_generation) - { - old_cache.info.scan_plan_digest = None; - } - let old_lkg = old_cache.info.snapshot_complete.then_some({ - ( - old_cache.info.next_cycle, - old_cache.info.last_update, - old_cache.info.leader_epoch, - old_cache.info.scan_plan_digest, - ) - }); - let prepare_outcome = match old_cache.prepare_for_scan( - DATA_USAGE_ROOT, - want_cycle, - leader_epoch, - source, - scan_plan_digest, - require_cache_source, - ) { - DataUsageCachePrepareOutcome::RejectedNewerCycle => { - cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel); - warn!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_CACHE_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - pool = self.pool_index, - set = self.set_index, - cache_name = DATA_USAGE_CACHE_NAME, - requested_cycle = want_cycle, - cached_cycle = old_cache.info.next_cycle, - state = "stale_cycle_rejected", - "Scanner rejected a set cache cycle regression" - ); - return Ok(()); + let mut cache = if let Some(cache) = scoped_cache.take() { + cache + } else { + // Fence a stale set aggregate before copying entries into per-bucket work caches. + if old_cache.info.next_cycle <= want_cycle + && old_cache.info.leader_epoch <= leader_epoch + && old_cache.info.tier_registry_generation != Some(tier_registry_generation) + { + old_cache.info.scan_plan_digest = None; } - DataUsageCachePrepareOutcome::RejectedNewerLeader => { - warn!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_CACHE_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - pool = self.pool_index, - set = self.set_index, - cache_name = DATA_USAGE_CACHE_NAME, - requested_epoch = leader_epoch, - cached_epoch = old_cache.info.leader_epoch, - state = "stale_leader_rejected", - "Scanner rejected work from an older leader epoch" - ); - return Ok(()); - } - outcome => outcome, - }; - if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused) - && let Some((cycle, last_update, epoch, digest)) = old_lkg - { - old_cache.info.lkg_snapshot_complete = true; - old_cache.info.lkg_next_cycle = Some(cycle); - old_cache.info.lkg_last_update = last_update; - old_cache.info.lkg_leader_epoch = Some(epoch); - old_cache.info.lkg_scan_plan_digest = digest; - } - - let mut cache = DataUsageCache { - info: DataUsageCacheInfo { - name: DATA_USAGE_ROOT.to_string(), - next_cycle: want_cycle, + let old_lkg = old_cache.info.snapshot_complete.then_some({ + ( + old_cache.info.next_cycle, + old_cache.info.last_update, + old_cache.info.leader_epoch, + old_cache.info.scan_plan_digest, + ) + }); + let prepare_outcome = match old_cache.prepare_for_scan( + DATA_USAGE_ROOT, + want_cycle, leader_epoch, - tier_registry_generation: Some(tier_registry_generation), - source: Some(source), - snapshot_complete: false, - scan_plan_digest: Some(scan_plan_digest), - cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, - ..Default::default() - }, - cache: HashMap::new(), + source, + scan_plan_digest, + require_cache_source, + ) { + DataUsageCachePrepareOutcome::RejectedNewerCycle => { + cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel); + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + pool = self.pool_index, + set = self.set_index, + cache_name = DATA_USAGE_CACHE_NAME, + requested_cycle = want_cycle, + cached_cycle = old_cache.info.next_cycle, + state = "stale_cycle_rejected", + "Scanner rejected a set cache cycle regression" + ); + return Ok(()); + } + DataUsageCachePrepareOutcome::RejectedNewerLeader => { + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_CACHE_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + pool = self.pool_index, + set = self.set_index, + cache_name = DATA_USAGE_CACHE_NAME, + requested_epoch = leader_epoch, + cached_epoch = old_cache.info.leader_epoch, + state = "stale_leader_rejected", + "Scanner rejected work from an older leader epoch" + ); + return Ok(()); + } + outcome => outcome, + }; + if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused) + && let Some((cycle, last_update, epoch, digest)) = old_lkg + { + old_cache.info.lkg_snapshot_complete = true; + old_cache.info.lkg_next_cycle = Some(cycle); + old_cache.info.lkg_last_update = last_update; + old_cache.info.lkg_leader_epoch = Some(epoch); + old_cache.info.lkg_scan_plan_digest = digest; + } + + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: want_cycle, + leader_epoch, + tier_registry_generation: Some(tier_registry_generation), + source: Some(source), + snapshot_complete: false, + scan_plan_digest: Some(scan_plan_digest), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + lkg_snapshot_complete: old_cache.info.lkg_snapshot_complete, + lkg_next_cycle: old_cache.info.lkg_next_cycle, + lkg_last_update: old_cache.info.lkg_last_update, + lkg_leader_epoch: old_cache.info.lkg_leader_epoch, + lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest, + ..Default::default() + }, + cache: HashMap::new(), + }; + cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); + for bucket in all_buckets.iter() { + cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); + } + cache }; - cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); - for bucket in all_buckets.iter() { - cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); - } let (bucket_tx, bucket_rx) = mpsc::channel::(buckets.len()); @@ -1257,11 +1386,6 @@ impl ScannerIOCache for SetDisks { incomplete_scope.info.snapshot_complete = false; incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest); incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT; - incomplete_scope.info.lkg_snapshot_complete = old_cache.info.lkg_snapshot_complete; - incomplete_scope.info.lkg_next_cycle = old_cache.info.lkg_next_cycle; - incomplete_scope.info.lkg_last_update = old_cache.info.lkg_last_update; - incomplete_scope.info.lkg_leader_epoch = old_cache.info.lkg_leader_epoch; - incomplete_scope.info.lkg_scan_plan_digest = old_cache.info.lkg_scan_plan_digest; if let Err(e) = updates.send(incomplete_scope).await { error!( target: "rustfs::scanner::io", diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index f51caf381..3b14e9ff2 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -63,6 +63,41 @@ pub(crate) async fn nsscanner_with_storage_status( where S: ScannerStorage, { + let request = ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle, + leader_epoch, + scan_mode, + scan_scope: ScannerBucketScanScope::default(), + }; + nsscanner_with_storage_status_scoped(store, request).await +} + +pub(crate) struct ScannerCycleRequest { + pub(crate) ctx: CancellationToken, + pub(crate) budget: Arc, + pub(crate) updates: mpsc::Sender, + pub(crate) want_cycle: u64, + pub(crate) leader_epoch: u64, + pub(crate) scan_mode: HealScanMode, + pub(crate) scan_scope: ScannerBucketScanScope, +} + +pub(crate) async fn nsscanner_with_storage_status_scoped(store: &S, request: ScannerCycleRequest) -> Result +where + S: ScannerStorage, +{ + let ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle, + leader_epoch, + scan_mode, + scan_scope, + } = request; let child_token = ctx.child_token(); let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch); @@ -280,6 +315,7 @@ where let scan_plan = ScannerBucketScanPlan { buckets: set_buckets, all_buckets: Arc::clone(&all_buckets), + scope: scan_scope.clone(), digest: scan_plan_digest, leader_epoch, tier_registry_generation, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 47f1e86a0..a59d620fc 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -765,6 +765,155 @@ fn bucket_usage_scan_order_prioritizes_dirty_buckets() { assert_eq!(names, vec!["dirty", "missing", "cached"]); } +fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsageScanPlanDigest) -> DataUsageCache { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: 7, + last_update: Some(SystemTime::now()), + leader_epoch: 11, + source: Some(DataUsageCacheSource::new(1, 2)), + snapshot_complete: true, + scan_plan_digest: Some(scan_plan_digest), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + tier_registry_generation: Some(13), + ..Default::default() + }, + ..Default::default() + }; + cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); + for (bucket, size) in buckets { + cache.replace( + bucket, + DATA_USAGE_ROOT, + DataUsageEntry { + size: *size, + objects: 1, + ..Default::default() + }, + ); + } + cache +} + +#[test] +fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() { + let baseline_digest = DataUsageScanPlanDigest([1; 32]); + let current_digest = DataUsageScanPlanDigest([2; 32]); + let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20), ("deleted", 30)], baseline_digest); + old_cache.replace( + "stable/prefix", + "stable", + DataUsageEntry { + size: 5, + objects: 1, + ..Default::default() + }, + ); + let all_buckets = vec![bucket_info("stable"), bucket_info("dirty")]; + let selected_buckets = Arc::new(HashSet::from(["dirty".to_string(), "deleted".to_string()])); + + let prepared = prepare_scoped_set_scan( + &old_cache, + &all_buckets, + &all_buckets, + &ScannerBucketScanScope { + selected_buckets: Some(selected_buckets), + baseline_scan_plan_digest: Some(baseline_digest), + }, + ScannerSetCacheGeneration { + want_cycle: 8, + leader_epoch: 11, + tier_registry_generation: 13, + source: DataUsageCacheSource::new(1, 2), + scan_plan_digest: current_digest, + }, + ) + .expect("complete matching set cache should support a scoped scan"); + + assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::>(), ["dirty"]); + let stable = prepared + .cache + .checked_flatten("stable") + .expect("unselected bucket subtree should be retained"); + assert_eq!((stable.size, stable.objects), (15, 2)); + assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0))); + assert!(prepared.cache.find("deleted").is_none()); + assert_eq!(prepared.cache.info.scan_plan_digest, Some(current_digest)); + assert_eq!(prepared.cache.info.next_cycle, 8); + assert!(!prepared.cache.info.snapshot_complete); + assert!(prepared.cache.info.lkg_snapshot_complete); + assert_eq!(prepared.cache.info.lkg_next_cycle, Some(7)); + assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest)); +} + +#[test] +fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() { + let baseline_digest = DataUsageScanPlanDigest([3; 32]); + let old_cache = complete_set_usage_cache(&[("stable", 10)], baseline_digest); + let all_buckets = vec![bucket_info("stable"), bucket_info("new")]; + + assert!( + prepare_scoped_set_scan( + &old_cache, + &all_buckets, + &all_buckets, + &ScannerBucketScanScope { + selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), + baseline_scan_plan_digest: Some(baseline_digest), + }, + ScannerSetCacheGeneration { + want_cycle: 8, + leader_epoch: 11, + tier_registry_generation: 13, + source: DataUsageCacheSource::new(1, 2), + scan_plan_digest: DataUsageScanPlanDigest([4; 32]), + }, + ) + .is_none() + ); +} + +#[test] +fn scoped_set_scan_requires_an_exact_complete_baseline() { + let baseline_digest = DataUsageScanPlanDigest([5; 32]); + let all_buckets = vec![bucket_info("dirty")]; + let scope = ScannerBucketScanScope { + selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), + baseline_scan_plan_digest: Some(baseline_digest), + }; + let generation = ScannerSetCacheGeneration { + want_cycle: 8, + leader_epoch: 11, + tier_registry_generation: 13, + source: DataUsageCacheSource::new(1, 2), + scan_plan_digest: DataUsageScanPlanDigest([6; 32]), + }; + + let mut incomplete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); + incomplete.info.snapshot_complete = false; + assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation).is_none()); + + let mut not_durable = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); + not_durable.info.last_update = None; + assert!(prepare_scoped_set_scan(¬_durable, &all_buckets, &all_buckets, &scope, generation).is_none()); + + let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); + wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32])); + assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none()); + + let empty_scope = ScannerBucketScanScope { + selected_buckets: Some(Arc::new(HashSet::new())), + baseline_scan_plan_digest: Some(baseline_digest), + }; + let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); + assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none()); + + let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); + future_cache.info.next_cycle = generation.want_cycle.saturating_add(1); + assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation).is_none()); +} + #[test] fn record_set_scan_failure_preserves_first_error() { let mut first = None;