fix: preserve data on self metadata copy (#2819)

Co-authored-by: cxymds <Cxymds@qq.com>

crates/e2e_test/src/copy_object_metadata_test.rs

@@ -0,0 +1,144 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! CopyObject metadata replacement regression tests.
+
+#[cfg(test)]
+mod tests {
+    use crate::common::{RustFSTestEnvironment, init_logging};
+    use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::types::MetadataDirective;
+    use serial_test::serial;
+    use tracing::info;
+
+    #[tokio::test]
+    #[serial]
+    async fn test_self_copy_replace_metadata_preserves_readable_object() {
+        init_logging();
+        info!("Issue #2789: self-copy metadata replacement must preserve object data");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = env.create_s3_client();
+        let bucket = "self-copy-metadata-replace-test";
+        let key = "assets/chunk-2F3R7JUG.js";
+        let content = b"console.log('metadata replacement should keep object data readable');";
+
+        client
+            .create_bucket()
+            .bucket(bucket)
+            .send()
+            .await
+            .expect("Failed to create bucket");
+
+        client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .content_type("text/javascript; charset=utf-8")
+            .metadata("mtime", "1777992333")
+            .metadata("stale", "must-be-removed")
+            .body(ByteStream::from_static(content))
+            .send()
+            .await
+            .expect("PUT failed");
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key(key)
+            .copy_source(format!("{bucket}/{key}"))
+            .metadata_directive(MetadataDirective::Replace)
+            .content_type("text/javascript; charset=utf-8")
+            .metadata("mtime", "1777992348")
+            .send()
+            .await
+            .expect("self CopyObject with metadata replacement failed");
+
+        let head_resp = client
+            .head_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("HEAD failed after self-copy");
+        assert_eq!(head_resp.content_length(), Some(content.len() as i64));
+        assert_eq!(
+            head_resp.metadata().and_then(|metadata| metadata.get("mtime")),
+            Some(&"1777992348".to_string()),
+            "HEAD should return replaced metadata"
+        );
+        assert_eq!(
+            head_resp.metadata().and_then(|metadata| metadata.get("stale")),
+            None,
+            "HEAD should not return metadata omitted by REPLACE"
+        );
+
+        let get_resp = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("GET failed after self-copy");
+        let body = get_resp
+            .body
+            .collect()
+            .await
+            .expect("Failed to collect GET body")
+            .into_bytes();
+        assert_eq!(body.as_ref(), content, "self-copy metadata replacement must not drop object data");
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key(key)
+            .copy_source(format!("{bucket}/{key}"))
+            .metadata_directive(MetadataDirective::Replace)
+            .send()
+            .await
+            .expect("self CopyObject with empty metadata replacement failed");
+
+        let empty_head_resp = client
+            .head_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("HEAD failed after empty metadata replacement");
+        assert_eq!(
+            empty_head_resp.metadata().and_then(|metadata| metadata.get("mtime")),
+            None,
+            "HEAD should not return metadata omitted by empty REPLACE"
+        );
+
+        let empty_get_resp = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("GET failed after empty metadata replacement");
+        let empty_body = empty_get_resp
+            .body
+            .collect()
+            .await
+            .expect("Failed to collect GET body after empty metadata replacement")
+            .into_bytes();
+        assert_eq!(empty_body.as_ref(), content, "empty metadata replacement must not drop object data");
+
+        env.stop_server();
+    }
+}

crates/e2e_test/src/lib.rs

@@ -107,6 +107,9 @@ mod group_delete_test;
 #[cfg(test)]
 mod head_object_range_test;
 
+#[cfg(test)]
+mod copy_object_metadata_test;
+
 // S3 dummy-compat bucket API tests
 #[cfg(test)]
 mod bucket_logging_test;

crates/ecstore/src/disk/local.rs

@@ -2665,7 +2665,7 @@ impl DiskAPI for LocalDisk {
 
             let mut xl_meta = FileMeta::load(buf.as_ref())?;
 
-            xl_meta.update_object_version(fi)?;
+            xl_meta.update_object_version_with_opts(fi, opts.replace_user_metadata)?;
 
             let wbuf = xl_meta.marshal_msg()?;
 

crates/ecstore/src/disk/mod.rs

@@ -559,6 +559,7 @@ pub struct CheckPartsResp {
 #[derive(Debug, Serialize, Deserialize, Default)]
 pub struct UpdateMetadataOpts {
     pub no_persistence: bool,
+    pub replace_user_metadata: bool,
 }
 
 pub struct DiskLocation {
@@ -914,9 +915,13 @@ mod tests {
     /// Test UpdateMetadataOpts structure
     #[test]
     fn test_update_metadata_opts() {
-        let opts = UpdateMetadataOpts { no_persistence: true };
+        let opts = UpdateMetadataOpts {
+            no_persistence: true,
+            ..Default::default()
+        };
 
         assert!(opts.no_persistence);
+        assert!(!opts.replace_user_metadata);
     }
 
     /// Test DiskOption structure

crates/ecstore/src/set_disk.rs

@@ -1446,7 +1446,6 @@ impl ObjectOperations for SetDisks {
             }
         };
 
-        let inline_data = fi.inline_data();
         fi.metadata = src_info.user_defined.clone();
 
         if let Some(etag) = &src_info.etag {
@@ -1454,27 +1453,50 @@ impl ObjectOperations for SetDisks {
         }
 
         let mod_time = OffsetDateTime::now_utc();
+        fi.mod_time = Some(mod_time);
+        fi.version_id = version_id;
+        fi.versioned = src_opts.versioned || src_opts.version_suspended;
 
-        for fi in metas.iter_mut() {
-            if fi.is_valid() {
-                fi.metadata = src_info.user_defined.clone();
-                fi.mod_time = Some(mod_time);
-                fi.version_id = version_id;
-                fi.versioned = src_opts.versioned || src_opts.version_suspended;
+        if src_info.version_only {
+            let inline_data = fi.inline_data();
 
-                if !fi.inline_data() {
-                    fi.data = None;
-                }
+            for fi in metas.iter_mut() {
+                if fi.is_valid() {
+                    fi.metadata = src_info.user_defined.clone();
+                    if let Some(etag) = &src_info.etag {
+                        fi.metadata.insert("etag".to_owned(), etag.clone());
+                    }
+                    fi.mod_time = Some(mod_time);
+                    fi.version_id = version_id;
+                    fi.versioned = src_opts.versioned || src_opts.version_suspended;
 
-                if inline_data {
-                    fi.set_inline_data();
+                    if !fi.inline_data() {
+                        fi.data = None;
+                    }
+
+                    if inline_data {
+                        fi.set_inline_data();
+                    }
                 }
             }
-        }
 
-        Self::write_unique_file_info(&online_disks, "", src_bucket, src_object, &metas, write_quorum)
+            Self::write_unique_file_info(&online_disks, "", src_bucket, src_object, &metas, write_quorum)
+                .await
+                .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
+        } else {
+            self.update_object_meta_with_opts(
+                src_bucket,
+                src_object,
+                fi.clone(),
+                &online_disks,
+                &UpdateMetadataOpts {
+                    replace_user_metadata: true,
+                    ..Default::default()
+                },
+            )
             .await
             .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
+        }
 
         Ok(ObjectInfo::from_file_info(
             &fi,

crates/ecstore/src/set_disk/write.rs

@@ -402,7 +402,7 @@ impl SetDisks {
         disks: &[Option<DiskStore>],
         opts: &UpdateMetadataOpts,
     ) -> disk::error::Result<()> {
-        if fi.metadata.is_empty() {
+        if fi.metadata.is_empty() && !opts.replace_user_metadata {
             return Ok(());
         }
 

crates/filemeta/src/filemeta.rs

@@ -201,6 +201,10 @@ impl FileMeta {
     }
 
     pub fn update_object_version(&mut self, fi: FileInfo) -> Result<()> {
+        self.update_object_version_with_opts(fi, false)
+    }
+
+    pub fn update_object_version_with_opts(&mut self, fi: FileInfo, replace_user_metadata: bool) -> Result<()> {
         for version in self.versions.iter_mut() {
             match version.header.version_type {
                 VersionType::Invalid | VersionType::Legacy => (),
@@ -213,6 +217,10 @@ impl FileMeta {
                         let mut ver = FileMetaVersion::try_from(version.meta.as_slice())?;
 
                         if let Some(ref mut obj) = ver.object {
+                            if replace_user_metadata {
+                                obj.meta_user.clear();
+                            }
+
                             for (k, v) in fi.metadata.iter() {
                                 // Split metadata into meta_user and meta_sys based on prefix
                                 // This logic must match From<FileInfo> for MetaObject