From 62d1f80dbf71d6b25f331006af7843e04d3020cf Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 25 May 2026 22:24:00 +0800 Subject: [PATCH] fix(data-usage): refresh admin usage after object changes (#3081) Co-authored-by: Henry Guo --- crates/ecstore/src/data_usage.rs | 292 ++++++++++++++++++++++++++-- crates/scanner/src/scanner.rs | 2 + rustfs/src/app/admin_usecase.rs | 3 +- rustfs/src/app/multipart_usecase.rs | 17 +- rustfs/src/app/object_usecase.rs | 52 +++-- 5 files changed, 327 insertions(+), 39 deletions(-) diff --git a/crates/ecstore/src/data_usage.rs b/crates/ecstore/src/data_usage.rs index ed80acff8..f7f30fe53 100644 --- a/crates/ecstore/src/data_usage.rs +++ b/crates/ecstore/src/data_usage.rs @@ -48,7 +48,14 @@ const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin"; pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin"; const DATA_USAGE_CACHE_TTL_SECS: u64 = 30; -type UsageMemoryCache = Arc>>; +#[derive(Debug, Clone)] +struct CachedBucketUsage { + usage: BucketUsageInfo, + refreshed_at: SystemTime, + usage_updated_at: SystemTime, +} + +type UsageMemoryCache = Arc>>; type CacheUpdating = Arc>; static USAGE_MEMORY_CACHE: OnceLock = OnceLock::new(); @@ -403,21 +410,90 @@ pub async fn compute_bucket_usage(store: Arc, bucket_name: &str) -> Res Ok(usage) } -/// Fast in-memory increment for immediate quota consistency -pub async fn increment_bucket_usage_memory(bucket: &str, size_increment: u64) { +async fn ensure_bucket_usage_cached(bucket: &str) { + let cache = memory_cache().read().await; + if cache.contains_key(bucket) { + return; + } + drop(cache); + + update_usage_cache_if_needed().await; +} + +fn cached_bucket_usage_from_backend(usage: BucketUsageInfo, updated_at: SystemTime) -> CachedBucketUsage { + CachedBucketUsage { + usage, + refreshed_at: SystemTime::now(), + usage_updated_at: updated_at, + } +} + +fn cached_bucket_usage_now(usage: BucketUsageInfo) -> CachedBucketUsage { + let now = SystemTime::now(); + CachedBucketUsage { + usage, + refreshed_at: now, + usage_updated_at: now, + } +} + +fn data_usage_info_updated_at(data_usage_info: &DataUsageInfo) -> SystemTime { + data_usage_info.last_update.unwrap_or(SystemTime::UNIX_EPOCH) +} + +/// Fast in-memory update for immediate quota and admin usage consistency. +pub async fn record_bucket_object_write_memory(bucket: &str, previous_current_size: Option, new_size: u64) { + ensure_bucket_usage_cached(bucket).await; + let mut cache = memory_cache().write().await; - let current = cache.entry(bucket.to_string()).or_insert_with(|| (0, SystemTime::now())); - current.0 += size_increment; - current.1 = SystemTime::now(); + let entry = cache + .entry(bucket.to_string()) + .or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default())); + + match previous_current_size { + Some(previous_size) => { + entry.usage.size = entry.usage.size.saturating_sub(previous_size).saturating_add(new_size); + } + None => { + entry.usage.size = entry.usage.size.saturating_add(new_size); + entry.usage.objects_count = entry.usage.objects_count.saturating_add(1); + entry.usage.versions_count = entry.usage.versions_count.saturating_add(1); + } + } + + let now = SystemTime::now(); + entry.refreshed_at = now; + entry.usage_updated_at = now; +} + +/// Fast in-memory increment for immediate quota consistency. +pub async fn increment_bucket_usage_memory(bucket: &str, size_increment: u64) { + record_bucket_object_write_memory(bucket, None, size_increment).await; +} + +/// Fast in-memory update for successful object deletes. +pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64, removed_current_object: bool) { + ensure_bucket_usage_cached(bucket).await; + + let mut cache = memory_cache().write().await; + let entry = cache + .entry(bucket.to_string()) + .or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default())); + + entry.usage.size = entry.usage.size.saturating_sub(deleted_size); + if removed_current_object { + entry.usage.objects_count = entry.usage.objects_count.saturating_sub(1); + entry.usage.versions_count = entry.usage.versions_count.saturating_sub(1); + } + + let now = SystemTime::now(); + entry.refreshed_at = now; + entry.usage_updated_at = now; } /// Fast in-memory decrement for immediate quota consistency pub async fn decrement_bucket_usage_memory(bucket: &str, size_decrement: u64) { - let mut cache = memory_cache().write().await; - if let Some(current) = cache.get_mut(bucket) { - current.0 = current.0.saturating_sub(size_decrement); - current.1 = SystemTime::now(); - } + record_bucket_object_delete_memory(bucket, size_decrement, size_decrement > 0).await; } /// Get bucket usage from in-memory cache @@ -425,7 +501,7 @@ pub async fn get_bucket_usage_memory(bucket: &str) -> Option { update_usage_cache_if_needed().await; let cache = memory_cache().read().await; - cache.get(bucket).map(|(usage, _)| *usage) + cache.get(bucket).map(|cached| cached.usage.size) } async fn update_usage_cache_if_needed() { @@ -434,7 +510,7 @@ async fn update_usage_cache_if_needed() { let now = SystemTime::now(); let cache = memory_cache().read().await; - let earliest_timestamp = cache.values().map(|(_, ts)| *ts).min(); + let earliest_timestamp = cache.values().map(|cached| cached.refreshed_at).min(); drop(cache); let age = match earliest_timestamp { @@ -461,8 +537,12 @@ async fn update_usage_cache_if_needed() { && let Ok(data_usage_info) = load_data_usage_from_backend(store.clone()).await { let mut cache = cache_clone.write().await; + let usage_updated_at = data_usage_info_updated_at(&data_usage_info); for (bucket_name, bucket_usage) in data_usage_info.buckets_usage.iter() { - cache.insert(bucket_name.clone(), (bucket_usage.size, SystemTime::now())); + cache.insert( + bucket_name.clone(), + cached_bucket_usage_from_backend(bucket_usage.clone(), usage_updated_at), + ); } } let mut updating = updating_clone.write().await; @@ -488,8 +568,12 @@ async fn update_usage_cache_if_needed() { && let Ok(data_usage_info) = load_data_usage_from_backend(store.clone()).await { let mut cache = memory_cache().write().await; + let usage_updated_at = data_usage_info_updated_at(&data_usage_info); for (bucket_name, bucket_usage) in data_usage_info.buckets_usage.iter() { - cache.insert(bucket_name.clone(), (bucket_usage.size, SystemTime::now())); + cache.insert( + bucket_name.clone(), + cached_bucket_usage_from_backend(bucket_usage.clone(), usage_updated_at), + ); } } @@ -497,15 +581,62 @@ async fn update_usage_cache_if_needed() { *updating = false; } +pub async fn replace_bucket_usage_memory_from_info(data_usage_info: &DataUsageInfo) { + let usage_updated_at = data_usage_info_updated_at(data_usage_info); + let mut cache = memory_cache().write().await; + let mut next_cache = HashMap::new(); + + for (bucket, bucket_usage) in data_usage_info.buckets_usage.iter() { + if let Some(existing) = cache.get(bucket) + && existing.usage_updated_at > usage_updated_at + { + next_cache.insert(bucket.clone(), existing.clone()); + continue; + } + + next_cache.insert(bucket.clone(), cached_bucket_usage_from_backend(bucket_usage.clone(), usage_updated_at)); + } + + for (bucket, existing) in cache.iter() { + if !data_usage_info.buckets_usage.contains_key(bucket) && existing.usage_updated_at > usage_updated_at { + next_cache.insert(bucket.clone(), existing.clone()); + } + } + + *cache = next_cache; +} + +pub async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut DataUsageInfo) { + let cache = memory_cache().read().await; + if cache.is_empty() { + return; + } + + let persisted_update = data_usage_info.last_update; + let mut changed = false; + + for (bucket, cached) in cache.iter() { + if persisted_update.is_some_and(|persisted| cached.usage_updated_at <= persisted) { + continue; + } + + data_usage_info.buckets_usage.insert(bucket.clone(), cached.usage.clone()); + data_usage_info.bucket_sizes.insert(bucket.clone(), cached.usage.size); + changed = true; + } + + if changed { + data_usage_info.buckets_count = data_usage_info.buckets_usage.len() as u64; + data_usage_info.calculate_totals(); + } +} + /// Sync memory cache with backend data (called by scanner) pub async fn sync_memory_cache_with_backend() -> Result<(), Error> { if let Some(store) = crate::global::GLOBAL_OBJECT_API.get() { match load_data_usage_from_backend(store.clone()).await { Ok(data_usage_info) => { - let mut cache = memory_cache().write().await; - for (bucket, bucket_usage) in data_usage_info.buckets_usage.iter() { - cache.insert(bucket.clone(), (bucket_usage.size, SystemTime::now())); - } + replace_bucket_usage_memory_from_info(&data_usage_info).await; } Err(e) => { debug!("Failed to sync memory cache with backend: {}", e); @@ -687,6 +818,32 @@ pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate: mod tests { use super::*; use rustfs_data_usage::BucketUsageInfo; + use serial_test::serial; + + async fn clear_usage_memory_cache_for_test() { + memory_cache().write().await.clear(); + *cache_updating().write().await = false; + } + + fn data_usage_info_for_test(bucket: &str, objects_count: u64, size: u64, last_update: SystemTime) -> DataUsageInfo { + let mut info = DataUsageInfo { + last_update: Some(last_update), + ..Default::default() + }; + info.buckets_usage.insert( + bucket.to_string(), + BucketUsageInfo { + objects_count, + versions_count: objects_count, + size, + ..Default::default() + }, + ); + info.bucket_sizes.insert(bucket.to_string(), size); + info.buckets_count = info.buckets_usage.len() as u64; + info.calculate_totals(); + info + } fn aggregate_for_test( inputs: Vec<(DiskUsageStatus, Result, Error>)>, @@ -772,4 +929,101 @@ mod tests { assert_eq!(aggregated.buckets_count, 1); assert_eq!(aggregated.buckets_usage.get("bucket-a").map(|b| (b.objects_count, b.size)), Some((3, 42))); } + + #[tokio::test] + #[serial] + async fn memory_overlay_reflects_recent_delete_before_scanner_persists() { + clear_usage_memory_cache_for_test().await; + + let persisted = data_usage_info_for_test("bucket-a", 1, 42, SystemTime::now() - Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&persisted).await; + record_bucket_object_delete_memory("bucket-a", 42, true).await; + + let mut response = persisted.clone(); + apply_bucket_usage_memory_overlay(&mut response).await; + + assert_eq!(response.objects_total_count, 0); + assert_eq!(response.objects_total_size, 0); + assert_eq!( + response + .buckets_usage + .get("bucket-a") + .map(|usage| (usage.objects_count, usage.size)), + Some((0, 0)) + ); + } + + #[tokio::test] + #[serial] + async fn memory_overlay_preserves_object_count_for_overwrite() { + clear_usage_memory_cache_for_test().await; + + let persisted = data_usage_info_for_test("bucket-a", 1, 10, SystemTime::now() - Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&persisted).await; + record_bucket_object_write_memory("bucket-a", Some(10), 20).await; + + let mut response = persisted.clone(); + apply_bucket_usage_memory_overlay(&mut response).await; + + assert_eq!(response.objects_total_count, 1); + assert_eq!(response.objects_total_size, 20); + assert_eq!( + response + .buckets_usage + .get("bucket-a") + .map(|usage| (usage.objects_count, usage.size)), + Some((1, 20)) + ); + } + + #[tokio::test] + #[serial] + async fn memory_overlay_does_not_replace_newer_persisted_usage() { + clear_usage_memory_cache_for_test().await; + + let now = SystemTime::now(); + let old_persisted = data_usage_info_for_test("bucket-a", 1, 42, now - Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&old_persisted).await; + record_bucket_object_delete_memory("bucket-a", 42, true).await; + + let mut newer_persisted = data_usage_info_for_test("bucket-a", 2, 84, now + Duration::from_secs(10)); + apply_bucket_usage_memory_overlay(&mut newer_persisted).await; + + assert_eq!(newer_persisted.objects_total_count, 2); + assert_eq!(newer_persisted.objects_total_size, 84); + assert_eq!( + newer_persisted + .buckets_usage + .get("bucket-a") + .map(|usage| (usage.objects_count, usage.size)), + Some((2, 84)) + ); + } + + #[tokio::test] + #[serial] + async fn scanner_sync_preserves_newer_memory_update() { + clear_usage_memory_cache_for_test().await; + + let now = SystemTime::now(); + let old_persisted = data_usage_info_for_test("bucket-a", 1, 42, now - Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&old_persisted).await; + record_bucket_object_delete_memory("bucket-a", 42, true).await; + + let scanner_snapshot = data_usage_info_for_test("bucket-a", 1, 42, now - Duration::from_secs(5)); + replace_bucket_usage_memory_from_info(&scanner_snapshot).await; + + let mut response = scanner_snapshot.clone(); + apply_bucket_usage_memory_overlay(&mut response).await; + + assert_eq!(response.objects_total_count, 0); + assert_eq!(response.objects_total_size, 0); + assert_eq!( + response + .buckets_usage + .get("bucket-a") + .map(|usage| (usage.objects_count, usage.size)), + Some((0, 0)) + ); + } } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 167c69cfe..40cb02918 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -403,6 +403,8 @@ pub async fn store_data_usage_in_backend( // Save main configuration if let Err(e) = save_config(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), data).await { error!("Failed to save data usage info to {}: {e}", DATA_USAGE_OBJ_NAME_PATH.as_str()); + } else { + rustfs_ecstore::data_usage::replace_bucket_usage_memory_from_info(&data_usage_info).await; } attempts += 1; diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 66b559916..778eff608 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -19,7 +19,7 @@ use crate::capacity::resolve_admin_used_capacity; use crate::error::ApiError; use rustfs_data_usage::DataUsageInfo; use rustfs_ecstore::admin_server_info::get_server_info; -use rustfs_ecstore::data_usage::load_data_usage_from_backend; +use rustfs_ecstore::data_usage::{apply_bucket_usage_memory_overlay, load_data_usage_from_backend}; use rustfs_ecstore::endpoints::EndpointServerPools; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::pools::{PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free}; @@ -154,6 +154,7 @@ impl DefaultAdminUsecase { error!("load_data_usage_from_backend failed {:?}", e); Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed") })?; + apply_bucket_usage_memory_overlay(&mut info).await; let storage_info = store.storage_info().await; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 2b8c064fc..f538d579f 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -339,14 +339,18 @@ impl DefaultMultipartUsecase { let current_opts = get_opts(&bucket, &key, None, None, &req.headers) .await .map_err(ApiError::from)?; - match store.get_object_info(&bucket, &key, ¤t_opts).await { - Ok(existing_obj_info) => validate_existing_object_lock_for_write(&existing_obj_info)?, + let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { + Ok(existing_obj_info) => { + validate_existing_object_lock_for_write(&existing_obj_info)?; + Some(existing_obj_info.size.max(0) as u64) + } Err(err) => { if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { return Err(ApiError::from(err).into()); } + None } - } + }; let multipart_info = store .get_multipart_info(&bucket, &key, &upload_id, &ObjectOptions::default()) @@ -395,7 +399,12 @@ impl DefaultMultipartUsecase { )); } // Update quota tracking after successful multipart upload - rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; + rustfs_ecstore::data_usage::record_bucket_object_write_memory( + &bucket, + previous_current_size, + obj_info.size.max(0) as u64, + ) + .await; } Err(e) => { warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 267ff93be..736395768 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1791,14 +1791,18 @@ impl DefaultObjectUsecase { let current_opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), None, &req.headers) .await .map_err(ApiError::from)?; - match store.get_object_info(&bucket, &key, ¤t_opts).await { - Ok(existing_obj_info) => validate_existing_object_lock_for_write(&existing_obj_info)?, + let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { + Ok(existing_obj_info) => { + validate_existing_object_lock_for_write(&existing_obj_info)?; + Some(existing_obj_info.size.max(0) as u64) + } Err(err) => { if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { return Err(ApiError::from(err).into()); } + None } - } + }; let actual_size = size; @@ -1913,8 +1917,13 @@ impl DefaultObjectUsecase { maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; - // Fast in-memory update for immediate quota consistency - rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; + // Fast in-memory update for immediate quota and admin usage consistency + rustfs_ecstore::data_usage::record_bucket_object_write_memory( + &bucket, + previous_current_size, + obj_info.size.max(0) as u64, + ) + .await; let raw_version = obj_info.version_id.map(|v| v.to_string()); @@ -2655,14 +2664,18 @@ impl DefaultObjectUsecase { let current_opts: ObjectOptions = get_opts(&bucket, &key, dest_version_id.clone(), None, &req.headers) .await .map_err(ApiError::from)?; - match store.get_object_info(&bucket, &key, ¤t_opts).await { - Ok(existing_obj_info) => validate_existing_object_lock_for_write(&existing_obj_info)?, + let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { + Ok(existing_obj_info) => { + validate_existing_object_lock_for_write(&existing_obj_info)?; + Some(existing_obj_info.size.max(0) as u64) + } Err(err) => { if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { return Err(ApiError::from(err).into()); } + None } - } + }; let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok(); let mut effective_sse = requested_sse.or_else(|| { @@ -2835,7 +2848,8 @@ impl DefaultObjectUsecase { // Update quota tracking after successful copy if has_bucket_metadata { - rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, oi.size as u64).await; + rustfs_ecstore::data_usage::record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64) + .await; } let raw_dest_version = oi.version_id.map(|v| v.to_string()); @@ -3091,10 +3105,13 @@ impl DefaultObjectUsecase { existing_object_infos[i].as_ref(), ) .await; - let size = object_sizes[i]; - if size > 0 { - rustfs_ecstore::data_usage::decrement_bucket_usage_memory(&bucket, size as u64).await; - } + let size = object_sizes[i].max(0) as u64; + rustfs_ecstore::data_usage::record_bucket_object_delete_memory( + &bucket, + size, + existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(), + ) + .await; continue; } @@ -3309,8 +3326,13 @@ impl DefaultObjectUsecase { enqueue_transitioned_delete_cleanup(&bucket, &key, &opts, existing_object_info.as_ref()).await; - // Fast in-memory update for immediate quota consistency - rustfs_ecstore::data_usage::decrement_bucket_usage_memory(&bucket, obj_info.size as u64).await; + // Fast in-memory update for immediate quota and admin usage consistency + rustfs_ecstore::data_usage::record_bucket_object_delete_memory( + &bucket, + obj_info.size.max(0) as u64, + opts.version_id.is_none(), + ) + .await; if obj_info.name.is_empty() { if replicate_force_delete {