diff --git a/crates/e2e_test/src/quota_test.rs b/crates/e2e_test/src/quota_test.rs index 1467de86d..87eee5f37 100644 --- a/crates/e2e_test/src/quota_test.rs +++ b/crates/e2e_test/src/quota_test.rs @@ -252,6 +252,7 @@ impl QuotaTestEnv { #[cfg(test)] mod integration_tests { use super::*; + use aws_sdk_s3::error::ProvideErrorMetadata; #[tokio::test] #[serial] @@ -963,9 +964,27 @@ mod integration_tests { .send() .await; - assert!(complete_result.is_err()); + let complete_error = complete_result.expect_err("multipart completion above quota must be rejected"); + assert_eq!(complete_error.as_service_error().and_then(|error| error.code()), Some("InvalidRequest")); assert!(!env.object_exists("over_quota.txt").await?); + let staged_parts = env + .client + .list_parts() + .bucket(&env.bucket_name) + .key("over_quota.txt") + .upload_id(upload_id2) + .send() + .await?; + assert_eq!(staged_parts.parts().len(), 2, "quota rejection must preserve the multipart upload"); + env.client + .abort_multipart_upload() + .bucket(&env.bucket_name) + .key("over_quota.txt") + .upload_id(upload_id2) + .send() + .await?; + env.cleanup_bucket().await?; Ok(()) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index d376cb0b7..50ae26736 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -308,6 +308,8 @@ pub mod config { } pub mod data_usage { + #[cfg(feature = "test-util")] + pub use crate::data_usage::seed_bucket_usage_memory_for_test; pub use crate::data_usage::{ DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, compute_bucket_usage, init_compression_total_memory_from_backend, invalidate_admin_data_usage_snapshot_cache, @@ -409,10 +411,10 @@ pub mod object { pub use crate::object_api::{ BLOCK_SIZE_V2, ERASURE_ALGORITHM, EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, NamespaceLockFence, ObjectEncryptionResolver, - ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, RangedDecompressReader, - ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest, StreamConsumer, get_object_body_cache_plaintext_len, - lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook, - unregister_get_object_body_cache_hook, unregister_object_mutation_hook, + ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission, + RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest, StreamConsumer, + get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, + register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook, }; pub use crate::store::{ PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError, diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 6bb0811a4..6ac52e87a 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -1638,7 +1638,7 @@ fn preserve_unknown_dirty_usage( Some(preserved) } -#[cfg(test)] +#[cfg(any(test, feature = "test-util"))] async fn replace_bucket_usage_memory_from_authoritative(bucket: &str, usage: BucketUsageInfo, refresh_started_at: SystemTime) { let mut cache = memory_cache().write().await; if let Some(existing) = cache.get(bucket) @@ -1650,6 +1650,19 @@ async fn replace_bucket_usage_memory_from_authoritative(bucket: &str, usage: Buc cache.insert(bucket.to_string(), cached_bucket_usage_from_backend(usage, refresh_started_at, true)); } +#[cfg(feature = "test-util")] +pub async fn seed_bucket_usage_memory_for_test(bucket: &str, size: u64) { + replace_bucket_usage_memory_from_authoritative( + bucket, + BucketUsageInfo { + size, + ..Default::default() + }, + SystemTime::now(), + ) + .await; +} + /// 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; diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index 7c13db86e..a5d4dd0e9 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -204,6 +204,8 @@ pub enum StorageError { required: usize, achieved: usize, }, + #[error("Bucket quota exceeded. Current usage: {current} bytes, limit: {limit} bytes")] + QuotaExceeded { current: u64, limit: u64 }, // ── Generic ────────────────────────────────────────────────────── #[error("Unexpected error")] @@ -540,6 +542,10 @@ impl Clone for StorageError { required: *required, achieved: *achieved, }, + StorageError::QuotaExceeded { current, limit } => StorageError::QuotaExceeded { + current: *current, + limit: *limit, + }, } } } @@ -627,6 +633,7 @@ impl StorageError { StorageError::NotModified => StorageErrorCode::NotModified, StorageError::InvalidPartNumber(_) => StorageErrorCode::InvalidPartNumber, StorageError::NamespaceLockQuorumUnavailable { .. } => StorageErrorCode::NamespaceLockQuorumUnavailable, + StorageError::QuotaExceeded { .. } => StorageErrorCode::QuotaExceeded, } } @@ -752,6 +759,10 @@ impl StorageError { required: Default::default(), achieved: Default::default(), }), + StorageErrorCode::QuotaExceeded => Some(StorageError::QuotaExceeded { + current: Default::default(), + limit: Default::default(), + }), } } } @@ -1301,6 +1312,7 @@ mod tests { .to_u32(), 0x42 ); + assert_eq!(StorageError::QuotaExceeded { current: 1, limit: 2 }.to_u32(), 0x53); } #[test] @@ -1319,6 +1331,10 @@ mod tests { StorageError::from_u32(0x42), Some(StorageError::NamespaceLockQuorumUnavailable { .. }) )); + assert!(matches!( + StorageError::from_u32(0x53), + Some(StorageError::QuotaExceeded { current: 0, limit: 0 }) + )); // Test invalid code returns None assert!(StorageError::from_u32(0xFF).is_none()); @@ -1549,6 +1565,7 @@ mod tests { StorageError::DecommissionAlreadyRunning, StorageError::RebalanceAlreadyRunning, StorageError::OperationCanceled, + StorageError::QuotaExceeded { current: 1, limit: 2 }, ]; for original_error in test_errors { diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 138b08b91..142ae9307 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -211,6 +211,26 @@ impl ObjectLockConfigSnapshot { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct QuotaAdmission { + current_usage: u64, + quota_limit: u64, +} + +impl QuotaAdmission { + pub(crate) fn current_usage(self) -> u64 { + self.current_usage + } + + pub(crate) fn quota_limit(self) -> u64 { + self.quota_limit + } + + pub(crate) fn remaining(self) -> u64 { + self.quota_limit - self.current_usage + } +} + #[derive(Debug, Default, Clone)] pub struct ObjectOptions { // Use the maximum parity (N/2), used when saving server configuration files @@ -275,12 +295,22 @@ pub struct ObjectOptions { pub want_checksum: Option, pub skip_verify_bitrot: bool, pub capacity_scope_token: Option, + /// Server-derived bucket-quota snapshot for commit-boundary admission. + pub quota_admission: Option, /// Storage-owned journal writer used by the atomic delete path. This is /// populated only by the `ECStore` wrapper that holds the namespace locks. pub tier_delete_journal_api: Option>, } impl ObjectOptions { + pub fn set_quota_admission(&mut self, current_usage: u64, quota_limit: u64) -> bool { + self.quota_admission = (current_usage <= quota_limit).then_some(QuotaAdmission { + current_usage, + quota_limit, + }); + self.quota_admission.is_some() + } + pub(crate) fn overwrites_existing_version(&self) -> bool { self.version_id.is_some() || !self.versioned || self.version_suspended } diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 74d7a45bb..0505526ec 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -1860,7 +1860,12 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } object_size += ext_part.size; - object_actual_size += ext_part.actual_size; + if opts.quota_admission.is_some() && ext_part.actual_size < 0 { + return Err(Error::PartMissingOrCorrupt); + } + object_actual_size = object_actual_size + .checked_add(ext_part.actual_size) + .ok_or(Error::PartMissingOrCorrupt)?; fi.parts.push(completed_multipart_object_part(p.part_num, ext_part)); } @@ -1889,6 +1894,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } } + if let Some(admission) = opts.quota_admission { + let quota_operation_size = u64::try_from(object_actual_size).map_err(|_| Error::PartMissingOrCorrupt)?; + if quota_operation_size > admission.remaining() { + return Err(Error::QuotaExceeded { + current: admission.current_usage(), + limit: admission.quota_limit(), + }); + } + } if let Some(rc_crc) = get_header_map(&opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC) { if let Ok(rc_crc_bytes) = base64_simd::STANDARD.decode_to_vec(&rc_crc) { fi.checksum = Some(Bytes::from(rc_crc_bytes)); @@ -2551,29 +2565,157 @@ mod tests { .new_multipart_upload(bucket, object, create_opts) .await .expect("multipart upload should be created"); + let part = put_test_part(set_disks, bucket, object, &upload.upload_id, 1, content, content.len() as i64).await; + (upload.upload_id, vec![part]) + } + + async fn put_test_part( + set_disks: &Arc, + bucket: &str, + object: &str, + upload_id: &str, + part_number: usize, + content: &[u8], + actual_size: i64, + ) -> CompletePart { let mut reader = PutObjReader::new( - HashReader::from_stream( - Cursor::new(content.to_vec()), - content.len() as i64, - content.len() as i64, - None, - None, - false, - ) - .expect("hash reader should be constructed"), + HashReader::from_stream(Cursor::new(content.to_vec()), content.len() as i64, actual_size, None, None, false) + .expect("hash reader should be constructed"), ); let part = set_disks - .put_object_part(bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default()) + .put_object_part(bucket, object, upload_id, part_number, &mut reader, &ObjectOptions::default()) .await .expect("uploading the part should succeed"); - ( - upload.upload_id, - vec![CompletePart { - part_num: part.part_num, - etag: part.etag, - ..Default::default() - }], - ) + CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + } + } + + #[tokio::test] + async fn complete_multipart_quota_rejection_preserves_destination_and_upload() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-quota-admission-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + let existing_payload = b"existing object"; + let mut existing_reader = PutObjReader::from_vec(existing_payload.to_vec()); + let existing = set_disks + .put_object(bucket, object, &mut existing_reader, &ObjectOptions::default()) + .await + .expect("existing object should be stored"); + + let payload = vec![0x51; 4096]; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &payload, &ObjectOptions::default()).await; + let mut denied_opts = ObjectOptions::default(); + assert!(denied_opts.set_quota_admission(100, 4195)); + + let err = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &denied_opts) + .await + .expect_err("completion larger than the remaining quota must be rejected"); + assert!(matches!( + err, + StorageError::QuotaExceeded { + current: 100, + limit: 4195 + } + )); + + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("quota rejection must preserve the existing destination"); + assert_eq!(current.etag, existing.etag); + assert!( + set_disks + .check_upload_id_exists(bucket, object, &upload_id, false) + .await + .is_ok(), + "quota rejection must leave the multipart upload retryable" + ); + + let mut allowed_opts = ObjectOptions::default(); + assert!(allowed_opts.set_quota_admission(100, 4196)); + let completed = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts, &allowed_opts) + .await + .expect("completion at the exact remaining-quota boundary should succeed"); + assert_eq!(completed.get_actual_size().expect("completed logical size should resolve"), 4096); + } + + #[tokio::test] + async fn complete_multipart_quota_uses_compressed_logical_size() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-compressed-quota-bucket"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + let mut create_opts = ObjectOptions::default(); + insert_str(&mut create_opts.user_defined, SUFFIX_COMPRESSION, "S2".to_string()); + let upload = set_disks + .new_multipart_upload(bucket, object, &create_opts) + .await + .expect("multipart upload should be created"); + let part = put_test_part(&set_disks, bucket, object, &upload.upload_id, 1, &[0x52; 128], 8192).await; + let mut complete_opts = ObjectOptions::default(); + assert!(complete_opts.set_quota_admission(0, 4096)); + + let err = set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload.upload_id, vec![part], &complete_opts) + .await + .expect_err("logical size above the remaining quota must be rejected"); + assert!(matches!(err, StorageError::QuotaExceeded { current: 0, limit: 4096 })); + assert!( + set_disks + .check_upload_id_exists(bucket, object, &upload.upload_id, false) + .await + .is_ok(), + "quota rejection must leave compressed parts retryable" + ); + } + + #[tokio::test] + async fn complete_multipart_quota_rejects_invalid_logical_sizes() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-invalid-logical-size-bucket"; + make_bucket_on_all(&disk_stores, bucket).await; + + let mut create_opts = ObjectOptions::default(); + insert_str(&mut create_opts.user_defined, SUFFIX_COMPRESSION, "S2".to_string()); + let mut complete_opts = ObjectOptions::default(); + assert!(complete_opts.set_quota_admission(0, u64::MAX)); + + let negative_upload = set_disks + .new_multipart_upload(bucket, "negative", &create_opts) + .await + .expect("negative-size upload should be created"); + let negative_part = put_test_part(&set_disks, bucket, "negative", &negative_upload.upload_id, 1, &[0x53], -1).await; + let negative_err = set_disks + .clone() + .complete_multipart_upload(bucket, "negative", &negative_upload.upload_id, vec![negative_part], &complete_opts) + .await + .expect_err("negative logical size must fail closed"); + assert!(matches!(negative_err, StorageError::PartMissingOrCorrupt)); + + let overflow_upload = set_disks + .new_multipart_upload(bucket, "overflow", &create_opts) + .await + .expect("overflow upload should be created"); + let first = put_test_part(&set_disks, bucket, "overflow", &overflow_upload.upload_id, 1, &[0x54], i64::MAX).await; + let second = put_test_part(&set_disks, bucket, "overflow", &overflow_upload.upload_id, 2, &[0x55], 1).await; + let overflow_err = set_disks + .clone() + .complete_multipart_upload(bucket, "overflow", &overflow_upload.upload_id, vec![first, second], &complete_opts) + .await + .expect_err("overflowing logical size must fail closed"); + assert!(matches!(overflow_err, StorageError::PartMissingOrCorrupt)); } async fn assert_complete_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) { diff --git a/crates/storage-api/src/error.rs b/crates/storage-api/src/error.rs index 0e7d03c7a..6238efe4e 100644 --- a/crates/storage-api/src/error.rs +++ b/crates/storage-api/src/error.rs @@ -103,6 +103,7 @@ pub enum StorageErrorCode { SourceStalled, Timeout, InvalidPath, + QuotaExceeded, } impl StorageErrorCode { @@ -188,6 +189,7 @@ impl StorageErrorCode { Self::SourceStalled => 0x50, Self::Timeout => 0x51, Self::InvalidPath => 0x52, + Self::QuotaExceeded => 0x53, } } @@ -273,6 +275,7 @@ impl StorageErrorCode { 0x50 => Some(Self::SourceStalled), 0x51 => Some(Self::Timeout), 0x52 => Some(Self::InvalidPath), + 0x53 => Some(Self::QuotaExceeded), _ => None, } } @@ -347,6 +350,7 @@ mod tests { (StorageErrorCode::RebalanceAlreadyRunning, 0x40), (StorageErrorCode::OperationCanceled, 0x41), (StorageErrorCode::NamespaceLockQuorumUnavailable, 0x42), + (StorageErrorCode::QuotaExceeded, 0x53), ]; const DISK_PRESERVATION_ERROR_CODES: &[(StorageErrorCode, u32)] = &[ diff --git a/rustfs/src/admin/site_replication_state.rs b/rustfs/src/admin/site_replication_state.rs index 7270a4c58..6d924a218 100644 --- a/rustfs/src/admin/site_replication_state.rs +++ b/rustfs/src/admin/site_replication_state.rs @@ -42,8 +42,8 @@ //! mutex, then state object lock) -> per-bucket metadata. use crate::admin::storage_api::runtime::ECStore; +use crate::admin::storage_api::s3::{S3Error, S3ErrorCode, S3Result}; use crate::storage::storage_api::with_config_object_write_lock; -use s3s::{S3Error, S3ErrorCode, S3Result}; use std::sync::Arc; use super::runtime_sources::current_object_store_handle; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index f0b88002a..68f073b3b 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -903,6 +903,10 @@ pub(crate) mod runtime { pub(crate) use super::{Endpoint, Endpoints, PoolEndpoints}; } +pub(crate) mod s3 { + pub(crate) use s3s::{S3Error, S3ErrorCode, S3Result}; +} + pub(crate) mod tier { pub(crate) use super::{ AdminError, DailyAllTierStats, ERR_TIER_ALREADY_EXISTS, ERR_TIER_BACKEND_IN_USE, ERR_TIER_BACKEND_NOT_EMPTY, diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index bcbd27bf4..865b93412 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -58,7 +58,9 @@ use super::storage_api::multipart_usecase::sse::{ get_buffer_size_opt_in, load_bucket_object_lock_config_state, map_get_object_reader_error, mark_encrypted_multipart_metadata, sse_decryption, sse_prepare_encryption, }; -use super::storage_api::multipart_usecase::{StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader}; +use super::storage_api::multipart_usecase::{ + StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader, +}; use crate::app::object_data_cache::{ ObjectDataCacheAdapter, invalidate_object_data_cache_after_complete_multipart_success, invalidate_object_data_cache_after_delete_success, invalidate_object_data_cache_before_mutation, @@ -223,6 +225,18 @@ fn internal_object_info_lookup_opts(mut opts: ObjectOptions) -> ObjectOptions { opts } +fn logical_object_size(info: &ObjectInfo) -> Result { + u64::try_from(info.get_actual_size()?).map_err(|_| StorageError::PartMissingOrCorrupt) +} + +fn quota_accounting_object_size(info: &ObjectInfo, fail_closed: bool) -> S3Result { + match logical_object_size(info) { + Ok(size) => Ok(size), + Err(err) if fail_closed => Err(ApiError::from(err).into()), + Err(_) => Ok(info.size.max(0) as u64), + } +} + fn encode_s3_path(path: &str) -> String { path.split('/') .map(|part| encode(part).to_string()) @@ -488,10 +502,16 @@ impl DefaultMultipartUsecase { .await .map_err(ApiError::from)?, ); - let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { + let previous_current_sizes = match store.get_object_info(&bucket, &key, ¤t_opts).await { Ok(existing_obj_info) => { validate_existing_object_lock_for_write(&existing_obj_info, ¤t_opts)?; - Some(existing_obj_info.size.max(0) as u64) + let physical_size = existing_obj_info.size.max(0) as u64; + let logical_size = if opts.replication_request { + Ok(physical_size) + } else { + logical_object_size(&existing_obj_info) + }; + Some((physical_size, logical_size)) } Err(err) => { if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { @@ -541,9 +561,33 @@ impl DefaultMultipartUsecase { let quota_metadata_sys = self.bucket_metadata_sys(); if let Some(metadata_sys) = quota_metadata_sys.as_ref() { let quota_checker = QuotaChecker::new(metadata_sys.clone()); - map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)?; + let check_result = + map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)?; + // Ciphertext-passthrough replication parts use a different size basis and retain + // the existing post-commit accounting path until they carry a trusted logical-size proof. + if !opts.replication_request + && let Some(quota_limit) = check_result.quota_limit + { + let installed = check_result + .current_usage + .is_some_and(|current_usage| opts.set_quota_admission(current_usage, quota_limit)); + if !installed { + return Err(S3Error::with_message( + S3ErrorCode::ServiceUnavailable, + "Bucket quota check temporarily unavailable, please retry".to_string(), + )); + } + } } + let previous_current_size = match previous_current_sizes { + Some((physical_size, _)) if opts.replication_request => Some(physical_size), + Some((_, Ok(logical_size))) => Some(logical_size), + Some((_, Err(err))) if opts.quota_admission.is_some() => return Err(ApiError::from(err).into()), + Some((physical_size, Err(_))) => Some(physical_size), + None => None, + }; + let obj_info = store .clone() .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts) @@ -552,18 +596,14 @@ impl DefaultMultipartUsecase { let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await; record_capacity_write(Some(capacity_scope_token)).await; - if let Some(metadata_sys) = quota_metadata_sys { - let quota_checker = QuotaChecker::new(metadata_sys); - - match quota_checker - .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64) - .await - { - Ok(check_result) => { - if !check_result.allowed { - // Preserve the established compensation behavior for - // a known over-quota result. Unknown usage is rejected - // by the preflight check before the upload is committed. + if let Some(metadata_sys) = quota_metadata_sys.as_ref() { + if opts.replication_request { + let quota_checker = QuotaChecker::new(metadata_sys.clone()); + match quota_checker + .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64) + .await + { + Ok(check_result) if !check_result.allowed => { let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await; let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; return Err(S3Error::with_message( @@ -575,16 +615,22 @@ impl DefaultMultipartUsecase { ), )); } - } - Err(err) => { - warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err); + Err(err) => { + warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err); + } + Ok(_) => {} } } - if versioned { - record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + let committed_size = if opts.replication_request { + obj_info.size.max(0) as u64 } else { - record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())? + }; + if versioned { + record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await; + } else { + record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await; } } @@ -1591,6 +1637,37 @@ mod tests { DefaultMultipartUsecase::without_context() } + #[test] + fn quota_accounting_uses_logical_size_when_available() { + let mut metadata = HashMap::new(); + insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "S2".to_string()); + insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "8192".to_string()); + let info = ObjectInfo { + size: 128, + user_defined: Arc::new(metadata), + ..Default::default() + }; + + assert_eq!(quota_accounting_object_size(&info, true).expect("logical size should resolve"), 8192); + assert_eq!(quota_accounting_object_size(&info, false).expect("logical size should resolve"), 8192); + } + + #[test] + fn quota_accounting_fails_closed_only_when_quota_is_configured() { + let mut metadata = HashMap::new(); + insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "S2".to_string()); + insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "-1".to_string()); + let info = ObjectInfo { + size: 128, + user_defined: Arc::new(metadata), + ..Default::default() + }; + + let err = quota_accounting_object_size(&info, true).expect_err("invalid logical size must fail closed"); + assert_eq!(err.code(), &S3ErrorCode::InternalError); + assert_eq!(quota_accounting_object_size(&info, false).expect("physical fallback should resolve"), 128); + } + #[test] fn test_build_complete_multipart_location_uses_forwarded_proto_and_encodes_key() { let mut headers = HeaderMap::new(); @@ -1978,6 +2055,77 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); } + #[tokio::test] + #[serial_test::serial] + async fn compressed_complete_records_logical_quota_usage_and_overwrite_delta() { + use crate::app::storage_api::multipart_usecase::bucket::quota::BucketQuota; + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::app::storage_api::test::data_usage::seed_bucket_usage_memory_for_test; + + let store = crate::app::gating_test_env::shared_gating_ecstore().await; + crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await; + + let bucket = format!("compressed-complete-quota-{}", Uuid::new_v4()); + let object = "object"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create compressed quota bucket"); + seed_bucket_usage_memory_for_test(&bucket, 0).await; + + let usecase = DefaultMultipartUsecase::from_global(); + let metadata_sys = usecase + .bucket_metadata_sys() + .expect("test app context should expose bucket metadata"); + let mut quota_checker = QuotaChecker::new(metadata_sys); + quota_checker + .set_quota_config(&bucket, BucketQuota::new(Some(16_384))) + .await + .expect("configure bucket quota"); + + for (actual_size, payload_byte) in [(8192_i64, 0x61), (4096_i64, 0x62)] { + let mut create_opts = ObjectOptions::default(); + insert_str(&mut create_opts.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "S2".to_string()); + let upload = store + .new_multipart_upload(&bucket, object, &create_opts) + .await + .expect("create compressed multipart upload"); + let payload = vec![payload_byte; 128]; + let mut part_reader = PutObjReader::new( + HashReader::from_stream(Cursor::new(payload), 128, actual_size, None, None, false) + .expect("construct compressed part reader"), + ); + let staged_part = store + .put_object_part(&bucket, object, &upload.upload_id, 1, &mut part_reader, &ObjectOptions::default()) + .await + .expect("write compressed multipart part"); + let staged_etag = staged_part.etag.expect("staged compressed part should have an ETag"); + let input = CompleteMultipartUploadInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .upload_id(upload.upload_id) + .multipart_upload(Some(CompletedMultipartUpload { + parts: Some(vec![CompletedPart { + part_number: Some(1), + e_tag: Some(to_s3s_etag(&staged_etag)), + ..Default::default() + }]), + })) + .build() + .expect("complete multipart input should build"); + usecase + .execute_complete_multipart_upload(build_request(input, Method::POST)) + .await + .expect("compressed multipart completion should succeed"); + + let quota = quota_checker + .check_quota(&bucket, QuotaOperation::PutObject, 0) + .await + .expect("read live quota usage"); + assert_eq!(quota.current_usage, Some(actual_size as u64)); + } + } + #[tokio::test] #[serial_test::serial] async fn rejected_empty_parts_preserve_existing_object_and_staging() { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 91579de7d..cfb5c5493 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -67,7 +67,7 @@ pub(crate) mod data_usage { // Test-only observables for the rustfs/backlog#1306 revert detector. #[cfg(test)] pub(crate) use crate::storage::storage_api::ecstore_data_usage::{ - compute_bucket_usage, live_bucket_usage_computations, store_data_usage_in_backend, + compute_bucket_usage, live_bucket_usage_computations, seed_bucket_usage_memory_for_test, store_data_usage_in_backend, }; pub(crate) async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64, removed_current_object: bool) { @@ -597,6 +597,8 @@ pub(crate) mod bucket { pub(crate) type QuotaChecker = crate::storage::storage_api::ecstore_bucket::quota::checker::QuotaChecker; } + #[cfg(test)] + pub(crate) type BucketQuota = crate::storage::storage_api::ecstore_bucket::quota::BucketQuota; pub(crate) type QuotaOperation = crate::storage::storage_api::ecstore_bucket::quota::QuotaOperation; pub(crate) type QuotaCheckResult = crate::storage::storage_api::ecstore_bucket::quota::QuotaCheckResult; pub(crate) type QuotaError = crate::storage::storage_api::ecstore_bucket::quota::QuotaError; @@ -1149,7 +1151,7 @@ pub(crate) mod multipart_usecase { } pub(crate) use super::{access, bucket, data_usage, error, helper, io, object_utils, options, s3_api, set_disk, sse}; - pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectOptions, StoragePutObjReader}; + pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader}; } pub(crate) mod select_object { diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 7d27a5241..49ac36319 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -298,6 +298,7 @@ impl From for ApiError { | StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::SlowDown, StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable, + StorageError::QuotaExceeded { .. } => S3ErrorCode::InvalidRequest, StorageError::Lock(_) => S3ErrorCode::ServiceUnavailable, StorageError::DecommissionNotStarted => S3ErrorCode::InvalidRequest, StorageError::DecommissionAlreadyRunning => S3ErrorCode::InvalidRequest, @@ -325,7 +326,7 @@ impl From for ApiError { _ => S3ErrorCode::InternalError, }; - let message = if code == S3ErrorCode::InternalError { + let message = if matches!(&err, StorageError::QuotaExceeded { .. }) || code == S3ErrorCode::InternalError { err.to_string() } else if let StorageError::InvalidArgument(_, _, reason) = &err && !reason.is_empty() @@ -623,6 +624,7 @@ mod tests { (StorageError::DecommissionNotStarted, S3ErrorCode::InvalidRequest), (StorageError::DecommissionAlreadyRunning, S3ErrorCode::InvalidRequest), (StorageError::RebalanceAlreadyRunning, S3ErrorCode::InvalidRequest), + (StorageError::QuotaExceeded { current: 5, limit: 10 }, S3ErrorCode::InvalidRequest), (StorageError::PrefixAccessDenied("test".into(), "test".into()), S3ErrorCode::AccessDenied), (StorageError::ObjectNotFound("test".into(), "test".into()), S3ErrorCode::NoSuchKey), (StorageError::ConfigNotFound, S3ErrorCode::NoSuchKey), @@ -708,6 +710,14 @@ mod tests { assert_eq!(s3_error.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE)); } + #[test] + fn quota_exceeded_preserves_existing_s3_error_contract() { + let api_error: ApiError = StorageError::QuotaExceeded { current: 5, limit: 10 }.into(); + + assert_eq!(api_error.code, S3ErrorCode::InvalidRequest); + assert_eq!(api_error.message, "Bucket quota exceeded. Current usage: 5 bytes, limit: 10 bytes"); + } + #[test] fn test_api_error_to_s3_error_without_source() { let api_error = ApiError { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index b765b9069..d5dfc3af5 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -435,7 +435,8 @@ pub(crate) mod ecstore_data_usage { // Test-only observables for the rustfs/backlog#1306 revert detector. #[cfg(test)] pub(crate) use rustfs_ecstore::api::data_usage::{ - compute_bucket_usage, live_bucket_usage_computations, load_data_usage_from_backend_cached, store_data_usage_in_backend, + compute_bucket_usage, live_bucket_usage_computations, load_data_usage_from_backend_cached, + seed_bucket_usage_memory_for_test, store_data_usage_in_backend, }; }