gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-10 06:05:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): bind cold scoped cache reuse (#7565)

Browse Source 
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 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-09 14:20:16 +08:00
committed by GitHub
parent 929d906131
commit 5eecd657e7
2 changed files with 131 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner_io/io_cache.rs
+54 -18
View File
@@ -26,6 +26,21 @@ pub(super) struct ScannerSetCacheGeneration {
pub(super) struct PreparedScopedSetScan {
pub(super) buckets: Vec<BucketInfo>,
pub(super) cache: DataUsageCache,
pub(super) cold_bucket_reuse_proof: Option<ScopedColdBucketReuseProof>,
}
pub(super) struct ScopedColdBucketReuseProof {
pub(super) baseline_scan_plan_digest: DataUsageScanPlanDigest,
pub(super) source: DataUsageCacheSource,
pub(super) bucket_incarnations: HashMap<String, uuid::Uuid>,
}
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<String>,
current_bucket_incarnations: Option<&HashMap<String, uuid::Uuid>>,
) -> bool {
let Some(current_bucket_incarnations) = current_bucket_incarnations else {
return all_buckets.iter().all(|bucket| selected_buckets.contains(&bucket.name));
};
all_buckets
) -> Option<HashMap<String, uuid::Uuid>> {
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<HashMap<String, uuid::Uuid>> {
@@ -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());
}
crates/scanner/src/scanner_io/tests.rs
+77
View File
@@ -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(&current_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]