diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index e5eccf385..761ec5ea2 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -64,6 +64,8 @@ use uuid::Uuid; const DEFAULT_HEALTH_CHECK_DURATION: Duration = Duration::from_secs(5); const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60); +const REPLICATION_REQUEST_TRUE: HeaderValue = HeaderValue::from_static("true"); + pub static GLOBAL_BUCKET_TARGET_SYS: OnceLock = OnceLock::new(); #[derive(Debug, Clone)] @@ -999,10 +1001,14 @@ impl PutObjectOptions { if self.internal.source_mtime.unix_timestamp() != 0 { header.insert( RUSTFS_BUCKET_SOURCE_MTIME, - HeaderValue::from_str(&self.internal.source_mtime.unix_timestamp().to_string()).expect("err"), + HeaderValue::from_str(&self.internal.source_mtime.format(&Rfc3339).unwrap_or_default()).expect("err"), ); } + if self.internal.replication_request { + header.insert(RUSTFS_BUCKET_REPLICATION_REQUEST, REPLICATION_REQUEST_TRUE); + } + header } @@ -1211,6 +1217,9 @@ impl TargetClient { { headers.insert(RUSTFS_BUCKET_SOURCE_VERSION_ID, header_value); } + if opts.internal.replication_request { + headers.insert(RUSTFS_BUCKET_REPLICATION_REQUEST, REPLICATION_REQUEST_TRUE); + } match self .client diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index d7dea95ea..e21db787d 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -1684,10 +1684,6 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, storage: if let Some(ref s) = new_replication_internal { eval_metadata.insert(format!("{RESERVED_METADATA_PREFIX_LOWER}replication-status"), s.clone()); } - eval_metadata.insert( - format!("{RESERVED_METADATA_PREFIX_LOWER}replication-timestamp"), - OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(), - ); let popts = ObjectOptions { version_id: roi.version_id.map(|v| v.to_string()), eval_metadata: Some(eval_metadata), @@ -1840,6 +1836,36 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } + let mut replication_action = replication_action; + match tgt_client + .head_object(&tgt_client.bucket, &object, self.version_id.map(|v| v.to_string())) + .await + { + Ok(oi) => { + replication_action = get_replication_action(&object_info, &oi, self.op_type); + if replication_action == ReplicationAction::None { + rinfo.replication_status = ReplicationStatusType::Completed; + rinfo.replication_resynced = true; + rinfo.replication_action = ReplicationAction::None; + rinfo.size = size; + return rinfo; + } + } + Err(e) => { + if let Some(se) = e.as_service_error() { + if !se.is_not_found() { + rinfo.error = Some(e.to_string()); + warn!("replication head_object failed bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + return rinfo; + } + } else { + rinfo.error = Some(e.to_string()); + warn!("replication head_object failed bucket:{} arn:{} error:{}", bucket, tgt_client.arn, e); + return rinfo; + } + } + } + rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; rinfo.size = size; diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index a35b35366..2a74d15c8 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -4786,7 +4786,9 @@ impl StorageAPI for SetDisks { } } - fi.mod_time = opts.mod_time; + if opts.mod_time.is_some() { + fi.mod_time = opts.mod_time; + } if let Some(ref version_id) = opts.version_id { fi.version_id = Uuid::parse_str(version_id).ok(); } diff --git a/crates/utils/src/http/headers.rs b/crates/utils/src/http/headers.rs index 9d5199f5c..243672556 100644 --- a/crates/utils/src/http/headers.rs +++ b/crates/utils/src/http/headers.rs @@ -175,7 +175,7 @@ pub const RUSTFS_REPLICATION_RESET_STATUS: &str = "X-Rustfs-Replication-Reset-St pub const RUSTFS_REPLICATION_ACTUAL_OBJECT_SIZE: &str = "X-Rustfs-Replication-Actual-Object-Size"; pub const RUSTFS_BUCKET_SOURCE_VERSION_ID: &str = "X-Rustfs-Source-Version-Id"; -pub const RUSTFS_BUCKET_SOURCE_MTIME: &str = "X-Rustfs-Source-Mtime"; +pub const RUSTFS_BUCKET_SOURCE_MTIME: &str = "X-RustFS-Source-Mtime"; pub const RUSTFS_BUCKET_SOURCE_ETAG: &str = "X-Rustfs-Source-Etag"; pub const RUSTFS_BUCKET_REPLICATION_DELETE_MARKER: &str = "X-Rustfs-Source-DeleteMarker"; pub const RUSTFS_BUCKET_REPLICATION_PROXY_REQUEST: &str = "X-Rustfs-Source-Proxy-Request"; diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index ad48fb64a..2372bc70d 100644 --- a/rustfs/src/storage/options.rs +++ b/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) -> 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, metadata: HashMap) -> Result { - 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();