fix(site-replication): only a repair settles snapshot-escalated retry entries

Second review round: every iam-item / bucket-meta delivery shares a
constant path, so any later successful single-item delivery (a Bob
update) dequeued the escalated marker recording a possibly-unreplayed
deletion (a failed Alice delete) while the entity still existed
remotely. Ordinary settlement now skips escalated entries; only the
repair path — the operator's explicit accountability transfer — clears
them via dequeue_..._including_escalated. A new hook failure still
overwrites the marker and re-arms the drain. Regression covers
survive-ordinary-dequeue and repair-clears.

crates/ecstore/src/api/mod.rs

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

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

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

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

@@ -704,12 +704,6 @@ impl ReplicationStats {
         } else {
             BucketReplicationStats::new()
         };
-        // Stamp the serializable failure windows from the live samples: the
-        // samples themselves do not cross the peer-RPC wire, so this snapshot
-        // is what cluster aggregation and the metrics endpoints see.
-        for stat in replication_stats.stats.values_mut() {
-            stat.fail_stats.refresh_windows();
-        }
         let uptime = if cache.contains_key(bucket) {
             SystemTime::now()
                 .duration_since(SystemTime::UNIX_EPOCH)

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

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

crates/replication/src/stats.rs

@@ -520,14 +520,6 @@ struct FailureSample {
 pub struct FailStats {
     pub count: i64,
     pub size: i64,
-    /// Rolling-window snapshots refreshed at collection time
-    /// ([`Self::refresh_windows`]). The raw samples (`recent`) are process
-    /// local (serde-skipped), so these fields are what survives the peer-RPC
-    /// wire and [`Self::merge`]-based cluster aggregation.
-    #[serde(default)]
-    pub last_minute: FailedMetric,
-    #[serde(default)]
-    pub last_hour: FailedMetric,
     #[serde(skip)]
     recent: VecDeque<FailureSample>,
 }
@@ -545,17 +537,6 @@ impl FailStats {
         self.prune(observed_at);
     }
 
-    /// Recompute the serializable rolling-window snapshots from the local
-    /// samples. Called at the collection point (per-node stats snapshot),
-    /// never on the failure hot path — the two deque scans are O(window) and
-    /// `add_size` runs under the bucket-stats write lock. Only meaningful on
-    /// the live per-node struct: a deserialized or merged struct has no
-    /// samples, and refreshing it would wipe the aggregated windows.
-    pub fn refresh_windows(&mut self) {
-        self.last_minute = self.recent_since(Duration::from_secs(60));
-        self.last_hour = self.recent_since(Duration::from_secs(3600));
-    }
-
     fn prune(&mut self, observed_at: Instant) {
         while self
             .recent
@@ -584,16 +565,6 @@ impl FailStats {
         Self {
             count: self.count.saturating_add(other.count),
             size: self.size.saturating_add(other.size),
-            // The window snapshots sum across nodes; the raw samples do not
-            // travel and stay empty on aggregated structs.
-            last_minute: FailedMetric {
-                count: self.last_minute.count.saturating_add(other.last_minute.count),
-                size: self.last_minute.size.saturating_add(other.last_minute.size),
-            },
-            last_hour: FailedMetric {
-                count: self.last_hour.count.saturating_add(other.last_hour.count),
-                size: self.last_hour.size.saturating_add(other.last_hour.size),
-            },
             recent: VecDeque::new(),
         }
     }
@@ -665,9 +636,7 @@ impl BucketReplicationStat {
     }
 
     pub fn update_xfer_rate(&mut self, size: i64, duration: Duration) {
-        // Same boundary as the worker-pool split and minio-go's
-        // Large/Small transfer-summary labels: >= 128 MiB is "large".
-        if size >= crate::runtime::MIN_LARGE_OBJ_SIZE {
+        if size > 1024 * 1024 {
             self.xfer_rate_lrg.add_size(size, duration);
         } else {
             self.xfer_rate_sml.add_size(size, duration);

rustfs/src/admin/handlers/replication.rs

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

rustfs/src/admin/handlers/site_replication.rs

@@ -3004,6 +3004,9 @@ 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;
     })
 }
 
@@ -3953,7 +3956,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(&mut state.retry_queue, &peer, &path);
+                dequeue_site_replication_retry_events_including_escalated(&mut state.retry_queue, &peer, &path);
             }
         }
         Ok(())
@@ -6041,6 +6044,20 @@ 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
@@ -6060,6 +6077,13 @@ 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,
@@ -6137,7 +6161,12 @@ 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| {
-        upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation);
+        // 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);
+        }
         Ok(())
     })
     .await;
@@ -6171,6 +6200,420 @@ 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.
@@ -11427,6 +11870,213 @@ 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-----";
@@ -16272,17 +16922,31 @@ 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/mod.rs

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

rustfs/src/admin/replication_metrics_wire.rs

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

rustfs/src/admin/router.rs

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

rustfs/src/admin/storage_api.rs

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