fix(quota): reject oversized multipart completion (#5958)

* fix(quota): reject oversized multipart completion

* fix(arch): route quota test through app facade

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(())

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,

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<u64>, new_size: u64) {
     record_bucket_object_write_memory_inner(bucket, previous_current_size, new_size, false).await;

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 {

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<Checksum>,
     pub skip_verify_bitrot: bool,
     pub capacity_scope_token: Option<Uuid>,
+    /// Server-derived bucket-quota snapshot for commit-boundary admission.
+    pub quota_admission: Option<QuotaAdmission>,
     /// 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<Arc<crate::store::ECStore>>,
 }
 
 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
     }

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<SetDisks>,
+        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) {

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)] = &[

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;

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,

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, StorageError> {
+    u64::try_from(info.get_actual_size()?).map_err(|_| StorageError::PartMissingOrCorrupt)
+}
+
+fn quota_accounting_object_size(info: &ObjectInfo, fail_closed: bool) -> S3Result<u64> {
+    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, &current_opts).await {
+        let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
             Ok(existing_obj_info) => {
                 validate_existing_object_lock_for_write(&existing_obj_info, &current_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() {

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 {

rustfs/src/error.rs

@@ -298,6 +298,7 @@ impl From<StorageError> 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<StorageError> 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 {

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,
     };
 }