fix(admin): gate, bound, and null-map the mrf stream

Second review round:

- Authorization: the default stream enumerates object names and version
  ids, which a metrics-only principal must not see — it now requires
  admin:ReplicationDiff (MinIO parity, route policy updated);
  ?aggregate=true carries no object identities and keeps
  admin:GetReplicationMetrics.
- The nil UUID is RustFS's in-memory null-version sentinel and now
  leaves as the S3 wire token 'null' instead of a zero UUID (a
  pre-versioning object scanned after versioning + existing-object
  replication can persist it into the ledger).
- The durable ledger is not bounded by the in-memory pending cap and
  the body is buffered before send; the stream now stops at 10,000
  documents and signals truncation via
  x-rustfs-replication-mrf-truncated (mirroring the diff endpoint)
  plus a warn event, instead of staging an unbounded body.

crates/e2e_test/src/fake_s3_target/mod.rs

@@ -76,18 +76,6 @@ 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"];
 
@@ -130,25 +118,6 @@ 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 {
@@ -162,7 +131,6 @@ 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>,
 }
 
@@ -568,15 +536,7 @@ impl S3Access for FaultAccess {
             .get(CONTENT_LENGTH)
             .and_then(|value| value.to_str().ok())
             .and_then(|value| value.parse().ok());
-        let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
-        let fault = record_request(
-            &self.control,
-            operation,
-            context.method().clone(),
-            parsed,
-            content_length,
-            replication_timestamps,
-        );
+        let fault = record_request(&self.control, operation, context.method().clone(), parsed, content_length);
         if let Some(RequestFault {
             action: FaultAction::Status(status),
             ..
@@ -629,7 +589,6 @@ fn record_request(
     method: Method,
     parsed: ParsedRequest,
     content_length: Option<u64>,
-    replication_timestamps: ReplicationTimestampHeaders,
 ) -> Option<RequestFault> {
     let mut state = lock(control);
     let action = parsed
@@ -654,7 +613,6 @@ fn record_request(
         part_number: parsed.part_number,
         content_length,
         consumed_bytes: None,
-        replication_timestamps,
         fault: action.clone(),
     });
     action.map(|action| RequestFault { sequence, action })
@@ -1741,52 +1699,6 @@ 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;
@@ -3073,7 +2985,6 @@ mod tests {
                         part_number: None,
                     },
                     Some(0),
-                    ReplicationTimestampHeaders::default(),
                 );
             }
             let records = lock(&control).requests.clone();
@@ -3095,7 +3006,6 @@ mod tests {
                     part_number: None,
                 },
                 None,
-                ReplicationTimestampHeaders::default(),
             );
             {
                 let bounded_records = lock(&bounded_control);

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -58,9 +58,7 @@ 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_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
-    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
-    insert_header,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
 };
 use rustls_pki_types::pem::PemObject;
 use serde::{Deserialize, Serialize};
@@ -1478,12 +1476,9 @@ impl Default for AdvancedPutOptions {
             replication_status: ReplicationStatusType::Pending,
             source_mtime: OffsetDateTime::now_utc(),
             replication_request: false,
-            // 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,
+            retention_timestamp: OffsetDateTime::now_utc(),
+            tagging_timestamp: OffsetDateTime::now_utc(),
+            legalhold_timestamp: OffsetDateTime::now_utc(),
             replication_validity_check: false,
         }
     }
@@ -1680,16 +1675,6 @@ 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");
         }
@@ -2857,57 +2842,6 @@ 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/replication_target_boundary.rs

@@ -259,23 +259,15 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
 
         if !tags.is_empty() {
             put_options.user_tags = tags;
+            put_options.internal.tagging_timestamp =
+                if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP) {
+                    OffsetDateTime::parse(&timestamp, &Rfc3339)
+                        .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
+                } else {
+                    object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
+                };
         }
     }
-    // Load the stored tagging timestamp independently of whether any tags
-    // remain: DeleteObjectTagging leaves the object tagless but stamps this
-    // key, and the deletion's LWW timestamp must still reach the replica.
-    // With no stored key, fall back to mod_time only while tags exist
-    // (MinIO parity); a tagless object without the key was never tagged and
-    // keeps the epoch default (no header).
-    put_options.internal.tagging_timestamp = if let Some(timestamp) = get_str(&object_info.user_defined, SUFFIX_TAGGING_TIMESTAMP)
-    {
-        OffsetDateTime::parse(&timestamp, &Rfc3339)
-            .map_err(|err| Error::other(format!("Failed to parse tagging timestamp: {err}")))?
-    } else if !put_options.user_tags.is_empty() {
-        object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-    } else {
-        OffsetDateTime::UNIX_EPOCH
-    };
 
     let metadata = &*object_info.user_defined;
 
