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/api/mod.rs

@@ -184,18 +184,17 @@ pub mod bucket {
             mrf_backlog_observability_snapshot,
         };
         pub use crate::bucket::replication::{
-            BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats,
-            DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric,
-            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION,
-            REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE,
-            REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS,
-            ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
-            ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
-            ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
-            ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
-            ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
-            ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
-            VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
+            BucketReplicationResyncStatus, BucketReplicationStats, BucketStats, DeleteReplicationConfigSnapshot,
+            DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry,
+            MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS,
+            REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION,
+            REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo,
+            ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt,
+            ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge,
+            ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission,
+            ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage,
+            ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog,
+            TargetReplicationResyncStatus, VersionPurgeStatusType, commit_force_delete_intent, complete_force_delete_intent,
             delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
             get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
             persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,

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/bucket/replication/mod.rs

@@ -81,6 +81,6 @@ pub use replication_queue_boundary::{
 pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
 pub use replication_scanner_bridge::ReplicationScannerBridge;
 pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog};
-pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
+pub use replication_stats_boundary::{BucketReplicationStats, BucketStats};
 pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
 pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;

crates/ecstore/src/bucket/replication/replication_stats_boundary.rs

@@ -15,9 +15,7 @@
 #[cfg(test)]
 pub(crate) use rustfs_replication::FailStats;
 pub(crate) use rustfs_replication::{
-    ActiveWorkerStat, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope, SRMetricsSummary,
+    ActiveWorkerStat, BucketReplicationStat, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache, ReplicationMetricScope,
+    SRMetricsSummary, XferStats,
 };
-// Public so the admin wire DTOs (rustfs/src/admin/replication_metrics_wire.rs)
-// can project the internal stats onto the minio-go response shapes through
-// the storage_api facade chain.
-pub use rustfs_replication::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats};
+pub use rustfs_replication::{BucketReplicationStats, BucketStats};

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

rustfs/src/admin/handlers/replication.rs

@@ -535,13 +535,8 @@ impl Operation for GetReplicationMetricsHandler {
 
         let bucket_stats = cluster_replication_stats(bucket, app_context_from_req(&req)).await;
 
-        // Same minio-go `replication.Metrics` wire shape as
-        // `?replication-metrics` — the internal snake_case stats are the peer
-        // RPC wire format and must not leak here.
-        let data = serde_json::to_vec(&crate::admin::replication_metrics_wire::MetricsWire::from(
-            &bucket_stats.replication_stats,
-        ))
-        .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "serialize failed"))?;
+        let data = serde_json::to_vec(&bucket_stats.replication_stats)
+            .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "serialize failed"))?;
         let mut headers = HeaderMap::new();
         headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
         Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))

rustfs/src/admin/mod.rs

@@ -19,7 +19,6 @@ pub mod handlers;
 mod plugin_contract;
 // Contract inventory is validated by tests before later runtime integration.
 #[allow(dead_code)]
-pub(crate) mod replication_metrics_wire;
 pub(crate) mod route_policy;
 pub mod router;
 pub(crate) mod runtime_sources;

rustfs/src/admin/replication_metrics_wire.rs

