fix(swift): enforce cumulative container quotas (#5378)

crates/protocols/src/swift/container.rs

@@ -22,12 +22,19 @@ use super::storage_api::container::{
 };
 use super::types::Container;
 use super::{SwiftError, SwiftResult};
-use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata};
+use super::{get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata};
 use rustfs_credentials::Credentials;
 use s3s::dto::{Tag, Tagging};
 use sha2::{Digest, Sha256};
 use tracing::{debug, error};
 
+fn required_bucket_usage(usage: &std::collections::HashMap<String, (u64, u64)>, bucket: &str) -> SwiftResult<(u64, u64)> {
+    usage
+        .get(bucket)
+        .copied()
+        .ok_or_else(|| SwiftError::ServiceUnavailable("Container usage is unavailable".to_string()))
+}
+
 const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
 const LOG_SUBSYSTEM_SWIFT_CONTAINER: &str = "swift_container";
 const EVENT_SWIFT_CONTAINER_STORAGE_STATE: &str = "swift_container_storage_state";
@@ -252,13 +259,24 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
         .list_bucket(&BucketOptions::default())
         .await
         .map_err(|e| sanitize_storage_error("Container listing", e))?;
+    let usage = get_swift_bucket_usage()
+        .await
+        .map_err(|e| sanitize_storage_error("Container usage retrieval", e))?
+        .ok_or_else(|| SwiftError::ServiceUnavailable("Container usage is unavailable".to_string()))?;
 
     // Filter and convert buckets to containers
     let containers: Vec<Container> = bucket_infos
         .iter()
         .filter(|info| mapper.bucket_belongs_to_project(&info.name, &project_id))
