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/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/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/handlers/site_replication.rs

@@ -3004,9 +3004,6 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Fut
                 "admin site replication state"
             );
         }
-        // Failed peer deliveries recorded in the retry queue; runs behind the
-        // same lifecycle guard and pending_* gates as the reconcilers above.
-        drain_site_replication_retry_queue().await;
     })
 }
 
@@ -3956,7 +3953,7 @@ async fn persist_site_replication_repair_task(
         match failure.as_deref() {
             Some(error) => upsert_site_replication_retry_event(&mut state.retry_queue, &peer, &path, error, None),
             None => {
-                dequeue_site_replication_retry_events_including_escalated(&mut state.retry_queue, &peer, &path);
+                dequeue_site_replication_retry_events(&mut state.retry_queue, &peer, &path);
             }
         }
         Ok(())
@@ -6044,20 +6041,6 @@ fn dequeue_site_replication_retry_events(queue: &mut Vec<SiteReplicationRetryEve
     settle_site_replication_retry_events(queue, peer, path, None)
 }
 
-/// Repair-path settlement: also clears snapshot-escalated entries. Running a
-/// repair is the operator's explicit accountability transfer for the
-/// possibly-unreplayed deletion the marker records; ordinary delivery
-/// successes must not clear it (see [`settle_site_replication_retry_events`]).
-fn dequeue_site_replication_retry_events_including_escalated(
-    queue: &mut Vec<SiteReplicationRetryEvent>,
-    peer: &PeerInfo,
-    path: &str,
-) -> usize {
-    let before = queue.len();
-    queue.retain(|event| !retry_event_matches(event, peer, path));
-    before.saturating_sub(queue.len())
-}
-
 /// Remove the retry events for (peer, path) that `generation` is entitled to
 /// settle. A successful delivery only proves the peer reached the state the
 /// delivery carried: while it was in flight another edit can commit, fail its
@@ -6077,13 +6060,6 @@ fn settle_site_replication_retry_events(
         if !retry_event_matches(event, peer, path) {
             return true;
         }
-        // A snapshot-escalated entry records a possibly-unreplayed deletion.
-        // Collapsed paths are shared by every entity, so a later successful
-        // delivery of a DIFFERENT item proves nothing about the deleted one —
-        // only a repair settles it (dequeue_..._including_escalated).
-        if event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER {
-            return true;
-        }
         match (generation, event.edit_generation) {
             (Some(settled), Some(failed)) => failed > settled,
             _ => false,
@@ -6161,12 +6137,7 @@ async fn enqueue_site_replication_retry_event_for_generation(
     let path_owned = path.to_string();
     let error_text = error.to_string();
     let result = update_site_replication_state(move |state| {
-        // A peer that left the state can never drain its entries again
-        // (remove_sites already pruned them); recording a late failure for it
-        // would only pollute retry_stats until the queue cap evicts it.
-        if state.peers.contains_key(&peer_owned.deployment_id) {
-            upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
-        }
+        upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
         Ok(())
     })
     .await;
@@ -6200,420 +6171,6 @@ fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEvent) -> bool
     )
 }
 
-/// Exponential backoff base for the background retry drain, aligned with the
-/// reconcile cadence (`site_replication_reconcile::RECONCILE_INTERVAL`).
-const SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS: i64 = 600;
-/// Backoff ceiling: a permanently failed peer is still probed daily.
-const SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS: i64 = 86_400;
-
-/// What the background drain may do for one retry event. Everything not
-/// representable here is operator territory (manual repair).
-#[derive(Debug, Clone, PartialEq, Eq)]
-enum RetryDrainAction {
-    /// Constant-path IAM item deliveries collapse into one queue entry per
-    /// peer and their bodies are not persisted; the only faithful replay is
-    /// the current IAM snapshot from the bootstrap plan.
-    IamSnapshot,
-    /// Same collapse for bucket-meta deliveries: replay the bucket metadata
-    /// snapshot from the bootstrap plan.
-    BucketMetadataSnapshot,
-    /// A self-contained bucket op the bootstrap plan can re-derive for its
-    /// bucket (`make-with-versioning` / `configure-replication`).
-    BucketOpReplay { operation: String, bucket: String },
-    /// Re-send the current peer records under a fresh edit generation.
-    PeerEdit,
-}
-
-fn classify_site_replication_retry_event(event: &SiteReplicationRetryEvent) -> Option<RetryDrainAction> {
-    if event.path.starts_with("internal:") {
-        // Marker records store payloads in `last_error` (legacy
-        // pending-endpoint-refresh backup); they are not delivery failures.
-        return None;
-    }
-    if event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER {
-        // Already snapshot-replayed once for this failure episode; a possible
-        // deletion cannot be replayed from a snapshot, so re-sending daily
-        // proves nothing. A new hook failure overwrites the marker.
-        return None;
-    }
-    let base_path = event.path.split_once('?').map(|(base, _)| base).unwrap_or(&event.path);
-    match base_path {
-        "/rustfs/admin/v3/site-replication/peer/iam-item" => Some(RetryDrainAction::IamSnapshot),
-        "/rustfs/admin/v3/site-replication/peer/bucket-meta" => Some(RetryDrainAction::BucketMetadataSnapshot),
-        SITE_REPLICATION_PEER_EDIT_PATH => Some(RetryDrainAction::PeerEdit),
-        SITE_REPLICATION_PEER_BUCKET_OPS_PATH => {
-            let operation = retry_bucket_operation(&event.path)?;
-            if !matches!(
-                operation.as_str(),
-                SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING | SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION
-            ) {
-                // Destructive ops (delete-bucket / force-delete-bucket) are
-                // operator territory: replaying them against a peer whose
-                // bucket was since recreated is irreversible.
-                return None;
-            }
-            let bucket = retry_bucket_name(&event.path)?;
-            Some(RetryDrainAction::BucketOpReplay { operation, bucket })
-        }
-        _ => None,
-    }
-}
-
-fn retry_bucket_name(path: &str) -> Option<String> {
-    let (_, query) = path.split_once('?')?;
-    form_urlencoded::parse(query.as_bytes())
-        .find_map(|(key, value)| (key == "bucket" && !value.is_empty()).then(|| value.into_owned()))
-}
-
-/// A collapsed (constant-path) retry event after a successful snapshot
-/// resend is escalated with this marker instead of being cleared: the
-/// snapshot replays every entity that still exists, but a failed *deletion*
-/// leaves no task in the plan, so remote absence is unproven and the entry
-/// must stay operator-visible until a later full delivery or a manual repair
-/// settles it. The drain skips marked entries so the once-per-episode
-/// snapshot is not re-sent daily; a new hook failure overwrites the marker
-/// and re-arms the drain.
-const SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER: &str = "snapshot replayed; a failed deletion cannot be replayed from a snapshot — run site replication repair or re-deliver to settle";
-
-/// Escalate a collapsed retry event after its snapshot resend succeeded,
-/// unless a newer failure was recorded after `snapshot_updated_at` (that
-/// failure belongs to a newer local commit the snapshot did not contain and
-/// must keep the entry drain-eligible).
-fn escalate_site_replication_retry_events_up_to(
-    queue: &mut [SiteReplicationRetryEvent],
-    peer: &PeerInfo,
-    path: &str,
-    snapshot_updated_at: Option<OffsetDateTime>,
-) -> usize {
-    let mut escalated = 0usize;
-    for event in queue.iter_mut() {
-        if !retry_event_matches(event, peer, path) {
-            continue;
-        }
-        let newer_failure_recorded = match (event.updated_at, snapshot_updated_at) {
-            (Some(current), Some(seen)) => current > seen,
-            (Some(_), None) => true,
-            (None, _) => false,
-        };
-        if newer_failure_recorded {
-            continue;
-        }
-        event.failed = true;
-        event.retry_count = event.retry_count.max(SITE_REPLICATION_RETRY_FAILED_AFTER);
-        event.last_error = SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER.to_string();
-        escalated += 1;
-    }
-    escalated
-}
-
-async fn escalate_site_replication_retry_event_up_to(peer: &PeerInfo, path: &str, snapshot_updated_at: Option<OffsetDateTime>) {
-    let peer_owned = peer.clone();
-    let path_owned = path.to_string();
-    let result = update_site_replication_state(move |state| {
-        escalate_site_replication_retry_events_up_to(&mut state.retry_queue, &peer_owned, &path_owned, snapshot_updated_at);
-        Ok(())
-    })
-    .await;
-
-    if let Err(err) = result {
-        warn!(
-            component = LOG_COMPONENT_ADMIN,
-            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-            peer = %peer.endpoint,
-            deployment_id = %peer.deployment_id,
-            path,
-            error = ?err,
-            "failed to escalate site replication retry event"
-        );
-    }
-}
-
-/// Whether the drain may attempt this event now.
-fn site_replication_retry_backoff_elapsed(event: &SiteReplicationRetryEvent, now: OffsetDateTime) -> bool {
-    let Some(updated_at) = event.updated_at else {
-        return true;
-    };
-    // 600 * 2^8 already exceeds the daily ceiling; capping the shift keeps
-    // the arithmetic overflow-free for any persisted retry_count.
-    let exponent = event.retry_count.saturating_sub(1).min(8);
-    let delay = (SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS << exponent).min(SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS);
-    now.unix_timestamp().saturating_sub(updated_at.unix_timestamp()) >= delay
-}
-
-/// The subset of the retry queue the background drain is allowed to touch.
-fn actionable_site_replication_retry_events(state: &SiteReplicationState, now: OffsetDateTime) -> Vec<SiteReplicationRetryEvent> {
-    state
-        .retry_queue
-        .iter()
-        .filter(|event| classify_site_replication_retry_event(event).is_some())
-        .filter(|event| state.peers.contains_key(&event.peer_deployment_id))
-        .filter(|event| site_replication_retry_backoff_elapsed(event, now))
-        .cloned()
-        .collect()
-}
-
-/// Background consumer for the retry queue, run from the reconcile tick.
-///
-/// Scope: this settles "delivered once and failed" entries whose replay is
-/// faithful (bucket ops, peer edits). Collapsed iam-item / bucket-meta
-/// entries are snapshot-resent and then *escalated*, not cleared — a failed
-/// deletion leaves no task in the snapshot, so remote absence stays unproven
-/// until a later delivery or a manual repair. A hook that never fired (crash
-/// between the local commit and the send) leaves no entry at all, so the
-/// drain is not a full cross-site diff-heal; manual repair remains the
-/// authoritative catch-all.
-async fn drain_site_replication_retry_queue() {
-    if let Err(err) = drain_site_replication_retry_queue_inner().await {
-        warn!(
-            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-            component = LOG_COMPONENT_ADMIN,
-            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-            result = "retry_drain_failed",
-            error = ?err,
-            "admin site replication state"
-        );
-    }
-}
-
-async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
-    let Some(runtime) = runtime_site_replication_targets().await? else {
-        return Ok(());
-    };
-    let actionable = actionable_site_replication_retry_events(&runtime.state, OffsetDateTime::now_utc());
-    if actionable.is_empty() {
-        return Ok(());
-    }
-    let Some(store) = current_object_store_handle() else {
-        return Ok(());
-    };
-    if runtime.state.pending_endpoint_refresh.is_some()
-        || runtime.state.pending_remove.is_some()
-        || runtime.state.pending_rotation.is_some()
-    {
-        // The tick-level gate ran before the reconcilers; a multi-step flow
-        // (endpoint refresh commits its pending marker without the lifecycle
-        // guard) may have started since. Re-check on the fresh state.
-        return Ok(());
-    }
-    // Serialize against operator repair execution. This does NOT close the
-    // dry-run -> execute window (dry-run takes no lock): a drain settling a
-    // replayable bucket-op entry in that window changes the preflight token
-    // and execute fails safe with "preflight is stale" — the operator
-    // re-runs the dry-run. Lock order matches repair: lifecycle guard (held
-    // by the reconcile tick) -> repair execution lock -> state object lock
-    // inside the send bookkeeping. An operator repair holding the lock makes
-    // this tick skip after the lock-acquire timeout.
-    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
-        drain_site_replication_retry_queue_locked(runtime, actionable).await
-    })
-    .await
-    .map_err(ApiError::from)?
-}
-
-async fn drain_site_replication_retry_queue_locked(
-    runtime: SiteReplicationRuntime,
-    events: Vec<SiteReplicationRetryEvent>,
-) -> S3Result<()> {
-    let needs_plan = events
-        .iter()
-        .any(|event| !matches!(classify_site_replication_retry_event(event), Some(RetryDrainAction::PeerEdit)));
-    // The plan is a full local snapshot (buckets + IAM); build it once per
-    // tick and only when a snapshot resend is actually due.
-    let plan = if needs_plan {
-        let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-        Some(site_replication_bootstrap_plan(&info)?)
-    } else {
-        None
-    };
-
-    let mut events_by_peer: BTreeMap<String, Vec<SiteReplicationRetryEvent>> = BTreeMap::new();
-    for event in events {
-        events_by_peer
-            .entry(event.peer_deployment_id.clone())
-            .or_default()
-            .push(event);
-    }
-
-    let mut settled = 0usize;
-    let mut failures = 0usize;
-    for (deployment_id, peer_events) in events_by_peer {
-        let Some(peer) = runtime.state.peers.get(&deployment_id) else {
-            continue;
-        };
-        if deployment_id == runtime.local_peer.deployment_id
-            || same_identity_endpoint(&peer.endpoint, &runtime.local_peer.endpoint)
-        {
-            continue;
-        }
-        let transport = match PeerTransport::for_runtime_peer(peer).await {
-            Ok(transport) => transport,
-            Err(err) => {
-                // Record the attempt so backoff advances for an unreachable
-                // peer instead of re-dialing it every tick.
-                for event in &peer_events {
-                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
-                }
-                failures += peer_events.len();
-                continue;
-            }
-        };
-        for event in peer_events {
-            let Some(action) = classify_site_replication_retry_event(&event) else {
-                continue;
-            };
-            match drain_one_site_replication_retry_event(&runtime, peer, &transport, &event, action, plan.as_ref()).await {
-                Ok(true) => settled += 1,
-                Ok(false) => {}
-                Err(_) => failures += 1,
-            }
-        }
-    }
-
-    if settled > 0 || failures > 0 {
-        info!(
-            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-            component = LOG_COMPONENT_ADMIN,
-            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-            result = "retry_drain_settled",
-            settled,
-            failures,
-            "admin site replication state"
-        );
-    }
-    Ok(())
-}
-
-/// Replay one retry event against its peer. Returns `Ok(true)` when the
-/// event was settled (delivered, or provably stale), `Ok(false)` when it was
-/// skipped, and `Err` after a failed delivery (already re-queued with an
-/// incremented retry count).
-async fn drain_one_site_replication_retry_event(
-    runtime: &SiteReplicationRuntime,
-    peer: &PeerInfo,
-    transport: &PeerTransport,
-    event: &SiteReplicationRetryEvent,
-    action: RetryDrainAction,
-    plan: Option<&SiteReplicationBootstrapPlan>,
-) -> S3Result<bool> {
-    let access_key = &runtime.state.service_account_access_key;
-    let secret_key = &runtime.service_account_secret_key;
-    match action {
-        RetryDrainAction::IamSnapshot | RetryDrainAction::BucketMetadataSnapshot => {
-            let Some(plan) = plan else {
-                return Ok(false);
-            };
-            let tasks: Vec<SiteReplicationRepairTask<'_>> = match action {
-                RetryDrainAction::IamSnapshot => plan.iam_items.iter().map(SiteReplicationRepairTask::Iam).collect(),
-                _ => plan
-                    .bucket_items
-                    .iter()
-                    .map(SiteReplicationRepairTask::BucketMetadata)
-                    .collect(),
-            };
-            for task in &tasks {
-                if let Err(err) = task.send(transport, access_key, secret_key).await {
-                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
-                    return Err(err);
-                }
-            }
-            // The snapshot replays every entity that still exists, but a
-            // failed *deletion* leaves no task in the plan — remote absence
-            // is unproven, so escalate (operator-visible, drain-idle) instead
-            // of clearing. Conditional on the snapshot timestamp: a hook
-            // failure recorded while this snapshot was in flight belongs to a
-            // newer commit and keeps the entry drain-eligible.
-            escalate_site_replication_retry_event_up_to(peer, &event.path, event.updated_at).await;
-            Ok(true)
-        }
-        RetryDrainAction::BucketOpReplay { operation, bucket } => {
-            let Some(plan) = plan else {
-                return Ok(false);
-            };
-            // Replay from the CURRENT plan, never the recorded path: the
-            // recorded query can carry an expired one-shot bootstrap token or
-            // a stale createdAt.
-            let make_op = operation == SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING;
-            let paths = if make_op {
-                &plan.bucket_make_ops
-            } else {
-                &plan.bucket_configure_ops
-            };
-            let tasks: Vec<SiteReplicationRepairTask<'_>> = paths
-                .iter()
-                .filter(|path| retry_bucket_name(path).as_deref() == Some(bucket.as_str()))
-                .map(|path| {
-                    if make_op {
-                        SiteReplicationRepairTask::BucketMake(path)
-                    } else {
-                        SiteReplicationRepairTask::Replication(path)
-                    }
-                })
-                .collect();
-            if tasks.is_empty() {
-                // The bucket left the plan (deleted, or replication no longer
-                // configured): the recorded intent is stale, settle it.
-                dequeue_site_replication_retry_event(peer, &event.path).await;
-                return Ok(true);
-            }
-            for task in &tasks {
-                if let Err(err) = task.send(transport, access_key, secret_key).await {
-                    enqueue_site_replication_retry_event(peer, &event.path, &err).await;
-                    return Err(err);
-                }
-            }
-            dequeue_site_replication_retry_event(peer, &event.path).await;
-            Ok(true)
-        }
-        RetryDrainAction::PeerEdit => {
-            // The recorded generation is stale by definition — the receiver
-            // fences it. Allocate a fresh generation and re-send the current
-            // peer records (a superset of the failed body; the receiver
-            // upserts), all inside one state transaction so the fence and the
-            // bodies agree.
-            let target_id = peer.deployment_id.clone();
-            let (generation, bodies) = update_site_replication_state(move |state| {
-                if !state.peers.contains_key(&target_id) {
-                    return Ok((None, Vec::new()));
-                }
-                Ok((Some(next_peer_edit_generation(state)), state.peers.values().cloned().collect::<Vec<_>>()))
-            })
-            .await?;
-            let Some(generation) = generation else {
-                // Peer left between the snapshot and now; the queue entry was
-                // already pruned by remove_sites.
-                return Ok(false);
-            };
-            let local_deployment_id = Some(runtime.local_peer.deployment_id.as_str()).filter(|id| !id.is_empty());
-            let edit_path = peer_edit_path_with_fence(local_deployment_id, generation);
-            let delivery_fence = local_deployment_id.is_some().then_some(generation);
-            for body in &bodies {
-                if let Err(err) = send_peer_admin_request_with_client(
-                    &transport.client,
-                    &transport.connection,
-                    &edit_path,
-                    access_key,
-                    secret_key,
-                    body,
-                )
-                .await
-                {
-                    enqueue_site_replication_retry_event_for_generation(
-                        peer,
-                        SITE_REPLICATION_PEER_EDIT_PATH,
-                        &err,
-                        delivery_fence,
-                    )
-                    .await;
-                    return Err(err);
-                }
-            }
-            dequeue_site_replication_retry_event_for_generation(peer, SITE_REPLICATION_PEER_EDIT_PATH, delivery_fence).await;
-            Ok(true)
-        }
-    }
-}
-
 /// Remove a retry event for (peer, path) from the queue on successful delivery.
 /// This is a no-op (load + no-op persist skipped) when no matching entry exists,
 /// avoiding unnecessary I/O on the common path.
