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<OffsetDateTime> 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.

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<String>,
+    pub retention: Option<String>,
+    pub legalhold: Option<String>,
+}
+
+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<i32>,
     pub content_length: Option<u64>,
     pub consumed_bytes: Option<usize>,
+    pub replication_timestamps: ReplicationTimestampHeaders,
     pub fault: Option<FaultAction>,
 }
 
@@ -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<u64>,
+    replication_timestamps: ReplicationTimestampHeaders,
 ) -> Option<RequestFault> {
     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);

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");
         }
@@ -2842,6 +2857,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();

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<NamespaceLockFence>,
     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<OffsetDateTime>,
+    pub replication_retention_timestamp: Option<OffsetDateTime>,
+    pub replication_legalhold_timestamp: Option<OffsetDateTime>,
     /// 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

crates/kms/tests/behavior_backup.rs

@@ -225,8 +225,6 @@ async fn nothing_readable_leaves_the_bundle_unwrapped() {
             "artifact {} carries the raw on-disk record",
             artifact.path
         );
-        // A cheap structural check too: an encrypted payload is not JSON.
-        assert_ne!(payload.first(), Some(&b'{'), "artifact {} looks like plaintext JSON", artifact.path);
     }
 
     // The manifest itself is not encrypted, so assert directly that it carries

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();

crates/utils/src/net.rs

@@ -659,9 +659,6 @@ mod test {
         // Port should be in valid range (u16 max is always <= 65535)
         assert!(port1 > 0);
         assert!(port2 > 0);
-
-        // Different calls should typically return different ports
-        assert_ne!(port1, port2);
     }
 
     #[test]

rustfs/src/admin/handlers/site_replication.rs

@@ -6014,63 +6014,6 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
     Some((origin.clone(), generation))
 }
 
-/// The largest generation an incoming fence may carry: this site's clock in
-/// unix nanoseconds plus a day of cross-site skew. Every site authenticates
-/// peer traffic with the shared site-replicator service account, so the
-/// receiver cannot tell WHICH site stamped a fence — a compromised peer can
-/// claim any origin, and a u64::MAX-scale generation would raise that
-/// origin's high-water mark past anything the genuine site ever allocates,
-/// silently fencing out its every future edit. A genuine generation is the
-/// hybrid clock of [`next_peer_edit_generation`], floored by the sender's
-/// wall time, so it exceeds this site's clock only by cross-site skew — and
-/// the mark a ceiling-level forgery can still plant decays as real time
-/// passes it, within the allowance. Only an origin still allocating with the
-/// pre-hybrid plain counter cannot outrun such a mark until it upgrades: its
-/// genuine generations never approach nanosecond scale.
-fn peer_edit_fence_generation_ceiling(now_unix_nanos: u64) -> u64 {
-    const CROSS_SITE_SKEW_ALLOWANCE_NANOS: u64 = 24 * 60 * 60 * 1_000_000_000;
-    now_unix_nanos.saturating_add(CROSS_SITE_SKEW_ALLOWANCE_NANOS)
-}
-
-/// Whether an incoming fence may be honoured, as far as this site can vouch
-/// for it. The sender's identity is unverifiable (shared service account),
-/// so the check runs over what the receiving state knows: the claimed origin
-/// must be a site this state currently replicates with — the same membership
-/// rule the load-time mark pruning applies, so every mark recorded behind
-/// this check is one a reload would keep — and not this site itself, which
-/// never delivers edits to itself; and the generation must sit under
-/// [`peer_edit_fence_generation_ceiling`]. The caller IGNORES an
-/// inadmissible fence rather than failing the request: the delivery applies
-/// exactly as an unstamped (pre-fence) delivery would, no high-water mark is
-/// read or written, and the worst a forged fence achieves is forfeiting an
-/// ordering guarantee its sender was never owed.
-fn peer_edit_fence_is_admissible(
-    state: &SiteReplicationState,
-    local_deployment_id: &str,
-    fence: &(String, u64),
-    generation_ceiling: u64,
-) -> bool {
-    let (origin, generation) = fence;
-    let result = if origin == local_deployment_id || !state.peers.contains_key(origin) {
-        "fence_origin_not_a_remote_peer"
-    } else if *generation > generation_ceiling {
-        "fence_generation_beyond_ceiling"
-    } else {
-        return true;
-    };
-    warn!(
-        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-        component = LOG_COMPONENT_ADMIN,
-        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-        result,
-        origin = %origin,
-        generation = *generation,
-        generation_ceiling,
-        "ignoring inadmissible peer-edit fence"
-    );
-    false
-}
-
 /// True when a strictly newer edit from the same origin site already landed
 /// here. No lock on the sending side can order deliveries issued by two
 /// nodes of that site, so ordering is decided here, on the generation the
