From 73c9dd4c9dd143f00e3b1fc89ad3ed899a634d49 Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 29 Aug 2026 00:27:54 +0800 Subject: [PATCH] fix(data-usage): preserve cold buckets in partial admin usage (#6811) Merge newer partial observed usage into the complete authoritative admin baseline instead of replacing the full bucket set. Keep the merged view partial and non-converged so shared consumers do not treat it as quota-authoritative. Co-authored-by: heihutu --- crates/ecstore/src/data_usage/mod.rs | 136 +++++++++++++++++++++++++++ rustfs/src/app/admin_usecase.rs | 24 ++++- 2 files changed, 159 insertions(+), 1 deletion(-) diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 8488b1d22..4ec3371ac 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -1161,11 +1161,45 @@ fn select_admin_data_usage_snapshot( authoritative.usage_snapshot_converged = Some(true); } match observed { + Some(observed) + if observed.usage_snapshot_partial + && authoritative.is_complete_bucket_usage_snapshot() + && observed_data_usage_is_newer(&observed, &authoritative) => + { + (merge_partial_observation_for_admin(authoritative, observed), true) + } Some(observed) if observed_data_usage_is_newer(&observed, &authoritative) => (observed, true), _ => (authoritative, authoritative_format), } } +fn merge_partial_observation_for_admin(mut authoritative: DataUsageInfo, observed: DataUsageInfo) -> DataUsageInfo { + for (bucket, usage) in observed.buckets_usage { + authoritative.buckets_usage.insert(bucket, usage); + } + + authoritative.last_update = observed.last_update; + authoritative.scanner_cycle = observed.scanner_cycle; + authoritative.scanner_epoch = observed.scanner_epoch; + authoritative.usage_snapshot_complete = false; + authoritative.usage_snapshot_partial = true; + authoritative.usage_snapshot_converged = Some(false); + authoritative.usage_snapshot_authoritative_baseline = observed.usage_snapshot_authoritative_baseline; + authoritative.usage_snapshot_set_states = observed.usage_snapshot_set_states; + authoritative.usage_snapshot_bootstrap_pending = false; + authoritative.buckets_count = authoritative.buckets_usage.len() as u64; + authoritative.bucket_sizes = authoritative + .buckets_usage + .iter() + .map(|(bucket, usage)| (bucket.clone(), usage.size)) + .collect(); + authoritative.replication_info.clear(); + authoritative.tier_stats = None; + authoritative.unknown_tier_stats = None; + authoritative.calculate_totals(); + authoritative +} + async fn load_admin_data_usage_from_backend(store: Arc) -> Result { let (authoritative, source) = load_data_usage_snapshot(store.clone()).await?; let observed = load_observed_data_usage_snapshot(store).await; @@ -3217,6 +3251,108 @@ mod tests { assert_eq!(selected.buckets_usage.get("bucket").map(|usage| usage.size), Some(100)); } + #[test] + fn partial_admin_observation_preserves_authoritative_cold_buckets() { + let baseline_time = SystemTime::UNIX_EPOCH + Duration::from_secs(10); + let mut authoritative = data_usage_info_for_test("cold", 152_318, 80 * 1024 * 1024 * 1024, baseline_time); + authoritative.scanner_epoch = Some(4); + authoritative.scanner_cycle = Some(10); + authoritative.buckets_usage.insert( + "hot".to_string(), + BucketUsageInfo { + objects_count: 3_000, + versions_count: 3_000, + size: 400 * 1024 * 1024, + ..Default::default() + }, + ); + authoritative.buckets_count = 2; + authoritative.bucket_sizes = authoritative + .buckets_usage + .iter() + .map(|(bucket, usage)| (bucket.clone(), usage.size)) + .collect(); + authoritative.calculate_totals(); + authoritative.replication_info.insert( + "stale-target".to_string(), + BucketTargetUsageInfo { + replicated_size: 400 * 1024 * 1024, + replicated_count: 3_000, + ..Default::default() + }, + ); + authoritative.tier_stats = Some(rustfs_data_usage::AllTierStats { + tiers: HashMap::from([( + "WARM".to_string(), + rustfs_data_usage::TierStats { + total_size: 80 * 1024 * 1024 * 1024, + num_versions: 152_318, + num_objects: 152_318, + }, + )]), + }); + + let mut observed = DataUsageInfo { + last_update: Some(baseline_time + Duration::from_secs(1)), + scanner_epoch: Some(4), + scanner_cycle: Some(11), + usage_snapshot_complete: false, + usage_snapshot_partial: true, + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(authoritative.snapshot_identity()), + usage_snapshot_set_states: vec![rustfs_data_usage::DataUsageSnapshotSetState { + pool_index: 0, + set_index: 0, + scanner_cycle: Some(11), + scanner_epoch: Some(4), + scan_plan_digest: Some([1; 32]), + complete: true, + tombstone: false, + }], + ..Default::default() + }; + observed.buckets_usage.insert( + "hot".to_string(), + BucketUsageInfo { + objects_count: 34, + versions_count: 34, + size: 8 * 1024 * 1024, + ..Default::default() + }, + ); + observed.buckets_count = 1; + observed.bucket_sizes.insert("hot".to_string(), 8 * 1024 * 1024); + observed.calculate_totals(); + + let (selected, current_format) = select_admin_data_usage_snapshot(authoritative, true, Some(observed)); + + assert!(current_format); + assert!(!selected.usage_snapshot_complete); + assert!(selected.usage_snapshot_partial); + assert!(selected.is_valid_partial_snapshot()); + assert_eq!(selected.usage_snapshot_converged, Some(false)); + assert_eq!(selected.buckets_count, 2); + assert_eq!( + selected + .buckets_usage + .get("cold") + .map(|usage| (usage.objects_count, usage.size)), + Some((152_318, 80 * 1024 * 1024 * 1024)) + ); + assert_eq!( + selected + .buckets_usage + .get("hot") + .map(|usage| (usage.objects_count, usage.size)), + Some((34, 8 * 1024 * 1024)) + ); + assert_eq!(selected.objects_total_count, 152_352); + assert_eq!(selected.objects_total_size, 80 * 1024 * 1024 * 1024 + 8 * 1024 * 1024); + assert!(selected.replication_info.is_empty()); + assert!(selected.tier_stats.is_none()); + assert!(selected.unknown_tier_stats.is_none()); + } + #[tokio::test] async fn authoritative_save_cleanup_removes_observed_snapshot_best_effort() { let store = UsageCasStore::default(); diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 5660ec2d9..a4f6a1dc5 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -237,7 +237,7 @@ impl DefaultAdminUsecase { /// namespace still contains, carries no usable information and is dropped /// exactly as before. fn narrow_data_usage_snapshot_to_measured_buckets(info: &mut DataUsageInfo, buckets: impl IntoIterator) { - if !info.is_complete_bucket_usage_snapshot() { + if !info.is_complete_bucket_usage_snapshot() && !info.is_valid_partial_snapshot() { *info = DataUsageInfo::default(); return; } @@ -759,6 +759,28 @@ mod tests { DefaultAdminUsecase::narrow_data_usage_snapshot_to_measured_buckets(&mut info, ["bucket-a".to_string()]); assert_eq!(info, DataUsageInfo::default()); + // A structurally valid partial admin view can still carry useful + // conservative totals. + let mut info = measured("bucket-a"); + info.usage_snapshot_complete = false; + info.usage_snapshot_partial = true; + info.usage_snapshot_converged = Some(false); + info.scanner_cycle = Some(11); + info.scanner_epoch = Some(4); + info.usage_snapshot_set_states = vec![rustfs_data_usage::DataUsageSnapshotSetState { + pool_index: 0, + set_index: 0, + scanner_cycle: Some(11), + scanner_epoch: Some(4), + scan_plan_digest: Some([1; 32]), + complete: true, + tombstone: false, + }]; + DefaultAdminUsecase::narrow_data_usage_snapshot_to_measured_buckets(&mut info, ["bucket-a".to_string()]); + assert_eq!(info.usage_snapshot_converged, Some(false)); + assert_eq!(info.buckets_count, 1); + assert_eq!(info.objects_total_count, 7); + // An empty namespace with an empty snapshot stays a confirmed zero. let mut info = DataUsageInfo { last_update: Some(std::time::SystemTime::UNIX_EPOCH),