Merge branch 'main' into houseme/ilm-manual-transition-1476

crates/e2e_test/src/copy_object_metadata_test.rs

@@ -17,14 +17,17 @@
 #[cfg(test)]
 mod tests {
     use crate::common::{RustFSTestEnvironment, init_logging};
-    use aws_sdk_s3::primitives::ByteStream;
+    use aws_sdk_s3::error::ProvideErrorMetadata;
-    use aws_sdk_s3::types::MetadataDirective;
+    use aws_sdk_s3::primitives::{ByteStream, DateTime, DateTimeFormat};
+    use aws_sdk_s3::types::{
+        BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, MetadataDirective, StorageClass, VersioningConfiguration,
+    };
     use serial_test::serial;
     use tracing::info;
 
     #[tokio::test]
     #[serial]
-    async fn test_self_copy_replace_metadata_preserves_readable_object() {
+    async fn copy_object_standard_metadata_copy_replace_and_clear() {
         init_logging();
         info!("Issue #2789: self-copy metadata replacement must preserve object data");
 
@@ -35,6 +38,14 @@ mod tests {
         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');";
+        let source_expires = DateTime::from_secs(1_893_456_000);
+        let source_expires_http_date = source_expires
+            .fmt(DateTimeFormat::HttpDate)
+            .expect("Test timestamp should format as an HTTP date");
+        let replacement_expires = DateTime::from_secs(1_924_992_000);
+        let replacement_expires_http_date = replacement_expires
+            .fmt(DateTimeFormat::HttpDate)
+            .expect("Test timestamp should format as an HTTP date");
 
         client
             .create_bucket()
@@ -47,7 +58,14 @@ mod tests {
             .put_object()
             .bucket(bucket)
             .key(key)
+            .cache_control("max-age=60")
+            .content_disposition("inline; filename=source.js")
+            .content_encoding("br")
+            .content_language("en-US")
             .content_type("text/javascript; charset=utf-8")
+            .expires(source_expires)
+            .website_redirect_location("/source.html")
+            .storage_class(StorageClass::StandardIa)
             .metadata("mtime", "1777992333")
             .metadata("stale", "must-be-removed")
             .body(ByteStream::from_static(content))
@@ -55,13 +73,117 @@ mod tests {
             .await
             .expect("PUT failed");
 
+        let copied_key = "assets/default-copy.js";
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key(copied_key)
+            .copy_source(format!("{bucket}/{key}"))
+            .send()
+            .await
+            .expect("default CopyObject failed");
+
+        let copied_head = client
+            .head_object()
+            .bucket(bucket)
+            .key(copied_key)
+            .send()
+            .await
+            .expect("HEAD failed after default copy");
+        assert_eq!(copied_head.cache_control(), Some("max-age=60"));
+        assert_eq!(copied_head.content_disposition(), Some("inline; filename=source.js"));
+        assert_eq!(copied_head.content_encoding(), Some("br"));
+        assert_eq!(copied_head.content_language(), Some("en-US"));
+        assert_eq!(copied_head.content_type(), Some("text/javascript; charset=utf-8"));
+        assert_eq!(copied_head.expires_string(), Some(source_expires_http_date.as_str()));
+        assert_eq!(
+            copied_head.storage_class(),
+            None,
+            "CopyObject without a storage class should write STANDARD"
+        );
+        assert_eq!(
+            copied_head.website_redirect_location(),
+            Some("/source.html"),
+            "default CopyObject should preserve source metadata"
+        );
+        assert_eq!(
+            copied_head.metadata().and_then(|metadata| metadata.get("stale")),
+            Some(&"must-be-removed".to_string())
+        );
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key("assets/explicit-copy.js")
+            .copy_source(format!("{bucket}/{key}"))
+            .metadata_directive(MetadataDirective::Copy)
+            .send()
+            .await
+            .expect("explicit COPY directive failed");
+        let explicit_copy_head = client
+            .head_object()
+            .bucket(bucket)
+            .key("assets/explicit-copy.js")
+            .send()
+            .await
+            .expect("HEAD failed after explicit COPY");
+        assert_eq!(explicit_copy_head.cache_control(), Some("max-age=60"));
+        assert_eq!(
+            explicit_copy_head.website_redirect_location(),
+            None,
+            "explicit COPY does not inherit website redirect metadata"
+        );
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key("assets/explicit-copy-redirect.js")
+            .copy_source(format!("{bucket}/{key}"))
+            .metadata_directive(MetadataDirective::Copy)
+            .website_redirect_location("/explicit-copy.html")
+            .send()
+            .await
+            .expect("explicit COPY with redirect failed");
+        let explicit_redirect_head = client
+            .head_object()
+            .bucket(bucket)
+            .key("assets/explicit-copy-redirect.js")
+            .send()
+            .await
+            .expect("HEAD failed after explicit COPY with redirect");
+        assert_eq!(explicit_redirect_head.website_redirect_location(), Some("/explicit-copy.html"));
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key("assets/explicit-storage-class.js")
+            .copy_source(format!("{bucket}/{key}"))
+            .storage_class(StorageClass::StandardIa)
+            .send()
+            .await
+            .expect("CopyObject with an explicit storage class failed");
+        let explicit_storage_class_head = client
+            .head_object()
+            .bucket(bucket)
+            .key("assets/explicit-storage-class.js")
+            .send()
+            .await
+            .expect("HEAD failed after explicit storage class copy");
+        assert_eq!(explicit_storage_class_head.storage_class().map(StorageClass::as_str), Some("STANDARD_IA"));
+
         client
             .copy_object()
             .bucket(bucket)
             .key(key)
             .copy_source(format!("{bucket}/{key}"))
             .metadata_directive(MetadataDirective::Replace)
-            .content_type("text/javascript; charset=utf-8")
+            .cache_control("no-cache")
+            .content_disposition("attachment; filename=replaced.js")
+            .content_encoding("gzip")
+            .content_language("fr-FR")
+            .content_type("application/javascript")
+            .expires(replacement_expires)
+            .website_redirect_location("/replaced.html")
             .metadata("mtime", "1777992348")
             .send()
             .await
@@ -85,6 +207,14 @@ mod tests {
             None,
             "HEAD should not return metadata omitted by REPLACE"
         );
+        assert_eq!(head_resp.cache_control(), Some("no-cache"));
+        assert_eq!(head_resp.content_disposition(), Some("attachment; filename=replaced.js"));
+        assert_eq!(head_resp.content_encoding(), Some("gzip"));
+        assert_eq!(head_resp.content_language(), Some("fr-FR"));
+        assert_eq!(head_resp.content_type(), Some("application/javascript"));
+        assert_eq!(head_resp.expires_string(), Some(replacement_expires_http_date.as_str()));
+        assert_eq!(head_resp.website_redirect_location(), Some("/replaced.html"));
+        assert_eq!(head_resp.storage_class(), None, "REPLACE without a storage class should write STANDARD");
 
         let get_resp = client
             .get_object()
@@ -123,6 +253,13 @@ mod tests {
             None,
             "HEAD should not return metadata omitted by empty REPLACE"
         );
+        assert_eq!(empty_head_resp.cache_control(), None);
+        assert_eq!(empty_head_resp.content_disposition(), None);
+        assert_eq!(empty_head_resp.content_encoding(), None);
+        assert_eq!(empty_head_resp.content_language(), None);
+        assert_eq!(empty_head_resp.content_type(), None);
+        assert_eq!(empty_head_resp.expires_string(), None);
+        assert_eq!(empty_head_resp.website_redirect_location(), None);
 
         let empty_get_resp = client
             .get_object()
@@ -141,4 +278,333 @@ mod tests {
 
         env.stop_server();
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn copy_object_replace_accepts_each_standard_field_independently() {
+        init_logging();
+        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 = "copy-object-metadata-fields";
+        let source = "source.txt";
+        client
+            .create_bucket()
+            .bucket(bucket)
+            .send()
+            .await
+            .expect("Failed to create bucket");
+        client
+            .put_object()
+            .bucket(bucket)
+            .key(source)
+            .cache_control("source-cache")
+            .content_disposition("inline")
+            .content_encoding("br")
+            .content_language("en")
+            .content_type("text/source")
+            .expires(DateTime::from_secs(1_893_456_000))
+            .body(ByteStream::from_static(b"field-by-field"))
+            .send()
+            .await
+            .expect("PUT failed");
+        let replacement_expires = DateTime::from_secs(1_924_992_000);
+        let replacement_expires_http_date = replacement_expires
+            .fmt(DateTimeFormat::HttpDate)
+            .expect("Test timestamp should format as an HTTP date");
+
+        for field in [
+            "cache-control",
+            "content-disposition",
+            "content-encoding",
+            "content-language",
+            "content-type",
+            "expires",
+            "website-redirect",
+        ] {
+            let destination = format!("{field}.txt");
+            let request = client
+                .copy_object()
+                .bucket(bucket)
+                .key(&destination)
+                .copy_source(format!("{bucket}/{source}"))
+                .metadata_directive(MetadataDirective::Replace);
+            let request = match field {
+                "cache-control" => request.cache_control("field-cache"),
+                "content-disposition" => request.content_disposition("attachment"),
+                "content-encoding" => request.content_encoding("gzip"),
+                "content-language" => request.content_language("de"),
+                "content-type" => request.content_type("text/field"),
+                "expires" => request.expires(replacement_expires),
+                "website-redirect" => request.website_redirect_location("/field.html"),
+                _ => unreachable!("field table contains only supported entries"),
+            };
+            request.send().await.expect("field-specific CopyObject failed");
+
+            let head = client
+                .head_object()
+                .bucket(bucket)
+                .key(&destination)
+                .send()
+                .await
+                .expect("HEAD failed");
+            assert_eq!(head.cache_control(), (field == "cache-control").then_some("field-cache"));
+            assert_eq!(head.content_disposition(), (field == "content-disposition").then_some("attachment"));
+            assert_eq!(head.content_encoding(), (field == "content-encoding").then_some("gzip"));
+            assert_eq!(head.content_language(), (field == "content-language").then_some("de"));
+            assert_eq!(head.content_type(), (field == "content-type").then_some("text/field"));
+            assert_eq!(
+                head.expires_string(),
+                (field == "expires").then_some(replacement_expires_http_date.as_str())
+            );
+            assert_eq!(head.website_redirect_location(), (field == "website-redirect").then_some("/field.html"));
+        }
+
+        client
+            .copy_object()
+            .bucket(bucket)
+            .key("user-metadata-collision.txt")
+            .copy_source(format!("{bucket}/{source}"))
+            .metadata_directive(MetadataDirective::Replace)
+            .metadata("content-type", "user-content-type")
+            .metadata("content-encoding", "user-content-encoding")
+            .send()
+            .await
+            .expect("CopyObject should preserve user metadata namespaces");
+        let collision_head = client
+            .head_object()
+            .bucket(bucket)
+            .key("user-metadata-collision.txt")
+            .send()
+            .await
+            .expect("HEAD failed for metadata collision case");
+        assert_eq!(collision_head.content_type(), None);
+        assert_eq!(collision_head.content_encoding(), None);
+        assert_eq!(
+            collision_head.metadata().and_then(|metadata| metadata.get("content-type")),
+            Some(&"user-content-type".to_string())
+        );
+        assert_eq!(
+            collision_head
+                .metadata()
+                .and_then(|metadata| metadata.get("content-encoding")),
+            Some(&"user-content-encoding".to_string())
+        );
+
+        env.stop_server();
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn copy_object_replace_handles_versioned_multipart_source() {
+        init_logging();
+        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 = "copy-object-metadata-multipart";
+        let source = "source.bin";
+        let multipart_body = b"multipart historical source";
+        client
+            .create_bucket()
+            .bucket(bucket)
+            .send()
+            .await
+            .expect("Failed to create bucket");
+        client
+            .put_bucket_versioning()
+            .bucket(bucket)
+            .versioning_configuration(
+                VersioningConfiguration::builder()
+                    .status(BucketVersioningStatus::Enabled)
+                    .build(),
+            )
+            .send()
+            .await
+            .expect("Failed to enable versioning");
+
+        let upload = client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(source)
+            .content_type("application/source")
+            .send()
+            .await
+            .expect("Failed to create multipart upload");
+        let upload_id = upload.upload_id().expect("Multipart upload should return an ID");
+        let part = client
+            .upload_part()
+            .bucket(bucket)
+            .key(source)
+            .upload_id(upload_id)
+            .part_number(1)
+            .body(ByteStream::from_static(multipart_body))
+            .send()
+            .await
+            .expect("Failed to upload multipart part");
+        let completed = client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(source)
+            .upload_id(upload_id)
+            .multipart_upload(
+                CompletedMultipartUpload::builder()
+                    .parts(
+                        CompletedPart::builder()
+                            .part_number(1)
+                            .e_tag(part.e_tag().expect("Uploaded part should return an ETag"))
+                            .build(),
+                    )
+                    .build(),
+            )
+            .send()
+            .await
+            .expect("Failed to complete multipart upload");
+        let historical_version = completed
+            .version_id()
+            .expect("Versioned multipart upload should return a version ID")
+            .to_string();
+
+        client
+            .put_object()
+            .bucket(bucket)
+            .key(source)
+            .body(ByteStream::from_static(b"new current version"))
+            .send()
+            .await
+            .expect("Failed to write current version");
+
+        let copy = client
+            .copy_object()
+            .bucket(bucket)
+            .key("restored.bin")
+            .copy_source(format!("{bucket}/{source}?versionId={historical_version}"))
+            .metadata_directive(MetadataDirective::Replace)
+            .content_type("application/replaced")
+            .send()
+            .await
+            .expect("Failed to copy historical multipart version");
+        assert_eq!(copy.copy_source_version_id(), Some(historical_version.as_str()));
+
+        let restored = client
+            .get_object()
+            .bucket(bucket)
+            .key("restored.bin")
+            .send()
+            .await
+            .expect("Failed to read copied multipart source");
+        assert_eq!(restored.content_type(), Some("application/replaced"));
+        assert_eq!(
+            restored
+                .body
+                .collect()
+                .await
+                .expect("Failed to collect restored body")
+                .into_bytes()
+                .as_ref(),
+            multipart_body
+        );
+
+        env.stop_server();
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn invalid_replacement_metadata_does_not_mutate_destination() {
+        init_logging();
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server_with_env(vec![], &[("RUSTFS_REJECT_ARCHIVE_CONTENT_ENCODING", "true")])
+            .await
+            .expect("Failed to start RustFS");
+
+        let client = env.create_s3_client();
+        let bucket = "copy-object-invalid-metadata";
+        let key = "destination.zip";
+        client
+            .create_bucket()
+            .bucket(bucket)
+            .send()
+            .await
+            .expect("Failed to create bucket");
+        client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .content_type("application/zip")
+            .metadata("state", "original")
+            .body(ByteStream::from_static(b"original destination"))
+            .send()
+            .await
+            .expect("Failed to write destination");
+
+        let error = client
+            .copy_object()
+            .bucket(bucket)
+            .key(key)
+            .copy_source(format!("{bucket}/{key}"))
+            .metadata_directive(MetadataDirective::Replace)
+            .content_type("application/zip")
+            .content_encoding("gzip")
+            .send()
+            .await
+            .expect_err("Invalid replacement metadata should be rejected");
+        assert_eq!(error.as_service_error().and_then(|err| err.code()), Some("InvalidArgument"));
+
+        let invalid_directive = client
+            .copy_object()
+            .bucket(bucket)
+            .key(key)
+            .copy_source(format!("{bucket}/{key}"))
+            .customize()
+            .mutate_request(|request| {
+                request.headers_mut().insert("x-amz-metadata-directive", "UNKNOWN");
+            })
+            .send()
+            .await
+            .expect_err("Unknown metadata directives should be rejected");
+        assert_eq!(
+            invalid_directive.as_service_error().and_then(|error| error.code()),
+            Some("InvalidArgument")
+        );
+
+        let ignored_replacement = client
+            .copy_object()
+            .bucket(bucket)
+            .key(key)
+            .copy_source(format!("{bucket}/{key}"))
+            .content_type("application/ignored")
+            .send()
+            .await
+            .expect_err("Replacement fields without REPLACE should be rejected");
+        assert_eq!(
+            ignored_replacement.as_service_error().and_then(|error| error.code()),
+            Some("InvalidRequest")
+        );
+
+        let unchanged = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .send()
+            .await
+            .expect("Destination should remain readable");
+        assert_eq!(unchanged.content_type(), Some("application/zip"));
+        assert_eq!(
+            unchanged.metadata().and_then(|metadata| metadata.get("state")),
+            Some(&"original".to_string())
+        );
+        assert_eq!(
+            unchanged
+                .body
+                .collect()
+                .await
+                .expect("Failed to collect destination body")
+                .into_bytes()
+                .as_ref(),
+            b"original destination"
+        );
+
+        env.stop_server();
+    }
 }

docs/testing/e2e-suite-inventory.md

@@ -35,7 +35,7 @@
 | console_smoke_test | 1 | ✅ |
 | content_encoding_test | 3 | ✅ |
 | copy_object_checksum_test | 3 |  |
-| copy_object_metadata_test | 1 | ✅ |
+| copy_object_metadata_test | 4 | ✅ |
 | copy_object_tagging_test | 2 | ✅ |
 | copy_object_version_restore_test | 2 |  |
 | copy_source_invalid_date_test | 1 | ✅ |
@@ -84,4 +84,4 @@
 
 `notification_webhook_test` also has 1 ignored store-and-forward regression tracked by rustfs#4852; ignored tests are excluded from the active counts above.
 
-**Total listed: 469 tests across 64 modules · PR smoke subset: 116 tests / 29 modules** (27 full modules + 4 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 27 tests** · generated 2026-07-23.
+**Total listed: 472 tests across 64 modules · PR smoke subset: 119 tests / 29 modules** (27 full modules + 4 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 27 tests** · generated 2026-07-24.

rustfs/src/app/object_usecase.rs

@@ -82,8 +82,8 @@ use super::storage_api::object_usecase::object_cache::{GetObjectBodyCacheHookLoo
 use super::storage_api::object_usecase::object_utils::to_s3s_etag;
 use super::storage_api::object_usecase::options::{
     copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
-    filter_object_metadata, get_content_sha256_with_query, get_opts, normalize_content_encoding_for_storage, put_opts,
+    filter_object_metadata, get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata,
-    validate_archive_content_encoding,
+    normalize_content_encoding_for_storage, put_opts, validate_archive_content_encoding,
 };
 use super::storage_api::object_usecase::request_context::{self, spawn_traced};
 use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params;
@@ -2695,6 +2695,52 @@ where
     Ok(())
 }
 
+fn insert_expires_metadata(metadata: &mut HashMap<String, String>, expires: Option<&Timestamp>) -> S3Result<()> {
+    if let Some(expires) = expires {
+        let mut formatted = Vec::new();
+        expires
+            .format(TimestampFormat::HttpDate, &mut formatted)
+            .map_err(|e| ApiError::from(StorageError::other(format!("Invalid expires timestamp: {e}"))))?;
+        metadata.insert("expires".to_string(), String::from_utf8_lossy(&formatted).into_owned());
+    }
+    Ok(())
+}
+
+#[allow(clippy::too_many_arguments)]
+fn apply_standard_object_metadata(
+    metadata: &mut HashMap<String, String>,
+    cache_control: Option<&str>,
+    content_disposition: Option<&str>,
+    content_encoding: Option<&str>,
+    content_language: Option<&str>,
+    content_type: Option<&str>,
+    expires: Option<&Timestamp>,
+    website_redirect_location: Option<&str>,
+) -> S3Result<()> {
+    if let Some(cache_control) = cache_control {
+        metadata.insert("cache-control".to_string(), cache_control.to_string());
+    }
+    if let Some(content_disposition) = content_disposition {
+        metadata.insert("content-disposition".to_string(), content_disposition.to_string());
+    }
+    if let Some(content_encoding) = content_encoding
+        && let Some(normalized_content_encoding) = normalize_content_encoding_for_storage(content_encoding)
+    {
+        metadata.insert("content-encoding".to_string(), normalized_content_encoding);
+    }
+    if let Some(content_language) = content_language {
+        metadata.insert("content-language".to_string(), content_language.to_string());
+    }
+    if let Some(content_type) = content_type {
+        metadata.insert("content-type".to_string(), content_type.to_string());
+    }
+    insert_expires_metadata(metadata, expires)?;
+    if let Some(website_redirect_location) = website_redirect_location {
+        metadata.insert(AMZ_WEBSITE_REDIRECT_LOCATION.to_string(), website_redirect_location.to_string());
+    }
+    Ok(())
+}
+
 #[allow(clippy::too_many_arguments)]
 fn apply_put_request_metadata(
     metadata: &mut HashMap<String, String>,
@@ -2710,33 +2756,16 @@ fn apply_put_request_metadata(
     tagging: Option<TaggingHeader>,
     storage_class: Option<StorageClass>,
 ) -> S3Result<()> {
-    if let Some(cache_control) = cache_control {
+    apply_standard_object_metadata(
-        metadata.insert("cache-control".to_string(), cache_control);
+        metadata,
-    }
+        cache_control.as_deref(),
-    if let Some(content_disposition) = content_disposition {
+        content_disposition.as_deref(),
-        metadata.insert("content-disposition".to_string(), content_disposition);
+        content_encoding.as_deref(),
-    }
+        content_language.as_deref(),
-    if let Some(content_encoding) = content_encoding
+        content_type.as_deref(),
-        && let Some(normalized_content_encoding) = normalize_content_encoding_for_storage(&content_encoding)
+        expires.as_ref(),
-    {
+        website_redirect_location.as_deref(),
-        metadata.insert("content-encoding".to_string(), normalized_content_encoding);
+    )?;
-    }
-    if let Some(content_language) = content_language {
-        metadata.insert("content-language".to_string(), content_language);
-    }
-    if let Some(content_type) = content_type {
-        metadata.insert("content-type".to_string(), content_type);
-    }
-    if let Some(expires) = expires {
-        let mut formatted = Vec::new();
-        expires
-            .format(TimestampFormat::HttpDate, &mut formatted)
-            .map_err(|e| ApiError::from(StorageError::other(format!("Invalid expires timestamp: {e}"))))?;
-        metadata.insert("expires".to_string(), String::from_utf8_lossy(&formatted).into_owned());
-    }
-    if let Some(website_redirect_location) = website_redirect_location {
-        metadata.insert(AMZ_WEBSITE_REDIRECT_LOCATION.to_string(), website_redirect_location);
-    }
     if let Some(tags) = tagging {
         metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags);
     }
@@ -5740,7 +5769,13 @@ impl DefaultObjectUsecase {
             tagging_directive,
             copy_source_if_match,
             copy_source_if_none_match,
+            cache_control,
+            content_disposition,
+            content_encoding,
+            content_language,
             content_type,
+            expires,
+            website_redirect_location,
             object_lock_legal_hold_status,
             object_lock_mode,
             object_lock_retain_until_date,
@@ -5784,6 +5819,47 @@ impl DefaultObjectUsecase {
         validate_object_key(&src_key, "COPY (source)")?;
         validate_object_key(&key, "COPY (dest)")?;
         validate_table_catalog_object_mutation(&bucket, &key).await?;
+        let replaces_metadata = match metadata_directive.as_ref().map(|directive| directive.as_str()) {
+            None | Some(MetadataDirective::COPY) => false,
+            Some(MetadataDirective::REPLACE) => true,
+            Some(_) => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidArgument,
+                    "The MetadataDirective header is invalid".to_string(),
+                ));
+            }
+        };
+        let has_replacement_metadata = metadata.is_some()
+            || cache_control.is_some()
+            || content_disposition.is_some()
+            || content_encoding.is_some()
+            || content_language.is_some()
+            || content_type.is_some()
+            || expires.is_some();
+        if has_replacement_metadata && !replaces_metadata {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                "Replacement metadata requires the REPLACE metadata directive".to_string(),
+            ));
+        }
+        let replacement_metadata = if replaces_metadata {
+            validate_archive_content_encoding(&key, content_type.as_deref(), content_encoding.as_deref())?;
+            let mut replacement_metadata = metadata.unwrap_or_default();
+            namespace_reserved_user_metadata(&mut replacement_metadata);
+            apply_standard_object_metadata(
+                &mut replacement_metadata,
+                cache_control.as_deref(),
+                content_disposition.as_deref(),
+                content_encoding.as_deref(),
+                content_language.as_deref(),
+                content_type.as_deref(),
+                expires.as_ref(),
+                website_redirect_location.as_deref(),
+            )?;
+            Some(replacement_metadata)
+        } else {
+            None
+        };
 
         // AWS S3 allows self-copy when metadata directive is REPLACE (used to update metadata in-place),
         // when an explicit storage class change is requested, or when restoring a specific historical
@@ -5794,7 +5870,7 @@ impl DefaultObjectUsecase {
             tagging_directive.as_ref(),
         )?;
 
-        if metadata_directive.as_ref().map(|d| d.as_str()) != Some(MetadataDirective::REPLACE)
+        if !replaces_metadata
             && tagging_directive.as_ref().map(TaggingDirective::as_str) != Some(TaggingDirective::REPLACE)
             && storage_class.is_none()
             && version_id.is_none()
@@ -5949,13 +6025,18 @@ impl DefaultObjectUsecase {
         // Extract user_defined from Arc for mutation; it will be re-wrapped after all edits.
         let mut user_defined = (*src_info.user_defined).clone();
         let effective_tags = replacement_tags.unwrap_or_else(|| (*src_info.user_tags).clone());
+        if !replaces_metadata {
+            let source_expires = src_info.expires.map(Timestamp::from);
+            insert_expires_metadata(&mut user_defined, source_expires.as_ref())?;
+        }
 
         strip_managed_encryption_metadata(&mut user_defined);
 
-        if let Some(ref sc) = storage_class {
+        let destination_storage_class = storage_class
-            src_info.storage_class = Some(sc.as_str().to_string());
+            .as_ref()
-            user_defined.insert(AMZ_STORAGE_CLASS.to_string(), sc.as_str().to_string());
+            .map(StorageClass::as_str)
-        }
+            .unwrap_or(storageclass::STANDARD);
+        src_info.storage_class = Some(destination_storage_class.to_string());
 
         let actual_size = src_info.get_actual_size().map_err(ApiError::from)?;
 
@@ -5981,17 +6062,23 @@ impl DefaultObjectUsecase {
         // Handle MetadataDirective REPLACE: replace user metadata while preserving system metadata.
         // System metadata (compression, encryption) is added after this block to ensure
         // it's not cleared by the REPLACE operation.
-        if metadata_directive.as_ref().map(|d| d.as_str()) == Some(MetadataDirective::REPLACE) {
+        if let Some(replacement_metadata) = replacement_metadata {
-            user_defined.clear();
+            user_defined = replacement_metadata;
-            if let Some(metadata) = metadata {
+            src_info.content_type = content_type.clone();
-                user_defined.extend(metadata);
+            src_info.content_encoding = content_encoding.as_deref().and_then(normalize_content_encoding_for_storage);
-            }
+            src_info.expires = expires.map(OffsetDateTime::from);
-            if let Some(ct) = content_type {
+        } else if metadata_directive.is_some() || website_redirect_location.is_some() {
-                src_info.content_type = Some(ct.clone());
+            user_defined.retain(|key, _| !key.eq_ignore_ascii_case(AMZ_WEBSITE_REDIRECT_LOCATION));
-                user_defined.insert("content-type".to_string(), ct);
+            if let Some(website_redirect_location) = website_redirect_location {
+                user_defined.insert(AMZ_WEBSITE_REDIRECT_LOCATION.to_string(), website_redirect_location);
             }
         }
 
+        user_defined.retain(|key, _| !key.eq_ignore_ascii_case(AMZ_STORAGE_CLASS));
+        if destination_storage_class != storageclass::STANDARD {
+            user_defined.insert(AMZ_STORAGE_CLASS.to_string(), destination_storage_class.to_string());
+        }
+
         user_defined.retain(|key, _| !key.eq_ignore_ascii_case(AMZ_OBJECT_TAGGING));
         if !effective_tags.is_empty() {
             user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), effective_tags.clone());

rustfs/src/app/storage_api.rs

@@ -875,8 +875,8 @@ pub(crate) mod options {
     pub(crate) use crate::storage::storage_api::options_consumer::{
         copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime,
         extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
-        get_content_sha256_with_query, get_opts, normalize_content_encoding_for_storage, parse_copy_source_range, put_opts,
+        get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
-        validate_archive_content_encoding,
+        parse_copy_source_range, put_opts, validate_archive_content_encoding,
     };
 }
 

rustfs/src/storage/options.rs

@@ -483,6 +483,31 @@ fn archive_content_encoding_strict_mode() -> bool {
     rustfs_utils::get_env_bool(ENV_REJECT_ARCHIVE_CONTENT_ENCODING, false)
 }
 
+const USER_METADATA_PREFIXES: &[&str] = &["x-amz-meta-", "x-rustfs-meta-", "x-minio-meta-"];
+const CANONICAL_USER_METADATA_PREFIX: &str = "x-amz-meta-";
+
+fn is_reserved_user_metadata_key(key: &str) -> bool {
+    SUPPORTED_HEADERS.iter().any(|header| key.eq_ignore_ascii_case(header))
+        || starts_with_ignore_ascii_case(key, "x-amz-")
+        || starts_with_ignore_ascii_case(key, RUSTFS_INTERNAL_PREFIX)
+        || starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX)
+}
+
+fn stored_user_metadata_key(key: &str) -> String {
+    if is_reserved_user_metadata_key(key) {
+        format!("{CANONICAL_USER_METADATA_PREFIX}{key}")
+    } else {
+        key.to_owned()
+    }
+}
+
+pub(crate) fn namespace_reserved_user_metadata(metadata: &mut HashMap<String, String>) {
+    *metadata = std::mem::take(metadata)
+        .into_iter()
+        .map(|(key, value)| (stored_user_metadata_key(&key), value))
+        .collect();
+}
+
 /// Extracts metadata from headers and returns it as a HashMap with object name for MIME type detection.
 pub fn extract_metadata_from_mime_with_object_name(
     headers: &HeaderMap<HeaderValue>,
@@ -490,8 +515,6 @@ pub fn extract_metadata_from_mime_with_object_name(
     skip_content_type: bool,
     object_name: Option<&str>,
 ) {
-    const USER_METADATA_PREFIXES: &[&str] = &["x-amz-meta-", "x-rustfs-meta-", "x-minio-meta-"];
-
     for (k, v) in headers.iter() {
         if k.as_str() == "content-type" && skip_content_type {
             continue;
@@ -505,7 +528,7 @@ pub fn extract_metadata_from_mime_with_object_name(
                 continue;
             }
 
-            metadata.insert(key.to_owned(), String::from_utf8_lossy(v.as_bytes()).to_string());
+            metadata.insert(stored_user_metadata_key(key), String::from_utf8_lossy(v.as_bytes()).to_string());
             continue;
         }
 
@@ -599,6 +622,26 @@ pub(crate) fn filter_object_metadata(metadata: &HashMap<String, String>) -> Opti
 
     let mut filtered_metadata = None;
     for (k, v) in metadata {
+        if starts_with_ignore_ascii_case(k, "x-amz-meta-internal-")
+            || k.eq_ignore_ascii_case(AMZ_META_UNENCRYPTED_CONTENT_MD5)
+            || k.eq_ignore_ascii_case(AMZ_META_UNENCRYPTED_CONTENT_LENGTH)
+        {
+            continue;
+        }
+
+        if let Some(key) = USER_METADATA_PREFIXES.iter().find_map(|prefix| {
+            k.get(..prefix.len())
+                .filter(|head| head.eq_ignore_ascii_case(prefix))
+                .map(|_| &k[prefix.len()..])
+        }) {
+            if !key.is_empty() {
+                filtered_metadata
+                    .get_or_insert_with(HashMap::new)
+                    .insert(key.to_owned(), v.clone());
+            }
+            continue;
+        }
+
         if should_skip_object_metadata_key(k, v, EXCLUDED_HEADERS) {
             continue;
         }
@@ -857,7 +900,8 @@ mod tests {
         ENV_REJECT_ARCHIVE_CONTENT_ENCODING, ReplicationStatusType, SUPPORTED_HEADERS, copy_dst_opts, copy_src_opts, del_opts,
         detect_content_type_from_object_name, extract_metadata, extract_metadata_from_mime,
         extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
-        get_default_opts, get_opts, parse_copy_source_range, put_opts, put_opts_from_headers, validate_archive_content_encoding,
+        get_default_opts, get_opts, namespace_reserved_user_metadata, parse_copy_source_range, put_opts, put_opts_from_headers,
+        validate_archive_content_encoding,
     };
     use http::{HeaderMap, HeaderValue};
     use rustfs_utils::http::{
@@ -1573,6 +1617,45 @@ mod tests {
         assert_eq!(filtered.get("custom-key"), Some(&"custom-value".to_string()));
     }
 
+    #[test]
+    fn test_user_metadata_cannot_shadow_standard_or_internal_metadata() {
+        let mut headers = HeaderMap::new();
+        headers.insert("content-encoding", HeaderValue::from_static("br"));
+        headers.insert("x-amz-meta-content-encoding", HeaderValue::from_static("user-encoding"));
+        headers.insert("x-amz-meta-x-rustfs-internal-healing", HeaderValue::from_static("user-value"));
+
+        let mut metadata = HashMap::new();
+        extract_metadata_from_mime(&headers, &mut metadata);
+
+        assert_eq!(metadata.get("content-encoding"), Some(&"br".to_string()));
+        assert_eq!(metadata.get("x-amz-meta-content-encoding"), Some(&"user-encoding".to_string()));
+        assert_eq!(metadata.get("x-amz-meta-x-rustfs-internal-healing"), Some(&"user-value".to_string()));
+        let filtered = filter_object_metadata(&metadata).expect("user metadata should remain");
+        assert_eq!(filtered.get("content-encoding"), Some(&"user-encoding".to_string()));
+        assert_eq!(filtered.get("x-rustfs-internal-healing"), Some(&"user-value".to_string()));
+
+        let mut copied_metadata = HashMap::from([
+            ("content-type".to_string(), "user-type".to_string()),
+            ("x-amz-meta-content-type".to_string(), "nested-user-type".to_string()),
+            ("x-amz-storage-class".to_string(), "user-class".to_string()),
+        ]);
+        namespace_reserved_user_metadata(&mut copied_metadata);
+        assert_eq!(copied_metadata.get("x-amz-meta-content-type"), Some(&"user-type".to_string()));
+        assert_eq!(
+            copied_metadata.get("x-amz-meta-x-amz-meta-content-type"),
+            Some(&"nested-user-type".to_string())
+        );
+        assert_eq!(copied_metadata.get("x-amz-meta-x-amz-storage-class"), Some(&"user-class".to_string()));
+
+        let legacy_metadata = HashMap::from([
+            ("x-amz-meta-internal-secret".to_string(), "must-not-leak".to_string()),
+            ("x-amz-meta-x-amz-unencrypted-content-md5".to_string(), "must-not-leak".to_string()),
+            ("x-amz-meta-project".to_string(), "rustfs".to_string()),
+        ]);
+        let filtered_legacy = filter_object_metadata(&legacy_metadata).expect("safe user metadata should remain");
+        assert_eq!(filtered_legacy, HashMap::from([("project".to_string(), "rustfs".to_string())]));
+    }
+
     #[test]
     fn test_detect_content_type_from_object_name() {
         // Test Parquet files (our custom handling)

rustfs/src/storage/storage_api.rs

@@ -183,8 +183,8 @@ pub(crate) mod options_consumer {
     pub(crate) use super::super::options::{
         copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime,
         extract_metadata_from_mime_with_object_name, filter_object_metadata, get_complete_multipart_upload_opts,
-        get_content_sha256_with_query, get_opts, normalize_content_encoding_for_storage, parse_copy_source_range, put_opts,
+        get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, normalize_content_encoding_for_storage,
-        validate_archive_content_encoding,
+        parse_copy_source_range, put_opts, validate_archive_content_encoding,
     };
 
     pub(crate) mod contract {