fix(quota): account compressed deletes by committed size (#6365)

crates/ecstore/src/set_disk/ops/object.rs

@@ -45,6 +45,7 @@ use crate::bucket::replication::{
     DeleteReplicationConfigSnapshot, ReplicationLifecycleBridge, ReplicationStatusType, VersionPurgeStatusType,
     replication_state_to_filemeta, replication_status_from_filemeta, version_purge_status_to_filemeta,
 };
+use crate::data_usage::quota_object_size;
 use crate::diagnostics::get::GetObjectFailureReason;
 use crate::disk::{DataDirDeleteStatus, OldCurrentSize};
 use crate::error::is_err_invalid_upload_id;
@@ -5655,7 +5656,18 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         objects: Vec<ObjectToDelete>,
         opts: ObjectOptions,
     ) -> (Vec<DeletedObject>, Vec<Option<Error>>) {
+        let (deleted, errors, _) = self.delete_objects_with_accounting(bucket, objects, opts).await;
+        (deleted, errors)
+    }
+
+    async fn delete_objects_with_accounting(
+        &self,
+        bucket: &str,
+        objects: Vec<ObjectToDelete>,
+        opts: ObjectOptions,
+    ) -> (Vec<DeletedObject>, Vec<Option<Error>>, Vec<Option<DeleteAccounting>>) {
         let mut del_objects = vec![DeletedObject::default(); objects.len()];
+        let mut accounting = vec![None; objects.len()];
         let delete_config_snapshot = opts
             .delete_replication_config_snapshot
             .clone()
@@ -5745,7 +5757,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                                 *item = Some(Error::other(message.clone()));
                             }
                         }
-                        return (del_objects, del_errs);
+                        return (del_objects, del_errs, accounting);
                     }
                 },
             }
@@ -5792,6 +5804,22 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             let source_missing = gerr
                 .as_ref()
                 .is_some_and(|err| is_err_object_not_found(err) || is_err_version_not_found(err));
+            // Resolve accounting from the generation selected under this
+            // object's write lock. A request-layer pre-stat is only an
+            // optimization and cannot identify a concurrent overwrite.
+            let (accounting_size, accounting_version_id, removed_current_object) = if source_missing
+                || dobj.synthetic_version_id
+                || set_disk_delete_creates_delete_marker(&check_opts)
+                || goi.delete_marker
+            {
+                (None, None, false)
+            } else {
+                (
+                    quota_object_size(&goi).ok(),
+                    goi.version_id.filter(|version_id| !version_id.is_nil()),
+                    (dobj.version_id.is_none() || is_explicit_null_version(dobj.version_id)) && !dobj.synthetic_version_id,
+                )
+            };
             // Normalize both sides before comparing. `goi.version_id` is the
             // client-facing identity, where `from_file_info` synthesizes
             // `Some(Uuid::nil())` for a null version on a versioned or
@@ -5920,7 +5948,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                     },
                     replication_state: vr.replication_state_internal.clone(),
                     ..Default::default()
-                }
+                };
+                accounting[i] = Some(DeleteAccounting {
+                    size: accounting_size,
+                    version_id: accounting_version_id,
+                    removed_current_object,
+                });
             }
 
             // Only add to vers_map if we hold the lock
@@ -5966,7 +5999,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                     });
                 }
             }
-            return (del_objects, del_errs);
+            return (del_objects, del_errs, accounting);
         }
 
         let mut persisted_journal_entries = Vec::with_capacity(journal_entries.len());
@@ -6204,7 +6237,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         }
 
-        (del_objects, del_errs)
+        // An accounting identity is actionable only when the delete result is
+        // successful. Never let a failed commit (including a partial quorum
+        // failure) reach the request-layer fast delta path.
+        for (index, err) in del_errs.iter().enumerate() {
+            if err.is_some() {
+                accounting[index] = None;
+            }
+        }
+
+        (del_objects, del_errs, accounting)
     }
 
     #[tracing::instrument(skip(self))]
@@ -6533,6 +6575,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
 
         let mut obj_info = ObjectInfo::from_file_info(&dfi, bucket, object, opts.versioned || opts.version_suspended);
         obj_info.size = goi.size;