@@ -702,44 +694,6 @@ mod tests {
         assert!(options.internal.replication_request);
     }
 
-    /// DeleteObjectTagging leaves the object tagless but stamps the
-    /// tagging-timestamp internal key; the deletion's LWW timestamp must
-    /// still be loaded (and therefore sent) so the replica can order the
-    /// deletion against concurrent tag edits.
-    #[test]
-    fn replication_put_options_carry_tagging_timestamp_after_tag_deletion() {
-        let mut metadata = std::collections::HashMap::new();
-        rustfs_utils::http::insert_str(&mut metadata, SUFFIX_TAGGING_TIMESTAMP, "2026-01-02T03:04:05Z".to_string());
-
-        let object_info = ObjectInfo {
-            user_defined: Arc::new(metadata),
-            user_tags: Arc::new(String::new()),
-            mod_time: Some(OffsetDateTime::UNIX_EPOCH),
-            version_id: Some(Uuid::nil()),
-            ..Default::default()
-        };
-
-        let (options, _) = replication_put_object_options("", &object_info).expect("build put options");
-
-        assert!(options.user_tags.is_empty());
-        assert_eq!(
-            options.internal.tagging_timestamp,
-            OffsetDateTime::parse("2026-01-02T03:04:05Z", &Rfc3339).expect("valid timestamp"),
-            "the stored tagging timestamp must load independently of remaining tags"
-        );
-
-        // A tagless object without the stored key was never tagged: the epoch
-        // default keeps the header unsent.
-        let untagged = ObjectInfo {
-            user_tags: Arc::new(String::new()),
-            mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
-            version_id: Some(Uuid::nil()),
-            ..Default::default()
-        };
-        let (options, _) = replication_put_object_options("", &untagged).expect("build put options");
-        assert_eq!(options.internal.tagging_timestamp, OffsetDateTime::UNIX_EPOCH);
-    }
-
     #[test]
     fn replication_put_options_strip_encryption_metadata_from_plaintext_objects() {
         use rustfs_utils::http::object_encryption_keys::{INTERNAL_ENCRYPTION_ORIGINAL_SIZE_HEADER, SSEC_ORIGINAL_SIZE_HEADER};

crates/ecstore/src/object_api/types.rs

@@ -277,12 +277,6 @@ 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,6 +225,8 @@ 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,14 +50,6 @@ 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.
@@ -204,27 +196,6 @@ 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,6 +659,9 @@ 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/replication.rs

@@ -1153,7 +1153,7 @@ struct MrfResponse {
 fn build_mrf_response(
     bucket: String,
     bucket_stats: &BucketStats,
-    durable: crate::admin::storage_api::replication::DurableMrfBacklog,
+    durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
 ) -> MrfResponse {
     let observation_scope = if bucket_stats.replication_stats.cluster_complete {
         "cluster_aggregated"
@@ -1223,7 +1223,10 @@ fn build_mrf_response(
         total_failed_size,
         queued_count: queued.count,
         queued_size: queued.bytes,
-        per_object_entries_available: false,
+        // The default (non-aggregate) response mode streams the durable
+        // backlog per object, so the enumerable API exists whenever the
+        // backlog is readable.
+        per_object_entries_available: durable.available,
         runtime_stats_available: bucket_stats.replication_stats.provider_available,
         cluster_complete: bucket_stats.replication_stats.cluster_complete,
         observed_node_count: bucket_stats.replication_stats.observed_node_count,
@@ -1235,23 +1238,155 @@ fn build_mrf_response(
     }
 }
 
+/// One durable MRF backlog entry rendered for the default (madmin-compatible)
+/// stream. Field names are the exact json tags of madmin-go `ReplicationMRF`
+/// (replication-api.go), which `mc replicate backlog` decodes one JSON
+/// document at a time. `Size` and `TargetARNs` are RustFS extension keys with
+/// no madmin counterpart; Go decoders ignore unknown keys.
+#[derive(Debug, Serialize)]
+struct MrfEntryDocument {
+    /// The durable backlog is a cluster-shared ledger with no per-node
+    /// attribution, so the madmin `nodeName` tag is always empty.
+    #[serde(rename = "nodeName")]
+    node_name: String,
+    #[serde(rename = "bucket")]
+    bucket: String,
+    #[serde(rename = "object")]
+    object: String,
+    #[serde(rename = "versionId")]
+    version_id: String,
+    #[serde(rename = "retryCount")]
+    retry_count: i32,
+    #[serde(rename = "Size")]
+    size: i64,
+    #[serde(rename = "TargetARNs", skip_serializing_if = "Vec::is_empty")]
+    target_arns: Vec<String>,
+}
+
+/// Upper bound on the number of documents one stream response emits. The
+/// durable ledger is not bounded by the in-memory pending cap (recovery can
+/// persist far larger generations), and the body is buffered before send, so
+/// an unbounded read could stage hundreds of MB per request. A truncated
+/// stream is signalled via `x-rustfs-replication-mrf-truncated`.
+const REPLICATION_MRF_MAX_STREAM_ENTRIES: usize = 10_000;
+
+/// Project the durable backlog into madmin `ReplicationMRF` documents,
+/// scoped to `bucket` when it is non-empty (madmin allows an empty bucket to
+/// mean "across all buckets"), bounded by
+/// [`REPLICATION_MRF_MAX_STREAM_ENTRIES`]. Returns the documents and whether
+/// the backlog was truncated.
+fn mrf_entry_documents(
+    bucket: &str,
+    durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
+) -> (Vec<MrfEntryDocument>, bool) {
+    let mut documents = Vec::new();
+    let mut truncated = false;
+    for entry in durable
+        .entries
+        .iter()
+        .filter(|entry| bucket.is_empty() || entry.bucket == bucket)
+    {
+        if documents.len() >= REPLICATION_MRF_MAX_STREAM_ENTRIES {
+            truncated = true;
+            break;
+        }
+        documents.push(MrfEntryDocument {
+            node_name: String::new(),
+            bucket: entry.bucket.clone(),
+            object: entry.object.clone(),
+            // Delete-marker purge entries track the marker version separately;
+            // fall back to it so those rows still carry a version identity.
+            // The nil UUID is RustFS's in-memory null-version sentinel and
+            // must leave as the S3 wire token, not a zero UUID.
+            version_id: entry
+                .version_id
+                .or(entry.delete_marker_version_id)
+                .map(|v| {
+                    if v.is_nil() {
+                        rustfs_filemeta::NULL_VERSION_ID.to_string()
+                    } else {
+                        v.to_string()
+                    }
+                })
+                .unwrap_or_default(),
+            retry_count: entry.retry_count,
+            size: entry.size,
+            target_arns: entry.target_arns.clone(),
+        });
+    }
+    (documents, truncated)
+}
+
+/// Render the MRF backlog as a response body.
+///
+/// Default (madmin-compatible) mode emits one `ReplicationMRF` JSON document
+/// per line with no envelope — madmin's `BucketReplicationMRF` reads the body
+/// with a `json.Decoder` loop, so an envelope object would decode as a single
+/// entry whose `"Bucket"` key case-insensitively matches
+/// `ReplicationMRF.Bucket` (a phantom row in `mc replicate backlog`), and an
+/// empty backlog must render an empty body so the loop ends on io.EOF with
+/// zero rows.
+///
+/// `aggregate=true` (RustFS extension) keeps the enveloped counter shape;
+/// backlog-source health (`RuntimeStatsAvailable`/`DurableBacklogAvailable`)
+/// is only representable there — an unreadable ledger fails the stream
+/// request outright in the handler (madmin only decodes the body of a 200,
+/// so an empty stream would read as a healthy zero-row backlog).
+fn render_mrf_backlog(
+    response: &MrfResponse,
+    durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
+    aggregate: bool,
+) -> Result<(Vec<u8>, bool), serde_json::Error> {
+    if aggregate {
+        return Ok((serde_json::to_vec(response)?, false));
+    }
+
+    let (documents, truncated) = mrf_entry_documents(&response.bucket, durable);
+    let mut data = Vec::new();
+    for entry in documents {
+        serde_json::to_writer(&mut data, &entry)?;
+        data.push(b'\n');
+    }
+    Ok((data, truncated))
+}
+
 /// `GET /v3/replication/mrf`
 ///
 /// Reports the failed-replication backlog (MinIO's MRF concept) for a bucket.
 ///
-/// Compatibility note: MinIO returns a stream of individual MRF entries. RustFS
-/// deliberately returns aggregate runtime and durable counters instead.
-/// `PerObjectEntriesAvailable` remains false until an enumerable API exists.
-/// `PerTargetDurableEntriesAvailable` is false when the durable backlog includes
-/// older entries that cannot be attributed to a target.
+/// The default response is a madmin-compatible stream of `ReplicationMRF`
+/// documents built from the durable backlog ledger (in-memory failures that
+/// have not been flushed yet — the persister runs every few seconds — are not
+/// visible). `?aggregate=true` (RustFS extension) returns the enveloped
+/// runtime + durable counter shape instead; `PerTargetDurableEntriesAvailable`
+/// is false there when the durable backlog includes older entries that cannot
+/// be attributed to a target.
+///
+/// The madmin `node` parameter is accepted but has no filtering effect: the
+/// durable ledger is cluster-shared with no per-node attribution, so every
+/// node serves the same (complete) backlog.
+///
+/// Authorization: the stream requires `admin:ReplicationDiff` (it enumerates
+/// object names and version ids, MinIO parity); `?aggregate=true` carries no
+/// object identities and requires only `admin:GetReplicationMetrics`.
 pub struct ReplicationMrfHandler {}
 
 #[async_trait::async_trait]
 impl Operation for ReplicationMrfHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        validate_replication_admin_request(&req, AdminAction::GetReplicationMetricsAction).await?;
-
         let queries = extract_query_params(&req.uri);
+        let aggregate = queries.get("aggregate").map(String::as_str) == Some("true");
+        // The default stream enumerates object names and version ids, which
+        // a metrics-only principal must not see; gate it on the same action
+        // MinIO uses for this endpoint. The aggregate counters carry no
+        // object identities and keep the metrics action.
+        let action = if aggregate {
+            AdminAction::GetReplicationMetricsAction
+        } else {
+            AdminAction::ReplicationDiff
+        };
+        validate_replication_admin_request(&req, action).await?;
+
         let Some(bucket) = queries.get("bucket").filter(|b| !b.is_empty()).cloned() else {
             return Err(s3_error!(InvalidRequest, "bucket is required"));
         };
@@ -1275,14 +1410,47 @@ impl Operation for ReplicationMrfHandler {
             return Err(ApiError::from(err).into());
         }
 
+        if let Some(node) = queries.get("node").filter(|node| !node.is_empty() && node.as_str() != "all") {
+            // The durable backlog ledger is cluster-shared with no per-node
+            // attribution, so a node-scoped request still sees the complete
+            // (superset) backlog.
+            debug!(node = %node, "replication mrf node filter has no effect on the cluster-shared backlog");
+        }
+
         let durable = crate::admin::storage_api::replication::read_durable_mrf_backlog(store).await;
         let bucket_stats = cluster_replication_stats(&bucket, app_context_from_req(&req)).await;
-        let response = build_mrf_response(bucket, &bucket_stats, durable);
+        let response = build_mrf_response(bucket, &bucket_stats, &durable);
 
-        let data = serde_json::to_vec(&response)
+        if !durable.available && !aggregate {
+            // The madmin stream has no envelope to carry source health, and
+            // madmin only decodes the body of a 200 — an empty stream would
+            // read as a clean, healthy zero-row backlog. Fail loudly instead;
+            // aggregate mode still reports the availability fields.
+            tracing::warn!(
+                bucket = %response.bucket,
+                "durable MRF backlog is unreadable; failing the stream request — use aggregate=true to see source health"
+            );
+            return Err(S3Error::with_message(
+                S3ErrorCode::ServiceUnavailable,
+                "durable MRF backlog is unreadable; retry, or use aggregate=true for source health".to_string(),
+            ));
+        }
+
+        let (data, truncated) = render_mrf_backlog(&response, &durable, aggregate)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?;
         let mut headers = HeaderMap::new();
         headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        if truncated {
+            // The madmin stream has no envelope to carry truncation; signal
+            // it out-of-band (madmin/mc ignore unknown headers), mirroring
+            // x-rustfs-replication-diff-truncated.
+            tracing::warn!(
+                bucket = %response.bucket,
+                max_entries = REPLICATION_MRF_MAX_STREAM_ENTRIES,
+                "replication mrf stream truncated; narrow with ?bucket= or drain the backlog"
+            );
+            headers.insert("x-rustfs-replication-mrf-truncated", HeaderValue::from_static("true"));
+        }
         Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))
     }
 }
