fix(admin): stream madmin ReplicationMRF documents from /v3/replication/mrf

The mrf endpoint returned a single aggregate envelope, which madmin's
json.Decoder loop decoded as one phantom row (empty object) in
'mc replicate backlog' (backlog#1675 P1-13, mrf half; the diff half was
fixed in #5799 and this mirrors its pattern).

- Default response is now a bare stream of ReplicationMRF documents
  (exact madmin json tags; Size/TargetARNs as ignored extension keys)
  built from the durable backlog ledger; an empty backlog renders an
  empty body, so mc shows zero rows instead of a phantom row.
- The aggregate counter envelope moves behind ?aggregate=true (RustFS
  extension) and now advertises PerObjectEntriesAvailable whenever the
  durable backlog is readable.
- An unreadable backlog is signalled out-of-band via
  x-rustfs-replication-mrf-backlog-unavailable (mirrors the diff
  truncation header) plus a warn event, since the bare stream cannot
  carry source health.
- The madmin node parameter is accepted but documented as a no-op: the
  durable ledger is cluster-shared with no per-node attribution.
- Delete-marker purge entries fall back to the marker version id so
  those rows keep a version identity.

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,15 +1238,105 @@ 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>,
+}
+
+/// 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").
+fn mrf_entry_documents(
+    bucket: &str,
+    durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
+) -> Vec<MrfEntryDocument> {
+    durable
+        .entries
+        .iter()
+        .filter(|entry| bucket.is_empty() || entry.bucket == bucket)
+        .map(|entry| 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.
+            version_id: entry
+                .version_id
+                .or(entry.delete_marker_version_id)
+                .map(|v| v.to_string())
+                .unwrap_or_default(),
+            retry_count: entry.retry_count,
+            size: entry.size,
+            target_arns: entry.target_arns.clone(),
+        })
+        .collect()
+}
+
+/// 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 and is signalled out-of-band for the stream.
+fn render_mrf_backlog(
+    response: &MrfResponse,
+    durable: &crate::admin::storage_api::replication::DurableMrfBacklog,
+    aggregate: bool,
+) -> Result<Vec<u8>, serde_json::Error> {
+    if aggregate {
+        return serde_json::to_vec(response);
+    }
+
+    let mut data = Vec::new();
+    for entry in mrf_entry_documents(&response.bucket, durable) {
+        serde_json::to_writer(&mut data, &entry)?;
+        data.push(b'\n');
+    }
+    Ok(data)
+}
+
 /// `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.
 pub struct ReplicationMrfHandler {}
 
 #[async_trait::async_trait]
@@ -1275,14 +1368,37 @@ 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 aggregate = queries.get("aggregate").map(String::as_str) == Some("true");
+        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; an
+            // unreadable backlog would otherwise be indistinguishable from an
+            // empty (healthy) one.
+            tracing::warn!(
+                bucket = %response.bucket,
+                "durable MRF backlog is unreadable; stream response is empty — use aggregate=true to see source health"
+            );
+        }
+
+        let data = 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"));
+        // Out-of-band source-health signal for the bare stream (madmin/mc
+        // ignore unknown headers), mirroring x-rustfs-replication-diff-truncated.
+        if !durable.available {
+            headers.insert("x-rustfs-replication-mrf-backlog-unavailable", HeaderValue::from_static("true"));
+        }
         Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers))
     }
 }
@@ -1292,7 +1408,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 +1627,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 +1639,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 +1688,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 +1706,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 +1719,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 +1732,97 @@ 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);
+    }
+
     #[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"