diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index d094f8cdb..c8ddcecf3 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -76,6 +76,18 @@ const SOURCE_MTIME_HEADERS: [&str; 2] = ["x-rustfs-source-mtime", "x-minio-sourc const SOURCE_REPLICATION_REQUEST_HEADERS: [&str; 2] = ["x-rustfs-source-replication-request", "x-minio-source-replication-request"]; const SOURCE_ETAG_HEADERS: [&str; 2] = ["x-rustfs-source-etag", "x-minio-source-etag"]; +const SOURCE_TAGGING_TIMESTAMP_HEADERS: [&str; 2] = [ + "x-rustfs-source-replication-tagging-timestamp", + "x-minio-source-replication-tagging-timestamp", +]; +const SOURCE_RETENTION_TIMESTAMP_HEADERS: [&str; 2] = [ + "x-rustfs-source-replication-retention-timestamp", + "x-minio-source-replication-retention-timestamp", +]; +const SOURCE_LEGALHOLD_TIMESTAMP_HEADERS: [&str; 2] = [ + "x-rustfs-source-replication-legalhold-timestamp", + "x-minio-source-replication-legalhold-timestamp", +]; const RESERVED_BUCKET_PREFIXES: [&str; 3] = ["xn--", "sthree-", "amzn-s3-demo-"]; const RESERVED_BUCKET_SUFFIXES: [&str; 6] = ["-s3alias", "--ol-s3", ".mrap", "--x-s3", "--table-s3", "-an"]; @@ -118,6 +130,25 @@ pub enum FaultAction { WrongEtag, } +/// Replication LWW timestamp headers observed on a request, journaled so +/// sender-side tests can assert what a real target would receive. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ReplicationTimestampHeaders { + pub tagging: Option, + pub retention: Option, + pub legalhold: Option, +} + +impl ReplicationTimestampHeaders { + fn from_headers(headers: &HeaderMap) -> Self { + Self { + tagging: header_value(headers, &SOURCE_TAGGING_TIMESTAMP_HEADERS).map(bounded_journal_value), + retention: header_value(headers, &SOURCE_RETENTION_TIMESTAMP_HEADERS).map(bounded_journal_value), + legalhold: header_value(headers, &SOURCE_LEGALHOLD_TIMESTAMP_HEADERS).map(bounded_journal_value), + } + } +} + /// Credential-free request metadata retained for deterministic assertions. #[derive(Debug, Clone, PartialEq, Eq)] pub struct RequestRecord { @@ -131,6 +162,7 @@ pub struct RequestRecord { pub part_number: Option, pub content_length: Option, pub consumed_bytes: Option, + pub replication_timestamps: ReplicationTimestampHeaders, pub fault: Option, } @@ -536,7 +568,15 @@ impl S3Access for FaultAccess { .get(CONTENT_LENGTH) .and_then(|value| value.to_str().ok()) .and_then(|value| value.parse().ok()); - let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length); + let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers()); + let fault = record_request( + &self.control, + operation, + context.method().clone(), + parsed, + content_length, + replication_timestamps, + ); if let Some(RequestFault { action: FaultAction::Status(status), .. @@ -589,6 +629,7 @@ fn record_request( method: Method, parsed: ParsedRequest, content_length: Option, + replication_timestamps: ReplicationTimestampHeaders, ) -> Option { let mut state = lock(control); let action = parsed @@ -613,6 +654,7 @@ fn record_request( part_number: parsed.part_number, content_length, consumed_bytes: None, + replication_timestamps, fault: action.clone(), }); action.map(|action| RequestFault { sequence, action }) @@ -1699,6 +1741,52 @@ mod tests { .await?) } + #[tokio::test] + async fn journals_replication_timestamp_headers() -> Result<(), BoxError> { + let target = FakeS3Target::start().await?; + target.create_bucket("target-bucket"); + let client = client(&target); + + client + .put_object() + .bucket("target-bucket") + .key("plain") + .body(ByteStream::from_static(b"plain")) + .send() + .await?; + client + .put_object() + .bucket("target-bucket") + .key("stamped") + .body(ByteStream::from_static(b"stamped")) + .customize() + .map_request(move |mut request| { + let headers = request.headers_mut(); + headers.insert("x-rustfs-source-replication-tagging-timestamp", "2026-01-02T03:04:05Z"); + headers.insert("x-minio-source-replication-retention-timestamp", "2026-01-02T03:04:06Z"); + headers.insert("x-rustfs-source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z"); + Ok::<_, std::convert::Infallible>(request) + }) + .send() + .await?; + + let requests = target.requests(); + let plain = requests + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("plain")) + .expect("plain PUT must be journaled"); + assert_eq!(plain.replication_timestamps, ReplicationTimestampHeaders::default()); + + let stamped = requests + .iter() + .find(|record| record.operation == Operation::PutObject && record.key.as_deref() == Some("stamped")) + .expect("stamped PUT must be journaled"); + assert_eq!(stamped.replication_timestamps.tagging.as_deref(), Some("2026-01-02T03:04:05Z")); + assert_eq!(stamped.replication_timestamps.retention.as_deref(), Some("2026-01-02T03:04:06Z")); + assert_eq!(stamped.replication_timestamps.legalhold.as_deref(), Some("2026-01-02T03:04:07Z")); + Ok(()) + } + macro_rules! assert_sdk_error { ($error:expr, $status:expr, $code:expr) => {{ let error = &$error; @@ -2985,6 +3073,7 @@ mod tests { part_number: None, }, Some(0), + ReplicationTimestampHeaders::default(), ); } let records = lock(&control).requests.clone(); @@ -3006,6 +3095,7 @@ mod tests { part_number: None, }, None, + ReplicationTimestampHeaders::default(), ); { let bounded_records = lock(&bounded_control); diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 1e1e18dc0..2e5232eea 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -2842,6 +2842,57 @@ mod tests { ); } + #[test] + fn put_object_headers_carry_replication_timestamp_headers() { + // MinIO receivers resolve concurrent tag/retention/legal-hold edits by + // last-writer-wins on these headers (object-api-options.go parses them + // as RFC3339); a replica without them loses every conflict resolution. + let mut opts = PutObjectOptions::default(); + opts.internal.replication_request = true; + let tagging = OffsetDateTime::from_unix_timestamp(1_700_000_001).expect("valid timestamp"); + let retention = OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("valid timestamp"); + let legalhold = OffsetDateTime::from_unix_timestamp(1_700_000_003).expect("valid timestamp"); + opts.internal.tagging_timestamp = tagging; + opts.internal.retention_timestamp = retention; + opts.internal.legalhold_timestamp = legalhold; + + let header = opts.header(); + for (suffix, expected) in [ + ("source-replication-tagging-timestamp", tagging), + ("source-replication-retention-timestamp", retention), + ("source-replication-legalhold-timestamp", legalhold), + ] { + assert_eq!( + rustfs_utils::http::get_header(&header, suffix).as_deref(), + Some(expected.format(&Rfc3339).expect("RFC3339 timestamp").as_str()), + "replication put requests must carry the {suffix} header" + ); + } + } + + #[test] + fn put_object_headers_omit_unset_replication_timestamps() { + // UNIX_EPOCH means "never modified on the source"; sending it would + // make the receiver treat an unset category as a fresh modification. + let mut opts = PutObjectOptions::default(); + opts.internal.replication_request = true; + opts.internal.tagging_timestamp = OffsetDateTime::UNIX_EPOCH; + opts.internal.retention_timestamp = OffsetDateTime::UNIX_EPOCH; + opts.internal.legalhold_timestamp = OffsetDateTime::UNIX_EPOCH; + + let header = opts.header(); + for suffix in [ + "source-replication-tagging-timestamp", + "source-replication-retention-timestamp", + "source-replication-legalhold-timestamp", + ] { + assert!( + rustfs_utils::http::get_header(&header, suffix).is_none(), + "unset {suffix} must not be sent to replication targets" + ); + } + } + #[tokio::test] async fn get_remote_target_client_internal_rejects_loopback_endpoint() { let sys = BucketTargetSys::default(); diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index 609b2ea62..f2bc64c6f 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -1599,6 +1599,81 @@ mod tests { assert!(opts_invalid.mod_time.is_none()); } + #[test] + fn test_put_opts_from_headers_gates_replication_timestamp_persistence_on_authorization() { + // Sender-side LWW state (replication_target_boundary.rs) is read back + // from these internal metadata keys, so an authorized replication PUT + // must persist the inbound timestamp headers; an unauthorized client + // must not be able to forge them. + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z"); + insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z"); + insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z"); + + let untrusted = put_opts_from_headers(&headers, HashMap::new()).expect("ordinary PUT options should be created"); + for suffix in [ + "tagging-timestamp", + "objectlock-retention-timestamp", + "objectlock-legalhold-timestamp", + ] { + assert!( + rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(), + "unauthorized clients must not persist the {suffix} internal key" + ); + } + + let trusted = put_opts_from_headers_with_replication_authorization(&headers, HashMap::new(), true) + .expect("authorized replication request should parse"); + for (suffix, expected) in [ + ("tagging-timestamp", "2026-01-02T03:04:05Z"), + ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"), + ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"), + ] { + assert_eq!( + rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(), + Some(expected), + "authorized replication must persist the {suffix} internal key" + ); + } + } + + #[test] + fn test_complete_multipart_opts_persist_replication_timestamps_when_authorized() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true"); + insert_header(&mut headers, "replication-actual-object-size", "1"); + insert_header(&mut headers, "source-replication-tagging-timestamp", "2026-01-02T03:04:05Z"); + insert_header(&mut headers, "source-replication-retention-timestamp", "2026-01-02T03:04:06Z"); + insert_header(&mut headers, "source-replication-legalhold-timestamp", "2026-01-02T03:04:07Z"); + + let untrusted = get_complete_multipart_upload_opts(&headers).expect("ordinary multipart options should be created"); + for suffix in [ + "tagging-timestamp", + "objectlock-retention-timestamp", + "objectlock-legalhold-timestamp", + ] { + assert!( + rustfs_utils::http::get_str(&untrusted.user_defined, suffix).is_none(), + "unauthorized multipart completes must not persist the {suffix} internal key" + ); + } + + let trusted = get_complete_multipart_upload_opts_with_replication_authorization(&headers, true) + .expect("authorized multipart complete should parse"); + for (suffix, expected) in [ + ("tagging-timestamp", "2026-01-02T03:04:05Z"), + ("objectlock-retention-timestamp", "2026-01-02T03:04:06Z"), + ("objectlock-legalhold-timestamp", "2026-01-02T03:04:07Z"), + ] { + assert_eq!( + rustfs_utils::http::get_str(&trusted.user_defined, suffix).as_deref(), + Some(expected), + "authorized multipart completes must persist the {suffix} internal key" + ); + } + } + #[test] fn test_put_opts_from_headers_with_replica_status() { let mut headers = HeaderMap::new();