test(filemeta): cover legacy delete marker decoding (#2333)

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

rustfs/src/app/object_usecase.rs

@@ -1631,13 +1631,13 @@ impl DefaultObjectUsecase {
     #[instrument(level = "debug", skip(self, _fs, req))]
     pub async fn execute_put_object(&self, _fs: &FS, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
         let start_time = std::time::Instant::now();
+        let mut req = req;
 
         if let Some(context) = &self.context {
             let _ = context.object_store();
         }
 
         let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req);
-        let mut helper = OperationHelper::new(&req, event_name, S3Operation::PutObject);
         if req.extensions.get::<PostObjectRequestMarker>().is_some() && is_post_object_sse_kms_requested(&req.input, &req.headers)
         {
             return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for POST object uploads"));
@@ -1651,7 +1651,7 @@ impl DefaultObjectUsecase {
             return self.execute_put_object_extract(req).await;
         }
 
-        let input = req.input;
+        let input = std::mem::take(&mut req.input);
 
         let PutObjectInput {
             body,
@@ -1870,6 +1870,8 @@ impl DefaultObjectUsecase {
             opts.want_checksum = reader.checksum();
         }
 
+        let mut helper = OperationHelper::new(&req, event_name, S3Operation::PutObject);
+
         // Apply encryption using unified SSE API.
         let encryption_request = EncryptionRequest {
             bucket: &bucket,
@@ -1885,7 +1887,16 @@ impl DefaultObjectUsecase {
             part_nonce: None,
         };
 
-        if let Some(material) = sse_encryption(encryption_request).await? {
+        let encryption_material = match sse_encryption(encryption_request).await {
+            Ok(material) => material,
+            Err(err) => {
+                let result = Err(err.into());
+                let _ = helper.complete(&result);
+                return result;
+            }
+        };
+
+        if let Some(material) = encryption_material {
             effective_sse = Some(material.server_side_encryption.clone());
             effective_kms_key_id = material.kms_key_id.clone();
 
@@ -1917,10 +1928,18 @@ impl DefaultObjectUsecase {
             );
         }
 
-        let obj_info = store
+        let obj_info = match store
             .put_object(&bucket, &key, &mut reader, &opts)
             .await
-            .map_err(ApiError::from)?;
+            .map_err(ApiError::from)
+        {
+            Ok(obj_info) => obj_info,
+            Err(err) => {
+                let result: S3Result<S3Response<PutObjectOutput>> = Err(err.into());
+                let _ = helper.complete(&result);
+                return result;
+            }
+        };
 
         maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await;
 

rustfs/src/capacity/capacity_manager.rs

@@ -15,6 +15,7 @@
 //! Hybrid Capacity Manager for efficient capacity statistics
 
 use crate::app::admin_usecase::calculate_data_dir_used_capacity;
+use futures::FutureExt;
 use rustfs_config::{
     DEFAULT_CAPACITY_ENABLE_DYNAMIC_TIMEOUT, DEFAULT_CAPACITY_FOLLOW_SYMLINKS, DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH,
     DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, DEFAULT_CAPACITY_STALL_TIMEOUT_SECS,
@@ -28,6 +29,7 @@ use rustfs_config::{
 use rustfs_io_metrics::{record_capacity_current_bytes, record_capacity_update_completed, record_capacity_write_operation};
 use rustfs_utils::{get_env_bool, get_env_u64, get_env_usize};
 use std::future::Future;
+use std::panic::AssertUnwindSafe;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::sync::{Mutex, RwLock, watch};
@@ -601,7 +603,10 @@ impl HybridCapacityManager {
                 .unwrap_or_else(|| Err("capacity refresh completed without a result".to_string()));
         }
 
-        let result = refresh_fn().await;
+        let result = AssertUnwindSafe(refresh_fn()).catch_unwind().await.unwrap_or_else(|err| {
+            warn!(error = ?err, "capacity refresh function panicked");
+            Err("capacity refresh panicked".to_string())
+        });
         if let Ok(update) = &result {
             self.update_capacity(update.clone(), source).await;
         }
@@ -638,7 +643,10 @@ impl HybridCapacityManager {
         }
 
         tokio::spawn(async move {
-            let result = refresh_fn().await;
+            let result = AssertUnwindSafe(refresh_fn()).catch_unwind().await.unwrap_or_else(|err| {
+                warn!(error = ?err, "capacity refresh function panicked");
+                Err("capacity refresh panicked".to_string())
+            });
             if let Ok(update) = &result {
                 self.update_capacity(update.clone(), source).await;
             }