fix: honor replication status metadata (#4178)

Co-authored-by: overtrue <anzhengchao@gmail.com>

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -1331,7 +1331,7 @@ impl PutObjectOptions {
         if !self.internal.source_version_id.is_empty() {
             insert_header(&mut header, SUFFIX_SOURCE_VERSION_ID, &self.internal.source_version_id);
         }
-        if self.internal.source_etag.is_empty() {
+        if !self.internal.source_etag.is_empty() {
             insert_header(&mut header, SUFFIX_SOURCE_ETAG, &self.internal.source_etag);
         }
         if self.internal.source_mtime.unix_timestamp() != 0 {
@@ -1926,6 +1926,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn put_object_headers_include_non_empty_source_etag_only() {
+        let mut opts = PutObjectOptions::default();
+
+        assert!(
+            rustfs_utils::http::get_header(&opts.header(), SUFFIX_SOURCE_ETAG).is_none(),
+            "empty source etag must not be sent to replication targets"
+        );
+
+        opts.internal.source_etag = "etag-1".to_string();
+
+        assert_eq!(
+            rustfs_utils::http::get_header(&opts.header(), SUFFIX_SOURCE_ETAG).as_deref(),
+            Some("etag-1"),
+            "replication targets need the source etag for idempotency checks"
+        );
+    }
+
     #[tokio::test]
     async fn get_remote_target_client_internal_rejects_loopback_endpoint() {
         let sys = BucketTargetSys::default();

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -155,11 +155,11 @@ impl ReplicationConfig {
 pub(crate) fn get_must_replicate_options(
     user_defined: &HashMap<String, String>,
     user_tags: String,
-    _status: ReplicationStatusType,
+    status: ReplicationStatusType,
     op_type: ReplicationType,
     opts: ObjectOptions,
 ) -> MustReplicateOptions {
-    MustReplicateOptions::new(user_defined, user_tags, op_type, opts.replication_request)
+    MustReplicateOptions::new(user_defined, user_tags, op_type, opts.replication_request).with_replication_status(status)
 }
 
 pub(crate) async fn check_replicate_delete(

crates/replication/src/operation.rs

@@ -32,6 +32,7 @@ use crate::{
 #[derive(Debug, Clone, Default)]
 pub struct MustReplicateOptions {
     meta: HashMap<String, String>,
+    status: ReplicationStatusType,
     op_type: ReplicationType,
     replication_request: bool,
 }
@@ -45,16 +46,25 @@ impl MustReplicateOptions {
 
         Self {
             meta,
+            status: ReplicationStatusType::default(),
             op_type,
             replication_request,
         }
     }
 
+    pub fn with_replication_status(mut self, status: ReplicationStatusType) -> Self {
+        self.status = status;
+        self
+    }
+
     pub fn replication_status(&self) -> ReplicationStatusType {
         if let Some(rs) = self.meta.get(AMZ_BUCKET_REPLICATION_STATUS) {
-            return ReplicationStatusType::from(rs.as_str());
+            let status = ReplicationStatusType::from(rs.as_str());
+            if !status.is_empty() {
+                return status;
+            }
         }
-        ReplicationStatusType::default()
+        self.status.clone()
     }
 
     pub fn is_existing_object_replication(&self) -> bool {
@@ -332,11 +342,20 @@ mod tests {
     #[test]
     fn must_replicate_options_reads_replication_status_header() {
         let meta = HashMap::from([(AMZ_BUCKET_REPLICATION_STATUS.to_string(), "COMPLETED".to_string())]);
-        let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Object, false);
+        let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Object, false)
+            .with_replication_status(ReplicationStatusType::Replica);
 
         assert_eq!(options.replication_status(), ReplicationStatusType::Completed);
     }
 
+    #[test]
+    fn must_replicate_options_falls_back_to_explicit_replication_status() {
+        let options = MustReplicateOptions::new(&HashMap::new(), String::new(), ReplicationType::Delete, false)
+            .with_replication_status(ReplicationStatusType::Replica);
+
+        assert_eq!(options.replication_status(), ReplicationStatusType::Replica);
+    }
+
     #[test]
     fn ssec_detection_uses_existing_metadata_headers() {
         let meta = HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]);