From a8327f89019c18d47097161741543119b72ec45e Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Fri, 26 Jun 2026 18:34:21 +0800 Subject: [PATCH] fix(scanner): account versioned delete markers in usage (#3904) Co-authored-by: Henry Guo --- crates/e2e_test/src/data_usage_test.rs | 146 ++++++++++++++- crates/ecstore/src/api/mod.rs | 3 +- crates/ecstore/src/data_usage.rs | 226 +++++++++++++++++++----- crates/scanner/src/data_usage_define.rs | 19 +- crates/scanner/src/scanner_folder.rs | 6 +- crates/scanner/src/scanner_io.rs | 72 ++++++++ rustfs/src/app/admin_usecase.rs | 7 +- rustfs/src/app/multipart_usecase.rs | 12 +- rustfs/src/app/object_usecase.rs | 59 ++++--- rustfs/src/app/storage_api.rs | 29 +++ rustfs/src/storage/storage_api.rs | 6 +- 11 files changed, 503 insertions(+), 82 deletions(-) diff --git a/crates/e2e_test/src/data_usage_test.rs b/crates/e2e_test/src/data_usage_test.rs index a64168397..1a1734bf5 100644 --- a/crates/e2e_test/src/data_usage_test.rs +++ b/crates/e2e_test/src/data_usage_test.rs @@ -13,11 +13,40 @@ // limitations under the License. use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration}; use rustfs_data_usage::DataUsageInfo; use serial_test::serial; +use tokio::time::{Duration, sleep}; use crate::common::{RustFSTestEnvironment, TEST_BUCKET, awscurl_get, init_logging}; +async fn get_data_usage_info(env: &RustFSTestEnvironment) -> Result> { + let url = format!("{}/rustfs/admin/v3/datausageinfo", env.url); + let resp = awscurl_get(&url, &env.access_key, &env.secret_key).await?; + Ok(serde_json::from_str(&resp)?) +} + +async fn wait_for_bucket_usage( + env: &RustFSTestEnvironment, + bucket: &str, + mut predicate: F, +) -> Result> +where + F: FnMut(&DataUsageInfo) -> bool, +{ + let mut last_usage = DataUsageInfo::default(); + for _ in 0..45 { + let usage = get_data_usage_info(env).await?; + if usage.buckets_usage.contains_key(bucket) && predicate(&usage) { + return Ok(usage); + } + last_usage = usage; + sleep(Duration::from_secs(2)).await; + } + + Err(format!("bucket usage did not converge for {bucket}; last usage: {last_usage:?}").into()) +} + /// Regression test for data usage accuracy (issue #1012). /// Launches rustfs, writes 1000 objects, then asserts admin data usage reports the full count. #[tokio::test(flavor = "multi_thread")] @@ -46,9 +75,7 @@ async fn data_usage_reports_all_objects() -> Result<(), Box Result<(), Box Result<(), Box> { + init_logging(); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + + let client = env.create_s3_client(); + let bucket = "data-usage-versioned"; + + client.create_bucket().bucket(bucket).send().await?; + client + .put_bucket_versioning() + .bucket(bucket) + .versioning_configuration( + VersioningConfiguration::builder() + .status(BucketVersioningStatus::Enabled) + .build(), + ) + .send() + .await?; + + client + .put_object() + .bucket(bucket) + .key("alpha") + .body(ByteStream::from_static(b"v1-alpha")) + .send() + .await?; + client + .put_object() + .bucket(bucket) + .key("alpha") + .body(ByteStream::from_static(b"v2-alpha-larger")) + .send() + .await?; + client + .put_object() + .bucket(bucket) + .key("beta") + .body(ByteStream::from_static(b"beta")) + .send() + .await?; + client.delete_object().bucket(bucket).key("alpha").send().await?; + + let listed_versions = client.list_object_versions().bucket(bucket).send().await?; + assert_eq!(listed_versions.versions().len(), 3, "S3 version listing should report stored versions"); + assert_eq!( + listed_versions.delete_markers().len(), + 1, + "S3 version listing should report the delete marker" + ); + + let usage = wait_for_bucket_usage(&env, bucket, |usage| { + usage + .buckets_usage + .get(bucket) + .map(|bucket_usage| { + bucket_usage.objects_count == 2 && bucket_usage.versions_count == 3 && bucket_usage.delete_markers_count == 1 + }) + .unwrap_or(false) + }) + .await?; + let bucket_usage = usage + .buckets_usage + .get(bucket) + .expect("bucket usage should exist after convergence"); + + assert_eq!( + bucket_usage.objects_count, 2, + "current object count should exclude the deleted current alpha" + ); + assert_eq!(bucket_usage.versions_count, 3, "version count should include all stored object versions"); + assert_eq!( + bucket_usage.delete_markers_count, 1, + "delete marker count should include the alpha marker" + ); + assert_eq!(usage.objects_total_count, 2, "total current object count should match bucket usage"); + assert_eq!(usage.versions_total_count, 3, "total version count should match bucket usage"); + assert_eq!(usage.delete_markers_total_count, 1, "total delete marker count should match bucket usage"); + + env.stop_server(); + env.start_rustfs_server(vec![]).await?; + + let restarted_usage = wait_for_bucket_usage(&env, bucket, |usage| { + usage + .buckets_usage + .get(bucket) + .map(|bucket_usage| { + bucket_usage.objects_count == 2 && bucket_usage.versions_count == 3 && bucket_usage.delete_markers_count == 1 + }) + .unwrap_or(false) + }) + .await?; + let restarted_bucket_usage = restarted_usage + .buckets_usage + .get(bucket) + .expect("bucket usage should exist after restart"); + assert_eq!(restarted_bucket_usage.objects_count, 2, "object count should persist after restart"); + assert_eq!(restarted_bucket_usage.versions_count, 3, "version count should persist after restart"); + assert_eq!( + restarted_bucket_usage.delete_markers_count, 1, + "delete marker count should persist after restart" + ); + + env.stop_server(); + Ok(()) +} diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index de8c4a1b1..6e8958073 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -69,7 +69,8 @@ pub mod config { pub mod data_usage { pub use crate::data_usage::{ DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, load_data_usage_from_backend, - record_bucket_object_delete_memory, record_bucket_object_write_memory, remove_bucket_usage_from_backend, + record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, + record_bucket_object_write_memory, refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, }; } diff --git a/crates/ecstore/src/data_usage.rs b/crates/ecstore/src/data_usage.rs index 3ad248d35..1414c30c3 100644 --- a/crates/ecstore/src/data_usage.rs +++ b/crates/ecstore/src/data_usage.rs @@ -16,7 +16,7 @@ pub mod local_snapshot; use crate::storage_api_contracts::{list::ListOperations as _, object::ObjectIO as _}; use crate::{ - bucket::metadata_sys::get_replication_config, + bucket::{metadata_sys::get_replication_config, versioning::VersioningApi as _, versioning_sys::BucketVersioningSys}, config::com::read_config, disk::DiskAPI, error::{Error, classify_system_path_failure_reason}, @@ -398,8 +398,9 @@ pub async fn aggregate_local_snapshots(store: Arc) -> Result<(Vec, bucket_name: &str) -> Result { - let mut continuation: Option = None; - let mut objects_count: u64 = 0; + let mut marker: Option = None; + let mut version_marker: Option = None; + let mut object_names: HashSet = HashSet::new(); let mut versions_count: u64 = 0; let mut total_size: u64 = 0; let mut delete_markers: u64 = 0; @@ -407,15 +408,13 @@ pub async fn compute_bucket_usage(store: Arc, bucket_name: &str) -> Res loop { let result = store .clone() - .list_objects_v2( + .list_object_versions( bucket_name, "", // prefix - continuation.clone(), - None, // delimiter - 1000, // max_keys - false, // fetch_owner - None, // start_after - false, // incl_deleted + marker.clone(), + version_marker.clone(), + None, // delimiter + 1000, // max_keys ) .await?; @@ -430,34 +429,27 @@ pub async fn compute_bucket_usage(store: Arc, bucket_name: &str) -> Res } let object_size = object.size.max(0) as u64; - objects_count = objects_count.saturating_add(1); + object_names.insert(object.name.clone()); total_size = total_size.saturating_add(object_size); - - let detected_versions = if object.num_versions > 0 { - object.num_versions as u64 - } else { - 1 - }; - versions_count = versions_count.saturating_add(detected_versions); + versions_count = versions_count.saturating_add(1); } if !result.is_truncated { break; } - continuation = result.next_continuation_token.clone(); - if continuation.is_none() { + marker = result.next_marker.clone(); + version_marker = result.next_version_idmarker.clone(); + if marker.is_none() { info!( - "Bucket {} listing marked truncated but no continuation token returned; stopping early", + "Bucket {} version listing marked truncated but no marker returned; stopping early", bucket_name ); break; } } - if versions_count == 0 { - versions_count = objects_count; - } + let objects_count = object_names.len() as u64; let usage = BucketUsageInfo { size: total_size, @@ -470,6 +462,42 @@ pub async fn compute_bucket_usage(store: Arc, bucket_name: &str) -> Res Ok(usage) } +pub async fn refresh_versioned_bucket_usage_from_object_layer(store: Arc, data_usage_info: &mut DataUsageInfo) { + let buckets = data_usage_info.buckets_usage.keys().cloned().collect::>(); + let mut changed = false; + + for bucket in buckets { + let Ok(versioning) = BucketVersioningSys::get(&bucket).await else { + continue; + }; + + if !versioning.enabled() && !versioning.suspended() { + continue; + } + + let usage = match compute_bucket_usage(store.clone(), &bucket).await { + Ok(usage) => usage, + Err(err) => { + debug!( + bucket = %bucket, + error = %err, + "failed to refresh versioned bucket usage from object layer" + ); + continue; + } + }; + + data_usage_info.bucket_sizes.insert(bucket.clone(), usage.size); + data_usage_info.buckets_usage.insert(bucket, usage); + changed = true; + } + + if changed { + set_buckets_count_from_usage(data_usage_info); + data_usage_info.calculate_totals(); + } +} + async fn ensure_bucket_usage_cached(bucket: &str) { let cache = memory_cache().read().await; if cache.contains_key(bucket) { @@ -514,6 +542,20 @@ fn bucket_usage_counts_match(left: &BucketUsageInfo, right: &BucketUsageInfo) -> /// 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) { + record_bucket_object_write_memory_inner(bucket, previous_current_size, new_size, false).await; +} + +/// Fast in-memory update for versioned object writes. +pub async fn record_bucket_object_version_write_memory(bucket: &str, previous_current_size: Option, new_size: u64) { + record_bucket_object_write_memory_inner(bucket, previous_current_size, new_size, true).await; +} + +async fn record_bucket_object_write_memory_inner( + bucket: &str, + previous_current_size: Option, + new_size: u64, + creates_new_version: bool, +) { ensure_bucket_usage_cached(bucket).await; let mut cache = memory_cache().write().await; @@ -521,14 +563,22 @@ pub async fn record_bucket_object_write_memory(bucket: &str, previous_current_si .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); + if creates_new_version { + entry.usage.size = entry.usage.size.saturating_add(new_size); + if previous_current_size.is_none() { entry.usage.objects_count = entry.usage.objects_count.saturating_add(1); - entry.usage.versions_count = entry.usage.versions_count.saturating_add(1); + } + entry.usage.versions_count = entry.usage.versions_count.saturating_add(1); + } else { + 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); + } } } @@ -566,6 +616,24 @@ pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64, entry.stale_snapshot_pending = false; } +/// Fast in-memory update for successful delete marker creation. +pub async fn record_bucket_delete_marker_memory(bucket: &str) { + 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.delete_markers_count = entry.usage.delete_markers_count.saturating_add(1); + + let now = SystemTime::now(); + entry.refreshed_at = now; + entry.usage_updated_at = now; + entry.dirty = true; + entry.stale_snapshot_pending = false; +} + /// Fast in-memory decrement for immediate quota consistency pub async fn decrement_bucket_usage_memory(bucket: &str, size_decrement: u64) { record_bucket_object_delete_memory(bucket, size_decrement, size_decrement > 0).await; @@ -605,20 +673,12 @@ async fn update_usage_cache_if_needed() { *updating = true; drop(updating); - let cache_clone = (*memory_cache()).clone(); let updating_clone = (*cache_updating()).clone(); tokio::spawn(async move { if let Some(store) = runtime_sources::object_store_handle() && 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(), - cached_bucket_usage_from_backend(bucket_usage.clone(), usage_updated_at), - ); - } + replace_bucket_usage_memory_from_info(&data_usage_info).await; } let mut updating = updating_clone.write().await; *updating = false; @@ -642,14 +702,7 @@ async fn update_usage_cache_if_needed() { if let Some(store) = runtime_sources::object_store_handle() && 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(), - cached_bucket_usage_from_backend(bucket_usage.clone(), usage_updated_at), - ); - } + replace_bucket_usage_memory_from_info(&data_usage_info).await; } let mut updating = cache_updating().write().await; @@ -1161,6 +1214,83 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn memory_overlay_counts_versioned_overwrite_as_new_version() { + 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_version_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.versions_total_count, 2); + assert_eq!(response.objects_total_size, 30); + assert_eq!( + response + .buckets_usage + .get("bucket-a") + .map(|usage| (usage.objects_count, usage.versions_count, usage.size)), + Some((1, 2, 30)) + ); + } + + #[tokio::test] + #[serial] + async fn memory_overlay_records_delete_marker_without_removing_versions() { + clear_usage_memory_cache_for_test().await; + + let persisted = data_usage_info_for_test("bucket-a", 2, 30, SystemTime::now() - Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&persisted).await; + record_bucket_delete_marker_memory("bucket-a").await; + + let mut response = persisted.clone(); + apply_bucket_usage_memory_overlay(&mut response).await; + + assert_eq!(response.objects_total_count, 2); + assert_eq!(response.versions_total_count, 2); + assert_eq!(response.delete_markers_total_count, 1); + assert_eq!(response.objects_total_size, 30); + assert_eq!( + response + .buckets_usage + .get("bucket-a") + .map(|usage| { (usage.objects_count, usage.versions_count, usage.delete_markers_count, usage.size,) }), + Some((2, 2, 1, 30)) + ); + } + + #[tokio::test] + #[serial] + async fn scanner_sync_preserves_dirty_delete_marker_with_later_snapshot() { + clear_usage_memory_cache_for_test().await; + + let now = SystemTime::now(); + let old_persisted = data_usage_info_for_test("bucket-a", 2, 30, now - Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&old_persisted).await; + record_bucket_delete_marker_memory("bucket-a").await; + + let scanner_without_marker = data_usage_info_for_test("bucket-a", 2, 30, now + Duration::from_secs(10)); + replace_bucket_usage_memory_from_info(&scanner_without_marker).await; + + let mut response = scanner_without_marker.clone(); + apply_bucket_usage_memory_overlay(&mut response).await; + + assert_eq!(response.objects_total_count, 2); + assert_eq!(response.versions_total_count, 2); + assert_eq!(response.delete_markers_total_count, 1); + assert_eq!( + response + .buckets_usage + .get("bucket-a") + .map(|usage| { (usage.objects_count, usage.versions_count, usage.delete_markers_count, usage.size,) }), + Some((2, 2, 1, 30)) + ); + } + #[tokio::test] #[serial] async fn memory_overlay_does_not_replace_newer_persisted_usage() { diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index fb6532666..c79f0cba3 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -172,6 +172,7 @@ impl SizeSummary { pub fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64) { if oi.delete_marker { self.delete_markers += 1; + return; } if oi.version_id.is_some_and(|v| !v.is_nil()) && size == actual_size { @@ -180,7 +181,7 @@ impl SizeSummary { self.total_size += if size > 0 { size as usize } else { 0 }; - if oi.delete_marker || oi.transitioned_object.free_version { + if oi.transitioned_object.free_version { return; } @@ -1065,6 +1066,22 @@ mod tests { assert_eq!(summary1.versions, 15); } + #[test] + fn size_summary_counts_delete_markers_separately_from_versions() { + let mut summary = SizeSummary::new(); + let marker = ObjectInfo { + delete_marker: true, + version_id: Some(uuid::Uuid::new_v4()), + ..Default::default() + }; + + summary.actions_accounting(&marker, 0, 0); + + assert_eq!(summary.delete_markers, 1); + assert_eq!(summary.versions, 0); + assert_eq!(summary.total_size, 0); + } + #[test] fn test_data_usage_entry_merge_sums_failed_objects() { let mut left = DataUsageEntry { diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index de3d9323b..3c83c0042 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -697,10 +697,10 @@ impl ScannerItem { component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_LIFECYCLE, bucket = %self.bucket, - state = "versioning_lookup_failed", - "Scanner lifecycle action skipped" + state = "versioning_lookup_failed_defaulting", + "Scanner lifecycle action falling back to default bucket versioning" ); - return; + Default::default() } }; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 2f7e71bb4..f2a081041 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -1543,8 +1543,10 @@ mod tests { use super::*; use crate::scanner_folder::ScannerItem; use crate::{DiskOption, Endpoint, new_disk, path2_bucket_object_with_base_path}; + use rustfs_filemeta::FileInfo; use serial_test::serial; use temp_env::with_var; + use time::OffsetDateTime; use uuid::Uuid; fn bucket_info(name: &str) -> BucketInfo { @@ -1770,6 +1772,76 @@ mod tests { let _ = tokio::fs::remove_dir_all(&temp_dir).await; } + #[tokio::test] + async fn get_size_counts_delete_markers_separately_from_versions() { + let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-versioned-usage-{}", Uuid::new_v4())); + let bucket = "bucket"; + let object = "object"; + let object_dir = temp_dir.join(bucket).join(object); + let metadata_path = object_dir.join(STORAGE_FORMAT_FILE); + + tokio::fs::create_dir_all(&object_dir) + .await + .expect("failed to create object directory"); + + let mut meta = FileMeta::new(); + for (size, timestamp) in [(10, 10), (20, 20)] { + let mut fi = FileInfo::new(object, 1, 1); + fi.version_id = Some(Uuid::new_v4()); + fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(timestamp).expect("timestamp should be valid")); + fi.size = size; + meta.add_version(fi).expect("object version should be added"); + } + + let mut delete_marker = FileInfo::new(object, 1, 1); + delete_marker.version_id = Some(Uuid::new_v4()); + delete_marker.mod_time = Some(OffsetDateTime::from_unix_timestamp(30).expect("timestamp should be valid")); + delete_marker.deleted = true; + meta.add_version(delete_marker).expect("delete marker should be added"); + + tokio::fs::write(&metadata_path, meta.marshal_msg().expect("metadata should marshal")) + .await + .expect("failed to write metadata"); + + let endpoint = Endpoint::try_from(temp_dir.to_string_lossy().as_ref()).expect("failed to create endpoint"); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("failed to open local disk"); + + let relative_path = metadata_path.to_string_lossy().to_string(); + let (_, scanner_path) = path2_bucket_object_with_base_path(temp_dir.to_string_lossy().as_ref(), relative_path.as_str()); + let file_type = tokio::fs::metadata(&metadata_path) + .await + .expect("failed to stat metadata") + .file_type(); + let item = ScannerItem { + path: scanner_path, + bucket: bucket.to_string(), + prefix: object.to_string(), + object_name: STORAGE_FORMAT_FILE.to_string(), + file_type, + lifecycle: None, + replication: None, + heal_enabled: false, + heal_bitrot: false, + debug: false, + }; + + let summary = disk.get_size(item).await.expect("scanner should read versioned metadata"); + + assert_eq!(summary.versions, 2); + assert_eq!(summary.delete_markers, 1); + assert_eq!(summary.total_size, 30); + + let _ = tokio::fs::remove_dir_all(&temp_dir).await; + } + #[test] fn cache_root_entry_info_flattens_bucket_children() { let mut cache = DataUsageCache { diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 8a45b7628..0958ff3c6 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -19,7 +19,10 @@ use super::storage_api::admin_usecase::admin::get_server_info; use super::storage_api::admin_usecase::capacity::{ PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity, get_total_usable_capacity_free, }; -use super::storage_api::admin_usecase::data_usage::{apply_bucket_usage_memory_overlay, load_data_usage_from_backend}; +use super::storage_api::admin_usecase::data_usage::{ + apply_bucket_usage_memory_overlay, load_data_usage_from_backend, refresh_versioned_bucket_usage_from_object_layer, + replace_bucket_usage_memory_from_info, +}; use super::storage_api::admin_usecase::{ECStore, EndpointServerPools}; use crate::app::runtime_sources::{ AppContext, current_app_context, resolve_endpoints_handle, resolve_object_store_handle_for_context, @@ -242,6 +245,8 @@ impl DefaultAdminUsecase { error!("load_data_usage_from_backend failed {:?}", e); Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed") })?; + refresh_versioned_bucket_usage_from_object_layer(store.clone(), &mut info).await; + replace_bucket_usage_memory_from_info(&info).await; apply_bucket_usage_memory_overlay(&mut info).await; let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 2aab93695..d6d738910 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -27,7 +27,9 @@ use super::storage_api::multipart_usecase::bucket::{ versioning_sys::BucketVersioningSys, }; use super::storage_api::multipart_usecase::compression::is_disk_compressible; -use super::storage_api::multipart_usecase::data_usage::record_bucket_object_write_memory; +use super::storage_api::multipart_usecase::data_usage::{ + record_bucket_object_version_write_memory, record_bucket_object_write_memory, +}; use super::storage_api::multipart_usecase::error::{StorageError, is_err_object_not_found, is_err_version_not_found}; use super::storage_api::multipart_usecase::helper::OperationHelper; #[cfg(test)] @@ -489,7 +491,13 @@ impl DefaultMultipartUsecase { )); } // Update quota tracking after successful multipart upload - record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + let mpu_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; + if mpu_versioned { + record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64) + .await; + } else { + 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 670e6bfa8..d715eab49 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -50,7 +50,10 @@ use super::storage_api::object_usecase::concurrency::{ self, ConcurrencyManager, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; -use super::storage_api::object_usecase::data_usage::{record_bucket_object_delete_memory, record_bucket_object_write_memory}; +use super::storage_api::object_usecase::data_usage::{ + record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, + record_bucket_object_write_memory, +}; use super::storage_api::object_usecase::deadlock_detector; use super::storage_api::object_usecase::ecfs::FS; use super::storage_api::object_usecase::error::{ @@ -2778,8 +2781,13 @@ impl DefaultObjectUsecase { maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; + let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; // Fast in-memory update for immediate quota and admin usage consistency - record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + if put_versioned { + record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + } else { + 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()); @@ -2788,11 +2796,7 @@ impl DefaultObjectUsecase { helper = helper.version_id(version_id.clone()); } - let put_version = if BucketVersioningSys::prefix_enabled(&bucket, &key).await { - raw_version - } else { - None - }; + let put_version = if put_versioned { raw_version } else { None }; let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)); @@ -3769,17 +3773,18 @@ impl DefaultObjectUsecase { maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await; + let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; // Update quota tracking after successful copy if has_bucket_metadata { - record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await; + if dest_versioned { + record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await; + } else { + 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()); - let dest_version = if BucketVersioningSys::prefix_enabled(&bucket, &key).await { - raw_dest_version - } else { - None - }; + let dest_version = if dest_versioned { raw_dest_version } else { None }; // warn!("copy_object oi {:?}", &oi); let object_info = oi.clone(); @@ -3994,6 +3999,7 @@ impl DefaultObjectUsecase { &bucket, object_to_delete.clone(), ObjectOptions { + versioned: version_cfg.enabled(), version_suspended: version_cfg.suspended(), ..Default::default() }, @@ -4047,13 +4053,20 @@ impl DefaultObjectUsecase { "failed to persist transitioned object cleanup journal" ); } - let size = object_sizes[i].max(0) as u64; - record_bucket_object_delete_memory( - &bucket, - size, - existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(), - ) - .await; + let creates_delete_marker = object_to_delete[i].version_id.is_none() + && version_cfg.prefix_enabled(object_to_delete[i].object_name.as_str()) + && !version_cfg.suspended(); + if creates_delete_marker { + record_bucket_delete_marker_memory(&bucket).await; + } else { + let size = object_sizes[i].max(0) as u64; + record_bucket_object_delete_memory( + &bucket, + size, + existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(), + ) + .await; + } continue; } @@ -4285,7 +4298,11 @@ impl DefaultObjectUsecase { } // Fast in-memory update for immediate quota and admin usage consistency - record_bucket_object_delete_memory(&bucket, obj_info.size.max(0) as u64, opts.version_id.is_none()).await; + if delete_creates_delete_marker(&opts) { + record_bucket_delete_marker_memory(&bucket).await; + } else { + 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 { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 483974d61..c0408f22b 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -55,11 +55,35 @@ pub(crate) mod data_usage { crate::storage::ecstore_data_usage::load_data_usage_from_backend(store).await } + pub(crate) async fn refresh_versioned_bucket_usage_from_object_layer( + store: Arc, + data_usage_info: &mut rustfs_data_usage::DataUsageInfo, + ) { + crate::storage::ecstore_data_usage::refresh_versioned_bucket_usage_from_object_layer(store, data_usage_info).await; + } + + pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) { + crate::storage::ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await; + } + pub(crate) async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64, removed_current_object: bool) { crate::storage::ecstore_data_usage::record_bucket_object_delete_memory(bucket, deleted_size, removed_current_object) .await; } + pub(crate) async fn record_bucket_delete_marker_memory(bucket: &str) { + crate::storage::ecstore_data_usage::record_bucket_delete_marker_memory(bucket).await; + } + + pub(crate) async fn record_bucket_object_version_write_memory( + bucket: &str, + previous_current_size: Option, + new_size: u64, + ) { + crate::storage::ecstore_data_usage::record_bucket_object_version_write_memory(bucket, previous_current_size, new_size) + .await; + } + pub(crate) async fn record_bucket_object_write_memory(bucket: &str, previous_current_size: Option, new_size: u64) { crate::storage::ecstore_data_usage::record_bucket_object_write_memory(bucket, previous_current_size, new_size).await; } @@ -214,11 +238,16 @@ pub(crate) mod bucket { } pub(crate) trait VersioningConfigExt { + fn enabled(&self) -> bool; fn prefix_enabled(&self, prefix: &str) -> bool; fn suspended(&self) -> bool; } impl VersioningConfigExt for s3s::dto::VersioningConfiguration { + fn enabled(&self) -> bool { + ::enabled(self) + } + fn prefix_enabled(&self, prefix: &str) -> bool { ::prefix_enabled( self, prefix, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 08b8c7d59..de1fcf1a2 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -81,8 +81,10 @@ pub(crate) mod ecstore_config { pub(crate) mod ecstore_data_usage { pub(crate) use rustfs_ecstore::api::data_usage::{ - apply_bucket_usage_memory_overlay, load_data_usage_from_backend, record_bucket_object_delete_memory, - record_bucket_object_write_memory, remove_bucket_usage_from_backend, + apply_bucket_usage_memory_overlay, load_data_usage_from_backend, record_bucket_delete_marker_memory, + record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory, + refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, + replace_bucket_usage_memory_from_info, }; }