-        .filter_map(|info| bucket_info_to_container(info, &mapper, &project_id))
-        .collect();
+        .filter_map(|info| {
+            bucket_info_to_container(info, &mapper, &project_id).map(|mut container| {
+                let (count, bytes) = required_bucket_usage(&usage, &info.name)?;
+                container.count = count;
+                container.bytes = bytes;
+                Ok(container)
+            })
+        })
+        .collect::<SwiftResult<Vec<_>>>()?;
 
     debug!(
         event = EVENT_SWIFT_CONTAINER_STORAGE_STATE,
@@ -368,6 +386,44 @@ pub struct ContainerMetadata {
     pub custom_metadata: std::collections::HashMap<String, String>,
 }
 
+async fn get_container_metadata_base(
+    account: &str,
+    container: &str,
+    credentials: &Credentials,
+) -> SwiftResult<(String, BucketInfo, std::collections::HashMap<String, String>)> {
+    let project_id = validate_account_access(account, credentials)?;
+    validate_container_name(container)?;
+    let bucket_name = ContainerMapper::default().swift_to_s3_bucket(container, &project_id);
+    let Some(store) = resolve_swift_object_store_handle() else {
+        return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
+    };
+    let bucket_info = store
+        .get_bucket_info(&bucket_name, &BucketOptions::default())
+        .await
+        .map_err(|e| {
+            if e.to_string().contains("not found") || e.to_string().contains("NoSuchBucket") {
+                SwiftError::NotFound(format!("Container '{}' not found", container))
+            } else {
+                sanitize_storage_error("Container metadata retrieval", e)
+            }
+        })?;
+    let custom_metadata = get_swift_bucket_metadata(&bucket_name)
+        .await
+        .ok()
+        .and_then(|bucket_meta| bucket_meta.tagging_config.as_ref().map(s3_tags_to_swift_metadata))
+        .unwrap_or_default();
+    Ok((bucket_name, bucket_info, custom_metadata))
+}
+
+pub(crate) async fn get_container_custom_metadata(
+    account: &str,
+    container: &str,
+    credentials: &Credentials,
+) -> SwiftResult<std::collections::HashMap<String, String>> {
+    let (_, _, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
+    Ok(custom_metadata)
+}
+
 /// Get container metadata (for HEAD operation)
 ///
 /// This function:
@@ -382,60 +438,22 @@ pub struct ContainerMetadata {
 /// - Returns 404 Not Found if container doesn't exist
 #[allow(dead_code)] // Used by handler
 pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<ContainerMetadata> {
-    // Validate account access and extract project_id
-    let project_id = validate_account_access(account, credentials)?;
+    let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
 
-    // Validate container name
-    validate_container_name(container)?;
-
-    // Create mapper with default config (tenant prefixing enabled)
-    let mapper = ContainerMapper::default();
-
-    // Convert Swift container name to S3 bucket name
-    let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
-
-    // Get storage layer
-    let Some(store) = resolve_swift_object_store_handle() else {
-        return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
-    };
-
-    // Get bucket info
-    let bucket_info = store
-        .get_bucket_info(&bucket_name, &BucketOptions::default())
+    // Ecstore serves the persisted scanner snapshot through a bounded in-process cache.
+    // Single-node deployments overlay successful in-process mutations; distributed
+    // deployments use only the cluster-wide persisted snapshot.
+    let usage = get_swift_bucket_usage()
         .await
-        .map_err(|e| {
-            // Check if bucket not found
-            if e.to_string().contains("not found") || e.to_string().contains("NoSuchBucket") {
-                SwiftError::NotFound(format!("Container '{}' not found", container))
-            } else {
-                sanitize_storage_error("Container metadata retrieval", e)
-            }
-        })?;
+        .map_err(|e| sanitize_storage_error("Container usage retrieval", e))?
+        .ok_or_else(|| SwiftError::ServiceUnavailable("Container usage is unavailable".to_string()))?;
+    let (object_count, bytes_used) = required_bucket_usage(&usage, &bucket_name)?;
 
-    // Load bucket metadata to get custom metadata from tags
-    let custom_metadata = match get_swift_bucket_metadata(&bucket_name).await {
-        Ok(bucket_meta) => {
-            if let Some(tagging) = &bucket_meta.tagging_config {
-                s3_tags_to_swift_metadata(tagging)
-            } else {
-                std::collections::HashMap::new()
-            }
-        }
-        Err(_) => {
-            // If metadata not available, return empty (container may be newly created)
-            std::collections::HashMap::new()
-        }
-    };
-
-    // Currently returns basic metadata with limitations:
-    // 1. Object count requires iterating all objects (expensive)
-    // 2. Bytes used requires summing all object sizes (expensive)
-    // 3. Custom metadata is now loaded from bucket tags ✅
     Ok(ContainerMetadata {
-        object_count: 0, // TODO: implement object counting in backend
-        bytes_used: 0,   // TODO: implement size aggregation in backend
+        object_count,
+        bytes_used,
         created: bucket_info.created,
-        custom_metadata, // ✅ Now populated from bucket tags!
+        custom_metadata,
     })
 }
 
@@ -1843,4 +1861,24 @@ mod tests {
         assert_eq!(tagging.tag_set.len(), 1);
         assert_eq!(tagging.tag_set[0].value.as_deref(), Some("new-archive"));
     }
+
+    #[test]
+    fn nonempty_container_usage_preserves_authoritative_totals() {
+        let usage = std::collections::HashMap::from([("tenant-container".to_string(), (7, 4097))]);
+
+        assert_eq!(
+            required_bucket_usage(&usage, "tenant-container").expect("authoritative bucket usage"),
+            (7, 4097)
+        );
+    }
+
+    #[test]
+    fn missing_container_usage_fails_closed() {
+        let usage = std::collections::HashMap::new();
+
+        assert!(matches!(
+            required_bucket_usage(&usage, "tenant-container"),
+            Err(SwiftError::ServiceUnavailable(_))
+        ));
+    }
 }

crates/protocols/src/swift/errors.rs

@@ -32,6 +32,8 @@ pub enum SwiftError {
     NotFound(String),
     /// 409 Conflict
     Conflict(String),
+    /// 411 Length Required
+    LengthRequired(String),
     /// 413 Request Entity Too Large (Payload Too Large)
     RequestEntityTooLarge(String),
     /// 422 Unprocessable Entity
@@ -54,6 +56,7 @@ impl fmt::Display for SwiftError {
             SwiftError::Forbidden(msg) => write!(f, "Forbidden: {}", msg),
             SwiftError::NotFound(msg) => write!(f, "Not Found: {}", msg),
             SwiftError::Conflict(msg) => write!(f, "Conflict: {}", msg),
+            SwiftError::LengthRequired(msg) => write!(f, "Length Required: {}", msg),
             SwiftError::RequestEntityTooLarge(msg) => write!(f, "Request Entity Too Large: {}", msg),
             SwiftError::UnprocessableEntity(msg) => write!(f, "Unprocessable Entity: {}", msg),
             SwiftError::TooManyRequests { retry_after, .. } => {
@@ -76,6 +79,7 @@ impl SwiftError {
             SwiftError::Forbidden(_) => StatusCode::FORBIDDEN,
             SwiftError::NotFound(_) => StatusCode::NOT_FOUND,
             SwiftError::Conflict(_) => StatusCode::CONFLICT,
+            SwiftError::LengthRequired(_) => StatusCode::LENGTH_REQUIRED,
             SwiftError::RequestEntityTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE,
             SwiftError::UnprocessableEntity(_) => StatusCode::UNPROCESSABLE_ENTITY,
             SwiftError::TooManyRequests { .. } => StatusCode::TOO_MANY_REQUESTS,

crates/protocols/src/swift/handler.rs

@@ -652,14 +652,11 @@ async fn handle_authenticated_request(
                             .await;
                     }
 
-                    // Check quota before upload (if Content-Length provided)
-                    if let Some(content_length) = headers.get("content-length")
-                        && let Ok(size_str) = content_length.to_str()
-                        && let Ok(object_size) = size_str.parse::<u64>()
-                    {
-                        // Check if upload would exceed quota
-                        super::quota::check_upload_quota(&account, &container, object_size, &credentials).await?;
-                    }
+                    let object_size = headers
+                        .get("content-length")
+                        .and_then(|value| value.to_str().ok())
+                        .and_then(|value| value.parse::<u64>().ok());
+                    super::quota::check_upload_quota(&account, &container, object_size, &credentials).await?;
 
                     // Check if versioning is enabled for this container
                     if let Some(archive_container) = container::get_versions_location(&account, &container, &credentials).await? {
@@ -1486,6 +1483,7 @@ fn swift_error_to_response(error: SwiftError) -> Response<Body> {
         SwiftError::Forbidden(msg) => (StatusCode::FORBIDDEN, msg.as_str()),
         SwiftError::NotFound(msg) => (StatusCode::NOT_FOUND, msg.as_str()),
         SwiftError::Conflict(msg) => (StatusCode::CONFLICT, msg.as_str()),
+        SwiftError::LengthRequired(msg) => (StatusCode::LENGTH_REQUIRED, msg.as_str()),
         SwiftError::RequestEntityTooLarge(msg) => (StatusCode::PAYLOAD_TOO_LARGE, msg.as_str()),
         SwiftError::UnprocessableEntity(msg) => (StatusCode::UNPROCESSABLE_ENTITY, msg.as_str()),
         SwiftError::TooManyRequests { .. } => (StatusCode::TOO_MANY_REQUESTS, "Rate limit exceeded"),

crates/protocols/src/swift/mod.rs

@@ -61,7 +61,7 @@ pub use router::{SwiftRoute, SwiftRouter};
 // Note: Container, Object, and SwiftMetadata types used by Swift implementation
 pub use storage_api::public_api::{SwiftGetObjectReader, SwiftObjectInfo, SwiftObjectOptions, SwiftPutObjReader};
 pub(crate) use storage_api::public_api::{
-    get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata,
+    get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata,
 };
 #[allow(unused_imports)]
 pub use types::{Container, Object, SwiftMetadata};

crates/protocols/src/swift/quota.rs

@@ -87,18 +87,17 @@ pub struct QuotaConfig {
 impl QuotaConfig {
     /// Load quota configuration from container metadata
     pub async fn load(account: &str, container_name: &str, credentials: &Credentials) -> SwiftResult<Self> {
-        // Get container metadata
-        let container_info = container::get_container_metadata(account, container_name, credentials).await?;
+        let custom_metadata = container::get_container_custom_metadata(account, container_name, credentials).await?;
 
         let mut config = QuotaConfig::default();
 
         // Parse Quota-Bytes
-        if let Some(quota_bytes_str) = container_info.custom_metadata.get("x-container-meta-quota-bytes") {
+        if let Some(quota_bytes_str) = custom_metadata.get("x-container-meta-quota-bytes") {
             config.quota_bytes = quota_bytes_str.parse().ok();
         }
 
         // Parse Quota-Count
-        if let Some(quota_count_str) = container_info.custom_metadata.get("x-container-meta-quota-count") {
+        if let Some(quota_count_str) = custom_metadata.get("x-container-meta-quota-count") {
             config.quota_count = quota_count_str.parse().ok();
         }
 
@@ -113,9 +112,12 @@ impl QuotaConfig {
     /// Check if adding an object would exceed quotas
     ///
     /// Returns Ok(()) if within quota, Err with 413 if exceeded
-    pub fn check_quota(&self, current_bytes: u64, current_count: u64, additional_bytes: u64) -> SwiftResult<()> {
+    pub fn check_quota(&self, current_bytes: u64, current_count: u64, additional_bytes: Option<u64>) -> SwiftResult<()> {
         // Check byte quota
         if let Some(max_bytes) = self.quota_bytes {
+            let additional_bytes = additional_bytes.ok_or_else(|| {
+                SwiftError::LengthRequired("Content-Length is required when a byte quota is configured".to_string())
+            })?;
             let new_bytes = current_bytes.saturating_add(additional_bytes);
             if new_bytes > max_bytes {
                 return Err(SwiftError::RequestEntityTooLarge(format!(
@@ -147,7 +149,7 @@ impl QuotaConfig {
 pub async fn check_upload_quota(
     account: &str,
     container_name: &str,
-    object_size: u64,
+    object_size: Option<u64>,
     credentials: &Credentials,
 ) -> SwiftResult<()> {
     // Load quota config
@@ -157,7 +159,6 @@ pub async fn check_upload_quota(
     if !quota.is_enabled() {
         return Ok(());
     }
-
     // Get current container usage
     let metadata = container::get_container_metadata(account, container_name, credentials).await?;
 
@@ -173,7 +174,7 @@ pub async fn check_upload_quota(
         quota_bytes = ?quota.quota_bytes,
         current_count = metadata.object_count,
         quota_count = ?quota.quota_count,
-        object_size,
+        object_size = ?object_size,
         "swift quota state changed"
     );
 
@@ -235,7 +236,7 @@ mod tests {
         };
 
         // Current: 500 bytes, adding 400 bytes = 900 total (within 1000 limit)
-        let result = config.check_quota(500, 0, 400);
+        let result = config.check_quota(500, 0, Some(400));
         assert!(result.is_ok());
     }
 
@@ -247,7 +248,7 @@ mod tests {
         };
 
         // Current: 500 bytes, adding 600 bytes = 1100 total (exceeds 1000 limit)
-        let result = config.check_quota(500, 0, 600);
+        let result = config.check_quota(500, 0, Some(600));
         assert!(result.is_err());
         match result {
             Err(SwiftError::RequestEntityTooLarge(msg)) => {
@@ -265,7 +266,7 @@ mod tests {
         };
 
         // Current: 500 bytes, adding 500 bytes = 1000 total (exactly at limit)
-        let result = config.check_quota(500, 0, 500);
+        let result = config.check_quota(500, 0, Some(500));
         assert!(result.is_ok());
     }
 
@@ -277,7 +278,7 @@ mod tests {
         };
 
         // Current: 5 objects, adding 1 = 6 total (within 10 limit)
-        let result = config.check_quota(0, 5, 100);
+        let result = config.check_quota(0, 5, Some(100));
         assert!(result.is_ok());
     }
 
@@ -289,7 +290,7 @@ mod tests {
         };
 
         // Current: 10 objects, adding 1 = 11 total (exceeds 10 limit)
-        let result = config.check_quota(0, 10, 100);
+        let result = config.check_quota(0, 10, Some(100));
         assert!(result.is_err());
         match result {
             Err(SwiftError::RequestEntityTooLarge(msg)) => {
@@ -307,7 +308,7 @@ mod tests {
         };
 
         // Current: 9 objects, adding 1 = 10 total (exactly at limit)
-        let result = config.check_quota(0, 9, 100);
+        let result = config.check_quota(0, 9, Some(100));
         assert!(result.is_ok());
     }
 
@@ -319,7 +320,7 @@ mod tests {
         };
 
         // Both within limits
-        let result = config.check_quota(500, 5, 400);
+        let result = config.check_quota(500, 5, Some(400));
         assert!(result.is_ok());
     }
 
@@ -331,7 +332,7 @@ mod tests {
         };
 
         // Bytes exceeded, count within
-        let result = config.check_quota(500, 5, 600);
+        let result = config.check_quota(500, 5, Some(600));
         assert!(result.is_err());
     }
 
@@ -343,7 +344,7 @@ mod tests {
         };
 
         // Count exceeded, bytes within
-        let result = config.check_quota(500, 10, 100);
+        let result = config.check_quota(500, 10, Some(100));
         assert!(result.is_err());
     }
 
@@ -355,7 +356,7 @@ mod tests {
         };
 
         // No limits, should always pass
-        let result = config.check_quota(999999, 999999, 999999);
+        let result = config.check_quota(999999, 999999, Some(999999));
         assert!(result.is_ok());
     }
 
@@ -367,7 +368,7 @@ mod tests {
         };
 
         // Zero limit means no uploads allowed
-        let result = config.check_quota(0, 0, 1);
+        let result = config.check_quota(0, 0, Some(1));
         assert!(result.is_err());
     }
 
@@ -379,7 +380,7 @@ mod tests {
         };
 
         // Zero limit means no objects allowed
-        let result = config.check_quota(0, 0, 100);
+        let result = config.check_quota(0, 0, Some(100));
         assert!(result.is_err());
     }
 
@@ -391,8 +392,28 @@ mod tests {
         };
 
         // Test saturating_add protection
-        let result = config.check_quota(u64::MAX - 100, 0, 200);
+        let result = config.check_quota(u64::MAX - 100, 0, Some(200));
         // Should saturate to u64::MAX and compare against quota
         assert!(result.is_ok());
     }
+
+    #[test]
+    fn byte_quota_rejects_upload_without_content_length() {
+        let config = QuotaConfig {
+            quota_bytes: Some(1000),
+            quota_count: None,
+        };
+
+        assert!(matches!(config.check_quota(500, 0, None), Err(SwiftError::LengthRequired(_))));
+    }
+
+    #[test]
+    fn count_only_quota_checks_upload_without_content_length() {
+        let config = QuotaConfig {
+            quota_bytes: None,
+            quota_count: Some(10),
+        };
+
+        assert!(matches!(config.check_quota(0, 10, None), Err(SwiftError::RequestEntityTooLarge(_))));
+    }
 }

crates/protocols/src/swift/storage_api.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 pub(crate) use rustfs_ecstore::api::bucket::metadata::BucketMetadata as SwiftBucketMetadata;
@@ -43,7 +44,9 @@ pub(crate) mod object {
 
 pub(crate) mod public_api {
     pub use super::{SwiftGetObjectReader, SwiftObjectInfo, SwiftObjectOptions, SwiftPutObjReader};
-    pub(crate) use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata};
+    pub(crate) use super::{
+        get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata,
+    };
 }
 
 pub(crate) mod versioning {
@@ -62,3 +65,18 @@ pub(crate) async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResul
 pub(crate) async fn set_swift_bucket_metadata(bucket: String, metadata: SwiftBucketMetadata) -> SwiftStorageResult<()> {
     set_swift_bucket_metadata_in_backend(bucket, metadata).await
 }
+
+pub(crate) async fn get_swift_bucket_usage() -> SwiftStorageResult<Option<HashMap<String, (u64, u64)>>> {
+    let Some(store) = resolve_swift_object_store_handle() else {
+        return Ok(None);
+    };
+    let mut data_usage = rustfs_ecstore::api::data_usage::load_data_usage_from_backend_cached(store).await?;
+    rustfs_ecstore::api::data_usage::apply_bucket_usage_memory_overlay(&mut data_usage).await;
+    Ok(Some(
+        data_usage
+            .buckets_usage
+            .into_iter()
+            .map(|(bucket, usage)| (bucket, (usage.objects_count, usage.size)))
+            .collect(),
+    ))
+}