@@ -1,511 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Serialize-only wire projections of the internal replication statistics
-//! onto the minio-go `replication.Metrics` / `replication.MetricsV2` json
-//! shapes consumed by `mc replicate status` (`?replication-metrics[=2]` and
-//! the admin `replicationmetrics` endpoint).
-//!
-//! Red line: the internal `BucketStats` family in
-//! `crates/replication/src/stats.rs` is ALSO the intra-cluster peer-RPC wire
-//! format — `node_service.rs` encodes it with `rmp_serde::to_vec_named`, so
-//! its Rust field names travel between nodes as msgpack map keys. Renaming
-//! those serde names would break mixed-version clusters mid rolling upgrade.
-//! All madmin/minio-go interop therefore happens in these DTOs; never add
-//! `#[serde(rename)]` to the internal structs instead.
-//!
-//! Field names below are the exact json tags of minio-go
-//! `pkg/replication/replication.go` (v7.0.91). Keys minio-go does not know
-//! are RustFS extensions; Go decoders ignore unknown keys. `max`/`peak` are
-//! both emitted for the queue peak because the MinIO server writes `max`
-//! while minio-go reads `peak` (an upstream drift); emitting both keeps every
-//! decoder working.
-
-use serde::Serialize;
-use std::collections::HashMap;
-use std::time::Duration;
-
-use crate::admin::storage_api::replication::{
-    BucketReplicationStat as InternalReplicationStat, BucketReplicationStats as InternalReplicationStats, BucketStats,
-    InQueueMetric as InternalInQueueMetric, XferStats as InternalXferStats,
-};
-
-/// minio-go `replication.RStat`.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct RStatWire {
-    #[serde(rename = "count")]
-    pub count: f64,
-    #[serde(rename = "bytes")]
-    pub bytes: i64,
-}
-
-/// minio-go `replication.TimedErrStats`.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct TimedErrStatsWire {
-    #[serde(rename = "lastMinute")]
-    pub last_minute: RStatWire,
-    #[serde(rename = "lastHour")]
-    pub last_hour: RStatWire,
-    #[serde(rename = "totals")]
-    pub totals: RStatWire,
-}
-
-impl TimedErrStatsWire {
-    fn add(self, other: TimedErrStatsWire) -> TimedErrStatsWire {
-        fn add(a: RStatWire, b: RStatWire) -> RStatWire {
-            RStatWire {
-                count: a.count + b.count,
-                bytes: a.bytes.saturating_add(b.bytes),
-            }
-        }
-        TimedErrStatsWire {
-            last_minute: add(self.last_minute, other.last_minute),
-            last_hour: add(self.last_hour, other.last_hour),
-            totals: add(self.totals, other.totals),
-        }
-    }
-}
-
-/// minio-go `replication.QStat`.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct QStatWire {
-    #[serde(rename = "count")]
-    pub count: f64,
-    #[serde(rename = "bytes")]
-    pub bytes: f64,
-}
-
-/// minio-go `replication.InQueueMetric`, with the queue peak emitted under
-/// both `peak` (minio-go tag) and `max` (MinIO server tag).
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct InQueueMetricWire {
-    #[serde(rename = "curr")]
-    pub curr: QStatWire,
-    #[serde(rename = "avg")]
-    pub avg: QStatWire,
-    #[serde(rename = "max")]
-    pub max: QStatWire,
-    #[serde(rename = "peak")]
-    pub peak: QStatWire,
-}
-
-impl From<&InternalInQueueMetric> for InQueueMetricWire {
-    fn from(metric: &InternalInQueueMetric) -> Self {
-        fn qstat(bytes: i64, count: i64) -> QStatWire {
-            QStatWire {
-                count: count as f64,
-                bytes: bytes as f64,
-            }
-        }
-        let peak = qstat(metric.max.bytes, metric.max.count);
-        InQueueMetricWire {
-            curr: qstat(metric.curr.bytes, metric.curr.count),
-            avg: qstat(metric.avg.bytes, metric.avg.count),
-            max: peak,
-            peak,
-        }
-    }
-}
-
-/// minio-go `replication.XferStats`.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct XferStatsWire {
-    #[serde(rename = "avgRate")]
-    pub avg_rate: f64,
-    #[serde(rename = "peakRate")]
-    pub peak_rate: f64,
-    #[serde(rename = "currRate")]
-    pub curr_rate: f64,
-}
-
-impl XferStatsWire {
-    fn merge(self, other: XferStatsWire) -> XferStatsWire {
-        XferStatsWire {
-            avg_rate: self.avg_rate + other.avg_rate,
-            peak_rate: self.peak_rate.max(other.peak_rate),
-            curr_rate: self.curr_rate + other.curr_rate,
-        }
-    }
-}
-
-impl From<&InternalXferStats> for XferStatsWire {
-    fn from(stats: &InternalXferStats) -> Self {
-        XferStatsWire {
-            avg_rate: stats.avg,
-            peak_rate: stats.peak,
-            curr_rate: stats.curr,
-        }
-    }
-}
-
-/// minio-go `replication.WorkerStat`. RustFS does not track per-bucket worker
-/// occupancy yet, so this always reports zeros.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct WorkerStatWire {
-    #[serde(rename = "curr")]
-    pub curr: i32,
-    #[serde(rename = "avg")]
-    pub avg: f32,
-    #[serde(rename = "max")]
-    pub max: i32,
-}
-
-/// minio-go `replication.ReplMRFStats`. RustFS does not track the 5-minute /
-/// dropped MRF windows, so this always reports zeros; the durable backlog is
-/// enumerable via `/v3/replication/mrf` instead.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct ReplMrfStatsWire {
-    #[serde(rename = "failedCount_last5min")]
-    pub last_failed_count: u64,
-    #[serde(rename = "droppedCount_since_uptime")]
-    pub total_dropped_count: u64,
-    #[serde(rename = "droppedBytes_since_uptime")]
-    pub total_dropped_bytes: u64,
-}
-
-/// minio-go `replication.CounterSummary`.
-#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub(crate) struct CounterSummaryWire {
-    #[serde(rename = "last1hr")]
-    pub last1hr: u64,
-    #[serde(rename = "last1m")]
-    pub last1m: u64,
-    #[serde(rename = "total")]
-    pub total: u64,
-}
-
-/// minio-go `replication.TargetMetrics` (one remote target / ARN).
-#[derive(Debug, Default, Serialize)]
-pub(crate) struct TargetMetricsWire {
-    #[serde(rename = "replicationCount")]
-    pub replicated_count: i64,
-    #[serde(rename = "completedReplicationSize")]
-    pub replicated_size: i64,
-    /// Bandwidth limit for this target. The tag says "bits" but both MinIO
-    /// and minio-go treat the value as bytes/sec; keep bytes/sec.
-    #[serde(rename = "limitInBits")]
-    pub bandwidth_limit_bytes_per_sec: i64,
-    #[serde(rename = "currentBandwidth")]
-    pub current_bandwidth_bytes_per_sec: f64,
-    #[serde(rename = "failed")]
-    pub failed: TimedErrStatsWire,
-    #[serde(rename = "failedReplicationSize")]
-    pub failed_size: i64,
-    #[serde(rename = "failedReplicationCount")]
-    pub failed_count: i64,
-}
-
-fn target_timed_err_stats(stat: &InternalReplicationStat) -> TimedErrStatsWire {
-    let last_minute = stat.fail_stats.recent_since(Duration::from_secs(60));
-    let last_hour = stat.fail_stats.recent_since(Duration::from_secs(3600));
-    TimedErrStatsWire {
-        last_minute: RStatWire {
-            count: last_minute.count as f64,
-            bytes: last_minute.size,
-        },
-        last_hour: RStatWire {
-            count: last_hour.count as f64,
-            bytes: last_hour.size,
-        },
-        totals: RStatWire {
-            count: stat.failed.count as f64,
-            bytes: stat.failed.size,
-        },
-    }
-}
-
-impl From<&InternalReplicationStat> for TargetMetricsWire {
-    fn from(stat: &InternalReplicationStat) -> Self {
-        TargetMetricsWire {
-            replicated_count: stat.replicated_count,
-            replicated_size: stat.replicated_size,
-            bandwidth_limit_bytes_per_sec: stat.bandwidth_limit_bytes_per_sec,
-            current_bandwidth_bytes_per_sec: stat.current_bandwidth_bytes_per_sec,
-            failed: target_timed_err_stats(stat),
-            failed_size: stat.failed.size,
-            failed_count: stat.failed.count,
-        }
-    }
-}
-
-/// minio-go `replication.Metrics` — the `currStats` member of `MetricsV2` and
-/// the whole v1 response body. The trailing snake_case fields are RustFS
-/// source-health extension keys (ignored by Go decoders) carried over from
-/// the previous response shape.
-#[derive(Debug, Default, Serialize)]
-pub(crate) struct MetricsWire {
-    #[serde(rename = "Stats")]
-    pub stats: HashMap<String, TargetMetricsWire>,
-    #[serde(rename = "completedReplicationSize")]
-    pub replicated_size: i64,
-    #[serde(rename = "replicaSize")]
-    pub replica_size: i64,
-    #[serde(rename = "replicaCount")]
-    pub replica_count: i64,
-    #[serde(rename = "replicationCount")]
-    pub replicated_count: i64,
-    #[serde(rename = "failed")]
-    pub failed: TimedErrStatsWire,
-    #[serde(rename = "queued")]
-    pub queued: InQueueMetricWire,
-    // RustFS extension keys (source health of the aggregation).
-    pub provider_available: bool,
-    pub cluster_complete: bool,
-    pub observed_node_count: u32,
-    pub expected_node_count: u32,
-}
-
-impl From<&InternalReplicationStats> for MetricsWire {
-    fn from(stats: &InternalReplicationStats) -> Self {
-        let mut failed = TimedErrStatsWire::default();
-        let mut targets = HashMap::with_capacity(stats.stats.len());
-        for (arn, stat) in &stats.stats {
-            let target = TargetMetricsWire::from(stat);
-            failed = failed.add(target.failed);
-            targets.insert(arn.clone(), target);
-        }
-        MetricsWire {
-            stats: targets,
-            replicated_size: stats.replicated_size,
-            replica_size: stats.replica_size,
-            replica_count: stats.replica_count,
-            replicated_count: stats.replicated_count,
-            failed,
-            queued: InQueueMetricWire::from(&stats.q_stat),
-            provider_available: stats.provider_available,
-            cluster_complete: stats.cluster_complete,
-            observed_node_count: stats.observed_node_count,
-            expected_node_count: stats.expected_node_count,
-        }
-    }
-}
-
-/// minio-go `replication.ReplQNodeStats`.
-#[derive(Debug, Default, Serialize)]
-pub(crate) struct ReplQNodeStatsWire {
-    #[serde(rename = "nodeName")]
-    pub node_name: String,
-    #[serde(rename = "uptime")]
-    pub uptime: i64,
-    #[serde(rename = "activeWorkers")]
-    pub workers: WorkerStatWire,
-    #[serde(rename = "transferSummary")]
-    pub xfer_stats: XferSummaryWire,
-    #[serde(rename = "tgtTransferStats")]
-    pub tgt_xfer_stats: TargetXferSummaryWire,
-    #[serde(rename = "queueStats")]
-    pub q_stats: InQueueMetricWire,
-    #[serde(rename = "mrfStats")]
-    pub mrf_stats: ReplMrfStatsWire,
-    #[serde(rename = "retries")]
-    pub retries: CounterSummaryWire,
-    #[serde(rename = "errors")]
-    pub errors: CounterSummaryWire,
-}
-
-/// minio-go `replication.ReplQueueStats`.
-#[derive(Debug, Default, Serialize)]
-pub(crate) struct ReplQueueStatsWire {
-    #[serde(rename = "nodes")]
-    pub nodes: Vec<ReplQNodeStatsWire>,
-}
-
-/// minio-go `replication.MetricsV2` — the `?replication-metrics=2` body.
-#[derive(Debug, Default, Serialize)]
-pub(crate) struct MetricsV2Wire {
-    #[serde(rename = "uptime")]
-    pub uptime: i64,
-    #[serde(rename = "currStats")]
-    pub current_stats: MetricsWire,
-    #[serde(rename = "queueStats")]
-    pub queue_stats: ReplQueueStatsWire,
-    #[serde(rename = "downtimeInfo")]
-    pub downtime_info: HashMap<String, serde_json::Value>,
-}
-
-/// `transferSummary` map keyed by minio-go `MetricName` (Large/Small/Total).
-type XferSummaryWire = HashMap<&'static str, XferStatsWire>;
-/// `tgtTransferStats` map keyed by target ARN.
-type TargetXferSummaryWire = HashMap<String, XferSummaryWire>;
-
-fn transfer_summaries(stats: &InternalReplicationStats) -> (XferSummaryWire, TargetXferSummaryWire) {
-    let mut summary: XferSummaryWire = HashMap::new();
-    let mut per_target: TargetXferSummaryWire = HashMap::new();
-    for (arn, stat) in &stats.stats {
-        let large = XferStatsWire::from(&stat.xfer_rate_lrg);
-        let small = XferStatsWire::from(&stat.xfer_rate_sml);
-        let total = large.merge(small);
-        per_target.insert(arn.clone(), HashMap::from([("Large", large), ("Small", small), ("Total", total)]));
-        for (key, value) in [("Large", large), ("Small", small), ("Total", total)] {
-            let entry = summary.entry(key).or_default();
-            *entry = entry.merge(value);
-        }
-    }
-    (summary, per_target)
-}
-
-impl MetricsV2Wire {
-    /// Project the aggregated internal stats onto the `MetricsV2` shape.
-    ///
-    /// The aggregation path leaves `queue_stats.nodes` empty today, so a
-    /// single node entry is synthesized from the bucket queue snapshot —
-    /// `mc replicate status` derives its queue/worker panels from
-    /// `queueStats.nodes` and treats an empty list as "no data".
-    pub(crate) fn from_stats(bucket_stats: &BucketStats, node_name: &str) -> Self {
-        let (xfer_stats, tgt_xfer_stats) = transfer_summaries(&bucket_stats.replication_stats);
-        let mut nodes: Vec<ReplQNodeStatsWire> = bucket_stats
-            .queue_stats
-            .nodes
-            .iter()
-            .map(|node| ReplQNodeStatsWire {
-                node_name: node_name.to_string(),
-                uptime: bucket_stats.uptime,
-                q_stats: InQueueMetricWire::from(&node.q_stats),
-                ..Default::default()
-            })
-            .collect();
-        if nodes.is_empty() {
-            nodes.push(ReplQNodeStatsWire {
-                node_name: node_name.to_string(),
-                uptime: bucket_stats.uptime,
-                q_stats: InQueueMetricWire::from(&bucket_stats.replication_stats.q_stat),
-                xfer_stats: xfer_stats.clone(),
-                tgt_xfer_stats: tgt_xfer_stats.clone(),
-                ..Default::default()
-            });
-        } else {
-            // Attach the transfer summaries to the first node; the internal
-            // snapshot does not attribute transfer rates per node.
-            if let Some(first) = nodes.first_mut() {
-                first.xfer_stats = xfer_stats.clone();
-                first.tgt_xfer_stats = tgt_xfer_stats.clone();
-            }
-        }
-
-        MetricsV2Wire {
-            uptime: bucket_stats.uptime,
-            current_stats: MetricsWire::from(&bucket_stats.replication_stats),
-            queue_stats: ReplQueueStatsWire { nodes },
-            downtime_info: HashMap::new(),
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    fn sample_bucket_stats() -> BucketStats {
-        let mut stats = BucketStats {
-            uptime: 42,
-            ..Default::default()
-        };
-        stats.replication_stats.replica_count = 2;
-        stats.replication_stats.replica_size = 128;
-        stats.replication_stats.replicated_count = 9;
-        stats.replication_stats.replicated_size = 4096;
-        let target = stats
-            .replication_stats
-            .stats
-            .entry("arn:minio:replication::t:b".to_string())
-            .or_default();
-        target.replicated_count = 9;
-        target.replicated_size = 4096;
-        target.failed.count = 3;
-        target.failed.size = 900;
-        target.bandwidth_limit_bytes_per_sec = 1024;
-        target.current_bandwidth_bytes_per_sec = 512.5;
-        stats
-            .replication_stats
-            .q_stat
-            .curr
-            .now_count
-            .store(4, std::sync::atomic::Ordering::Relaxed);
-        stats
-            .replication_stats
-            .q_stat
-            .curr
-            .now_bytes
-            .store(1200, std::sync::atomic::Ordering::Relaxed);
-        stats.replication_stats.q_stat = stats.replication_stats.q_stat.snapshot();
-        stats
-    }
-
-    #[test]
-    fn metrics_wire_matches_minio_go_tags() {
-        let stats = sample_bucket_stats();
-        let json = serde_json::to_value(MetricsWire::from(&stats.replication_stats)).expect("v1 wire should serialize");
-
-        assert_eq!(json["replicaCount"], 2);
-        assert_eq!(json["replicaSize"], 128);
-        assert_eq!(json["replicationCount"], 9);
-        assert_eq!(json["completedReplicationSize"], 4096);
-        assert_eq!(json["queued"]["curr"]["count"], 4.0);
-        assert_eq!(json["queued"]["curr"]["bytes"], 1200.0);
-        let target = &json["Stats"]["arn:minio:replication::t:b"];
-        assert_eq!(target["replicationCount"], 9);
-        assert_eq!(target["completedReplicationSize"], 4096);
-        assert_eq!(target["limitInBits"], 1024);
-        assert_eq!(target["currentBandwidth"], 512.5);
-        // failed is the madmin TimedErrStats envelope, not the internal
-        // {count,size} pair.
-        assert_eq!(target["failed"]["totals"]["count"], 3.0);
-        assert_eq!(target["failed"]["totals"]["bytes"], 900);
-        assert!(target["failed"].get("count").is_none());
-        // Aggregate failed mirrors the per-target totals.
-        assert_eq!(json["failed"]["totals"]["count"], 3.0);
-    }
-
-    #[test]
-    fn metrics_v2_wire_synthesizes_queue_node() {
-        let stats = sample_bucket_stats();
-        let json = serde_json::to_value(MetricsV2Wire::from_stats(&stats, "node-1:9000")).expect("v2 wire should serialize");
-
-        assert_eq!(json["uptime"], 42);
-        assert_eq!(json["currStats"]["replicaCount"], 2);
-        let node = &json["queueStats"]["nodes"][0];
-        assert_eq!(node["nodeName"], "node-1:9000");
-        assert_eq!(node["uptime"], 42);
-        assert_eq!(node["queueStats"]["curr"]["count"], 4.0);
-        // The queue peak is emitted under both the minio-go tag (`peak`) and
-        // the MinIO server tag (`max`).
-        assert_eq!(node["queueStats"]["peak"], node["queueStats"]["max"]);
-        assert!(node["activeWorkers"].get("curr").is_some());
-        assert!(node["transferSummary"].get("Total").is_some());
-        assert_eq!(json["downtimeInfo"], serde_json::json!({}));
-    }
-
-    /// Pin the intra-cluster peer-RPC wire format of the internal stats: it
-    /// is msgpack with the Rust field names as map keys
-    /// (`rmp_serde::to_vec_named` in node_service.rs). If someone "fixes"
-    /// the interop bug by renaming the internal serde fields instead of using
-    /// these DTOs, this test fails and points them here.
-    #[test]
-    fn internal_bucket_stats_rpc_wire_stays_snake_case() {
-        let stats = sample_bucket_stats();
-        let encoded = rmp_serde::to_vec_named(&stats).expect("internal stats should encode");
-        let value: serde_json::Value = rmp_serde::from_slice(&encoded).expect("named msgpack should decode generically");
-
-        assert!(
-            value.get("replication_stats").is_some(),
-            "peer RPC key replication_stats must not be renamed"
-        );
-        assert!(value["replication_stats"].get("q_stat").is_some());
-        assert!(value.get("queue_stats").is_some());
-        assert!(value.get("proxy_stats").is_some());
-
-        let decoded: BucketStats = rmp_serde::from_slice(&encoded).expect("round-trip through the peer RPC wire");
-        assert_eq!(decoded.replication_stats.replica_count, 2);
-    }
-}

rustfs/src/admin/router.rs

@@ -1548,8 +1548,7 @@ async fn build_replication_metrics_response(
     let bucket_stats = apply_replication_metrics_bandwidth_report(bucket_stats, collect_replication_metrics_bandwidth(bucket));
     let bucket_stats = apply_replication_metrics_runtime_fields(bucket_stats, route, replication_metrics_uptime_seconds());
 
-    let node_name = crate::runtime_sources::current_local_node_name().await.unwrap_or_default();
-    let body = serialize_replication_metrics_body(&bucket_stats, route, &node_name)?;
+    let body = serialize_replication_metrics_body(&bucket_stats, route)?;
 
     let mut resp = S3Response::with_status(Body::from(body), StatusCode::OK);
     resp.headers
@@ -1609,24 +1608,12 @@ fn apply_replication_metrics_runtime_fields(
     bucket_stats
 }
 
-/// Serialize the metrics body in the minio-go wire shapes
-/// (`replication.Metrics` for v1, `replication.MetricsV2` for v2). The
-/// internal `BucketStats` serde names are the intra-cluster peer-RPC wire
-/// format and must never appear here — see
-/// `crate::admin::replication_metrics_wire`.
-fn serialize_replication_metrics_body(
-    bucket_stats: &BucketStats,
-    route: ReplicationExtRoute,
-    node_name: &str,
-) -> S3Result<Vec<u8>> {
-    use crate::admin::replication_metrics_wire::{MetricsV2Wire, MetricsWire};
+fn serialize_replication_metrics_body(bucket_stats: &BucketStats, route: ReplicationExtRoute) -> S3Result<Vec<u8>> {
     match route {
         ReplicationExtRoute::MetricsV1 => {
-            serde_json::to_vec(&MetricsWire::from(&bucket_stats.replication_stats)).map_err(|e| s3_error!(InternalError, "{e}"))
-        }
-        ReplicationExtRoute::MetricsV2 => {
-            serde_json::to_vec(&MetricsV2Wire::from_stats(bucket_stats, node_name)).map_err(|e| s3_error!(InternalError, "{e}"))
+            serde_json::to_vec(&bucket_stats.replication_stats).map_err(|e| s3_error!(InternalError, "{e}"))
         }
+        ReplicationExtRoute::MetricsV2 => serde_json::to_vec(bucket_stats).map_err(|e| s3_error!(InternalError, "{e}")),
         ReplicationExtRoute::Check | ReplicationExtRoute::ResetStart | ReplicationExtRoute::ResetStatus => {
             Err(s3_error!(InternalError, "invalid route for metrics response"))
         }
@@ -4160,37 +4147,22 @@ mod tests {
         assert!(err.message().unwrap_or_default().contains("rule-stale"));
     }
 
-    /// The v1 body must decode into minio-go `replication.Metrics` (exact
-    /// json tags); Go's decoder matches case-insensitively but does not
-    /// ignore underscores, so the internal snake_case names read as all-zero.
     #[test]
-    fn serialize_replication_metrics_body_v1_returns_minio_go_metrics_shape() {
+    fn serialize_replication_metrics_body_v1_returns_replication_stats_only() {
         let mut stats = BucketStats {
             uptime: 99,
             ..Default::default()
         };
         stats.replication_stats.replica_count = 7;
-        stats.replication_stats.replicated_size = 2048;
-        stats
-            .replication_stats
-            .stats
-            .entry("arn:minio:replication::t:b".to_string())
-            .or_default()
-            .replicated_count = 5;
         stats.proxy_stats.put_total = 3;
 
-        let body = serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV1, "node-1:9000")
-            .expect("metrics v1 body should serialize");
+        let body =
+            serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV1).expect("metrics v1 body should serialize");
         let payload: serde_json::Value = serde_json::from_slice(&body).expect("body should be json");
 
-        assert_eq!(payload["replicaCount"], 7);
-        assert_eq!(payload["completedReplicationSize"], 2048);
-        assert_eq!(payload["Stats"]["arn:minio:replication::t:b"]["replicationCount"], 5);
+        assert_eq!(payload["replica_count"], 7);
         assert!(payload.get("uptime").is_none());
         assert!(payload.get("proxy_stats").is_none());
-        // The internal snake_case names must not leak into the wire body.
-        assert!(payload.get("replica_count").is_none());
-        assert!(payload.get("q_stat").is_none());
     }
 
     #[test]
@@ -4276,48 +4248,22 @@ mod tests {
         assert_eq!(target.current_bandwidth_bytes_per_sec, 3000.0);
     }
 
-    /// The v2 body must decode into minio-go `replication.MetricsV2`
-    /// (`uptime`/`currStats`/`queueStats`); `mc replicate status` reads
-    /// `currStats` and `queueStats.nodes` and silently shows zeros when the
-    /// keys do not match.
     #[test]
-    fn serialize_replication_metrics_body_v2_returns_minio_go_metrics_v2_shape() {
+    fn serialize_replication_metrics_body_v2_returns_full_bucket_stats() {
         let mut stats = BucketStats {
             uptime: 99,
             ..Default::default()
         };
         stats.replication_stats.replica_count = 7;
-        stats
-            .replication_stats
-            .q_stat
-            .curr
-            .now_count
-            .store(4, std::sync::atomic::Ordering::Relaxed);
-        stats
-            .replication_stats
-            .q_stat
-            .curr
-            .now_bytes
-            .store(1200, std::sync::atomic::Ordering::Relaxed);
-        stats.replication_stats.q_stat = stats.replication_stats.q_stat.snapshot();
         stats.proxy_stats.put_total = 3;
 
-        let body = serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV2, "node-1:9000")
-            .expect("metrics v2 body should serialize");
+        let body =
+            serialize_replication_metrics_body(&stats, ReplicationExtRoute::MetricsV2).expect("metrics v2 body should serialize");
         let payload: serde_json::Value = serde_json::from_slice(&body).expect("body should be json");
 
         assert_eq!(payload["uptime"], 99);
-        assert_eq!(payload["currStats"]["replicaCount"], 7);
-        assert_eq!(payload["currStats"]["queued"]["curr"]["count"], 4.0);
-        // The queue snapshot must surface at least one node: mc derives the
-        // worker/queue panels from queueStats.nodes and treats an empty list
-        // as "no data".
-        assert_eq!(payload["queueStats"]["nodes"][0]["queueStats"]["curr"]["count"], 4.0);
-        assert_eq!(payload["queueStats"]["nodes"][0]["uptime"], 99);
-        // The internal snake_case names must not leak into the wire body.
-        assert!(payload.get("replication_stats").is_none());
-        assert!(payload.get("queue_stats").is_none());
-        assert!(payload.get("proxy_stats").is_none());
+        assert_eq!(payload["replication_stats"]["replica_count"], 7);
+        assert_eq!(payload["proxy_stats"]["put_total"], 3);
     }
 
     #[test]

rustfs/src/admin/storage_api.rs

@@ -417,10 +417,6 @@ pub(crate) mod replication {
     };
     pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus;
     pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;
-    pub(crate) type BucketReplicationStats = super::ecstore_bucket::replication::BucketReplicationStats;
-    pub(crate) type BucketReplicationStat = super::ecstore_bucket::replication::BucketReplicationStat;
-    pub(crate) type InQueueMetric = super::ecstore_bucket::replication::InQueueMetric;
-    pub(crate) type XferStats = super::ecstore_bucket::replication::XferStats;
     pub(crate) type ReplicationStatusType = super::ecstore_bucket::replication::ReplicationStatusType;
     pub(crate) type ResyncOpts = super::ecstore_bucket::replication::ResyncOpts;
     pub(crate) type ResyncStatusType = super::ecstore_bucket::replication::ResyncStatusType;

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