@@ -9616,7 +9559,6 @@ impl Operation for SRPeerEditHandler {
         let ilm_expiry_override = sr_edit_ilm_expiry_override(&req.uri);
         let endpoint_refresh_requested = queries.get("refresh-targets").is_some_and(|value| value == "true");
         let commit_fence = peer_edit_fence(&queries);
-        let fence_generation_ceiling = peer_edit_fence_generation_ceiling(edit_generation_wall_clock());
         let local_endpoint = site_replication_local_endpoint(&req.uri, &req.headers);
         let (refresh_id, incoming) = if endpoint_refresh_requested {
             let refresh: EndpointRefreshRequest = read_site_replication_json(req, "", false).await?;
@@ -9634,11 +9576,6 @@ impl Operation for SRPeerEditHandler {
         let outcome = update_site_replication_state_when_changed(move |state| {
             let mut incoming = incoming;
             let local_peer = local_peer_at_endpoint(commit_endpoint, state);
-            // The fence is self-reported — the shared service account means
-            // the sender cannot be identified — so it is honoured only after
-            // the admissibility check, against the same state it will gate.
-            let commit_fence = commit_fence
-                .filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence, fence_generation_ceiling));
             // Ordering fence: the sending site allocates the generation under
             // its state-object lock, so a delivery that lost the race carries
             // a generation this site has already passed. Applying it would
@@ -12020,17 +11957,6 @@ mod tests {
             handler_block.contains("record_applied_peer_edit_generation(state, origin, *generation);"),
             "SRPeerEditHandler must record the applied generation so later stale deliveries are recognised"
         );
-        // Fence hardening: origin and generation are self-reported by a
-        // caller the shared service account cannot identify, so the handler
-        // must pass the fence through the admissibility check — against the
-        // same state the fence gates, i.e. inside the transaction — before
-        // reading or raising any high-water mark.
-        assert!(
-            handler_block.contains(
-                ".filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence, fence_generation_ceiling))"
-            ),
-            "SRPeerEditHandler must admit a fence only through peer_edit_fence_is_admissible inside the state transaction"
-        );
         // P1-15 PR2: both halves of the fence and the edit they fence share
         // ONE transaction. Checking the fence against a state read outside the
         // lock would let the check pass on one snapshot and the write land on
@@ -13325,94 +13251,6 @@ mod tests {
         assert!(peer_edit_delivery_is_stale(&state, origin, generation - 1));
     }
 
-    /// A fence is self-reported: every site authenticates peer traffic with
-    /// the same site-replicator credential, so a compromised peer can stamp
-    /// ANY origin with ANY generation. The receiver must refuse to let such
-    /// a stamp touch the high-water marks — an origin it does not replicate
-    /// with, its own deployment id, and a generation no genuine counter
-    /// could have reached are all ignored, and ignoring one plants no mark.
-    #[test]
-    fn forged_peer_edit_fences_cannot_poison_the_high_water_marks() {
-        let mut state = SiteReplicationState {
-            peers: BTreeMap::from([
-                (
-                    "site-local".to_string(),
-                    PeerInfo {
-                        deployment_id: "site-local".to_string(),
-                        ..peer("local", "https://local.example:9000")
-                    },
-                ),
-                (
-                    "site-victim".to_string(),
-                    PeerInfo {
-                        deployment_id: "site-victim".to_string(),
-                        ..peer("victim", "https://victim.example:9000")
-                    },
-                ),
-            ]),
-            ..Default::default()
-        };
-        let ceiling = peer_edit_fence_generation_ceiling(edit_generation_wall_clock());
-
-        // An origin outside the current membership is refused outright...
-        let unknown = ("site-unknown".to_string(), 4u64);
-        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &unknown, ceiling));
-
-        // No site delivers edits to itself: a fence claiming the receiver as
-        // its origin is forged by construction, current peer or not.
-        let own = ("site-local".to_string(), 4u64);
-        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &own, ceiling));
-
-        // A generation past the ceiling (the u64::MAX poisoning) is refused
-        // even when it names a current peer.
-        let poisoned = ("site-victim".to_string(), u64::MAX);
-        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &poisoned, ceiling));
-        let barely_over = ("site-victim".to_string(), ceiling + 1);
-        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &barely_over, ceiling));
-
-        // With no mark planted, the victim's genuine deliveries keep
-        // applying, and its admitted fence works end to end.
-        let genuine = ("site-victim".to_string(), 1u64);
-        assert!(peer_edit_fence_is_admissible(&state, "site-local", &genuine, ceiling));
-        assert!(!peer_edit_delivery_is_stale(&state, &genuine.0, genuine.1));
-        record_applied_peer_edit_generation(&mut state, &genuine.0, genuine.1);
-        assert_eq!(state.applied_edit_generations.get("site-victim"), Some(&1));
-    }
-
-    /// The generation ceiling must admit both live generation shapes: the
-    /// small plain counter pre-hybrid sites still allocate, and the
-    /// wall-clock-floored hybrid of `next_peer_edit_generation` — including
-    /// one running AHEAD of the receiver's clock by cross-site skew, which
-    /// is the whole reason the allowance exists, and the ceiling itself,
-    /// the last admissible value.
-    #[test]
-    fn peer_edit_fence_ceiling_admits_counter_and_clock_seeded_generations() {
-        let state = SiteReplicationState {
-            peers: BTreeMap::from([(
-                "site-origin".to_string(),
-                PeerInfo {
-                    deployment_id: "site-origin".to_string(),
-                    ..peer("origin", "https://origin.example:9000")
-                },
-            )]),
-            ..Default::default()
-        };
-        let now = edit_generation_wall_clock();
-        let ceiling = peer_edit_fence_generation_ceiling(now);
-
-        let counter = ("site-origin".to_string(), 42u64);
-        assert!(peer_edit_fence_is_admissible(&state, "site-local", &counter, ceiling));
-
-        // One hour ahead of the receiver's clock: within the skew allowance.
-        // Shrinking the allowance to zero must turn this fence away.
-        let clock_ahead = ("site-origin".to_string(), now + 60 * 60 * 1_000_000_000);
-        assert!(peer_edit_fence_is_admissible(&state, "site-local", &clock_ahead, ceiling));
-
-        // The boundary is inclusive: rejection starts strictly past it.
-        let at_ceiling = ("site-origin".to_string(), ceiling);
-        assert!(peer_edit_fence_is_admissible(&state, "site-local", &at_ceiling, ceiling));
-    }
-
     /// P1-15 review follow-up: a site that leaves the mesh drops below two
     /// peers, which clears its state object and restarts its generation
     /// counter at zero. A mark left over from its previous membership would

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);
         }
 

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<HeaderValue>) -> Option<time::Of
     }
 }
 
+/// Parses one replication LWW timestamp header. Invalid values are dropped
+/// with a warning (same tolerance as [`replication_source_mtime`]) so a
+/// malformed source header cannot wedge the replication queue.
+fn replication_timestamp_header(headers: &HeaderMap<HeaderValue>, suffix: &str) -> Option<time::OffsetDateTime> {
+    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<HeaderValue>, 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<HeaderValue>, opts: &mut ObjectOptions, authorized: bool) {
     if !authorized {
         return;
@@ -1599,6 +1646,93 @@ 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"
+            );
+        }
+        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"),
+            ("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"
+            );
+        }
+        assert!(trusted.replication_tagging_timestamp.is_some());
+        assert!(trusted.replication_retention_timestamp.is_some());
+        assert!(trusted.replication_legalhold_timestamp.is_some());
+    }
+
     #[test]
     fn test_put_opts_from_headers_with_replica_status() {
         let mut headers = HeaderMap::new();