@@ -1292,7 +1460,8 @@ mod tests {
     use super::{
         REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest,
         ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params,
-        parse_remote_target_update_ops, render_replication_diff, unique_replication_peers, validate_remote_target_tls_settings,
+        parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers,
+        validate_remote_target_tls_settings,
     };
     use crate::admin::storage_api::bucket::target::{BucketTarget, LatencyStat};
     use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
@@ -1510,7 +1679,7 @@ mod tests {
             ],
         };
 
-        let response = build_mrf_response("bucket-a".to_string(), &stats, durable);
+        let response = build_mrf_response("bucket-a".to_string(), &stats, &durable);
         let json = serde_json::to_value(response).expect("MRF response should serialize");
 
         assert_eq!(json["TotalFailedCount"], 3);
@@ -1522,7 +1691,9 @@ mod tests {
         assert_eq!(json["RuntimeStatsAvailable"], true);
         assert_eq!(json["ClusterComplete"], false);
         assert_eq!(json["Targets"][0]["ObservationScope"], "partial_cluster");
-        assert_eq!(json["PerObjectEntriesAvailable"], false);
+        // The bare stream enumerates the durable backlog per object, so a
+        // readable backlog advertises the enumerable API.
+        assert_eq!(json["PerObjectEntriesAvailable"], true);
         assert_eq!(json["PerTargetDurableEntriesAvailable"], true);
 
         let targets = json["Targets"].as_array().expect("targets should serialize as an array");
@@ -1569,7 +1740,7 @@ mod tests {
             }],
         };
 
