From 1a9952c3fe3994ccb721fb2ed7917b9e64ef4899 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 15:54:48 +0800 Subject: [PATCH] refactor(replication): move target option builders (#4186) --- .../replication/replication_resyncer.rs | 335 +--------------- .../replication_target_boundary.rs | 363 +++++++++++++++++- 2 files changed, 382 insertions(+), 316 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index c9943e794..bcfb5e8f8 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -32,24 +32,20 @@ use super::replication_storage_boundary::{ AdvancedGetOptions, DeletedObject, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions, }; -use super::replication_tagging_boundary::ReplicationTagFilter; use super::replication_target_boundary::{ - AdvancedPutOptions, BucketTargets, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, ReplicationTargetStore, - TargetClient, + BucketTargets, PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, + replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, + replication_force_delete_remove_options, replication_put_object_header_size, replication_put_object_options, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput}; use aws_sdk_s3::primitives::ByteStream; -use aws_sdk_s3::types::{CompletedPart, ObjectLockLegalHoldStatus}; +use aws_sdk_s3::types::CompletedPart; use aws_smithy_types::body::SdkBody; use futures::future::join_all; use futures::stream::StreamExt; -use headers::{ - AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_SERVER_SIDE_ENCRYPTION, - AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_LANGUAGE, CONTENT_TYPE, -}; use http::HeaderMap; use http_body::Frame; use http_body_util::StreamBody; @@ -67,13 +63,8 @@ use rustfs_replication::{ }; use rustfs_s3_types::EventName; use rustfs_utils::http::{ - AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, AMZ_TAGGING_DIRECTIVE, CONTENT_ENCODING, HeaderExt as _, - SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_RESET, - SUFFIX_REPLICATION_STATUS, SUFFIX_TAGGING_TIMESTAMP, headers, -}; -use rustfs_utils::http::{ - SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, get_str, has_internal_suffix, insert_header_map, - insert_str, is_internal_key, + AMZ_BUCKET_REPLICATION_STATUS, AMZ_TAGGING_DIRECTIVE, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, + has_internal_suffix, insert_str, }; use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; use s3s::dto::ReplicationConfiguration; @@ -1726,15 +1717,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb &tgt_client.bucket, &dobj.delete_object.object_name, Some(delete_marker_version_id.to_string()), - RemoveObjectOptions { - force_delete: false, - governance_bypass: false, - replication_delete_marker: false, - replication_mtime: dobj.delete_object.delete_marker_mtime, - replication_status: ReplicationStatusType::Replica, - replication_request: true, - replication_validity_check: false, - }, + replication_delete_marker_purge_remove_options(dobj.delete_object.delete_marker_mtime), ) .await; } @@ -1918,20 +1901,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted } if let Err(e) = tgt_client - .remove_object( - &tgt_client.bucket, - &object_name, - None, - RemoveObjectOptions { - force_delete: true, - governance_bypass: false, - replication_delete_marker: false, - replication_mtime: None, - replication_status: ReplicationStatusType::Replica, - replication_request: true, - replication_validity_check: false, - }, - ) + .remove_object(&tgt_client.bucket, &object_name, None, replication_force_delete_remove_options()) .await { error!( @@ -2055,15 +2025,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli &tgt_client.bucket, &dobj.delete_object.object_name, version_id.clone(), - RemoveObjectOptions { - force_delete: false, - governance_bypass: false, - replication_delete_marker: dobj.delete_object.delete_marker, - replication_mtime: dobj.delete_object.delete_marker_mtime, - replication_status: ReplicationStatusType::Replica, - replication_request: true, - replication_validity_check: false, - }, + replication_delete_remove_options(dobj.delete_object.delete_marker, dobj.delete_object.delete_marker_mtime), ) .await { @@ -2581,7 +2543,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { rinfo.size = size; rinfo.replication_action = replication_action; - let (put_opts, is_multipart) = match put_replication_opts(&tgt_client.storage_class, &object_info) { + let (put_opts, is_multipart) = match replication_put_object_options(&tgt_client.storage_class, &object_info) { Ok((put_opts, is_mp)) => (put_opts, is_mp), Err(e) => { warn!( @@ -2971,7 +2933,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { if replication_action != ReplicationAction::All { // TODO: copy object } else { - let (put_opts, is_multipart) = match put_replication_opts(&tgt_client.storage_class, &object_info) { + let (put_opts, is_multipart) = match replication_put_object_options(&tgt_client.storage_class, &object_info) { Ok((put_opts, is_mp)) => (put_opts, is_mp), Err(e) => { rinfo.error = Some(e.to_string()); @@ -3074,34 +3036,6 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } } -// Standard headers that needs to be extracted from User metadata. -static STANDARD_HEADERS: &[&str] = &[ - CONTENT_TYPE, - CACHE_CONTROL, - CONTENT_ENCODING, - CONTENT_LANGUAGE, - CONTENT_DISPOSITION, - AMZ_STORAGE_CLASS, - AMZ_OBJECT_TAGGING, - AMZ_BUCKET_REPLICATION_STATUS, - AMZ_OBJECT_LOCK_MODE, - AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, - AMZ_OBJECT_LOCK_LEGAL_HOLD, - AMZ_TAG_COUNT, - AMZ_SERVER_SIDE_ENCRYPTION, -]; - -fn calc_put_object_header_size(put_opts: &PutObjectOptions) -> usize { - let mut header_size: usize = 0; - for (key, value) in put_opts.header().iter() { - header_size += key.as_str().len(); - header_size += value.as_bytes().len(); - // Account for HTTP header formatting: ": " (2 bytes) and "\r\n" (2 bytes) - header_size += 4; - } - header_size -} - fn wrap_with_bandwidth_monitor_with_header( stream: Box, bucket: &str, @@ -3130,7 +3064,7 @@ fn wrap_with_bandwidth_monitor( bucket: &str, arn: &str, ) -> Box { - let header_size = calc_put_object_header_size(put_opts); + let header_size = replication_put_object_header_size(put_opts); wrap_with_bandwidth_monitor_with_header(stream, bucket, arn, header_size) } @@ -3142,219 +3076,6 @@ fn async_read_to_bytestream(reader: impl AsyncRead + Send + Sync + Unpin + 'stat ByteStream::new(SdkBody::from_body_1_x(body)) } -fn is_standard_header(k: &str) -> bool { - STANDARD_HEADERS.iter().any(|h| h.eq_ignore_ascii_case(k)) -} - -// Valid SSE replication headers mapping from internal to replication headers -static VALID_SSE_REPLICATION_HEADERS: &[(&str, &str)] = &[ - ( - "X-Rustfs-Internal-Server-Side-Encryption-Sealed-Key", - "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key", - ), - ( - "X-Rustfs-Internal-Server-Side-Encryption-Seal-Algorithm", - "X-Rustfs-Replication-Server-Side-Encryption-Seal-Algorithm", - ), - ( - "X-Rustfs-Internal-Server-Side-Encryption-Iv", - "X-Rustfs-Replication-Server-Side-Encryption-Iv", - ), - ("X-Rustfs-Internal-Encrypted-Multipart", "X-Rustfs-Replication-Encrypted-Multipart"), - ("X-Rustfs-Internal-Actual-Object-Size", "X-Rustfs-Replication-Actual-Object-Size"), -]; - -fn is_valid_sse_header(k: &str) -> Option<&str> { - VALID_SSE_REPLICATION_HEADERS - .iter() - .find(|(internal, _)| k.eq_ignore_ascii_case(internal)) - .map(|(_, replication)| *replication) -} - -fn put_replication_opts(sc: &str, object_info: &ObjectInfo) -> Result<(PutObjectOptions, bool)> { - use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; - use rustfs_utils::http::{ - AMZ_CHECKSUM_TYPE, AMZ_CHECKSUM_TYPE_FULL_OBJECT, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID, - }; - - let mut meta = HashMap::new(); - let is_ssec = rustfs_replication::is_ssec_encrypted(&object_info.user_defined); - - // Process user-defined metadata - for (k, v) in object_info.user_defined.iter() { - let has_valid_sse_header = is_valid_sse_header(k).is_some(); - - // In case of SSE-C objects copy the allowed internal headers as well - if !is_ssec || !has_valid_sse_header { - if is_internal_key(k) { - continue; - } - if is_standard_header(k) { - continue; - } - } - - if let Some(replication_header) = is_valid_sse_header(k) { - meta.insert(replication_header.to_string(), v.to_string()); - } else { - meta.insert(k.to_string(), v.to_string()); - } - } - - let mut is_multipart = object_info.is_multipart(); - - // Handle checksum - if let Some(checksum_data) = &object_info.checksum - && !checksum_data.is_empty() - { - // Add encrypted CRC to metadata for SSE-C objects - if is_ssec { - let encoded = BASE64_STANDARD.encode(checksum_data); - insert_header_map(&mut meta, SUFFIX_REPLICATION_SSEC_CRC, encoded); - } else { - // Get checksum metadata for non-SSE-C objects - let (cs_meta, is_mp) = object_info.decrypt_checksums(0, &HeaderMap::new())?; - is_multipart = is_mp; - - // Set object checksum metadata - for (k, v) in cs_meta.iter() { - if k != AMZ_CHECKSUM_TYPE { - meta.insert(k.clone(), v.clone()); - } - } - - // For objects where checksum is full object, use the cheaper PutObject replication - if !object_info.is_multipart() - && cs_meta - .get(AMZ_CHECKSUM_TYPE) - .map(|v| v.as_str() == AMZ_CHECKSUM_TYPE_FULL_OBJECT) - .unwrap_or(false) - { - is_multipart = false; - } - } - } - - // Handle storage class default - let storage_class = if sc.is_empty() { - let obj_sc = object_info.storage_class.as_deref().unwrap_or_default(); - if obj_sc == ReplicationConfigStore::STANDARD || obj_sc == ReplicationConfigStore::RRS { - obj_sc.to_string() - } else { - sc.to_string() - } - } else { - sc.to_string() - }; - - let mut put_op = PutObjectOptions { - user_metadata: meta, - content_type: object_info.content_type.clone().unwrap_or_default(), - content_encoding: object_info.content_encoding.clone().unwrap_or_default(), - expires: object_info.expires.unwrap_or(OffsetDateTime::UNIX_EPOCH), - storage_class, - internal: AdvancedPutOptions { - source_version_id: object_info.version_id.map(|v| v.to_string()).unwrap_or_default(), - source_etag: object_info.etag.clone().unwrap_or_default(), - source_mtime: object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH), - replication_status: ReplicationStatusType::Replica, // Changed from Pending to Replica - replication_request: true, // always set this to distinguish between replication and normal PUT operation - ..Default::default() - }, - ..Default::default() - }; - - if !object_info.user_tags.is_empty() { - let tags = ReplicationTagFilter::decode_tags_to_map(&object_info.user_tags); - - if !tags.is_empty() { - put_op.user_tags = tags; - // set tag timestamp in opts - put_op.internal.tagging_timestamp = if let Some(ts) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) { - OffsetDateTime::parse(&ts, &Rfc3339) - .map_err(|e| Error::other(format!("Failed to parse tagging timestamp: {}", e)))? - } else { - object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - }; - } - } - - // Use case-insensitive lookup for headers - let lk_map = &*object_info.user_defined; - - if let Some(lang) = lk_map.lookup(CONTENT_LANGUAGE) { - put_op.content_language = lang.to_string(); - } - - if let Some(cd) = lk_map.lookup(CONTENT_DISPOSITION) { - put_op.content_disposition = cd.to_string(); - } - - if let Some(v) = lk_map.lookup(CACHE_CONTROL) { - put_op.cache_control = v.to_string(); - } - - if let Some(v) = lk_map.lookup(AMZ_OBJECT_LOCK_MODE) { - let mode = v.to_string().to_uppercase(); - put_op.mode = Some(aws_sdk_s3::types::ObjectLockRetentionMode::from(mode.as_str())); - } - - if let Some(v) = lk_map.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) { - put_op.retain_until_date = - OffsetDateTime::parse(v, &Rfc3339).map_err(|e| Error::other(format!("Failed to parse retain until date: {}", e)))?; - // set retention timestamp in opts - put_op.internal.retention_timestamp = - if let Some(v) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) { - OffsetDateTime::parse(&v, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - }; - } - - if let Some(v) = lk_map.lookup(AMZ_OBJECT_LOCK_LEGAL_HOLD) { - let hold = v.to_uppercase(); - put_op.legalhold = Some(ObjectLockLegalHoldStatus::from(hold.as_str())); - // set legalhold timestamp in opts - put_op.internal.legalhold_timestamp = - if let Some(v) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP) { - OffsetDateTime::parse(&v, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - }; - } - - // Handle SSE-S3 encryption - if object_info - .user_defined - .get(AMZ_SERVER_SIDE_ENCRYPTION) - .map(|v| v.eq_ignore_ascii_case("AES256")) - .unwrap_or(false) - { - // SSE-S3 detected - set ServerSideEncryption - // Note: This requires the PutObjectOptions to support SSE - // TODO: Implement SSE-S3 support in PutObjectOptions if not already present - } - - // Handle SSE-KMS encryption - if object_info.user_defined.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_KMS_ID) { - // SSE-KMS detected - // If KMS key ID replication is enabled (as by default) - // we include the object's KMS key ID. In any case, we - // always set the SSE-KMS header. If no KMS key ID is - // specified, the server uses the default applicable - // config applies on the site or bucket. - // TODO: Implement SSE-KMS support with key ID replication - // let key_id = if kms::replicate_key_id() { - // object_info.kms_key_id() - // } else { - // None - // }; - // TODO: Set SSE-KMS encryption in put_op - } - - Ok((put_op, is_multipart)) -} - fn part_range_spec_from_actual_size(offset: i64, part_size: i64) -> std::io::Result<(HTTPRangeSpec, i64)> { if offset < 0 { return Err(std::io::Error::other("invalid part offset")); @@ -3423,9 +3144,11 @@ async fn replicate_object_with_multipart(ctx: MultipartR let mut uploaded_parts: Vec = Vec::new(); - let mut header_size = calc_put_object_header_size(&put_opts); + let mut header_size = replication_put_object_header_size(&put_opts); let mut offset: i64 = 0; for part_info in object_info.parts.iter() { + let part_number = i32::try_from(part_info.number) + .map_err(|_| std::io::Error::other(format!("part number {} overflows i32", part_info.number)))?; let part_size = part_info.actual_size; let (range_spec, next_offset) = part_range_spec_from_actual_size(offset, part_size)?; offset = next_offset; @@ -3444,7 +3167,7 @@ async fn replicate_object_with_multipart(ctx: MultipartR dst_bucket, object, &upload_id, - part_info.number as i32, + part_number, part_size, byte_stream, &PutObjectPartOptions { ..Default::default() }, @@ -3454,36 +3177,18 @@ async fn replicate_object_with_multipart(ctx: MultipartR let etag = object_part.e_tag.unwrap_or_default(); - uploaded_parts.push( - CompletedPart::builder() - .part_number(part_info.number as i32) - .e_tag(etag) - .build(), - ); + uploaded_parts.push(CompletedPart::builder().part_number(part_number).e_tag(etag).build()); } - let mut user_metadata = HashMap::new(); - - insert_header_map( - &mut user_metadata, - SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, - rustfs_utils::http::get_str(&object_info.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE).unwrap_or_default(), - ); + let actual_size = + rustfs_utils::http::get_str(&object_info.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE).unwrap_or_default(); cli.complete_multipart_upload( dst_bucket, object, &upload_id, uploaded_parts, - &PutObjectOptions { - user_metadata, - internal: AdvancedPutOptions { - replication_status: ReplicationStatusType::Replica, - replication_request: true, - ..Default::default() - }, - ..Default::default() - }, + &replication_complete_multipart_options(actual_size), ) .await .map_err(|e| std::io::Error::other(e.to_string()))?; diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 981853924..e394b93ee 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -12,19 +12,70 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::collections::HashMap; use std::sync::Arc; use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys}; +use aws_sdk_s3::types::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode}; +use http::HeaderMap; +use rustfs_utils::http::{ + AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, + AMZ_OBJECT_TAGGING, AMZ_SERVER_SIDE_ENCRYPTION, AMZ_STORAGE_CLASS, AMZ_TAG_COUNT, CACHE_CONTROL, CONTENT_DISPOSITION, + CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, HeaderExt as _, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, + SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, + SUFFIX_TAGGING_TIMESTAMP, get_str, insert_header_map, is_internal_key, +}; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; pub(crate) use crate::bucket::bucket_target_sys::{ AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient, }; pub(crate) use crate::bucket::target::BucketTargets; +use super::replication_config_store::ReplicationConfigStore; +use super::replication_error_boundary::{Error, Result}; +use super::replication_filemeta_boundary::ReplicationStatusType; +use super::replication_storage_boundary::ObjectInfo; +use super::replication_tagging_boundary::ReplicationTagFilter; + +static STANDARD_HEADERS: &[&str] = &[ + CONTENT_TYPE, + CACHE_CONTROL, + CONTENT_ENCODING, + CONTENT_LANGUAGE, + CONTENT_DISPOSITION, + AMZ_STORAGE_CLASS, + AMZ_OBJECT_TAGGING, + AMZ_BUCKET_REPLICATION_STATUS, + AMZ_OBJECT_LOCK_MODE, + AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, + AMZ_OBJECT_LOCK_LEGAL_HOLD, + AMZ_TAG_COUNT, + AMZ_SERVER_SIDE_ENCRYPTION, +]; + +static VALID_SSE_REPLICATION_HEADERS: &[(&str, &str)] = &[ + ( + "X-Rustfs-Internal-Server-Side-Encryption-Sealed-Key", + "X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key", + ), + ( + "X-Rustfs-Internal-Server-Side-Encryption-Seal-Algorithm", + "X-Rustfs-Replication-Server-Side-Encryption-Seal-Algorithm", + ), + ( + "X-Rustfs-Internal-Server-Side-Encryption-Iv", + "X-Rustfs-Replication-Server-Side-Encryption-Iv", + ), + ("X-Rustfs-Internal-Encrypted-Multipart", "X-Rustfs-Replication-Encrypted-Multipart"), + ("X-Rustfs-Internal-Actual-Object-Size", "X-Rustfs-Replication-Actual-Object-Size"), +]; + pub(crate) struct ReplicationTargetStore; impl ReplicationTargetStore { - pub(crate) async fn list_bucket_targets(bucket: &str) -> Result { + pub(crate) async fn list_bucket_targets(bucket: &str) -> std::result::Result { BucketTargetSys::get().list_bucket_targets(bucket).await } @@ -36,3 +87,313 @@ impl ReplicationTargetStore { BucketTargetSys::get().is_offline(&target_client.to_url()).await } } + +pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) -> Result<(PutObjectOptions, bool)> { + use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; + use rustfs_utils::http::{AMZ_CHECKSUM_TYPE, AMZ_CHECKSUM_TYPE_FULL_OBJECT}; + + let mut meta = HashMap::new(); + let is_ssec = rustfs_replication::is_ssec_encrypted(&object_info.user_defined); + + for (key, value) in object_info.user_defined.iter() { + let has_valid_sse_header = valid_sse_replication_header(key).is_some(); + + if !is_ssec || !has_valid_sse_header { + if is_internal_key(key) || is_standard_header(key) { + continue; + } + } + + if let Some(replication_header) = valid_sse_replication_header(key) { + meta.insert(replication_header.to_string(), value.to_string()); + } else { + meta.insert(key.to_string(), value.to_string()); + } + } + + let mut is_multipart = object_info.is_multipart(); + + if let Some(checksum_data) = &object_info.checksum + && !checksum_data.is_empty() + { + if is_ssec { + let encoded = BASE64_STANDARD.encode(checksum_data); + insert_header_map(&mut meta, SUFFIX_REPLICATION_SSEC_CRC, encoded); + } else { + let (checksum_meta, is_mp) = object_info.decrypt_checksums(0, &HeaderMap::new())?; + is_multipart = is_mp; + + for (key, value) in checksum_meta.iter() { + if key != AMZ_CHECKSUM_TYPE { + meta.insert(key.clone(), value.clone()); + } + } + + if !object_info.is_multipart() + && checksum_meta + .get(AMZ_CHECKSUM_TYPE) + .is_some_and(|value| value == AMZ_CHECKSUM_TYPE_FULL_OBJECT) + { + is_multipart = false; + } + } + } + + let storage_class = if sc.is_empty() { + let obj_sc = object_info.storage_class.as_deref().unwrap_or_default(); + if obj_sc == ReplicationConfigStore::STANDARD || obj_sc == ReplicationConfigStore::RRS { + obj_sc.to_string() + } else { + sc.to_string() + } + } else { + sc.to_string() + }; + + let mut put_options = PutObjectOptions { + user_metadata: meta, + content_type: object_info.content_type.clone().unwrap_or_default(), + content_encoding: object_info.content_encoding.clone().unwrap_or_default(), + expires: object_info.expires.unwrap_or(OffsetDateTime::UNIX_EPOCH), + storage_class, + internal: AdvancedPutOptions { + source_version_id: object_info.version_id.map(|value| value.to_string()).unwrap_or_default(), + source_etag: object_info.etag.clone().unwrap_or_default(), + source_mtime: object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH), + replication_status: ReplicationStatusType::Replica, + replication_request: true, + ..Default::default() + }, + ..Default::default() + }; + + if !object_info.user_tags.is_empty() { + let tags = ReplicationTagFilter::decode_tags_to_map(&object_info.user_tags); + + if !tags.is_empty() { + put_options.user_tags = tags; + put_options.internal.tagging_timestamp = + if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) { + OffsetDateTime::parse(×tamp, &Rfc3339) + .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))? + } else { + object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + }; + } + } + + let metadata = &*object_info.user_defined; + + if let Some(language) = metadata.lookup(CONTENT_LANGUAGE) { + put_options.content_language = language.to_string(); + } + + if let Some(content_disposition) = metadata.lookup(CONTENT_DISPOSITION) { + put_options.content_disposition = content_disposition.to_string(); + } + + if let Some(cache_control) = metadata.lookup(CACHE_CONTROL) { + put_options.cache_control = cache_control.to_string(); + } + + if let Some(mode) = metadata.lookup(AMZ_OBJECT_LOCK_MODE) { + put_options.mode = Some(ObjectLockRetentionMode::from(mode.to_uppercase().as_str())); + } + + if let Some(retain_until_date) = metadata.lookup(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE) { + put_options.retain_until_date = OffsetDateTime::parse(retain_until_date, &Rfc3339) + .map_err(|err| Error::other(format!("Failed to parse retain until date: {err}")))?; + put_options.internal.retention_timestamp = + if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP) { + OffsetDateTime::parse(×tamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + }; + } + + if let Some(legal_hold) = metadata.lookup(AMZ_OBJECT_LOCK_LEGAL_HOLD) { + put_options.legalhold = Some(ObjectLockLegalHoldStatus::from(legal_hold.to_uppercase().as_str())); + put_options.internal.legalhold_timestamp = + if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP) { + OffsetDateTime::parse(×tamp, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + }; + } + + Ok((put_options, is_multipart)) +} + +pub(crate) fn replication_put_object_header_size(put_options: &PutObjectOptions) -> usize { + put_options + .header() + .iter() + .map(|(key, value)| key.as_str().len() + value.as_bytes().len() + 4) + .sum() +} + +pub(crate) fn replication_delete_remove_options( + delete_marker: bool, + replication_mtime: Option, +) -> RemoveObjectOptions { + RemoveObjectOptions { + force_delete: false, + governance_bypass: false, + replication_delete_marker: delete_marker, + replication_mtime, + replication_status: ReplicationStatusType::Replica, + replication_request: true, + replication_validity_check: false, + } +} + +pub(crate) fn replication_delete_marker_purge_remove_options(replication_mtime: Option) -> RemoveObjectOptions { + RemoveObjectOptions { + force_delete: false, + governance_bypass: false, + replication_delete_marker: false, + replication_mtime, + replication_status: ReplicationStatusType::Replica, + replication_request: true, + replication_validity_check: false, + } +} + +pub(crate) fn replication_force_delete_remove_options() -> RemoveObjectOptions { + RemoveObjectOptions { + force_delete: true, + governance_bypass: false, + replication_delete_marker: false, + replication_mtime: None, + replication_status: ReplicationStatusType::Replica, + replication_request: true, + replication_validity_check: false, + } +} + +pub(crate) fn replication_complete_multipart_options(actual_size: String) -> PutObjectOptions { + let mut user_metadata = HashMap::new(); + insert_header_map(&mut user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, actual_size); + + PutObjectOptions { + user_metadata, + internal: AdvancedPutOptions { + replication_status: ReplicationStatusType::Replica, + replication_request: true, + ..Default::default() + }, + ..Default::default() + } +} + +fn is_standard_header(key: &str) -> bool { + STANDARD_HEADERS.iter().any(|header| header.eq_ignore_ascii_case(key)) +} + +fn valid_sse_replication_header(key: &str) -> Option<&str> { + VALID_SSE_REPLICATION_HEADERS + .iter() + .find(|(internal, _)| key.eq_ignore_ascii_case(internal)) + .map(|(_, replication)| *replication) +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_utils::http::{ + SSEC_ALGORITHM_HEADER, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, get_header_map, + }; + use std::sync::Arc; + use time::Duration; + use uuid::Uuid; + + #[test] + fn replication_remove_options_mark_replication_requests() { + let mtime = OffsetDateTime::UNIX_EPOCH + Duration::seconds(10); + + let delete = replication_delete_remove_options(true, Some(mtime)); + assert!(!delete.force_delete); + assert!(delete.replication_delete_marker); + assert_eq!(delete.replication_mtime, Some(mtime)); + assert_eq!(delete.replication_status, ReplicationStatusType::Replica); + assert!(delete.replication_request); + + let purge = replication_delete_marker_purge_remove_options(Some(mtime)); + assert!(!purge.force_delete); + assert!(!purge.replication_delete_marker); + assert_eq!(purge.replication_mtime, Some(mtime)); + assert_eq!(purge.replication_status, ReplicationStatusType::Replica); + assert!(purge.replication_request); + + let force = replication_force_delete_remove_options(); + assert!(force.force_delete); + assert!(!force.replication_delete_marker); + assert_eq!(force.replication_status, ReplicationStatusType::Replica); + assert!(force.replication_request); + } + + #[test] + fn replication_complete_multipart_options_sets_actual_size() { + let options = replication_complete_multipart_options("1024".to_string()); + + assert_eq!( + get_header_map(&options.user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE).as_deref(), + Some("1024") + ); + assert_eq!(options.internal.replication_status, ReplicationStatusType::Replica); + assert!(options.internal.replication_request); + } + + #[test] + fn replication_put_options_filter_and_map_metadata() { + let mut metadata = HashMap::new(); + metadata.insert(CONTENT_TYPE.to_string(), "text/plain".to_string()); + metadata.insert("x-user-meta".to_string(), "value".to_string()); + metadata.insert(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string()); + metadata.insert("X-Rustfs-Internal-Server-Side-Encryption-Sealed-Key".to_string(), "sealed".to_string()); + + let object_info = ObjectInfo { + user_defined: Arc::new(metadata), + user_tags: Arc::new("env=prod".to_string()), + content_type: Some("text/plain".to_string()), + content_encoding: Some("gzip".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + version_id: Some(Uuid::nil()), + etag: Some("0123456789abcdef0123456789abcdef".to_string()), + ..Default::default() + }; + + let (options, is_multipart) = replication_put_object_options("STANDARD", &object_info).expect("build put options"); + + assert!(!is_multipart); + assert_eq!(options.user_metadata.get("x-user-meta"), Some(&"value".to_string())); + assert!(!options.user_metadata.contains_key(CONTENT_TYPE)); + assert_eq!( + options + .user_metadata + .get("X-Rustfs-Replication-Server-Side-Encryption-Sealed-Key"), + Some(&"sealed".to_string()) + ); + assert_eq!(options.content_type, "text/plain"); + assert_eq!(options.content_encoding, "gzip"); + assert_eq!(options.user_tags.get("env"), Some(&"prod".to_string())); + assert_eq!(options.internal.source_version_id, Uuid::nil().to_string()); + assert_eq!(options.internal.source_etag, "0123456789abcdef0123456789abcdef"); + assert_eq!(options.internal.replication_status, ReplicationStatusType::Replica); + assert!(options.internal.replication_request); + } + + #[test] + fn replication_put_options_adds_ssec_checksum_metadata() { + let metadata = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]); + let object_info = ObjectInfo { + user_defined: Arc::new(metadata), + checksum: Some(bytes::Bytes::from_static(b"checksum")), + ..Default::default() + }; + + let (options, _) = replication_put_object_options("", &object_info).expect("build put options"); + + assert!(get_header_map(&options.user_metadata, SUFFIX_REPLICATION_SSEC_CRC).is_some()); + } +}