From 5eecd657e706e81c21c23cca0a6e0753152b49cd Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 9 Sep 2026 14:20:16 +0800 Subject: [PATCH] fix(scanner): bind cold scoped cache reuse (#7565) Record an explicit cold bucket reuse proof for scoped set scans and treat fully cold sets as valid cold-zero-walk reuse when the baseline and bucket incarnations are bound. Co-authored-by: zhi22915 --- crates/scanner/src/scanner_io/io_cache.rs | 72 +++++++++++++++------ crates/scanner/src/scanner_io/tests.rs | 77 +++++++++++++++++++++++ 2 files changed, 131 insertions(+), 18 deletions(-) diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index e457a70ea..d1a8ea028 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -26,6 +26,21 @@ pub(super) struct ScannerSetCacheGeneration { pub(super) struct PreparedScopedSetScan { pub(super) buckets: Vec, pub(super) cache: DataUsageCache, + pub(super) cold_bucket_reuse_proof: Option, +} + +pub(super) struct ScopedColdBucketReuseProof { + pub(super) baseline_scan_plan_digest: DataUsageScanPlanDigest, + pub(super) source: DataUsageCacheSource, + pub(super) bucket_incarnations: HashMap, +} + +impl ScopedColdBucketReuseProof { + fn authorizes(&self, source: DataUsageCacheSource, baseline_scan_plan_digest: DataUsageScanPlanDigest) -> bool { + self.source == source + && self.baseline_scan_plan_digest == baseline_scan_plan_digest + && !self.bucket_incarnations.is_empty() + } } pub(super) fn prepare_scoped_set_scan( @@ -51,10 +66,16 @@ 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; } + let unselected_bucket_incarnations = + unselected_bucket_incarnation_bindings(old_cache, all_buckets, selected_buckets, current_bucket_incarnations)?; + let cold_bucket_reuse_proof = (!unselected_bucket_incarnations.is_empty()).then(|| ScopedColdBucketReuseProof { + baseline_scan_plan_digest, + source: generation.source, + bucket_incarnations: unselected_bucket_incarnations, + }); let mut cache = DataUsageCache { info: DataUsageCacheInfo { @@ -100,35 +121,41 @@ pub(super) fn prepare_scoped_set_scan( .cloned() .collect(), cache, + cold_bucket_reuse_proof, }) } -fn unselected_bucket_incarnations_match( +fn unselected_bucket_incarnation_bindings( 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 +) -> Option> { + let mut unselected_buckets = 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 + .peekable(); + let Some(current_bucket_incarnations) = current_bucket_incarnations else { + return unselected_buckets.peek().is_none().then(HashMap::new); + }; + let mut bound_incarnations = HashMap::new(); + for bucket in unselected_buckets { + let current = current_bucket_incarnations + .get(&bucket.name) + .filter(|incarnation| !incarnation.is_nil())?; + if old_cache.find(&bucket.name).is_none() + || old_cache .info .scan_bucket_incarnations .get(&bucket.name) .filter(|cached| !cached.is_nil()) - == Some(current) - }) + != Some(current) + { + return None; + } + bound_incarnations.insert(bucket.name.clone(), *current); + } + Some(bound_incarnations) } async fn scanner_current_bucket_incarnations(set: &SetDisks, all_buckets: &[BucketInfo]) -> Option> { @@ -222,7 +249,13 @@ impl ScannerIOCache for SetDisks { current_bucket_incarnations.as_ref(), ); let cold_zero_walk_reuse_candidate = scoped_scan.as_ref().is_some_and(|prepared| { - old_cache.info.next_cycle < want_cycle && !prepared.buckets.is_empty() && prepared.buckets.len() < all_buckets.len() + old_cache.info.next_cycle < want_cycle + && scope.baseline_scan_plan_digest.is_some_and(|baseline_scan_plan_digest| { + prepared + .cold_bucket_reuse_proof + .as_ref() + .is_some_and(|proof| proof.authorizes(source, baseline_scan_plan_digest)) + }) }); let mut scoped_cache = scoped_scan.map(|mut prepared| { buckets = prepared.buckets; @@ -268,6 +301,9 @@ impl ScannerIOCache for SetDisks { cache.info.lkg_last_update = None; cache.info.lkg_leader_epoch = None; cache.info.lkg_scan_plan_digest = None; + if cold_zero_walk_reuse_candidate { + cold_zero_walk_reuse_observed.store(true, Ordering::Release); + } if cache.find(DATA_USAGE_ROOT).is_none() { cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default()); } diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 5ab52886e..c2d9ac752 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -2677,6 +2677,58 @@ fn scoped_set_scan_reuses_unselected_buckets_with_matching_incarnations() { 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); + let proof = prepared + .cold_bucket_reuse_proof + .as_ref() + .expect("cold bucket reuse should be explicitly bound"); + assert_eq!(proof.baseline_scan_plan_digest, baseline_digest); + assert_eq!(proof.source, DataUsageCacheSource::new(1, 2)); + assert_eq!(proof.bucket_incarnations, HashMap::from([("stable".to_string(), Uuid::from_u128(1))])); +} + +#[test] +fn scoped_set_scan_reuses_all_cold_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), ("archive", 20)], baseline_digest); + old_cache.info.scan_bucket_incarnations = test_bucket_incarnations(&["stable", "archive"]); + 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("archive"), + ]; + + let prepared = prepare_scoped_set_scan( + &old_cache, + &all_buckets, + &all_buckets, + &ScannerBucketScanScope { + selected_buckets: Some(Arc::new(HashSet::from(["dirty-on-another-set".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("a set with only cold buckets should reuse the complete bound baseline"); + + assert!(prepared.buckets.is_empty()); + assert_eq!(prepared.cache.find("stable").map(|entry| entry.size), Some(10)); + assert_eq!(prepared.cache.find("archive").map(|entry| entry.size), Some(20)); + assert_eq!( + prepared + .cold_bucket_reuse_proof + .as_ref() + .expect("all cold bucket reuse should carry incarnation proof") + .bucket_incarnations, + current_incarnations + ); } #[test] @@ -2749,6 +2801,31 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() { ) .is_none() ); + + let mut missing_entry = complete_set_usage_cache(&[("stable", 10)], baseline_digest); + missing_entry.info.scan_bucket_incarnations = test_bucket_incarnations(&["stable", "new"]); + assert!( + prepare_scoped_set_scan( + &missing_entry, + &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: DataUsageScanPlanDigest([4; 32]), + }, + Some(&missing_entry.info.scan_bucket_incarnations), + ) + .is_none(), + "an incarnation without a durable bucket entry must not authorize a cold skip" + ); } #[test]