-        let response = build_mrf_response("bucket-a".to_string(), &stats, durable);
+        let response = build_mrf_response("bucket-a".to_string(), &stats, &durable);
         let json = serde_json::to_value(response).expect("MRF response should serialize");
 
         assert_eq!(json["DurableBacklogAvailable"], true);
@@ -1587,7 +1758,7 @@ mod tests {
 
     #[test]
     fn mrf_response_distinguishes_unavailable_sources_from_valid_zero() {
-        let unavailable = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), DurableMrfBacklog::default());
+        let unavailable = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &DurableMrfBacklog::default());
         let unavailable_json = serde_json::to_value(unavailable).expect("unavailable response should serialize");
         assert_eq!(unavailable_json["RuntimeStatsAvailable"], false);
         assert_eq!(unavailable_json["DurableBacklogAvailable"], false);
@@ -1600,7 +1771,7 @@ mod tests {
         let valid_empty = build_mrf_response(
             "bucket-a".to_string(),
             &valid_empty_stats,
-            DurableMrfBacklog {
+            &DurableMrfBacklog {
                 available: true,
                 entries: Vec::new(),
             },
@@ -1613,6 +1784,151 @@ mod tests {
         assert_eq!(valid_empty_json["PerTargetDurableEntriesAvailable"], true);
     }
 
+    fn sample_durable_backlog() -> DurableMrfBacklog {
+        DurableMrfBacklog {
+            available: true,
+            entries: vec![
+                MrfReplicateEntry {
+                    bucket: "bucket-a".to_string(),
+                    object: "object-a".to_string(),
+                    version_id: Some(uuid::Uuid::from_u128(7)),
+                    retry_count: 2,
+                    size: 250,
+                    op: MrfOpKind::Object,
+                    target_arns: vec!["arn-a".to_string()],
+                    ..Default::default()
+                },
+                MrfReplicateEntry {
+                    bucket: "other-bucket".to_string(),
+                    object: "object-b".to_string(),
+                    version_id: None,
+                    retry_count: 0,
+                    size: 999,
+                    op: MrfOpKind::Object,
+                    target_arns: Vec::new(),
+                    ..Default::default()
+                },
+            ],
+        }
+    }
+
+    /// madmin's `BucketReplicationMRF` decodes the body one `ReplicationMRF`
+    /// JSON document at a time; the default response must therefore be a bare
+    /// document stream with madmin's exact json tags, not an envelope.
+    #[test]
+    fn mrf_stream_renders_bare_madmin_documents() {
+        let durable = sample_durable_backlog();
+        let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
+
+        let (body, _) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
+        let text = String::from_utf8(body).expect("body should be utf-8");
+        let lines: Vec<&str> = text.lines().filter(|line| !line.trim().is_empty()).collect();
+
+        // Only the entry matching the requested bucket is streamed.
+        assert_eq!(lines.len(), 1, "expected one MRF document, got: {text}");
+        let doc: serde_json::Value = serde_json::from_str(lines[0]).expect("each line should be a JSON document");
+        assert_eq!(doc["bucket"], "bucket-a");
+        assert_eq!(doc["object"], "object-a");
+        assert_eq!(doc["versionId"], uuid::Uuid::from_u128(7).to_string());
+        assert_eq!(doc["retryCount"], 2);
+        // madmin `ReplicationMRF` has a `nodeName` tag; the durable backlog is
+        // cluster-shared, so RustFS reports an empty node name.
+        assert_eq!(doc["nodeName"], "");
+        // The envelope keys must not leak into the stream: a `"Bucket"` key
+        // would case-insensitively populate `ReplicationMRF.Bucket` and render
+        // a phantom row in `mc replicate backlog`.
+        assert!(doc.get("Bucket").is_none());
+        assert!(doc.get("Targets").is_none());
+    }
+
+    /// An empty backlog must produce an empty body: madmin's decoder loop then
+    /// terminates on io.EOF with zero rows instead of one phantom row.
+    #[test]
+    fn mrf_stream_renders_empty_body_for_no_entries() {
+        let durable = DurableMrfBacklog {
+            available: true,
+            entries: Vec::new(),
+        };
+        let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
+
+        let (body, _) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
+        assert!(
+            body.is_empty(),
+            "empty backlog must serialize to an empty body, got: {}",
+            String::from_utf8_lossy(&body)
+        );
+    }
+
+    /// `?aggregate=true` (RustFS extension) keeps the enveloped counter shape.
+    #[test]
+    fn mrf_aggregate_envelope_retains_counters() {
+        let durable = sample_durable_backlog();
+        let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
+
+        let (body, _) = render_mrf_backlog(&response, &durable, true).expect("aggregate body should serialize");
+        let json: serde_json::Value = serde_json::from_slice(&body).expect("aggregate body should be one JSON object");
+        assert_eq!(json["Bucket"], "bucket-a");
+        assert_eq!(json["DurableCount"], 1);
+        assert_eq!(json["DurableBacklogAvailable"], true);
+        // The bare stream is an enumerable per-object API, so the aggregate
+        // shell now truthfully advertises it whenever the backlog is readable.
+        assert_eq!(json["PerObjectEntriesAvailable"], true);
+    }
+
+    /// The nil UUID is RustFS's in-memory null-version sentinel; the wire
+    /// token is `null`, never the zero UUID (second review round).
+    #[test]
+    fn mrf_stream_maps_nil_version_to_null_token() {
+        let durable = DurableMrfBacklog {
+            available: true,
+            entries: vec![MrfReplicateEntry {
+                bucket: "bucket-a".to_string(),
+                object: "null-version-object".to_string(),
+                version_id: Some(uuid::Uuid::nil()),
+                retry_count: 1,
+                size: 10,
+                op: MrfOpKind::Object,
+                ..Default::default()
+            }],
+        };
+        let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
+
+        let (body, truncated) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
+        assert!(!truncated);
+        let doc: serde_json::Value =
+            serde_json::from_str(String::from_utf8(body).expect("utf-8").lines().next().expect("one line"))
+                .expect("line should be a JSON document");
+        assert_eq!(doc["versionId"], "null");
+    }
+
+    /// The durable ledger is not bounded by the in-memory pending cap; the
+    /// stream must stop at the documented bound and signal truncation
+    /// (second review round).
+    #[test]
+    fn mrf_stream_truncates_at_the_documented_bound() {
+        let entries = (0..super::REPLICATION_MRF_MAX_STREAM_ENTRIES + 1)
+            .map(|index| MrfReplicateEntry {
+                bucket: "bucket-a".to_string(),
+                object: format!("object-{index}"),
+                retry_count: 1,
+                op: MrfOpKind::Object,
+                ..Default::default()
+            })
+            .collect();
+        let durable = DurableMrfBacklog {
+            available: true,
+            entries,
+        };
+        let response = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), &durable);
+
+        let (body, truncated) = render_mrf_backlog(&response, &durable, false).expect("stream body should serialize");
+        assert!(truncated, "one entry past the bound must signal truncation");
+        assert_eq!(
+            String::from_utf8(body).expect("utf-8").lines().count(),
+            super::REPLICATION_MRF_MAX_STREAM_ENTRIES
+        );
+    }
+
     #[test]
     fn test_extract_query_params_decodes_percent_encoded_values() {
         let uri: Uri = "/rustfs/admin/v3/list-remote-targets?bucket=foo%2Fbar&flag=a+b"

rustfs/src/admin/route_policy.rs

@@ -1459,10 +1459,13 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         REPLICATION_DIFF,
         RouteRiskLevel::Sensitive,
     ),
+    // The default stream enumerates object names/version ids and requires
+    // ReplicationDiff (MinIO parity); only ?aggregate=true relaxes to
+    // GetReplicationMetrics in the handler.
     admin(
         HttpMethod::Get,
         "/rustfs/admin/v3/replication/mrf",
-        GET_REPLICATION_METRICS,
+        REPLICATION_DIFF,
         RouteRiskLevel::Sensitive,
     ),
 ];

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, SUFFIX_TAGGING_TIMESTAMP, insert_str};
+use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str};
 use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error};
 use std::collections::HashMap;
 use std::fmt::Debug;
