From 8d191c7653c947fad11e218e6ba9efb558dca621 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sat, 15 Aug 2026 10:40:54 +0800 Subject: [PATCH] fix(replication): transport and persist LWW timestamps for tag, retention, and legal hold Active-active conflict resolution for concurrent tag/retention/legal-hold edits needs the source's per-category modification times on both sides of the wire; the three AdvancedPutOptions timestamp fields were dead and the headers were neither sent nor parsed. - Emit x-{rustfs,minio}-source-replication-{tagging,retention,legalhold}- timestamp from PutObjectOptions::header(); names and RFC3339 values interoperate with MinIO (minio-go constants.go, object-api-options.go), pinned by a header_compat wire-name test. - Default the three AdvancedPutOptions timestamps to UNIX_EPOCH and skip epoch values in header(), so "never modified" is not sent as a modification made now. - Parse the headers only on authorized replication PUTs and multipart completes, expose them as Option on ObjectOptions, and persist them into the dual-prefix internal metadata keys so the outbound pass (replication_target_boundary) reads the source's timestamps instead of the mod_time fallback. - Record the local tagging timestamp in the PutObjectTagging and DeleteObjectTagging eval metadata, mirroring the object-lock handlers; without it the sender only ever had the mod_time fallback to offer. Receiver-side LWW comparison (keep newer stored category metadata over a stale inbound copy) is left as a TODO at the parse site. --- .../ecstore/src/bucket/bucket_target_sys.rs | 23 +++++-- crates/ecstore/src/object_api/types.rs | 6 ++ crates/utils/src/http/header_compat.rs | 29 ++++++++ rustfs/src/storage/ecfs.rs | 12 +++- rustfs/src/storage/options.rs | 67 +++++++++++++++++-- 5 files changed, 128 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 2e5232eea..c6bcc5742 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -58,7 +58,9 @@ use rustfs_utils::http::{ }; use rustfs_utils::http::{ SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, - SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header, + SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, + SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, + insert_header, }; use rustls_pki_types::pem::PemObject; use serde::{Deserialize, Serialize}; @@ -1476,9 +1478,12 @@ impl Default for AdvancedPutOptions { replication_status: ReplicationStatusType::Pending, source_mtime: OffsetDateTime::now_utc(), replication_request: false, - retention_timestamp: OffsetDateTime::now_utc(), - tagging_timestamp: OffsetDateTime::now_utc(), - legalhold_timestamp: OffsetDateTime::now_utc(), + // UNIX_EPOCH means "never modified": header() must not emit a + // timestamp header for it, otherwise a receiver would treat an + // unset category as a modification made right now. + retention_timestamp: OffsetDateTime::UNIX_EPOCH, + tagging_timestamp: OffsetDateTime::UNIX_EPOCH, + legalhold_timestamp: OffsetDateTime::UNIX_EPOCH, replication_validity_check: false, } } @@ -1675,6 +1680,16 @@ impl PutObjectOptions { ); } + for (suffix, timestamp) in [ + (SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, self.internal.tagging_timestamp), + (SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, self.internal.retention_timestamp), + (SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, self.internal.legalhold_timestamp), + ] { + if timestamp.unix_timestamp() != 0 { + insert_header(&mut header, suffix, timestamp.format(&Rfc3339).unwrap_or_default()); + } + } + if self.internal.replication_request { insert_header(&mut header, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); } diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 291b9039a..624cb7593 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -277,6 +277,12 @@ pub struct ObjectOptions { /// fence avoids recursively acquiring the read lock behind a queued writer. pub bucket_lifecycle_lock_fence: Option, pub replication_request: bool, + /// Source-cluster LWW timestamps carried by an authorized replication + /// request; None when the source never modified the category. Only the + /// replication-authorized options builders may set these. + pub replication_tagging_timestamp: Option, + pub replication_retention_timestamp: Option, + pub replication_legalhold_timestamp: Option, /// Authorized SSE-C replication passthrough: the body is already /// ciphertext, so the write path must not encrypt or compress it and /// stores the restored encryption metadata verbatim. Only the diff --git a/crates/utils/src/http/header_compat.rs b/crates/utils/src/http/header_compat.rs index 38486ca07..c36632d8e 100644 --- a/crates/utils/src/http/header_compat.rs +++ b/crates/utils/src/http/header_compat.rs @@ -50,6 +50,14 @@ pub const SUFFIX_SOURCE_DELETEMARKER: &str = "source-deletemarker"; pub const SUFFIX_SOURCE_PROXY_REQUEST: &str = "source-proxy-request"; pub const SUFFIX_SOURCE_REPLICATION_REQUEST: &str = "source-replication-request"; pub const SUFFIX_SOURCE_REPLICATION_CHECK: &str = "source-replication-check"; +// LWW timestamps for replicated tag/retention/legal-hold modifications. MinIO +// declares these with mixed case (internal/http/headers.go: +// X-Minio-Source-Replication-Tagging-Timestamp / -Retention-Timestamp / +// -LegalHold-Timestamp); HTTP header names compare case-insensitively, so the +// lowercase suffix forms interoperate. Values are RFC3339 on the wire. +pub const SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP: &str = "source-replication-tagging-timestamp"; +pub const SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP: &str = "source-replication-retention-timestamp"; +pub const SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP: &str = "source-replication-legalhold-timestamp"; pub const SUFFIX_REPLICATION_SSEC_CRC: &str = "replication-ssec-crc"; /// Returns true if the key is object-encryption metadata understood by RustFS or MinIO. @@ -196,6 +204,27 @@ mod tests { assert_eq!(get_object_encryption_original_size(&metadata).expect("valid size"), Some(42)); } + #[test] + fn replication_timestamp_headers_match_minio_wire_names() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z"); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, "2026-01-02T03:04:06Z"); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, "2026-01-02T03:04:07Z"); + + // The exact names MinIO's object-api-options.go reads (its Get() + // canonicalizes case, so a case-insensitive match is wire-equivalent). + for name in [ + "X-Minio-Source-Replication-Tagging-Timestamp", + "X-Minio-Source-Replication-Retention-Timestamp", + "X-Minio-Source-Replication-LegalHold-Timestamp", + "x-rustfs-source-replication-tagging-timestamp", + "x-rustfs-source-replication-retention-timestamp", + "x-rustfs-source-replication-legalhold-timestamp", + ] { + assert!(headers.contains_key(name), "replication timestamp header {name} must be written"); + } + } + #[test] fn test_get_header() { let mut headers = HeaderMap::new(); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index ce4be49d5..9369d7224 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -45,7 +45,7 @@ use rustfs_targets::EventName; use rustfs_utils::http::headers::{ AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, }; -use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str}; +use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, insert_str}; use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error}; use std::collections::HashMap; use std::fmt::Debug; @@ -461,6 +461,11 @@ impl S3 for FS { let mut eval_metadata = HashMap::new(); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default()); + insert_str( + &mut eval_metadata, + SUFFIX_TAGGING_TIMESTAMP, + OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(), + ); opts.eval_metadata = Some(eval_metadata); } @@ -1645,6 +1650,11 @@ impl S3 for FS { let mut eval_metadata = HashMap::new(); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default()); + insert_str( + &mut eval_metadata, + SUFFIX_TAGGING_TIMESTAMP, + OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_default(), + ); opts.eval_metadata = Some(eval_metadata); } diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index f2bc64c6f..4028afa64 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -17,11 +17,13 @@ use crate::storage::storage_api::options_consumer::contract::{object::HTTPPrecon use http::header::{IF_MATCH, IF_NONE_MATCH}; use http::{HeaderMap, HeaderValue}; use rustfs_utils::http::{ - AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, - SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_REQUEST, - SUFFIX_SOURCE_VERSION_ID, get_header, + AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, + SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC, + SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, + SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header, header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX}, - insert_header_map, + insert_header_map, insert_str, metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX}, }; use rustfs_utils::http::{ @@ -422,6 +424,9 @@ pub fn get_complete_multipart_upload_opts_with_replication_authorization( preserve_etag, ..Default::default() }; + if replication_request { + apply_replication_timestamps_from_headers(headers, &mut opts); + } apply_replica_status_from_headers(headers, &mut opts, replication_request_authorized); fill_conditional_writes_opts_from_header(headers, &mut opts)?; @@ -479,6 +484,7 @@ pub fn put_opts_from_headers_with_replication_authorization( if let Some(crc) = get_header(headers, SUFFIX_REPLICATION_SSEC_CRC) { insert_header_map(&mut opts.user_defined, SUFFIX_REPLICATION_SSEC_CRC, crc.into_owned()); } + apply_replication_timestamps_from_headers(headers, &mut opts); } Ok(opts) } @@ -504,6 +510,47 @@ fn replication_source_mtime(headers: &HeaderMap) -> Option, suffix: &str) -> Option { + let value = get_header(headers, suffix)?; + let value = value.trim(); + match time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339) { + Ok(timestamp) => Some(timestamp), + Err(err) => { + tracing::warn!("Invalid {} value '{}' (replication request=true): {}", suffix, value, err); + None + } + } +} + +/// Callers must gate on an authorized replication request: these headers are +/// trusted source-cluster state, not client input. +fn apply_replication_timestamps_from_headers(headers: &HeaderMap, opts: &mut ObjectOptions) { + opts.replication_tagging_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP); + opts.replication_retention_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP); + opts.replication_legalhold_timestamp = replication_timestamp_header(headers, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP); + + // Persist into the internal metadata keys so a later outbound replication + // pass (replication_target_boundary) reads the source's modification + // times instead of falling back to mod_time. + // TODO(P1-6): receiver-side LWW is still missing — when the stored + // per-category timestamp is newer than the inbound one, the existing + // tags/retention/legal-hold should win instead of being overwritten. + for (timestamp, suffix) in [ + (opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP), + (opts.replication_retention_timestamp, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP), + (opts.replication_legalhold_timestamp, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP), + ] { + if let Some(timestamp) = timestamp + && let Ok(value) = timestamp.format(&time::format_description::well_known::Rfc3339) + { + insert_str(&mut opts.user_defined, suffix, value); + } + } +} + fn apply_replica_status_from_headers(headers: &HeaderMap, opts: &mut ObjectOptions, authorized: bool) { if !authorized { return; @@ -1622,9 +1669,18 @@ mod tests { "unauthorized clients must not persist the {suffix} internal key" ); } + assert!(untrusted.replication_tagging_timestamp.is_none()); + assert!(untrusted.replication_retention_timestamp.is_none()); + assert!(untrusted.replication_legalhold_timestamp.is_none()); let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) .expect("authorized replication request should parse"); + let parse = |value: &str| { + time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).expect("valid RFC3339") + }; + assert_eq!(trusted.replication_tagging_timestamp, Some(parse("2026-01-02T03:04:05Z"))); + assert_eq!(trusted.replication_retention_timestamp, Some(parse("2026-01-02T03:04:06Z"))); + assert_eq!(trusted.replication_legalhold_timestamp, Some(parse("2026-01-02T03:04:07Z"))); for (suffix, expected) in [ ("tagging-timestamp", "2026-01-02T03:04:05Z"), ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"), @@ -1672,6 +1728,9 @@ mod tests { "authorized multipart completes must persist the {suffix} internal key" ); } + assert!(trusted.replication_tagging_timestamp.is_some()); + assert!(trusted.replication_retention_timestamp.is_some()); + assert!(trusted.replication_legalhold_timestamp.is_some()); } #[test]