@@ -11870,213 +11427,6 @@ mod tests {
         assert!(target_state.peers["remote"].skip_tls_verify);
     }
 
-    fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option<OffsetDateTime>) -> SiteReplicationRetryEvent {
-        SiteReplicationRetryEvent {
-            id: format!("evt-{peer}"),
-            peer_deployment_id: peer.to_string(),
-            peer_endpoint: format!("https://{peer}.example.com"),
-            path: path.to_string(),
-            retry_count,
-            failed: retry_count >= SITE_REPLICATION_RETRY_FAILED_AFTER,
-            last_error: "remote-operation-failed".to_string(),
-            updated_at,
-            edit_generation: None,
-        }
-    }
-
-    /// P1-3 red-light: the drain must only ever act on deliveries it can
-    /// replay faithfully. IAM / bucket-meta entries collapse per (peer, path)
-    /// with no body persisted — only a snapshot resend is truthful; bucket
-    /// makes/replication configs are re-derivable; destructive bucket ops and
-    /// `internal:` marker records (the pending-endpoint-refresh backup store)
-    /// are never background-replayed.
-    #[test]
-    fn test_classify_site_replication_retry_event_actions() {
-        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
-        let classify = |path: &str| classify_site_replication_retry_event(&drain_event("remote", path, 1, Some(now)));
-
-        assert_eq!(
-            classify("/rustfs/admin/v3/site-replication/peer/iam-item"),
-            Some(RetryDrainAction::IamSnapshot)
-        );
-        assert_eq!(
-            classify("/rustfs/admin/v3/site-replication/peer/bucket-meta"),
-            Some(RetryDrainAction::BucketMetadataSnapshot)
-        );
-        assert_eq!(classify(SITE_REPLICATION_PEER_EDIT_PATH), Some(RetryDrainAction::PeerEdit));
-        assert_eq!(
-            classify(
-                "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning&createdAt=1"
-            ),
-            Some(RetryDrainAction::BucketOpReplay {
-                operation: SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING.to_string(),
-                bucket: "photos".to_string(),
-            })
-        );
-        assert_eq!(
-            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=configure-replication"),
-            Some(RetryDrainAction::BucketOpReplay {
-                operation: SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION.to_string(),
-                bucket: "photos".to_string(),
-            })
-        );
-        // Destructive ops are operator territory: replaying a bucket delete
-        // against a peer whose bucket was since recreated is irreversible.
-        assert_eq!(
-            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket"),
-            None
-        );
-        assert_eq!(
-            classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=force-delete-bucket"),
-            None
-        );
-        // `internal:` records store payloads in `last_error`, not failures.
-        assert_eq!(classify(SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH), None);
-        assert_eq!(classify("internal:some-future-marker"), None);
-        assert_eq!(classify("/rustfs/admin/v3/site-replication/peer/unknown"), None);
-    }
-
-    /// Exponential backoff gates every attempt: without it a dead peer's
-    /// entries hit `failed` (retry_count >= 3) within 30 minutes of reconcile
-    /// ticks and the retry stats lose their signal.
-    #[test]
-    fn test_site_replication_retry_backoff_schedule() {
-        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
-        let at = |secs_ago: i64| Some(now - time::Duration::seconds(secs_ago));
-        let elapsed = |retry_count: u32, secs_ago: i64| {
-            site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", retry_count, at(secs_ago)), now)
-        };
-
-        // No record of when it failed: attempt now.
-        assert!(site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", 1, None), now));
-        // First failure: one reconcile interval.
-        assert!(!elapsed(1, 599));
-        assert!(elapsed(1, 601));
-        // Third failure: 600 * 2^2 = 2400s.
-        assert!(!elapsed(3, 1200));
-        assert!(elapsed(3, 2401));
-        // Ceiling: a long-dead peer is still probed daily, never less often.
-        assert!(!elapsed(30, 86_000));
-        assert!(elapsed(30, 86_401));
-    }
-
-    /// The actionable subset respects classification, peer membership and
-    /// backoff; everything else stays untouched in the queue.
-    #[test]
-    fn test_actionable_site_replication_retry_events_filters() {
-        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
-        let old = Some(now - time::Duration::seconds(700));
-        let mut state = SiteReplicationState::default();
-        state
-            .peers
-            .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
-
-        state.retry_queue = vec![
-            // Eligible: known peer, replayable, past backoff.
-            drain_event("remote", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old),
-            // Not yet due.
-            drain_event("remote", "/rustfs/admin/v3/site-replication/peer/bucket-meta", 2, Some(now)),
-            // Unknown peer (removed since the failure was recorded).
-            drain_event("gone", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old),
-            // Marker record, not a delivery failure.
-            drain_event("remote", SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH, 0, old),
-            // Destructive op: operator-only.
-            drain_event(
-                "remote",
-                "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket",
-                1,
-                old,
-            ),
-        ];
-
-        let actionable = actionable_site_replication_retry_events(&state, now);
-        assert_eq!(actionable.len(), 1, "only the due, replayable, known-peer event is actionable");
-        assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/iam-item");
-    }
-
-    /// The drain settles a peer-edit success under a freshly allocated
-    /// generation; legacy queue entries carry `edit_generation: None` and
-    /// must be cleared by that generation-scoped settlement (`(Some, None)`
-    /// falls through to removal), or the drain would spin on them forever.
-    #[test]
-    fn test_settle_clears_legacy_none_generation_event_for_generation_scoped_success() {
-        let target = peer("remote", "https://remote.example.com");
-        let mut queue = vec![drain_event("remote", SITE_REPLICATION_PEER_EDIT_PATH, 1, None)];
-        assert!(queue[0].edit_generation.is_none());
-
-        let settled = settle_site_replication_retry_events(&mut queue, &target, SITE_REPLICATION_PEER_EDIT_PATH, Some(42));
-
-        assert_eq!(settled, 1, "a legacy None-generation event must settle under a newer generation");
-        assert!(queue.is_empty());
-    }
-
-    /// A successful snapshot resend cannot prove a failed *deletion* was
-    /// replayed, so the collapsed entry is escalated (operator-visible,
-    /// drain-idle) instead of cleared — unless a newer failure was stamped
-    /// during the delivery window, which keeps the entry drain-eligible.
-    #[test]
-    fn test_escalate_up_to_marks_snapshot_replayed_and_keeps_newer_failures() {
-        let target = peer("remote", "https://remote.example.com");
-        let path = "/rustfs/admin/v3/site-replication/peer/iam-item";
-        let snapshot_at = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
-
-        // Failure re-stamped after the snapshot: untouched, still eligible.
-        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at + time::Duration::seconds(5)))];
-        assert_eq!(
-            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
-            0
-        );
-        assert!(!queue[0].failed);
-        assert!(
-            classify_site_replication_retry_event(&queue[0]).is_some(),
-            "a newer failure must stay drain-eligible"
-        );
-
-        // Unchanged since the snapshot: escalated, kept, drain-idle.
-        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at))];
-        assert_eq!(
-            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
-            1
-        );
-        assert_eq!(queue.len(), 1, "the entry must survive until remote absence is proven");
-        assert!(queue[0].failed);
-        assert_eq!(queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER);
-        assert!(
-            classify_site_replication_retry_event(&queue[0]).is_none(),
-            "a snapshot-replayed entry must not be re-sent daily"
-        );
-        // Ordinary success dequeues must not clear the marker: collapsed
-        // paths are shared by every entity, so a successful Bob update
-        // proves nothing about a failed Alice deletion (second review
-        // round).
-        assert_eq!(dequeue_site_replication_retry_events(&mut queue, &target, path), 0);
-        assert_eq!(queue.len(), 1, "an escalated entry must survive an ordinary delivery success");
-        // Only a repair — the operator's accountability transfer — settles it.
-        assert_eq!(dequeue_site_replication_retry_events_including_escalated(&mut queue, &target, path), 1);
-        assert!(queue.is_empty());
-
-        // A later hook failure overwrites the marker and re-arms the drain.
-        let mut queue = vec![drain_event("remote", path, 2, Some(snapshot_at))];
-        escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at));
-        upsert_site_replication_retry_event(&mut queue, &target, path, "peer offline", None);
-        assert!(classify_site_replication_retry_event(&queue[0]).is_some());
-
-        // Legacy entry without a timestamp: escalated.
-        let mut queue = vec![drain_event("remote", path, 2, None)];
-        assert_eq!(
-            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
-            1
-        );
-
-        // Other (peer, path) entries are untouched.
-        let mut queue = vec![drain_event("other", path, 2, Some(snapshot_at))];
-        assert_eq!(
-            escalate_site_replication_retry_events_up_to(&mut queue, &target, path, Some(snapshot_at)),
-            0
-        );
-        assert!(!queue[0].failed);
-    }
-
     #[test]
     fn test_pending_endpoint_refresh_retry_summary_redacts_pem() {
         let pem = "-----BEGIN CERTIFICATE-----\nsecret-marker\n-----END CERTIFICATE-----";
@@ -16922,31 +16272,17 @@ mod tests {
     async fn test_retry_event_persist_must_not_wipe_concurrent_locked_rmw() {
         publish_ready_iam_context().await;
 
-        const ROUNDS: usize = 8;
         let seed = SiteReplicationState {
             pending_rotation: Some(PendingRotation {
                 id: "rot-1".to_string(),
                 access_key: "svc-account".to_string(),
                 ..Default::default()
             }),
-            // Retry events are only recorded for current peers; seed them so
-            // the concurrency assertion below exercises the persist path.
-            peers: (0..ROUNDS)
-                .map(|round| {
-                    let deployment_id = format!("peer-{round}-deployment");
-                    (
-                        deployment_id.clone(),
-                        PeerInfo {
-                            endpoint: format!("https://peer-{round}.example:9000"),
-                            deployment_id,
-                            ..Default::default()
-                        },
-                    )
-                })
-                .collect(),
             ..Default::default()
         };
         save_site_replication_state(&seed).await.expect("seed state");
+
+        const ROUNDS: usize = 8;
         for round in 0..ROUNDS {
             let peer = PeerInfo {
                 endpoint: format!("https://peer-{round}.example:9000"),

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,
     ),
 ];