@@ -461,11 +461,6 @@ 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);
         }
 
@@ -1650,11 +1645,6 @@ 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,13 +17,11 @@ 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_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,
+    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,
     header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
-    insert_header_map, insert_str,
+    insert_header_map,
     metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
 };
 use rustfs_utils::http::{
@@ -424,9 +422,6 @@ 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)?;
@@ -484,7 +479,6 @@ 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)
 }
@@ -510,47 +504,6 @@ 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;
@@ -710,13 +663,6 @@ fn is_reserved_user_metadata_key(key: &str) -> bool {
         || starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX)
         || starts_with_ignore_ascii_case(key, RUSTFS_ENCRYPTION_PREFIX)
         || starts_with_ignore_ascii_case(key, MINIO_ENCRYPTION_PREFIX)
-        // Replication transport names (source-replication timestamps,
-        // source-mtime/-etag/-version-id, ...). A bare stored key with one of
-        // these names is forwarded verbatim by the outbound replication
-        // header builder on a server-authorized request, so the receiver
-        // would persist attacker-chosen values as trusted internal LWW state.
-        || starts_with_ignore_ascii_case(key, "x-rustfs-source-")
-        || starts_with_ignore_ascii_case(key, "x-minio-source-")
 }
 
 fn stored_user_metadata_key(key: &str) -> String {
@@ -1653,136 +1599,6 @@ mod tests {
         assert!(opts_invalid.mod_time.is_none());
     }
 
-    /// A client PUT must not materialize the replication transport names as
-    /// bare stored user-metadata keys: the outbound replication header
-    /// builder forwards user metadata verbatim on a server-authorized
-    /// request, so a bare `x-rustfs-source-replication-*-timestamp` key would
-    /// deliver an attacker-chosen value into the replica's trusted internal
-    /// LWW state (for a tagless object nothing later overwrites it).
-    #[test]
-    fn test_replication_transport_names_cannot_be_forged_via_user_metadata() {
-        let mut headers = HeaderMap::new();
-        for name in [
-            "x-amz-meta-x-rustfs-source-replication-tagging-timestamp",
-            "x-amz-meta-x-minio-source-replication-legalhold-timestamp",
-            "x-rustfs-meta-x-rustfs-source-replication-retention-timestamp",
-            "x-amz-meta-x-rustfs-source-mtime",
-        ] {
-            headers.insert(
-                http::header::HeaderName::from_static(name),
-                HeaderValue::from_static("2026-01-02T03:04:05Z"),
-            );
-        }
-
-        let metadata = extract_metadata(&headers);
-
-        for forged in [
-            "x-rustfs-source-replication-tagging-timestamp",
-            "x-minio-source-replication-legalhold-timestamp",
-            "x-rustfs-source-replication-retention-timestamp",
-            "x-rustfs-source-mtime",
-        ] {
-            assert!(
-                metadata.get(forged).is_none(),
-                "{forged} must not be storable as a bare user-metadata key"
-            );
-        }
-        // The values survive, namespaced back under the user-metadata prefix.
-        assert_eq!(
-            metadata
-                .get("x-amz-meta-x-rustfs-source-replication-tagging-timestamp")
-                .map(String::as_str),
-            Some("2026-01-02T03:04:05Z")
-        );
-    }
-
-    #[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();