fix(replication): avoid re-replication loop in Active-Active replication (#1751)

Co-authored-by: loverustfs <hello@rustfs.com>

rustfs/src/storage/options.rs

@@ -32,8 +32,10 @@ use rustfs_utils::http::RESERVED_METADATA_PREFIX_LOWER;
 use rustfs_utils::http::RUSTFS_BUCKET_REPLICATION_DELETE_MARKER;
 use rustfs_utils::http::RUSTFS_BUCKET_REPLICATION_REQUEST;
 use rustfs_utils::http::RUSTFS_BUCKET_REPLICATION_SSEC_CHECKSUM;
+use rustfs_utils::http::RUSTFS_BUCKET_SOURCE_MTIME;
 use rustfs_utils::http::RUSTFS_BUCKET_SOURCE_VERSION_ID;
 use rustfs_utils::http::RUSTFS_ENCRYPTION_LOWER;
+use rustfs_utils::http::RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE;
 use rustfs_utils::path::is_dir_object;
 use s3s::{S3Result, s3_error};
 use std::collections::HashMap;
@@ -45,6 +47,8 @@ use crate::auth::AuthType;
 use crate::auth::get_request_auth_type;
 use crate::auth::is_request_presigned_signature_v4;
 
+const REPLICATION_REQUEST_TRUE: HeaderValue = HeaderValue::from_static("true");
+
 /// Creates options for deleting an object in a bucket.
 pub async fn del_opts(
     bucket: &str,
@@ -240,12 +244,16 @@ pub fn get_complete_multipart_upload_opts(headers: &HeaderMap<HeaderValue>) -> s
     let mut user_defined = HashMap::new();
 
     let mut replication_request = false;
-    if let Some(v) = headers.get(RUSTFS_BUCKET_REPLICATION_REQUEST) {
-        user_defined.insert(
-            format!("{RESERVED_METADATA_PREFIX_LOWER}Actual-Object-Size"),
-            v.to_str().unwrap_or_default().to_owned(),
-        );
+    if headers.get(RUSTFS_BUCKET_REPLICATION_REQUEST) == Some(&REPLICATION_REQUEST_TRUE) {
         replication_request = true;
+        if let Some(actual_size_str) = headers
+            .get(RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE)
+            .and_then(|h| h.to_str().ok())
+        {
+            user_defined.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}Actual-Object-Size"), actual_size_str.to_string());
+        } else {
+            tracing::warn!("Failed to get or parse {} header", RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE);
+        }
     }
 
     if let Some(v) = headers.get(RUSTFS_BUCKET_REPLICATION_SSEC_CHECKSUM) {
@@ -282,7 +290,33 @@ pub fn copy_src_opts(_bucket: &str, _object: &str, headers: &HeaderMap<HeaderVal
 }
 
 pub fn put_opts_from_headers(headers: &HeaderMap<HeaderValue>, metadata: HashMap<String, String>) -> Result<ObjectOptions> {
-    get_default_opts(headers, metadata, false)
+    let mut opts = get_default_opts(headers, metadata, false)?;
+    if headers.get(RUSTFS_BUCKET_REPLICATION_REQUEST) == Some(&REPLICATION_REQUEST_TRUE) {
+        opts.replication_request = true;
+        if let Some(v) = headers.get(RUSTFS_BUCKET_SOURCE_MTIME) {
+            match v.to_str() {
+                Ok(s) => {
+                    let trimmed_s = s.trim();
+                    match time::OffsetDateTime::parse(trimmed_s, &time::format_description::well_known::Rfc3339) {
+                        Ok(mtime) => opts.mod_time = Some(mtime),
+                        Err(e) => {
+                            tracing::warn!(
+                                "Invalid X-RustFS-Source-Mtime value '{}' (replication request=true): {}",
+                                trimmed_s,
+                                e
+                            );
+                            opts.mod_time = None;
+                        }
+                    }
+                }
+                Err(e) => {
+                    tracing::warn!("X-RustFS-Source-Mtime header is not valid UTF-8 (replication request=true): {}", e);
+                    opts.mod_time = None;
+                }
+            }
+        }
+    }
+    Ok(opts)
 }
 
 /// Creates default options for getting an object from a bucket.
@@ -891,6 +925,35 @@ mod tests {
         assert_eq!(user_defined.get("key2"), Some(&"value2".to_string()));
     }
 
+    #[test]
+    fn test_put_opts_from_headers_with_replication_request() {
+        let mut headers = HeaderMap::new();
+        headers.insert(RUSTFS_BUCKET_REPLICATION_REQUEST, REPLICATION_REQUEST_TRUE.clone());
+        let valid_mtime = "2024-05-20T10:30:00+08:00";
+        headers.insert(RUSTFS_BUCKET_SOURCE_MTIME, HeaderValue::from_static(valid_mtime));
+
+        let metadata = HashMap::new();
+
+        let result = put_opts_from_headers(&headers, metadata);
+
+        assert!(result.is_ok());
+        let opts = result.unwrap();
+
+        assert!(opts.replication_request);
+
+        let expected_mtime = time::OffsetDateTime::parse(valid_mtime, &time::format_description::well_known::Rfc3339).unwrap();
+        assert_eq!(opts.mod_time, Some(expected_mtime));
+
+        let mut headers_invalid_mtime = HeaderMap::new();
+        headers_invalid_mtime.insert(RUSTFS_BUCKET_REPLICATION_REQUEST, REPLICATION_REQUEST_TRUE.clone());
+        headers_invalid_mtime.insert(RUSTFS_BUCKET_SOURCE_MTIME, HeaderValue::from_static("invalid-time"));
+        let result_invalid = put_opts_from_headers(&headers_invalid_mtime, HashMap::new());
+        assert!(result_invalid.is_ok());
+        let opts_invalid = result_invalid.unwrap();
+        assert!(opts_invalid.replication_request);
+        assert!(opts_invalid.mod_time.is_none());
+    }
+
     #[test]
     fn test_get_default_opts_with_metadata() {
         let headers = create_test_headers();