+        // Keep the committed source metadata on the internal delete result so
+        // the request layer can derive canonical accounting for this exact
+        // generation. Delete responses do not expose these fields.
+        obj_info.actual_size = goi.actual_size;
+        obj_info.user_defined = Arc::clone(&goi.user_defined);
+        obj_info.parts = Arc::clone(&goi.parts);
         obj_info.user_tags = Arc::clone(&goi.user_tags);
         self.invalidate_get_object_metadata_cache(bucket, object).await;
         Ok(obj_info)
@@ -7824,6 +7872,113 @@ mod replication_quota_safety_tests {
         assert_eq!(stored.get_actual_size().expect("stored logical size should parse"), 1);
     }
 
+    #[tokio::test]
+    async fn delete_returns_canonical_compressed_accounting_size() {
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "compressed-delete-accounting";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut user_defined = HashMap::new();
+        insert_str(
+            &mut user_defined,
+            rustfs_utils::http::SUFFIX_COMPRESSION,
+            "klauspost/compress/s2".to_string(),
+        );
+        insert_str(&mut user_defined, SUFFIX_ACTUAL_SIZE, "1000".to_string());
+        let mut reader = PutObjReader::new(
+            HashReader::from_stream(Cursor::new(vec![0x5a; 400]), 400, 1000, None, None, false)
+                .expect("compressed fixture reader should be valid"),
+        );
+        set_disks
+            .put_object(
+                bucket,
+                "object",
+                &mut reader,
+                &ObjectOptions {
+                    user_defined,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("compressed object should be written");
+
+        let (deleted, errors, accounting) = set_disks
+            .delete_objects_with_accounting(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: "object".to_string(),
+                    ..Default::default()
+                }],
+                ObjectOptions {
+                    object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                        ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await;
+
+        assert!(errors[0].is_none(), "compressed delete should succeed: {:?}", errors[0]);
+        assert!(deleted[0].found, "the committed object must be reported as found");
+        assert_eq!(accounting[0].as_ref().and_then(|value| value.size), Some(1000));
+        assert!(accounting[0].as_ref().is_some_and(|value| value.version_id.is_none()));
+        assert!(accounting[0].as_ref().is_some_and(|value| value.removed_current_object));
+    }
+
+    #[tokio::test]
+    async fn suspended_delete_marker_does_not_return_body_accounting() {
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "suspended-delete-accounting";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut user_defined = HashMap::new();
+        insert_str(
+            &mut user_defined,
+            rustfs_utils::http::SUFFIX_COMPRESSION,
+            "klauspost/compress/s2".to_string(),
+        );
+        insert_str(&mut user_defined, SUFFIX_ACTUAL_SIZE, "1000".to_string());
+        let mut reader = PutObjReader::new(
+            HashReader::from_stream(Cursor::new(vec![0x5a; 400]), 400, 1000, None, None, false)
+                .expect("compressed fixture reader should be valid"),
+        );
+        let suspended_opts = ObjectOptions {
+            version_suspended: true,
+            delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::from_configs_for_test(
+                s3s::dto::VersioningConfiguration {
+                    status: Some(s3s::dto::BucketVersioningStatus::from_static(s3s::dto::BucketVersioningStatus::SUSPENDED)),
+                    ..Default::default()
+                },
+                None,
+            ))),
+            user_defined,
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
+            ..Default::default()
+        };
+        set_disks
+            .put_object(bucket, "object", &mut reader, &suspended_opts)
+            .await
+            .expect("compressed object should be written");
+
+        let (deleted, errors, accounting) = set_disks
+            .delete_objects_with_accounting(
+                bucket,
+                vec![ObjectToDelete {
+                    object_name: "object".to_string(),
+                    ..Default::default()
+                }],
+                suspended_opts,
+            )
+            .await;
+        assert!(errors[0].is_none(), "suspended delete should create a marker: {:?}", errors[0]);
+        assert!(deleted[0].delete_marker);
+        assert!(accounting[0].is_none(), "a delete marker must not carry body accounting");
+    }
+
     #[tokio::test]
     async fn direct_put_cannot_persist_a_tiny_logical_size() {
         let (_temp_dirs, disks, set_disks) = hermetic_set_disks(4).await;