From d5a1530409e76130fc893cb83b5889981077f734 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 5 Sep 2026 12:53:40 +0800 Subject: [PATCH] fix(scanner): require complete publication coverage Refs rustfs/backlog#2261 and rustfs/backlog#2240. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/scanner/src/data_usage_define.rs | 23 +++ crates/scanner/src/scanner/tests.rs | 30 ++++ crates/scanner/src/scanner_io.rs | 24 +++ crates/scanner/src/scanner_io/cache.rs | 109 +++++++++--- crates/scanner/src/scanner_io/io_cache.rs | 19 +- crates/scanner/src/scanner_io/io_cycle.rs | 13 +- .../src/scanner_io/publish_gate_tests.rs | 165 +++++++++++++++++- crates/scanner/src/scanner_io/tests.rs | 71 +++++++- 8 files changed, 402 insertions(+), 52 deletions(-) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 35229ba6b..bf5d2392c 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -774,6 +774,29 @@ impl DataUsageCache { (visited == expected_entries).then_some(entry) } + pub(crate) fn has_complete_root_inventory(&self, bucket_keys: &HashSet) -> bool { + let Some(root) = self.find(DATA_USAGE_ROOT) else { + return false; + }; + // Set roots only connect bucket entries. Scalar data at the root, an + // extra bucket, or an orphan must not disappear during bucket folding. + root.children.len() == bucket_keys.len() + && bucket_keys.iter().all(|key| root.children.contains(key)) + && root.size == 0 + && root.objects == 0 + && root.versions == 0 + && root.delete_markers == 0 + && root.failed_objects == 0 + && !root.compacted + && root.obj_sizes.is_empty() + && root.obj_versions.is_empty() + && root.replication_stats.is_none() + && root.all_tier_stats.is_none() + && root.unknown_tier_stats.is_none() + && root.tier_accounting_proof.is_none() + && self.checked_flatten_complete(DATA_USAGE_ROOT).is_some() + } + fn checked_flatten_inner(&self, path: &str) -> Option<(DataUsageEntry, usize)> { let root_key = hash_path(path).key(); let (root_key, root) = self.cache.get_key_value(&root_key)?; diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index b9b4cd299..84a39c62a 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -4758,6 +4758,28 @@ async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() { #[serial] async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() { let (_temp_dir, store) = setup_scanner_cycle_store().await; + let quota_ledger_path = "config/quota-ledger/reserved-bucket.json"; + let quota_ledger = serde_json::to_vec(&serde_json::json!({ + "version": 1, + "bucket_incarnation": "00000000-0000-0000-0000-000000000001", + "quota_revision_unix_nanos": 1, + "accounted_usage": 100, + "reservations": { + "00000000-0000-0000-0000-000000000002": { + "object": "pending-object", + "old_size": 0, + "new_size": 64, + "created_at": 1, + "pool_index": 0, + "set_index": 0, + "commit_started": true + } + } + })) + .expect("quota ledger fixture should encode"); + save_config(store.clone(), quota_ledger_path, quota_ledger.clone()) + .await + .expect("independent quota reservations should persist"); let cycle = CurrentCycle { current: 41, next: 42, @@ -4816,6 +4838,14 @@ async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() { assert!(!data_usage_info_has_persisted_baseline_identity(&usage)); assert_eq!(usage.scanner_epoch, Some(9)); + assert_eq!( + read_config(store.clone(), quota_ledger_path) + .await + .expect("quota ledger must remain readable after scanner reset"), + quota_ledger, + "scanner reset must preserve incarnation and outstanding reserved bytes exactly" + ); + for path in [ usage_backup_path.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 0d5e82eeb..0b049bb61 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -312,6 +312,30 @@ fn scanner_bucket_plan_digest(buckets: &[BucketInfo], activity_digest: [u8; 32]) DataUsageScanPlanDigest(hasher.finalize().into()) } +fn scanner_bucket_inventory_is_complete( + all_buckets: &[BucketInfo], + buckets_by_source: &HashMap>, +) -> bool { + let inventory = all_buckets + .iter() + .map(|bucket| (bucket.name.as_str(), bucket.created)) + .collect::>(); + if inventory.len() != all_buckets.len() || inventory.keys().any(|name| name.is_empty() || *name == DATA_USAGE_ROOT) { + return false; + } + let mut covered = HashSet::with_capacity(inventory.len()); + for buckets in buckets_by_source.values() { + let mut set_names = HashSet::with_capacity(buckets.len()); + for bucket in buckets { + if !set_names.insert(bucket.name.as_str()) || inventory.get(bucket.name.as_str()) != Some(&bucket.created) { + return false; + } + covered.insert(bucket.name.as_str()); + } + } + covered.len() == inventory.len() +} + fn scanner_bucket_cache_digest( scan_plan_digest: DataUsageScanPlanDigest, dirty_generation: Option, diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 7ff684c99..3d1efe1fb 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -213,10 +213,85 @@ pub(super) fn cache_snapshot_is_current( ) } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct ScannerSnapshotIdentity { + pub(super) cycle: u64, + pub(super) leader_epoch: u64, + pub(super) plan_digest: DataUsageScanPlanDigest, + pub(super) tier_registry_generation: Option, +} + +pub(super) struct ScannerSnapshotScope<'a> { + pub(super) sources: &'a HashSet, + pub(super) buckets: &'a [String], + pub(super) identity: ScannerSnapshotIdentity, +} + +#[derive(Debug, PartialEq, Eq, thiserror::Error)] +pub(super) enum ScannerSnapshotValidationError { + #[error("scanner snapshot does not cover the expected complete sets")] + IncompleteSets, + #[error("scanner snapshot does not match the requested generation")] + GenerationMismatch, + #[error("scanner snapshot bucket inventory is invalid")] + InvalidInventory, + #[error("scanner snapshot root is incomplete or corrupt")] + InvalidRoot, +} + +struct ValidatedScannerSnapshot<'a> { + results: &'a [DataUsageCache], + last_update: SystemTime, +} + +impl<'a> ValidatedScannerSnapshot<'a> { + fn validate( + results: &'a [DataUsageCache], + scope: &ScannerSnapshotScope<'_>, + ) -> std::result::Result { + if !scanner_results_form_complete_snapshot(results, scope.sources) { + return Err(ScannerSnapshotValidationError::IncompleteSets); + } + let bucket_keys = scope + .buckets + .iter() + .map(|bucket| crate::hash_path(bucket).key()) + .collect::>(); + if bucket_keys.len() != scope.buckets.len() + || scope + .buckets + .iter() + .any(|bucket| bucket.is_empty() || bucket == DATA_USAGE_ROOT) + { + return Err(ScannerSnapshotValidationError::InvalidInventory); + } + for result in results { + if result.info.next_cycle != scope.identity.cycle + || result.info.leader_epoch != scope.identity.leader_epoch + || result.info.scan_plan_digest != Some(scope.identity.plan_digest) + || result.info.tier_registry_generation != scope.identity.tier_registry_generation + { + return Err(ScannerSnapshotValidationError::GenerationMismatch); + } + if result.info.name != DATA_USAGE_ROOT + || result.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT + || !result.has_complete_root_inventory(&bucket_keys) + { + return Err(ScannerSnapshotValidationError::InvalidRoot); + } + } + let last_update = results + .iter() + .filter_map(|result| result.info.last_update) + .max() + .ok_or(ScannerSnapshotValidationError::IncompleteSets)?; + Ok(Self { results, last_update }) + } +} + pub(super) fn completed_data_usage_info( results: &[DataUsageCache], - expected_sources: &HashSet, - all_buckets: &[String], + scope: &ScannerSnapshotScope<'_>, tier_registry_names: &[String], bucket_plan_complete: bool, budget_elapsed: bool, @@ -229,26 +304,10 @@ pub(super) fn completed_data_usage_info( if !should_publish_completed_snapshot(completed_set_count, results.len(), budget_elapsed, cancelled) { return None; } - if !scanner_results_form_complete_snapshot(results, expected_sources) { - return None; - } - - // A generation is comparable across nodes because it is derived from the - // frozen registry names. Cycle and leader fencing remain separate cache - // metadata. Legacy peers omit the generation; an all-legacy result remains - // readable, but mixing legacy and new (or two new generations) would make - // the per-tier accounting ambiguous. - let registry_generation = results.first()?.info.tier_registry_generation; - if results.iter().any(|result| match registry_generation { - Some(generation) => result.info.tier_registry_generation != Some(generation), - None => result.info.tier_registry_generation.is_some(), - }) { - return None; - } - - if results.iter().any(|result| result.root().is_none()) { - return None; - } + let validated = ValidatedScannerSnapshot::validate(results, scope).ok()?; + let results = validated.results; + let all_buckets = scope.buckets; + let registry_generation = scope.identity.tier_registry_generation; let mut total = DataUsageEntry::default(); let mut bucket_entries = HashMap::with_capacity(all_buckets.len()); @@ -273,7 +332,7 @@ pub(super) fn completed_data_usage_info( return None; } - let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?; + let merged_last_update = validated.last_update; let buckets_usage = bucket_entries .iter() .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?))) @@ -300,8 +359,8 @@ pub(super) fn completed_data_usage_info( usage_snapshot_set_states.sort_by_key(|state| (state.pool_index, state.set_index)); let data_usage_info = DataUsageInfo { last_update: Some(merged_last_update), - scanner_cycle: Some(results.first()?.info.next_cycle), - scanner_epoch: Some(results.first()?.info.leader_epoch), + scanner_cycle: Some(scope.identity.cycle), + scanner_epoch: Some(scope.identity.leader_epoch), objects_total_count: u64::try_from(total.objects).ok()?, versions_total_count: u64::try_from(total.versions).ok()?, delete_markers_total_count: u64::try_from(total.delete_markers).ok()?, diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index a8686a1e4..e3066e134 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -39,6 +39,12 @@ pub(super) fn prepare_scoped_set_scan( else { return None; }; + // The existing cache does not bind each bucket to a durable incarnation. + // Listing creation times can come from volume metadata, so even Some(time) + // cannot prove that an unselected same-name bucket is the cached bucket. + if all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name)) { + return None; + } if selected_buckets.is_empty() || !old_cache.info.snapshot_complete || old_cache.info.last_update.is_none() @@ -49,7 +55,7 @@ pub(super) fn prepare_scoped_set_scan( || 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() + || !old_cache.has_complete_root_inventory(&old_cache.find(DATA_USAGE_ROOT)?.children) { return None; } @@ -74,21 +80,12 @@ pub(super) fn prepare_scoped_set_scan( 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)?; + cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); } Some(PreparedScopedSetScan { diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 2947d638d..95b4ed1bb 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -260,6 +260,7 @@ where } } bucket_plan_complete &= buckets_by_source.keys().copied().collect::>() == *expected_sources; + bucket_plan_complete &= scanner_bucket_inventory_is_complete(&all_buckets, &buckets_by_source); let scan_plan_digest = scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before)); let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); @@ -536,8 +537,16 @@ where let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::>(); let completed_usage = completed_data_usage_info( &results, - &expected_sources, - &all_bucket_names, + &ScannerSnapshotScope { + sources: &expected_sources, + buckets: &all_bucket_names, + identity: ScannerSnapshotIdentity { + cycle: want_cycle, + leader_epoch, + plan_digest: scan_plan_digest, + tier_registry_generation: Some(tier_registry_generation), + }, + }, &tier_registry.names, bucket_plan_complete, budget_elapsed, diff --git a/crates/scanner/src/scanner_io/publish_gate_tests.rs b/crates/scanner/src/scanner_io/publish_gate_tests.rs index f131cf437..ac65670ab 100644 --- a/crates/scanner/src/scanner_io/publish_gate_tests.rs +++ b/crates/scanner/src/scanner_io/publish_gate_tests.rs @@ -18,6 +18,32 @@ use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccount const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]); +#[test] +fn scanner_bucket_inventory_requires_exact_unique_set_union() { + let first = BucketInfo { + name: "first".to_string(), + ..Default::default() + }; + let second = BucketInfo { + name: "second".to_string(), + ..Default::default() + }; + let source = DataUsageCacheSource::new(0, 0); + let mut sets = HashMap::from([(source, vec![first.clone()])]); + assert!(scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets)); + assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), second.clone()], &sets)); + assert!(!scanner_bucket_inventory_is_complete(&[], &sets)); + assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), first.clone()], &sets)); + sets.insert(source, vec![first.clone(), first.clone()]); + assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets)); + sets.insert(source, vec![second]); + assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets)); + let mut recreated = first.clone(); + recreated.created = Some(OffsetDateTime::UNIX_EPOCH); + sets.insert(source, vec![recreated]); + assert!(!scanner_bucket_inventory_is_complete(&[first], &sets)); +} + #[test] fn should_publish_completed_snapshot_requires_full_clean_cycle() { assert!(should_publish_completed_snapshot(3, 3, false, false)); @@ -108,7 +134,133 @@ fn completed_data_usage_info_for_test( cancelled: bool, ) -> Option<(DataUsageInfo, SystemTime)> { let expected_sources = results.iter().filter_map(|result| result.info.source).collect::>(); - completed_data_usage_info(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled) + completed_usage_for_scope(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled) +} + +fn completed_usage_for_scope( + results: &[DataUsageCache], + expected_sources: &HashSet, + all_buckets: &[String], + tier_registry_names: &[String], + bucket_plan_complete: bool, + budget_elapsed: bool, + cancelled: bool, +) -> Option<(DataUsageInfo, SystemTime)> { + let first = results.first()?; + completed_data_usage_info( + results, + &ScannerSnapshotScope { + sources: expected_sources, + buckets: all_buckets, + identity: ScannerSnapshotIdentity { + cycle: first.info.next_cycle, + leader_epoch: first.info.leader_epoch, + plan_digest: TEST_PLAN_DIGEST, + tier_registry_generation: first.info.tier_registry_generation, + }, + }, + tier_registry_names, + bucket_plan_complete, + budget_elapsed, + cancelled, + ) +} + +#[test] +fn completed_data_usage_info_rejects_duplicate_bucket_inventory() { + let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0)); + let buckets = vec!["bucket".to_string(), "bucket".to_string()]; + assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none()); +} + +#[test] +fn completed_data_usage_info_rejects_extra_or_detached_bucket_data() { + let buckets = vec!["bucket".to_string()]; + let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0)); + set.replace( + "unlisted", + DATA_USAGE_ROOT, + DataUsageEntry { + objects: 1, + ..Default::default() + }, + ); + assert!(completed_data_usage_info_for_test(&[set.clone()], &buckets, false, false).is_none()); + set.cache + .get_mut(DATA_USAGE_ROOT) + .expect("set root") + .children + .remove(&hash_path("unlisted").key()); + assert!( + completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none(), + "orphaned data must not disappear from authoritative accounting" + ); +} + +#[test] +fn completed_data_usage_info_rejects_disconnected_expected_bucket() { + let buckets = vec!["bucket".to_string()]; + let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0)); + set.cache.get_mut(DATA_USAGE_ROOT).expect("set root").children.clear(); + assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none()); +} + +#[test] +fn completed_data_usage_info_rejects_root_scalar_data_and_unknown_key_format() { + let buckets = vec!["bucket".to_string()]; + let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0)); + let mut scalar_root = set.clone(); + scalar_root.cache.get_mut(DATA_USAGE_ROOT).expect("set root").size = 10; + assert!(completed_data_usage_info_for_test(&[scalar_root], &buckets, false, false).is_none()); + let mut future_format = set; + future_format.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT + 1; + assert!(completed_data_usage_info_for_test(&[future_format], &buckets, false, false).is_none()); +} + +#[test] +fn completed_data_usage_info_binds_all_results_to_requested_identity() { + let buckets = vec!["bucket".to_string()]; + let source = DataUsageCacheSource::new(0, 0); + let sources = HashSet::from([source]); + let set = completed_root_cache("bucket", 2, 10, source); + let identity = ScannerSnapshotIdentity { + cycle: 0, + leader_epoch: 0, + plan_digest: TEST_PLAN_DIGEST, + tier_registry_generation: None, + }; + let results = [set]; + for expected in [ + ScannerSnapshotIdentity { cycle: 1, ..identity }, + ScannerSnapshotIdentity { + leader_epoch: 1, + ..identity + }, + ScannerSnapshotIdentity { + plan_digest: DataUsageScanPlanDigest([9; 32]), + ..identity + }, + ScannerSnapshotIdentity { + tier_registry_generation: Some(1), + ..identity + }, + ] { + let scope = ScannerSnapshotScope { + sources: &sources, + buckets: &buckets, + identity: expected, + }; + assert!(completed_data_usage_info(&results, &scope, &[], true, false, false).is_none()); + } + let scope = ScannerSnapshotScope { + sources: &sources, + buckets: &buckets, + identity, + }; + let (usage, _) = completed_data_usage_info(&results, &scope, &[], true, false, false) + .expect("the requested complete scope remains publishable"); + assert_eq!(usage.objects_total_count, 2); + assert!(usage.is_complete_bucket_usage_snapshot()); } fn lkg_root_cache(bucket: &str, objects: usize, source: DataUsageCacheSource) -> DataUsageCache { @@ -136,7 +288,7 @@ fn partial_usage_is_observational_not_authoritative_for_quota() { let expected = HashSet::from([current_source, stalled_source]); assert!( - completed_data_usage_info(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false) + completed_usage_for_scope(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false) .is_none() ); let (observed, _) = observational_data_usage_info(&[current, stalled], &expected, &all_buckets, &[], TEST_PLAN_DIGEST, 8, 3) @@ -509,7 +661,7 @@ fn completed_data_usage_info_accepts_unknown_only_with_current_registry_generati let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]); assert!( - completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some() + completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some() ); } @@ -547,7 +699,7 @@ fn completed_data_usage_info_rejects_non_registry_tier_in_current_generation() { let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]); assert!( - completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none() + completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none() ); } @@ -698,6 +850,7 @@ fn completed_data_usage_info_publishes_confirmed_empty_namespace() { source: Some(DataUsageCacheSource::new(0, 0)), snapshot_complete: true, scan_plan_digest: Some(TEST_PLAN_DIGEST), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, ..Default::default() }, ..Default::default() @@ -855,7 +1008,7 @@ fn completed_data_usage_info_requires_exact_topology_sources() { let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(1, 0)]); assert!( - completed_data_usage_info(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false) + completed_usage_for_scope(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false) .is_none() ); } @@ -866,7 +1019,7 @@ fn completed_data_usage_info_rejects_incomplete_bucket_plan() { let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0)); let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]); - assert!(completed_data_usage_info(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none()); + assert!(completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none()); } #[test] diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index ec2c1ad65..500f8493d 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -986,8 +986,15 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() } } +fn bucket_info_with_created_time(name: &str) -> BucketInfo { + BucketInfo { + created: Some(time::OffsetDateTime::UNIX_EPOCH), + ..bucket_info(name) + } +} + #[test] -fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() { +fn scoped_set_scan_rebuilds_selected_buckets_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); @@ -1000,8 +1007,11 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() { ..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 all_buckets = vec![ + bucket_info_with_created_time("stable"), + bucket_info_with_created_time("dirty"), + ]; + let selected_buckets = Arc::new(HashSet::from(["stable".to_string(), "dirty".to_string(), "deleted".to_string()])); let prepared = prepare_scoped_set_scan( &old_cache, @@ -1021,12 +1031,16 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() { ) .expect("complete matching set cache should support a scoped scan"); - assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::>(), ["dirty"]); + assert_eq!( + prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::>(), + ["stable", "dirty"] + ); let stable = prepared .cache .checked_flatten("stable") - .expect("unselected bucket subtree should be retained"); - assert_eq!((stable.size, stable.objects), (15, 2)); + .expect("selected bucket placeholder should exist"); + assert_eq!((stable.size, stable.objects), (0, 0)); + assert!(prepared.cache.find("stable/prefix").is_none()); 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)); @@ -1037,11 +1051,41 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() { assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest)); } +#[test] +fn scoped_set_scan_rejects_unbound_bucket_incarnations() { + let baseline_digest = DataUsageScanPlanDigest([1; 32]); + let old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest); + 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([2; 32]), + }; + for created in [ + None, + Some(OffsetDateTime::UNIX_EPOCH), + Some(OffsetDateTime::UNIX_EPOCH + time::Duration::days(1)), + ] { + let mut stable = bucket_info("stable"); + stable.created = created; + let buckets = vec![stable, bucket_info_with_created_time("dirty")]; + assert!( + prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation).is_none(), + "missing identity, volume timestamps and same-name recreation must all rebuild" + ); + } +} + #[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")]; + let all_buckets = vec![bucket_info_with_created_time("stable"), bucket_info_with_created_time("new")]; assert!( prepare_scoped_set_scan( @@ -1067,7 +1111,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() { #[test] fn scoped_set_scan_requires_an_exact_complete_baseline() { let baseline_digest = DataUsageScanPlanDigest([5; 32]); - let all_buckets = vec![bucket_info("dirty")]; + let all_buckets = vec![bucket_info_with_created_time("dirty")]; let scope = ScannerBucketScanScope { selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), baseline_scan_plan_digest: Some(baseline_digest), @@ -1088,6 +1132,10 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() { not_durable.info.last_update = None; assert!(prepare_scoped_set_scan(¬_durable, &all_buckets, &all_buckets, &scope, generation).is_none()); + let mut unscoped_usage = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); + unscoped_usage.cache.get_mut(DATA_USAGE_ROOT).expect("set root").objects = 1; + assert!(prepare_scoped_set_scan(&unscoped_usage, &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()); @@ -1098,6 +1146,13 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() { }; 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()); + assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation).is_some()); + + let unidentified_buckets = vec![bucket_info("dirty")]; + assert!( + prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation).is_some(), + "fully selected buckets are rebuilt without reusing an unproven incarnation" + ); let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest); future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);