diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index d969452da..52193e834 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -536,6 +536,10 @@ pub struct DataUsageEntryInfo { pub name: String, pub parent: String, pub entry: DataUsageEntry, + /// Durable bucket incarnation that produced this bucket root. Missing + /// values are legacy/unproven and must not authorize cold-bucket reuse. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bucket_incarnation: Option, /// Registry generation used to classify this root entry. Older remote /// workers omit it; callers must reject that result when a frozen cycle /// requires generation fencing. @@ -653,6 +657,11 @@ pub struct DataUsageCacheInfo { /// structural plan remains reusable across ordinary bucket writes. #[serde(default)] pub scan_execution_digest: Option, + /// Durable bucket incarnations captured for a complete set aggregate. + /// Missing or nil entries are legacy/unproven and cannot authorize + /// skipping an unselected bucket in a later scoped set scan. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub scan_bucket_incarnations: HashMap, } impl Serialize for DataUsageCacheInfo { @@ -676,7 +685,8 @@ impl Serialize for DataUsageCacheInfo { + usize::from(self.lkg_last_update.is_some()) + usize::from(self.lkg_leader_epoch.is_some()) + usize::from(self.lkg_scan_plan_digest.is_some()) - + usize::from(self.scan_execution_digest.is_some()); + + usize::from(self.scan_execution_digest.is_some()) + + usize::from(!self.scan_bucket_incarnations.is_empty()); let mut state = serializer.serialize_map(Some(field_count))?; state.serialize_entry("name", &self.name)?; state.serialize_entry("next_cycle", &self.next_cycle)?; @@ -736,6 +746,9 @@ impl Serialize for DataUsageCacheInfo { if let Some(scan_execution_digest) = self.scan_execution_digest { state.serialize_entry("scan_execution_digest", &scan_execution_digest)?; } + if !self.scan_bucket_incarnations.is_empty() { + state.serialize_entry("scan_bucket_incarnations", &self.scan_bucket_incarnations)?; + } state.end() } } diff --git a/crates/scanner/src/remote_scanner/stream/tests.rs b/crates/scanner/src/remote_scanner/stream/tests.rs index ba0b71e58..c2fb3c372 100644 --- a/crates/scanner/src/remote_scanner/stream/tests.rs +++ b/crates/scanner/src/remote_scanner/stream/tests.rs @@ -195,6 +195,7 @@ fn test_usage(bucket: &str, objects: usize) -> DataUsageEntryInfo { name: bucket.to_string(), parent: crate::DATA_USAGE_ROOT.to_string(), entry, + bucket_incarnation: Some(Uuid::from_u128(7)), tier_registry_generation: Some(0), } } diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index af6b4ac58..0add00423 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -741,6 +741,11 @@ pub(crate) fn cache_root_entry_info(cache: &DataUsageCache) -> std::result::Resu name: cache.info.name.clone(), parent: DATA_USAGE_ROOT.to_string(), entry, + bucket_incarnation: cache + .info + .scan_identity + .map(|identity| identity.bucket_incarnation) + .filter(|incarnation| !incarnation.is_nil()), tier_registry_generation: cache.info.tier_registry_generation, }) } @@ -752,6 +757,14 @@ fn apply_bucket_result_to_cache(cache: &mut DataUsageCache, result: DataUsageEnt // forces the caller to re-account it under one frozen registry. return false; } + match result.bucket_incarnation { + Some(incarnation) if !incarnation.is_nil() => { + cache.info.scan_bucket_incarnations.insert(result.name.clone(), incarnation); + } + _ => { + cache.info.scan_bucket_incarnations.remove(&result.name); + } + } cache.replace(&result.name, &result.parent, result.entry); cache.info.last_update = Some(update_time); true diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index 4cd995ebf..2cb970d8e 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -34,17 +34,12 @@ pub(super) fn prepare_scoped_set_scan( all_buckets: &[BucketInfo], scope: &ScannerBucketScanScope, generation: ScannerSetCacheGeneration, + current_bucket_incarnations: Option<&HashMap>, ) -> Option { let (Some(selected_buckets), Some(baseline_scan_plan_digest)) = (&scope.selected_buckets, scope.baseline_scan_plan_digest) 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() @@ -56,6 +51,7 @@ pub(super) fn prepare_scoped_set_scan( || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest) || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT || !old_cache.has_complete_root_inventory(&old_cache.find(DATA_USAGE_ROOT)?.children) + || !unselected_bucket_incarnations_match(old_cache, all_buckets, selected_buckets, current_bucket_incarnations) { return None; } @@ -75,6 +71,7 @@ pub(super) fn prepare_scoped_set_scan( 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, + scan_bucket_incarnations: old_cache.info.scan_bucket_incarnations.clone(), ..Default::default() }, cache: HashMap::new(), @@ -85,7 +82,15 @@ pub(super) fn prepare_scoped_set_scan( if !current_bucket_names.insert(bucket.name.as_str()) { return None; } - cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); + if selected_buckets.contains(&bucket.name) { + cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default()); + } else { + cache.copy_with_children( + old_cache, + &rustfs_data_usage::hash_path(&bucket.name), + &Some(rustfs_data_usage::hash_path(DATA_USAGE_ROOT)), + ); + } } Some(PreparedScopedSetScan { @@ -98,6 +103,47 @@ pub(super) fn prepare_scoped_set_scan( }) } +fn unselected_bucket_incarnations_match( + old_cache: &DataUsageCache, + all_buckets: &[BucketInfo], + selected_buckets: &HashSet, + current_bucket_incarnations: Option<&HashMap>, +) -> bool { + let Some(current_bucket_incarnations) = current_bucket_incarnations else { + return all_buckets.iter().all(|bucket| selected_buckets.contains(&bucket.name)); + }; + all_buckets + .iter() + .filter(|bucket| !selected_buckets.contains(&bucket.name)) + .all(|bucket| { + let Some(current) = current_bucket_incarnations + .get(&bucket.name) + .filter(|incarnation| !incarnation.is_nil()) + else { + return false; + }; + old_cache + .info + .scan_bucket_incarnations + .get(&bucket.name) + .filter(|cached| !cached.is_nil()) + == Some(current) + }) +} + +async fn scanner_current_bucket_incarnations(set: &SetDisks, all_buckets: &[BucketInfo]) -> Option> { + let mut incarnations = HashMap::with_capacity(all_buckets.len()); + for bucket in all_buckets { + let Ok(incarnation) = set.bucket_incarnation_id_from_disk(&bucket.name).await else { + return None; + }; + if incarnation.is_nil() || incarnations.insert(bucket.name.clone(), incarnation).is_some() { + return None; + } + } + Some(incarnations) +} + #[async_trait::async_trait] impl ScannerIOCache for SetDisks { #[tracing::instrument(skip(self, budget, scan_plan, updates))] @@ -158,6 +204,7 @@ impl ScannerIOCache for SetDisks { None } }; + let current_bucket_incarnations = scanner_current_bucket_incarnations(self.as_ref(), &all_buckets).await; let scoped_scan = prepare_scoped_set_scan( &old_cache, &buckets, @@ -170,6 +217,7 @@ impl ScannerIOCache for SetDisks { source, scan_plan_digest, }, + current_bucket_incarnations.as_ref(), ); let mut scoped_cache = scoped_scan.map(|mut prepared| { buckets = prepared.buckets; @@ -191,6 +239,7 @@ impl ScannerIOCache for SetDisks { scan_plan_digest: Some(scan_plan_digest), scan_coverage_digest: Some(bucket_coverage_digest), cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(), ..Default::default() }, cache: HashMap::new(), @@ -486,6 +535,7 @@ impl ScannerIOCache for SetDisks { 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, + scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(), ..Default::default() }, cache: HashMap::new(), diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 5189af0d0..1e01eebf9 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -1232,6 +1232,19 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa cache } +fn test_bucket_incarnations(buckets: &[&str]) -> HashMap { + buckets + .iter() + .enumerate() + .map(|(index, bucket)| { + ( + (*bucket).to_string(), + Uuid::from_u128(u128::try_from(index).expect("test index should fit") + 1), + ) + }) + .collect() +} + #[tokio::test] #[serial] async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers() { @@ -2082,6 +2095,7 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() { source: DataUsageCacheSource::new(1, 2), scan_plan_digest: current_digest, }, + None, ) .expect("complete matching set cache should support a scoped scan"); @@ -2105,6 +2119,57 @@ fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() { assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest)); } +#[test] +fn scoped_set_scan_reuses_unselected_buckets_with_matching_incarnations() { + let baseline_digest = DataUsageScanPlanDigest([1; 32]); + let current_digest = DataUsageScanPlanDigest([2; 32]); + let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest); + old_cache.replace( + "stable/prefix", + "stable", + DataUsageEntry { + size: 5, + objects: 1, + ..Default::default() + }, + ); + old_cache.info.scan_bucket_incarnations = test_bucket_incarnations(&["stable", "dirty"]); + let current_incarnations = old_cache.info.scan_bucket_incarnations.clone(); + let all_buckets = vec![ + bucket_info_with_created_time("stable"), + bucket_info_with_created_time("dirty"), + ]; + + let prepared = prepare_scoped_set_scan( + &old_cache, + &all_buckets, + &all_buckets, + &ScannerBucketScanScope { + selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))), + selected_bucket_prefixes: None, + 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, + }, + Some(¤t_incarnations), + ) + .expect("matching bucket incarnations should authorize cold bucket reuse"); + + assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::>(), ["dirty"]); + let stable = prepared + .cache + .checked_flatten("stable") + .expect("unselected stable bucket should be copied with children"); + assert_eq!((stable.size, stable.objects), (15, 2)); + assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0))); + assert_eq!(prepared.cache.info.scan_bucket_incarnations, current_incarnations); +} + #[test] fn scoped_set_scan_rejects_unbound_bucket_incarnations() { let baseline_digest = DataUsageScanPlanDigest([1; 32]); @@ -2130,10 +2195,22 @@ fn scoped_set_scan_rejects_unbound_bucket_incarnations() { 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(), + prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation, None).is_none(), "missing identity, volume timestamps and same-name recreation must all rebuild" ); } + let mut mismatched = test_bucket_incarnations(&["stable", "dirty"]); + mismatched.insert("stable".to_string(), Uuid::from_u128(99)); + let mut old_cache = old_cache; + old_cache.info.scan_bucket_incarnations = test_bucket_incarnations(&["stable", "dirty"]); + let buckets = vec![ + bucket_info_with_created_time("stable"), + bucket_info_with_created_time("dirty"), + ]; + assert!( + prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation, Some(&mismatched)).is_none(), + "a same-name unselected bucket with a different incarnation must rebuild" + ); } #[test] @@ -2159,6 +2236,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() { source: DataUsageCacheSource::new(1, 2), scan_plan_digest: DataUsageScanPlanDigest([4; 32]), }, + Some(&test_bucket_incarnations(&["stable", "new"])), ) .is_none() ); @@ -2183,19 +2261,19 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() { 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()); + assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation, None).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()); + assert!(prepare_scoped_set_scan(¬_durable, &all_buckets, &all_buckets, &scope, generation, None).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()); + assert!(prepare_scoped_set_scan(&unscoped_usage, &all_buckets, &all_buckets, &scope, generation, None).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()); + assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation, None).is_none()); let empty_scope = ScannerBucketScanScope { selected_buckets: Some(Arc::new(HashSet::new())), @@ -2203,18 +2281,18 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() { 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()); - assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation).is_some()); + assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation, None).is_none()); + assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation, None).is_some()); let unidentified_buckets = vec![bucket_info("dirty")]; assert!( - prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation).is_some(), + prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation, None).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); - assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation).is_none()); + assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation, None).is_none()); } #[test] @@ -3083,6 +3161,7 @@ fn apply_bucket_result_to_cache_updates_bucket_entry() { objects: 2, ..Default::default() }, + bucket_incarnation: Some(Uuid::from_u128(7)), tier_registry_generation: None, }, update_time, @@ -3092,6 +3171,7 @@ fn apply_bucket_result_to_cache_updates_bucket_entry() { let entry = cache.find("bucket").expect("bucket entry should remain present"); assert_eq!(entry.size, 10); assert_eq!(entry.objects, 2); + assert_eq!(cache.info.scan_bucket_incarnations.get("bucket"), Some(&Uuid::from_u128(7))); } #[test] @@ -3122,6 +3202,7 @@ fn apply_bucket_result_to_cache_rejects_a_different_tier_generation() { size: 11, ..Default::default() }, + bucket_incarnation: Some(Uuid::from_u128(7)), tier_registry_generation: Some(8), }, SystemTime::now(), diff --git a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs index 31de76153..c6bc58c3e 100644 --- a/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs +++ b/crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs @@ -82,8 +82,8 @@ async fn persist_baseline(store: &Arc, baseline: &DataUsageInfo) { .expect("fixture baseline should persist"); } -// Every invocation uses the production default scope. The expected walker set -// comes from storage's per-source inventory, not the resolver's selected names. +// Every invocation uses the production default scope. Once durable bucket +// incarnations are present, the expected walker set follows the resolved scope. async fn run_entry(store: &Arc, cycle: u64, selected: Option<&str>, expect_walks: bool) -> DataUsageInfo { let drives = drive_identities(store).await; let inventory = store @@ -99,6 +99,7 @@ async fn run_entry(store: &Arc, cycle: u64, selected: Option<&str>, exp let source = DataUsageCacheSource::new(set.pool_index, set.set_index); set.buckets.into_iter().map(move |bucket| ((source, bucket.name), 1_u64)) }) + .filter(|((_, bucket), _)| selected.is_none_or(|selected| bucket == selected)) .collect::>() } else { HashMap::new() @@ -201,8 +202,8 @@ async fn scoped_entry_fallback_distinguishes_planned_scope_from_real_cold_walks( let baseline = run_entry(&store, 1, None, true).await; persist_baseline(&store, &baseline).await; - // A same-intent, same-cycle Current cache is a retry, not proof that a - // later cycle may reuse unselected buckets without durable incarnation. + // Same-cycle Current remains a retry. The later cycle may skip the cold + // bucket only after the prior complete set cache has durable incarnations. run_entry(&store, 1, Some(&hot), false).await; let usage = run_entry(&store, 2, Some(&hot), true).await; assert_eq!(usage.buckets_usage[&hot].objects_count, 1);