fix(replication): report authoritative backlog metrics (#5209)

crates/ecstore/src/api/mod.rs

@@ -149,18 +149,19 @@ pub mod bucket {
 
     pub mod replication {
         pub use crate::bucket::replication::{
-            BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, MustReplicateOptions,
-            ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationConfig,
-            ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource,
-            ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait,
-            ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats,
-            ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, ReplicationType, ResyncOpts,
-            ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType, delete_replication_state_from_config,
-            delete_replication_version_id, get_global_replication_pool, get_global_replication_stats,
-            init_background_replication, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map,
-            replication_target_arns, should_remove_replication_target, should_schedule_delete_replication,
-            should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
-            validate_replication_config_target_arns, version_purge_status_to_filemeta,
+            BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool,
+            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
+            ReplicateObjectInfo, ReplicationConfig, ReplicationConfigurationExt, ReplicationDeleteScheduleInput,
+            ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
+            ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
+            ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
+            ReplicationType, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType,
+            delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
+            get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, replication_state_to_filemeta,
+            replication_status_to_filemeta, replication_statuses_map, replication_target_arns, should_remove_replication_target,
+            should_schedule_delete_replication, should_use_existing_delete_replication_info,
+            should_use_existing_delete_replication_source, validate_replication_config_target_arns,
+            version_purge_status_to_filemeta,
         };
     }
 

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

@@ -52,9 +52,9 @@ pub use replication_config_boundary::{
 pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision;
 pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
 pub use replication_filemeta_boundary::{
-    REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType,
-    VersionPurgeStatusType, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map,
-    version_purge_status_to_filemeta,
+    MrfOpKind, MrfReplicateEntry, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState,
+    ReplicationStatusType, ReplicationType, VersionPurgeStatusType, replication_state_to_filemeta,
+    replication_status_to_filemeta, replication_statuses_map, version_purge_status_to_filemeta,
 };
 pub(crate) use replication_filemeta_boundary::{
     replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
@@ -69,8 +69,8 @@ pub use replication_object_decision_boundary::{
     should_use_existing_delete_replication_source,
 };
 pub use replication_pool::{
-    DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats,
-    init_background_replication,
+    DurableMrfBacklog, DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats,
+    init_background_replication, read_durable_mrf_backlog,
 };
 pub use replication_queue_boundary::{
     DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority,

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

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub(crate) use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
+pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
 pub(crate) use rustfs_replication::{
     REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos,
     ReplicatedTargetInfo, ReplicationAction, ReplicationWorkerOperation, ResyncDecision, get_replication_state,

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

@@ -68,6 +68,33 @@ const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_ski
 const EVENT_REPLICATION_RESYNC_RECOVERED: &str = "replication_resync_recovered";
 const EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE: &str = "replication_mrf_queue_unavailable";
 
+#[derive(Debug, Default)]
+pub struct DurableMrfBacklog {
+    pub available: bool,
+    pub entries: Vec<MrfReplicateEntry>,
+}
+
+fn durable_mrf_backlog_from_read(result: Result<Vec<u8>, EcstoreError>) -> DurableMrfBacklog {
+    match result {
+        Ok(data) => match decode_mrf_file(&data) {
+            Ok(entries) if entries.iter().all(|entry| entry.size >= 0) => DurableMrfBacklog {
+                available: true,
+                entries,
+            },
+            Ok(_) | Err(_) => DurableMrfBacklog::default(),
+        },
+        Err(EcstoreError::ConfigNotFound) => DurableMrfBacklog {
+            available: true,
+            entries: Vec::new(),
+        },
+        Err(_) => DurableMrfBacklog::default(),
+    }
+}
+
+pub async fn read_durable_mrf_backlog<S: ReplicationObjectIO>(storage: Arc<S>) -> DurableMrfBacklog {
+    durable_mrf_backlog_from_read(ReplicationConfigStore::read(storage, ReplicationMetadataStore::MRF_REPLICATION_FILE).await)
+}
+
 /// Main replication pool structure
 #[derive(Debug)]
 pub struct ReplicationPool<S: ReplicationStorage> {
@@ -2233,4 +2260,53 @@ mod tests {
         // None so replay falls back to the current time (backlog#867 backward compatibility).
         assert_eq!(entry.delete_marker_mtime, None, "missing deleteMarkerMtime key must default to None");
     }
+
+    #[test]
+    fn durable_mrf_snapshot_reads_restart_backlog_and_valid_empty_state() {
+        let entries = vec![MrfReplicateEntry {
+            bucket: "restart-bucket".to_string(),
+            object: "object".to_string(),
+            version_id: None,
+            retry_count: 1,
+            size: 512,
+            op: MrfOpKind::Object,
+            delete_marker_version_id: None,
+            delete_marker: false,
+            delete_marker_mtime: None,
+        }];
+        let encoded = encode_mrf_file(&entries).expect("durable MRF backlog should encode");
+
+        let recovered = durable_mrf_backlog_from_read(Ok(encoded));
+        assert!(recovered.available);
+        assert_eq!(recovered.entries.len(), 1);
+        assert_eq!(recovered.entries[0].bucket, "restart-bucket");
+        assert_eq!(recovered.entries[0].size, 512);
+
+        let missing_file = durable_mrf_backlog_from_read(Err(EcstoreError::ConfigNotFound));
+        assert!(missing_file.available);
+        assert!(missing_file.entries.is_empty());
+    }
+
+    #[test]
+    fn durable_mrf_snapshot_marks_corrupt_or_invalid_data_unavailable() {
+        let corrupt = durable_mrf_backlog_from_read(Ok(vec![0, 1, 2]));
+        assert!(!corrupt.available);
+        assert!(corrupt.entries.is_empty());
+
+        let negative = encode_mrf_file(&[MrfReplicateEntry {
+            bucket: "bucket".to_string(),
+            object: "object".to_string(),
+            version_id: None,
+            retry_count: 0,
+            size: -1,
+            op: MrfOpKind::Object,
+            delete_marker_version_id: None,
+            delete_marker: false,
+            delete_marker_mtime: None,
+        }])
+        .expect("invalid persisted entry should still encode for boundary testing");
+        let invalid = durable_mrf_backlog_from_read(Ok(negative));
+        assert!(!invalid.available);
+        assert!(invalid.entries.is_empty());
+    }
 }

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

@@ -15,9 +15,11 @@
 use super::replication_error_boundary::Error;
 use super::replication_filemeta_boundary::{ReplicatedTargetInfo, ReplicationStatusType, ReplicationType};
 use super::replication_resync_boundary::ResyncStatusType;
+#[cfg(test)]
+use super::replication_stats_boundary::FailStats;
 use super::replication_stats_boundary::{
     ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, ProxyMetric, ProxyStatsCache,
-    QueueCache, SRMetricsSummary, XferStats,
+    QueueCache, ReplicationMetricScope, SRMetricsSummary, XferStats,
 };
 use super::runtime_boundary as runtime_sources;
 use std::collections::HashMap;
@@ -361,10 +363,12 @@ impl ReplicationStats {
                 if rs.transfer_duration > Duration::default() {
                     stat.latency.update(rs.transfer_size, rs.transfer_duration);
                     stat.update_xfer_rate(rs.transfer_size, rs.transfer_duration);
+                    stat.latency_scope = ReplicationMetricScope::NodeLocal;
                 }
             }
             (false, true, false) => {
                 stat.fail_stats.add_size(rs.transfer_size, rs.err.as_ref());
+                stat.failed = stat.fail_stats.to_metric();
             }
             (false, false, true) => {
                 // Pending status, no processing for now
@@ -379,7 +383,10 @@ impl ReplicationStats {
         let mut result = HashMap::with_capacity(cache.len());
 
         for (bucket, stats) in cache.iter() {
-            result.insert(bucket.clone(), stats.clone_stats());
+            let mut snapshot = stats.clone_stats();
+            snapshot.mark_node_local_provider_available();
+            snapshot.queue_scope = ReplicationMetricScope::NodeLocal;
+            result.insert(bucket.clone(), snapshot);
         }
         drop(cache);
 
@@ -388,6 +395,8 @@ impl ReplicationStats {
             for (bucket, queue_stats) in &q_cache.bucket_stats {
                 let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new);
                 bucket_stats.q_stat = queue_stats.snapshot();
+                bucket_stats.mark_node_local_provider_available();
+                bucket_stats.queue_scope = ReplicationMetricScope::NodeLocal;
             }
         }
 
@@ -405,9 +414,13 @@ impl ReplicationStats {
     pub async fn get(&self, bucket: &str) -> BucketReplicationStats {
         let cache = self.cache.read().await;
         if let Some(stats) = cache.get(bucket) {
-            stats.clone_stats()
+            let mut snapshot = stats.clone_stats();
+            snapshot.mark_node_local_provider_available();
+            snapshot
         } else {
-            BucketReplicationStats::new()
+            let mut snapshot = BucketReplicationStats::new();
+            snapshot.mark_node_local_provider_available();
+            snapshot
         }
     }
 
@@ -453,11 +466,15 @@ impl ReplicationStats {
         let mut tq = InQueueMetric::default();
 
         for bucket_stat in &bucket_stats {
-            tot_replica_size += bucket_stat.replication_stats.replica_size;
-            tot_replica_count += bucket_stat.replication_stats.replica_count;
+            tot_replica_size = tot_replica_size.saturating_add(bucket_stat.replication_stats.replica_size);
+            tot_replica_count = tot_replica_count.saturating_add(bucket_stat.replication_stats.replica_count);
 
-            for q in &bucket_stat.queue_stats.nodes {
-                tq = tq.merge(&q.q_stats);
+            if bucket_stat.replication_stats.queue_scope != ReplicationMetricScope::Unavailable {
+                tq = tq.merge(&bucket_stat.replication_stats.q_stat);
+            } else {
+                for q in &bucket_stat.queue_stats.nodes {
+                    tq = tq.merge(&q.q_stats);
+                }
             }
 
             for (arn, stat) in &bucket_stat.replication_stats.stats {
@@ -470,22 +487,38 @@ impl ReplicationStats {
                 let f_stats = stat.fail_stats.merge(&old_stat.fail_stats);
                 let lrg = old_stat.xfer_rate_lrg.merge(&stat.xfer_rate_lrg);
                 let sml = old_stat.xfer_rate_sml.merge(&stat.xfer_rate_sml);
+                let latency_available = stat.latency_scope != ReplicationMetricScope::Unavailable
+                    || old_stat.latency_scope != ReplicationMetricScope::Unavailable;
+                let bandwidth_available = stat.bandwidth_scope != ReplicationMetricScope::Unavailable
+                    || old_stat.bandwidth_scope != ReplicationMetricScope::Unavailable;
 
                 *old_stat = BucketReplicationStat {
                     failed: f_stats.to_metric(),
                     fail_stats: f_stats,
-                    replicated_size: stat.replicated_size + old_stat.replicated_size,
-                    replicated_count: stat.replicated_count + old_stat.replicated_count,
+                    replicated_size: stat.replicated_size.saturating_add(old_stat.replicated_size),
+                    replicated_count: stat.replicated_count.saturating_add(old_stat.replicated_count),
                     latency: stat.latency.merge(&old_stat.latency),
                     xfer_rate_lrg: lrg,
                     xfer_rate_sml: sml,
-                    bandwidth_limit_bytes_per_sec: stat.bandwidth_limit_bytes_per_sec,
+                    bandwidth_limit_bytes_per_sec: stat
+                        .bandwidth_limit_bytes_per_sec
+                        .saturating_add(old_stat.bandwidth_limit_bytes_per_sec),
                     current_bandwidth_bytes_per_sec: stat.current_bandwidth_bytes_per_sec
                         + old_stat.current_bandwidth_bytes_per_sec,
+                    latency_scope: if latency_available {
+                        ReplicationMetricScope::ClusterAggregated
+                    } else {
+                        ReplicationMetricScope::Unavailable
+                    },
+                    bandwidth_scope: if bandwidth_available {
+                        ReplicationMetricScope::ClusterAggregated
+                    } else {
+                        ReplicationMetricScope::Unavailable
+                    },
                 };
 
-                tot_replicated_size += stat.replicated_size;
-                tot_replicated_count += stat.replicated_count;
+                tot_replicated_size = tot_replicated_size.saturating_add(stat.replicated_size);
+                tot_replicated_count = tot_replicated_count.saturating_add(stat.replicated_count);
             }
         }
 
@@ -499,23 +532,28 @@ impl ReplicationStats {
             resync_started_count: bucket_stats
                 .iter()
                 .map(|stats| stats.replication_stats.resync_started_count)
-                .sum(),
+                .fold(0i64, i64::saturating_add),
             resync_completed_count: bucket_stats
                 .iter()
                 .map(|stats| stats.replication_stats.resync_completed_count)
-                .sum(),
+                .fold(0i64, i64::saturating_add),
             resync_failed_count: bucket_stats
                 .iter()
                 .map(|stats| stats.replication_stats.resync_failed_count)
-                .sum(),
+                .fold(0i64, i64::saturating_add),
             resync_canceled_count: bucket_stats
                 .iter()
                 .map(|stats| stats.replication_stats.resync_canceled_count)
-                .sum(),
+                .fold(0i64, i64::saturating_add),
             resync_duration_ms: bucket_stats
                 .iter()
                 .map(|stats| stats.replication_stats.resync_duration_ms)
-                .sum(),
+                .fold(0i64, i64::saturating_add),
+            provider_available: true,
+            cluster_complete: true,
+            observed_node_count: u32::try_from(bucket_stats.len()).unwrap_or(u32::MAX),
+            expected_node_count: u32::try_from(bucket_stats.len()).unwrap_or(u32::MAX),
+            queue_scope: ReplicationMetricScope::ClusterAggregated,
         };
 
         let qs = Default::default();
@@ -547,6 +585,33 @@ impl ReplicationStats {
         bs
     }
 
+    pub async fn aggregate_bucket_replication_stats(
+        &self,
+        bucket: &str,
+        bucket_stats: Vec<BucketStats>,
+        expected_node_count: u32,
+    ) -> BucketStats {
+        let mut aggregated = self.calculate_bucket_replication_stats(bucket, bucket_stats).await;
+        let observed_node_count = aggregated.replication_stats.observed_node_count;
+        let complete = observed_node_count == expected_node_count;
+        aggregated.replication_stats.expected_node_count = expected_node_count;
+        aggregated.replication_stats.cluster_complete = complete;
+        aggregated.replication_stats.queue_scope = if complete {
+            ReplicationMetricScope::ClusterAggregated
+        } else {
+            ReplicationMetricScope::PartialCluster
+        };
+        for stat in aggregated.replication_stats.stats.values_mut() {
+            if stat.latency_scope != ReplicationMetricScope::Unavailable {
+                stat.latency_scope = aggregated.replication_stats.queue_scope;
+            }
+            if stat.bandwidth_scope != ReplicationMetricScope::Unavailable {
+                stat.bandwidth_scope = aggregated.replication_stats.queue_scope;
+            }
+        }
+        aggregated
+    }
+
     /// Get latest replication statistics
     pub async fn get_latest_replication_stats(&self, bucket: &str) -> BucketStats {
         // In actual implementation, statistics would be obtained from cluster
@@ -567,6 +632,15 @@ impl ReplicationStats {
         };
         drop(cache);
 
+        {
+            let q_cache = self.q_cache.lock().await;
+            if let Some(queue_stats) = q_cache.bucket_stats.get(bucket) {
+                replication_stats.q_stat = queue_stats.snapshot();
+            }
+        }
+        replication_stats.mark_node_local_provider_available();
+        replication_stats.queue_scope = ReplicationMetricScope::NodeLocal;
+
         if let Some(monitor) = runtime_sources::bucket_monitor() {
             let bw_report = monitor.get_report(|name| name == bucket);
             for (opts, bw) in bw_report.bucket_stats {
@@ -578,8 +652,7 @@ impl ReplicationStats {
                         xfer_rate_sml: XferStats::new(),
                         ..Default::default()
                     });
-                stat.bandwidth_limit_bytes_per_sec = bw.limit_bytes_per_sec;
-                stat.current_bandwidth_bytes_per_sec = bw.current_bandwidth_bytes_per_sec;
+                stat.set_node_local_bandwidth(bw.limit_bytes_per_sec, bw.current_bandwidth_bytes_per_sec);
             }
         }
 
@@ -724,6 +797,132 @@ mod tests {
         assert_eq!(stat.replicated_count, 1);
     }
 
+    #[tokio::test]
+    async fn latest_stats_include_queue_until_drained() {
+        let stats = ReplicationStats::new();
+
+        stats.inc_q("queued-bucket", 4096, false, ReplicationType::Object).await;
+        let queued = stats.get_latest_replication_stats("queued-bucket").await;
+        assert!(queued.replication_stats.provider_available);
+        assert_eq!(queued.replication_stats.q_stat.curr.count, 1);
+        assert_eq!(queued.replication_stats.q_stat.curr.bytes, 4096);
+        assert_eq!(queued.replication_stats.queue_scope, ReplicationMetricScope::NodeLocal);
+
+        stats.dec_q("queued-bucket", 4096, false, ReplicationType::Object).await;
+        let drained = stats.get_latest_replication_stats("queued-bucket").await;
+        assert_eq!(drained.replication_stats.q_stat.curr.count, 0);
+        assert_eq!(drained.replication_stats.q_stat.curr.bytes, 0);
+    }
+
+    #[tokio::test]
+    async fn failed_metric_matches_authoritative_fail_stats() {
+        let stats = ReplicationStats::new();
+        let target_info = ReplicatedTargetInfo {
+            arn: "failed-arn".to_string(),
+            size: 2048,
+            duration: Duration::from_millis(25),
+            op_type: ReplicationType::Object,
+            error: Some("target unavailable".to_string()),
+            ..Default::default()
+        };
+
+        stats
+            .update(
+                "failed-bucket",
+                &target_info,
+                ReplicationStatusType::Failed,
+                ReplicationStatusType::Pending,
+            )
+            .await;
+
+        let snapshot = stats.get_latest_replication_stats("failed-bucket").await;
+        let target = &snapshot.replication_stats.stats["failed-arn"];
+        assert_eq!(target.failed.count, target.fail_stats.count);
+        assert_eq!(target.failed.size, target.fail_stats.size);
+        assert_eq!(target.failed.count, 1);
+        assert_eq!(target.failed.size, 2048);
+    }
+
+    #[tokio::test]
+    async fn valid_empty_provider_is_not_reported_as_unavailable() {
+        let stats = ReplicationStats::new();
+
+        let snapshot = stats.get_latest_replication_stats("empty-bucket").await;
+
+        assert!(snapshot.replication_stats.provider_available);
+        assert!(snapshot.replication_stats.cluster_complete);
+        assert_eq!(snapshot.replication_stats.observed_node_count, 1);
+        assert_eq!(snapshot.replication_stats.expected_node_count, 1);
+        assert!(snapshot.replication_stats.stats.is_empty());
+    }
+
+    #[tokio::test]
+    async fn cluster_aggregation_counts_each_node_once_and_marks_partial() {
+        let stats = ReplicationStats::new();
+        let node = |failed_count, failed_size, queued_count, queued_size| {
+            let mut fail_stats = FailStats::new();
+            fail_stats.count = failed_count;
+            fail_stats.size = failed_size;
+            let mut targets = HashMap::new();
+            targets.insert(
+                "arn".to_string(),
+                BucketReplicationStat {
+                    fail_stats,
+                    latency_scope: ReplicationMetricScope::NodeLocal,
+                    ..Default::default()
+                },
+            );
+            let q_stat = InQueueMetric::default();
+            q_stat.curr.now_count.store(queued_count, Ordering::Relaxed);
+            q_stat.curr.now_bytes.store(queued_size, Ordering::Relaxed);
+            let q_stat = q_stat.snapshot();
+            BucketStats {
+                replication_stats: BucketReplicationStats {
+                    stats: targets,
+                    q_stat,
+                    provider_available: true,
+                    queue_scope: ReplicationMetricScope::NodeLocal,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }
+        };
+
+        let aggregated = stats
+            .aggregate_bucket_replication_stats("bucket", vec![node(1, 10, 2, 20), node(3, 30, 4, 40)], 3)
+            .await;
+
+        let target = &aggregated.replication_stats.stats["arn"];
+        assert_eq!(target.failed.count, 4);
+        assert_eq!(target.failed.size, 40);
+        assert_eq!(aggregated.replication_stats.q_stat.curr.count, 6);
+        assert_eq!(aggregated.replication_stats.q_stat.curr.bytes, 60);
+        assert_eq!(aggregated.replication_stats.observed_node_count, 2);
+        assert_eq!(aggregated.replication_stats.expected_node_count, 3);
+        assert!(!aggregated.replication_stats.cluster_complete);
+        assert_eq!(aggregated.replication_stats.queue_scope, ReplicationMetricScope::PartialCluster);
+        assert_eq!(target.latency_scope, ReplicationMetricScope::PartialCluster);
+    }
+
+    #[tokio::test]
+    async fn concurrent_queue_updates_are_visible_without_lost_counts() {
+        let stats = Arc::new(ReplicationStats::new());
+        let mut tasks = Vec::with_capacity(32);
+        for _ in 0..32 {
+            let stats = Arc::clone(&stats);
+            tasks.push(tokio::spawn(async move {
+                stats.inc_q("concurrent-bucket", 7, false, ReplicationType::Object).await;
+            }));
+        }
+        for task in tasks {
+            task.await.expect("queue update task should complete");
+        }
+
+        let snapshot = stats.get_latest_replication_stats("concurrent-bucket").await;
+        assert_eq!(snapshot.replication_stats.q_stat.curr.count, 32);
+        assert_eq!(snapshot.replication_stats.q_stat.curr.bytes, 224);
+    }
+
     #[tokio::test]
     async fn test_get_all_includes_proxy_only_bucket() {
         let stats = ReplicationStats::new();

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

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 pub use rustfs_replication::BucketStats;
+#[cfg(test)]
+pub(crate) use rustfs_replication::FailStats;
 pub(crate) use rustfs_replication::{
     ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache,
-    SRMetricsSummary, XferStats,
+    ReplicationMetricScope, SRMetricsSummary, XferStats,
 };

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -19,6 +19,7 @@ use crate::cluster::rpc::client::{
 use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof};
 use crate::error::{Error, Result};
 use crate::{
+    bucket::replication::BucketStats,
     disk::disk_store::{get_drive_active_check_interval, get_drive_active_check_timeout},
     layout::endpoints::EndpointServerPools,
     runtime::sources as runtime_sources,
@@ -34,14 +35,15 @@ use rustfs_madmin::{
 };
 use rustfs_protos::proto_gen::node_service::{
     BackgroundHealStatusRequest, CancelDecommissionRequest, ClearDecommissionRequest, DeleteBucketMetadataRequest,
-    DeletePolicyRequest, DeleteServiceAccountRequest, DeleteUserRequest, GetCpusRequest, GetLiveEventsRequest, GetMemInfoRequest,
-    GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest,
-    GetSysConfigRequest, GetSysErrorsRequest, HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest,
-    LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
-    LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest,
-    ReloadSiteReplicationConfigRequest, ScannerActivityRequest, ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest,
-    StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest,
-    TierMutationControlResponse, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
+    DeletePolicyRequest, DeleteServiceAccountRequest, DeleteUserRequest, GetBucketStatsDataRequest, GetBucketStatsDataResponse,
+    GetCpusRequest, GetLiveEventsRequest, GetMemInfoRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest,
+    GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, GetSysConfigRequest, GetSysErrorsRequest,
+    HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest,
+    LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest,
+    LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ScannerActivityRequest,
+    ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, StartDecommissionRequest, StartProfilingRequest,
+    StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse,
+    TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
     tier_mutation_control_service_client::TierMutationControlServiceClient,
 };
 use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection};
@@ -74,6 +76,26 @@ const HEAL_CONTROL_PAYLOAD_MAX_SIZE: usize = 64 * 1024;
 const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
 const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
+const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024;
+
+fn decode_bucket_stats_response(response: GetBucketStatsDataResponse) -> Result<BucketStats> {
+    if !response.success {
+        return Err(Error::other(
+            response
+                .error_info
+                .unwrap_or_else(|| "peer replication statistics provider is unavailable".to_string()),
+        ));
+    }
+    if response.bucket_stats.len() > REPLICATION_STATS_MAX_MESSAGE_SIZE {
+        return Err(Error::other("peer replication statistics response exceeds size limit"));
+    }
+    let mut buf = Deserializer::new(Cursor::new(response.bucket_stats));
+    let stats = BucketStats::deserialize(&mut buf).map_err(Error::from)?;
+    if !stats.replication_stats.provider_available {
+        return Err(Error::other("peer replication statistics provider is unavailable"));
+    }
+    Ok(stats)
+}
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ScannerPeerActivity {
@@ -819,9 +841,26 @@ impl PeerRestClient {
         Err(Error::NotImplemented)
     }
 
-    pub async fn get_bucket_stats(&self) -> Result<()> {
-        warn!("get_bucket_stats is not implemented in PeerRestClient");
-        Err(Error::NotImplemented)
+    pub async fn get_bucket_stats(&self, bucket: &str) -> Result<BucketStats> {
+        let response = self
+            .finalize_result(
+                async {
+                    let mut client = self
+                        .get_client()
+                        .await?
+                        .max_decoding_message_size(REPLICATION_STATS_MAX_MESSAGE_SIZE);
+                    let response = client
+                        .get_bucket_stats(Request::new(GetBucketStatsDataRequest {
+                            bucket: bucket.to_string(),
+                        }))
+                        .await?
+                        .into_inner();
+                    Ok(response)
+                }
+                .await,
+            )
+            .await?;
+        decode_bucket_stats_response(response)
     }
 
     pub async fn get_sr_metrics(&self) -> Result<()> {
@@ -1508,6 +1547,50 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
 
+    #[test]
+    fn replication_stats_response_decodes_valid_empty_provider() {
+        let mut stats = BucketStats::default();
+        stats.replication_stats.provider_available = true;
+        let payload = rmp_serde::to_vec_named(&stats).expect("bucket statistics should encode");
+
+        let decoded = decode_bucket_stats_response(GetBucketStatsDataResponse {
+            success: true,
+            bucket_stats: payload.into(),
+            error_info: None,
+        })
+        .expect("valid bucket statistics should decode");
+
+        assert!(decoded.replication_stats.provider_available);
+        assert!(decoded.replication_stats.stats.is_empty());
+    }
+
+    #[test]
+    fn replication_stats_response_rejects_unavailable_malformed_and_oversized_payloads() {
+        let unavailable = decode_bucket_stats_response(GetBucketStatsDataResponse {
+            success: false,
+            bucket_stats: Bytes::new(),
+            error_info: Some("provider unavailable".to_string()),
+        })
+        .expect_err("unavailable provider must not become a zero snapshot");
+        assert!(unavailable.to_string().contains("provider unavailable"));
+
+        let malformed = decode_bucket_stats_response(GetBucketStatsDataResponse {
+            success: true,
+            bucket_stats: Bytes::from_static(b"not-msgpack"),
+            error_info: None,
+        })
+        .expect_err("malformed peer statistics must fail closed");
+        assert!(!malformed.to_string().is_empty());
+
+        let oversized = decode_bucket_stats_response(GetBucketStatsDataResponse {
+            success: true,
+            bucket_stats: Bytes::from(vec![0; REPLICATION_STATS_MAX_MESSAGE_SIZE + 1]),
+            error_info: None,
+        })
+        .expect_err("oversized peer statistics must fail closed");
+        assert!(oversized.to_string().contains("size limit"));
+    }
+
     #[derive(Clone, Default)]
     struct CapturedLogs {
         buffer: Arc<Mutex<Vec<u8>>>,

crates/replication/src/lib.rs

@@ -82,7 +82,7 @@ pub use runtime::{
 pub use stats::{
     ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, BucketStats, ExponentialMovingAverage, FailStats,
     FailedMetric, InQueueMetric, InQueueStats, LatencyStats, ProxyMetric, ProxyStatsCache, QueueCache, QueueNode, QueueStats,
-    SRMetricsSummary, XferStats,
+    ReplicationMetricScope, SRMetricsSummary, XferStats,
 };
 pub use storage_api::{DeletedObject, ObjectToDelete};
 pub use tagging::{ReplicationTagFilter, decode_tags_to_map};

crates/replication/src/stats.rs

@@ -313,18 +313,24 @@ impl InQueueMetric {
     pub fn merge(&self, other: &InQueueMetric) -> Self {
         Self {
             curr: InQueueStats {
-                bytes: self.curr.bytes + other.curr.bytes,
-                count: self.curr.count + other.curr.count,
+                bytes: self.curr.bytes.saturating_add(other.curr.bytes),
+                count: self.curr.count.saturating_add(other.curr.count),
                 now_bytes: AtomicI64::new(
-                    self.curr.now_bytes.load(Ordering::Relaxed) + other.curr.now_bytes.load(Ordering::Relaxed),
+                    self.curr
+                        .now_bytes
+                        .load(Ordering::Relaxed)
+                        .saturating_add(other.curr.now_bytes.load(Ordering::Relaxed)),
                 ),
                 now_count: AtomicI64::new(
-                    self.curr.now_count.load(Ordering::Relaxed) + other.curr.now_count.load(Ordering::Relaxed),
+                    self.curr
+                        .now_count
+                        .load(Ordering::Relaxed)
+                        .saturating_add(other.curr.now_count.load(Ordering::Relaxed)),
                 ),
             },
             avg: InQueueStats {
-                bytes: (self.avg.bytes + other.avg.bytes) / 2,
-                count: (self.avg.count + other.avg.count) / 2,
+                bytes: self.avg.bytes.saturating_add(other.avg.bytes) / 2,
+                count: self.avg.count.saturating_add(other.avg.count) / 2,
                 ..Default::default()
             },
             max: InQueueStats {
@@ -333,8 +339,8 @@ impl InQueueMetric {
                 ..Default::default()
             },
             last_minute: InQueueStats {
-                bytes: self.last_minute.bytes + other.last_minute.bytes,
-                count: self.last_minute.count + other.last_minute.count,
+                bytes: self.last_minute.bytes.saturating_add(other.last_minute.bytes),
+                count: self.last_minute.count.saturating_add(other.last_minute.count),
                 ..Default::default()
             },
             samples: VecDeque::new(),
@@ -495,8 +501,8 @@ impl FailStats {
 
     pub fn add_size<E>(&mut self, size: i64, _err: Option<&E>) {
         let observed_at = Instant::now();
-        self.count += 1;
-        self.size += size;
+        self.count = self.count.saturating_add(1);
+        self.size = self.size.saturating_add(size);
         self.recent.push_back(FailureSample { observed_at, size });
         self.prune(observed_at);
     }
@@ -527,8 +533,8 @@ impl FailStats {
 
     pub fn merge(&self, other: &FailStats) -> Self {
         Self {
-            count: self.count + other.count,
-            size: self.size + other.size,
+            count: self.count.saturating_add(other.count),
+            size: self.size.saturating_add(other.size),
             recent: VecDeque::new(),
         }
     }
@@ -588,6 +594,10 @@ pub struct BucketReplicationStat {
     pub xfer_rate_sml: XferStats,
     pub bandwidth_limit_bytes_per_sec: i64,
     pub current_bandwidth_bytes_per_sec: f64,
+    #[serde(default)]
+    pub latency_scope: ReplicationMetricScope,
+    #[serde(default)]
+    pub bandwidth_scope: ReplicationMetricScope,
 }
 
 impl BucketReplicationStat {
@@ -602,6 +612,22 @@ impl BucketReplicationStat {
             self.xfer_rate_sml.add_size(size, duration);
         }
     }
+
+    pub fn set_node_local_bandwidth(&mut self, limit_bytes_per_sec: i64, current_bytes_per_sec: f64) {
+        self.bandwidth_limit_bytes_per_sec = limit_bytes_per_sec;
+        self.current_bandwidth_bytes_per_sec = current_bytes_per_sec;
+        self.bandwidth_scope = ReplicationMetricScope::NodeLocal;
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ReplicationMetricScope {
+    #[default]
+    Unavailable,
+    NodeLocal,
+    ClusterAggregated,
+    PartialCluster,
 }
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -632,6 +658,16 @@ pub struct BucketReplicationStats {
     pub resync_canceled_count: i64,
     #[serde(default)]
     pub resync_duration_ms: i64,
+    #[serde(default)]
+    pub provider_available: bool,
+    #[serde(default)]
+    pub cluster_complete: bool,
+    #[serde(default)]
+    pub observed_node_count: u32,
+    #[serde(default)]
+    pub expected_node_count: u32,
+    #[serde(default)]
+    pub queue_scope: ReplicationMetricScope,
 }
 
 impl BucketReplicationStats {
@@ -662,7 +698,18 @@ impl BucketReplicationStats {
     }
 
     pub fn clone_stats(&self) -> Self {
-        self.clone()
+        let mut snapshot = self.clone();
+        for stat in snapshot.stats.values_mut() {
+            stat.failed = stat.fail_stats.to_metric();
+        }
+        snapshot
+    }
+
+    pub fn mark_node_local_provider_available(&mut self) {
+        self.provider_available = true;
+        self.cluster_complete = true;
+        self.observed_node_count = 1;
+        self.expected_node_count = 1;
     }
 
     pub fn record_resync_status(&mut self, status: ResyncStatusType, duration: Option<Duration>) {
@@ -791,6 +838,27 @@ mod tests {
         assert_eq!(last_hour.size, 96);
     }
 
+    #[test]
+    fn fail_stats_saturate_instead_of_wrapping() {
+        let mut stats = FailStats {
+            count: i64::MAX,
+            size: i64::MAX,
+            ..Default::default()
+        };
+
+        stats.add_size(1, None::<&()>);
+        let merged = stats.merge(&FailStats {
+            count: 1,
+            size: 1,
+            ..Default::default()
+        });
+
+        assert_eq!(stats.count, i64::MAX);
+        assert_eq!(stats.size, i64::MAX);
+        assert_eq!(merged.count, i64::MAX);
+        assert_eq!(merged.size, i64::MAX);
+    }
+
     #[test]
     fn active_worker_stat_update_tracks_rolling_avg_and_max() {
         let mut stats = ActiveWorkerStat::default();

rustfs/src/admin/handlers/replication.rs

@@ -15,7 +15,10 @@
 use crate::admin::auth::validate_admin_request;
 use crate::admin::handlers::site_replication::site_replication_peer_deployment_id_for_endpoint;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
-use crate::admin::runtime_sources::{current_replication_stats_handle, current_runtime_port, object_store_from_req};
+use crate::admin::runtime_sources::{
+    AppContext, app_context_from_req, current_notification_system_for_context, current_replication_stats_handle_for_context,
+    current_runtime_port, object_store_from_req,
+};
 use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE;
 use crate::admin::storage_api::bucket::metadata_sys;
 use crate::admin::storage_api::bucket::metadata_sys::get_replication_config;
@@ -25,6 +28,7 @@ use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTar
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
 use crate::admin::storage_api::contract::list::ListOperations as _;
 use crate::admin::storage_api::error::StorageError;
+use crate::admin::storage_api::runtime::PeerRestClient;
 use crate::admin::utils::read_compatible_admin_body;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::error::ApiError;
@@ -39,7 +43,8 @@ use rustfs_policy::policy::action::{Action, AdminAction};
 use s3s::header::CONTENT_TYPE;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
 use std::time::Duration;
 use time::OffsetDateTime;
 use tracing::{debug, error, info, warn};
@@ -283,6 +288,39 @@ fn is_local_host(_host: String) -> bool {
     false
 }
 
+pub(crate) async fn cluster_replication_stats(bucket: &str, context: Option<Arc<AppContext>>) -> BucketStats {
+    let Some(stats) = current_replication_stats_handle_for_context(context.clone()) else {
+        return BucketStats::default();
+    };
+
+    let local = stats.get_latest_replication_stats(bucket).await;
+    let Some(notification_system) = current_notification_system_for_context(context.as_deref()) else {
+        return local;
+    };
+
+    let (peers, expected_node_count) = unique_replication_peers(&notification_system.peer_clients);
+    let peer_results = futures_util::future::join_all(peers.into_iter().map(|peer| peer.get_bucket_stats(bucket))).await;
+    let mut snapshots = Vec::with_capacity(peer_results.len().saturating_add(1));
+    snapshots.push(local);
+    snapshots.extend(peer_results.into_iter().filter_map(Result::ok));
+
+    stats
+        .aggregate_bucket_replication_stats(bucket, snapshots, expected_node_count)
+        .await
+}
+
+fn unique_replication_peers(peer_clients: &[Option<PeerRestClient>]) -> (Vec<&PeerRestClient>, u32) {
+    let mut seen_grid_hosts = HashSet::new();
+    let peers: Vec<_> = peer_clients
+        .iter()
+        .filter_map(|peer| peer.as_ref())
+        .filter(|peer| seen_grid_hosts.insert(peer.grid_host.clone()))
+        .collect();
+    let unavailable_slots = peer_clients.iter().filter(|peer| peer.is_none()).count();
+    let expected_node_count = u32::try_from(peers.len().saturating_add(unavailable_slots).saturating_add(1)).unwrap_or(u32::MAX);
+    (peers, expected_node_count)
+}
+
 //awscurl --service s3 --region us-east-1 --access_key rustfsadmin --secret_key rustfsadmin "http://:9000/rustfs/admin/v3/replicationmetrics?bucket=1"
 pub struct GetReplicationMetricsHandler {}
 
@@ -322,13 +360,7 @@ impl Operation for GetReplicationMetricsHandler {
             return Err(ApiError::from(err).into());
         }
 
-        // TODO cluster cache
-        // In actual implementation, statistics would be obtained from cluster
-        // This is simplified to get from local cache
-        let bucket_stats = match current_replication_stats_handle() {
-            Some(s) => s.get_latest_replication_stats(bucket).await,
-            None => BucketStats::default(),
-        };
+        let bucket_stats = cluster_replication_stats(bucket, app_context_from_req(&req)).await;
 
         let data = serde_json::to_vec(&bucket_stats.replication_stats)
             .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "serialize failed"))?;
@@ -779,9 +811,8 @@ impl Operation for ReplicationDiffHandler {
     }
 }
 
-/// Failed-replication backlog for one remote target (ARN), summarised from the
-/// replication stats. RustFS tracks failed replications as aggregate counters,
-/// not an enumerable per-object MRF queue, so this reports the aggregate.
+/// Failed-replication totals for one remote target (ARN), summarised from the
+/// runtime replication statistics.
 #[derive(Debug, Serialize)]
 struct MrfTargetBacklog {
     #[serde(rename = "ARN")]
@@ -790,14 +821,14 @@ struct MrfTargetBacklog {
     failed_count: i64,
     #[serde(rename = "FailedSize")]
     failed_size: i64,
+    #[serde(rename = "ObservationScope")]
+    observation_scope: &'static str,
 }
 
 /// Response body for `GET /v3/replication/mrf`.
 ///
-/// MinIO streams individual MRF (most-recently-failed) entries. RustFS persists
-/// MRF entries to disk but does not expose a runtime query over that queue, so
-/// this endpoint reports the aggregate failed-replication backlog and the
-/// in-queue counters instead. See the handler docs and the PR limitations note.
+/// Runtime failed/queued totals and the durable MRF recovery backlog are kept
+/// separate because persisted MRF entries do not contain a target ARN.
 #[derive(Debug, Serialize)]
 struct MrfResponse {
     #[serde(rename = "Bucket")]
@@ -814,18 +845,89 @@ struct MrfResponse {
     queued_size: i64,
     #[serde(rename = "PerObjectEntriesAvailable")]
     per_object_entries_available: bool,
+    #[serde(rename = "RuntimeStatsAvailable")]
+    runtime_stats_available: bool,
+    #[serde(rename = "ClusterComplete")]
+    cluster_complete: bool,
+    #[serde(rename = "ObservedNodeCount")]
+    observed_node_count: u32,
+    #[serde(rename = "ExpectedNodeCount")]
+    expected_node_count: u32,
+    #[serde(rename = "DurableBacklogAvailable")]
+    durable_backlog_available: bool,
+    #[serde(rename = "DurableCount")]
+    durable_count: i64,
+    #[serde(rename = "DurableSize")]
+    durable_size: i64,
+    #[serde(rename = "PerTargetDurableEntriesAvailable")]
+    per_target_durable_entries_available: bool,
+}
+
+fn build_mrf_response(
+    bucket: String,
+    bucket_stats: &BucketStats,
+    durable: crate::admin::storage_api::replication::DurableMrfBacklog,
+) -> MrfResponse {
+    let observation_scope = if bucket_stats.replication_stats.cluster_complete {
+        "cluster_aggregated"
+    } else {
+        "partial_cluster"
+    };
+    let mut targets: Vec<MrfTargetBacklog> = Vec::with_capacity(bucket_stats.replication_stats.stats.len());
+    let mut total_failed_count: i64 = 0;
+    let mut total_failed_size: i64 = 0;
+    for (arn, stat) in &bucket_stats.replication_stats.stats {
+        total_failed_count = total_failed_count.saturating_add(stat.failed.count);
+        total_failed_size = total_failed_size.saturating_add(stat.failed.size);
+        targets.push(MrfTargetBacklog {
+            arn: arn.clone(),
+            failed_count: stat.failed.count,
+            failed_size: stat.failed.size,
+            observation_scope,
+        });
+    }
+    targets.sort_by(|a, b| a.arn.cmp(&b.arn));
+
+    let queued = &bucket_stats.replication_stats.q_stat.curr;
+    let (durable_count, durable_size) = if durable.available {
+        durable
+            .entries
+            .iter()
+            .filter(|entry| entry.bucket == bucket)
+            .fold((0i64, 0i64), |(count, size), entry| {
+                (count.saturating_add(1), size.saturating_add(entry.size))
+            })
+    } else {
+        (0, 0)
+    };
+
+    MrfResponse {
+        bucket,
+        targets,
+        total_failed_count,
+        total_failed_size,
+        queued_count: queued.count,
+        queued_size: queued.bytes,
+        per_object_entries_available: false,
+        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,
+        expected_node_count: bucket_stats.replication_stats.expected_node_count,
+        durable_backlog_available: durable.available,
+        durable_count,
+        durable_size,
+        per_target_durable_entries_available: false,
+    }
 }
 
 /// `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 drained
-/// from its in-memory MRF queue. RustFS records failed replications as aggregate
-/// counters in the replication stats and persists MRF entries to disk without a
-/// runtime query API, so this handler returns the aggregate failed and queued
-/// counts per target instead of per-object rows. `PerObjectEntriesAvailable` is
-/// always `false` to make that limitation explicit to clients.
+/// Compatibility note: MinIO returns a stream of individual MRF entries. RustFS
+/// deliberately returns aggregate runtime and durable counters instead.
+/// `PerObjectEntriesAvailable` and `PerTargetDurableEntriesAvailable` remain
+/// false until an enumerable API and target-bearing durable format exist.
 pub struct ReplicationMrfHandler {}
 
 #[async_trait::async_trait]
@@ -857,35 +959,9 @@ impl Operation for ReplicationMrfHandler {
             return Err(ApiError::from(err).into());
         }
 
-        let bucket_stats = match current_replication_stats_handle() {
-            Some(s) => s.get_latest_replication_stats(&bucket).await,
-            None => BucketStats::default(),
-        };
-
-        let mut targets: Vec<MrfTargetBacklog> = Vec::new();
-        let mut total_failed_count: i64 = 0;
-        let mut total_failed_size: i64 = 0;
-        for (arn, stat) in &bucket_stats.replication_stats.stats {
-            total_failed_count += stat.failed.count;
-            total_failed_size += stat.failed.size;
-            targets.push(MrfTargetBacklog {
-                arn: arn.clone(),
-                failed_count: stat.failed.count,
-                failed_size: stat.failed.size,
-            });
-        }
-        targets.sort_by(|a, b| a.arn.cmp(&b.arn));
-
-        let queued = &bucket_stats.replication_stats.q_stat.curr;
-        let response = MrfResponse {
-            bucket,
-            targets,
-            total_failed_count,
-            total_failed_size,
-            queued_count: queued.count,
-            queued_size: queued.bytes,
-            per_object_entries_available: false,
-        };
+        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 data = serde_json::to_vec(&response)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?;
@@ -897,8 +973,12 @@ impl Operation for ReplicationMrfHandler {
 
 #[cfg(test)]
 mod tests {
-    use super::{RemoteTargetRequest, extract_query_params, validate_remote_target_tls_settings};
+    use super::{
+        RemoteTargetRequest, build_mrf_response, extract_query_params, unique_replication_peers,
+        validate_remote_target_tls_settings,
+    };
     use crate::admin::storage_api::bucket::target::BucketTarget;
+    use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
     use http::Uri;
 
     fn valid_remote_target_request() -> serde_json::Value {
@@ -914,6 +994,115 @@ mod tests {
         })
     }
 
+    #[test]
+    fn cluster_peer_plan_deduplicates_nodes_and_counts_unavailable_slots() {
+        let peer = crate::admin::storage_api::runtime::PeerRestClient::new(
+            rustfs_utils::XHost::try_from("127.0.0.1:9000".to_string()).expect("peer host should parse"),
+            "node-a.example.com:9001".to_string(),
+        );
+        let slots = vec![Some(peer.clone()), Some(peer), None];
+
+        let (peers, expected_node_count) = unique_replication_peers(&slots);
+
+        assert_eq!(peers.len(), 1);
+        assert_eq!(peers[0].grid_host, "node-a.example.com:9001");
+        assert_eq!(expected_node_count, 3);
+    }
+
+    #[test]
+    fn mrf_response_keeps_runtime_and_durable_truth_separate() {
+        let mut stats = BucketStats::default();
+        stats.replication_stats.provider_available = true;
+        stats.replication_stats.cluster_complete = false;
+        stats.replication_stats.observed_node_count = 2;
+        stats.replication_stats.expected_node_count = 3;
+        let target = stats.replication_stats.stats.entry("arn-a".to_string()).or_default();
+        target.failed.count = 3;
+        target.failed.size = 900;
+        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();
+        let durable = DurableMrfBacklog {
+            available: true,
+            entries: vec![
+                MrfReplicateEntry {
+                    bucket: "bucket-a".to_string(),
+                    object: "object-a".to_string(),
+                    version_id: None,
+                    retry_count: 0,
+                    size: 250,
+                    op: MrfOpKind::Object,
+                    delete_marker_version_id: None,
+                    delete_marker: false,
+                    delete_marker_mtime: None,
+                },
+                MrfReplicateEntry {
+                    bucket: "other-bucket".to_string(),
+                    object: "object-b".to_string(),
+                    version_id: None,
+                    retry_count: 0,
+                    size: 999,
+                    op: MrfOpKind::Object,
+                    delete_marker_version_id: None,
+                    delete_marker: false,
+                    delete_marker_mtime: None,
+                },
+            ],
+        };
+
+        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);
+        assert_eq!(json["TotalFailedSize"], 900);
+        assert_eq!(json["QueuedCount"], 4);
+        assert_eq!(json["QueuedSize"], 1200);
+        assert_eq!(json["DurableCount"], 1);
+        assert_eq!(json["DurableSize"], 250);
+        assert_eq!(json["RuntimeStatsAvailable"], true);
+        assert_eq!(json["ClusterComplete"], false);
+        assert_eq!(json["Targets"][0]["ObservationScope"], "partial_cluster");
+        assert_eq!(json["PerObjectEntriesAvailable"], false);
+        assert_eq!(json["PerTargetDurableEntriesAvailable"], false);
+    }
+
+    #[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_json = serde_json::to_value(unavailable).expect("unavailable response should serialize");
+        assert_eq!(unavailable_json["RuntimeStatsAvailable"], false);
+        assert_eq!(unavailable_json["DurableBacklogAvailable"], false);
+
+        let mut valid_empty_stats = BucketStats::default();
+        valid_empty_stats.replication_stats.provider_available = true;
+        valid_empty_stats.replication_stats.cluster_complete = true;
+        valid_empty_stats.replication_stats.observed_node_count = 1;
+        valid_empty_stats.replication_stats.expected_node_count = 1;
+        let valid_empty = build_mrf_response(
+            "bucket-a".to_string(),
+            &valid_empty_stats,
+            DurableMrfBacklog {
+                available: true,
+                entries: Vec::new(),
+            },
+        );
+        let valid_empty_json = serde_json::to_value(valid_empty).expect("valid empty response should serialize");
+        assert_eq!(valid_empty_json["RuntimeStatsAvailable"], true);
+        assert_eq!(valid_empty_json["DurableBacklogAvailable"], true);
+        assert_eq!(valid_empty_json["TotalFailedCount"], 0);
+        assert_eq!(valid_empty_json["DurableCount"], 0);
+    }
+
     #[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/router.rs

@@ -28,8 +28,8 @@ use super::storage_api::runtime::PeerRestClient;
 use crate::admin::console::{is_console_path, make_console_server};
 use crate::admin::handlers::oidc::is_oidc_path;
 use crate::admin::runtime_sources::{
-    ServerContextSlot, current_boot_time, current_bucket_monitor_handle, current_deployment_id, current_notification_system,
-    current_object_store_handle, current_region, current_replication_pool_handle, current_replication_stats_handle,
+    ServerContextSlot, app_context_from_req, current_boot_time, current_bucket_monitor_handle, current_deployment_id,
+    current_notification_system, current_object_store_handle, current_region, current_replication_pool_handle,
     current_server_config, default_object_usecase,
 };
 use crate::admin::storage_api::access::{ReqInfo, authorize_request, spawn_traced};
@@ -1509,11 +1509,12 @@ async fn ensure_replication_config_exists(bucket: &str) -> S3Result<()> {
     }
 }
 
-async fn build_replication_metrics_response(bucket: &str, route: ReplicationExtRoute) -> S3Result<S3Response<Body>> {
-    let bucket_stats = match current_replication_stats_handle() {
-        Some(stats) => stats.get_latest_replication_stats(bucket).await,
-        None => BucketStats::default(),
-    };
+async fn build_replication_metrics_response(
+    bucket: &str,
+    route: ReplicationExtRoute,
+    context: Option<Arc<crate::admin::runtime_sources::AppContext>>,
+) -> S3Result<S3Response<Body>> {
+    let bucket_stats = crate::admin::handlers::replication::cluster_replication_stats(bucket, context).await;
     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());
 
@@ -1555,10 +1556,12 @@ fn apply_replication_metrics_bandwidth_report(
     mut bucket_stats: BucketStats,
     bandwidth_report: HashMap<String, BandwidthDetails>,
 ) -> BucketStats {
+    if bucket_stats.replication_stats.expected_node_count > 1 {
+        return bucket_stats;
+    }
     for (arn, details) in bandwidth_report {
         let stat = bucket_stats.replication_stats.stats.entry(arn).or_default();
-        stat.bandwidth_limit_bytes_per_sec = details.limit_bytes_per_sec;
-        stat.current_bandwidth_bytes_per_sec = details.current_bandwidth_bytes_per_sec;
+        stat.set_node_local_bandwidth(details.limit_bytes_per_sec, details.current_bandwidth_bytes_per_sec);
     }
 
     bucket_stats
@@ -2519,7 +2522,7 @@ async fn handle_replication_extension_request(
     match ext_req.route {
         ReplicationExtRoute::MetricsV1 | ReplicationExtRoute::MetricsV2 => {
             ensure_replication_config_exists(&ext_req.bucket).await?;
-            build_replication_metrics_response(&ext_req.bucket, ext_req.route).await
+            build_replication_metrics_response(&ext_req.bucket, ext_req.route, app_context_from_req(req)).await
         }
         ReplicationExtRoute::Check => {
             let (versioning, _) = metadata_sys::get_versioning_config(&ext_req.bucket)
@@ -3738,6 +3741,10 @@ mod tests {
         assert_eq!(stat.replicated_count, 3);
         assert_eq!(stat.bandwidth_limit_bytes_per_sec, 2048);
         assert_eq!(stat.current_bandwidth_bytes_per_sec, 1536.5);
+        assert_eq!(
+            serde_json::to_value(stat).expect("target stats should serialize")["bandwidth_scope"],
+            "node_local"
+        );
     }
 
     #[test]
@@ -3761,6 +3768,34 @@ mod tests {
         assert_eq!(stat.current_bandwidth_bytes_per_sec, 1024.25);
     }
 
+    #[test]
+    fn apply_replication_metrics_bandwidth_report_does_not_replace_cluster_values() {
+        let mut stats = BucketStats::default();
+        stats.replication_stats.expected_node_count = 2;
+        let target = stats
+            .replication_stats
+            .stats
+            .entry("arn:replication:a".to_string())
+            .or_default();
+        target.bandwidth_limit_bytes_per_sec = 8192;
+        target.current_bandwidth_bytes_per_sec = 3000.0;
+
+        let updated = apply_replication_metrics_bandwidth_report(
+            stats,
+            HashMap::from([(
+                "arn:replication:a".to_string(),
+                BandwidthDetails {
+                    limit_bytes_per_sec: 2048,
+                    current_bandwidth_bytes_per_sec: 1000.0,
+                },
+            )]),
+        );
+        let target = &updated.replication_stats.stats["arn:replication:a"];
+
+        assert_eq!(target.bandwidth_limit_bytes_per_sec, 8192);
+        assert_eq!(target.current_bandwidth_bytes_per_sec, 3000.0);
+    }
+
     #[test]
     fn serialize_replication_metrics_body_v2_returns_full_bucket_stats() {
         let mut stats = BucketStats {

rustfs/src/admin/runtime_sources.rs

@@ -26,8 +26,8 @@ pub(crate) use crate::runtime_sources::{
     current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_service,
     current_iam_handle, current_kms_runtime_service_manager, current_notification_system_for_context,
     current_object_data_cache_handle_for_context, current_object_store_handle_for_context, current_ready_iam_handle,
-    current_region, current_replication_pool_handle, current_replication_stats_handle, current_server_config_for_context,
-    current_token_signing_key,
+    current_region, current_replication_pool_handle, current_replication_stats_handle,
+    current_replication_stats_handle_for_context, current_server_config_for_context, current_token_signing_key,
 };
 #[cfg(test)]
 pub(crate) use crate::runtime_sources::{IamInterface, KmsInterface, ServerConfigInterface, StorageClassInterface};

rustfs/src/admin/storage_api.rs

@@ -340,6 +340,15 @@ pub(crate) mod replication {
     pub(crate) type ResyncOpts = super::ecstore_bucket::replication::ResyncOpts;
     pub(crate) type ResyncStatusType = super::ecstore_bucket::replication::ResyncStatusType;
     pub(crate) type TargetReplicationResyncStatus = super::ecstore_bucket::replication::TargetReplicationResyncStatus;
+    pub(crate) type DurableMrfBacklog = super::ecstore_bucket::replication::DurableMrfBacklog;
+    #[cfg(test)]
+    pub(crate) type MrfReplicateEntry = super::ecstore_bucket::replication::MrfReplicateEntry;
+    #[cfg(test)]
+    pub(crate) type MrfOpKind = super::ecstore_bucket::replication::MrfOpKind;
+
+    pub(crate) async fn read_durable_mrf_backlog(api: std::sync::Arc<super::ECStore>) -> DurableMrfBacklog {
+        super::ecstore_bucket::replication::read_durable_mrf_backlog(api).await
+    }
 
     pub(crate) fn resync_opts(
         bucket: &str,

rustfs/src/app/context.rs

@@ -366,7 +366,7 @@ fn resolve_replication_pool_handle_with(context: Option<Arc<AppContext>>) -> Opt
     context.and_then(|context| context.replication_pool().handle())
 }
 
-fn resolve_replication_stats_handle_with(context: Option<Arc<AppContext>>) -> Option<Arc<ReplicationStats>> {
+pub(crate) fn resolve_replication_stats_handle_with(context: Option<Arc<AppContext>>) -> Option<Arc<ReplicationStats>> {
     context.and_then(|context| context.replication_stats().handle())
 }
 

rustfs/src/runtime_sources.rs

@@ -44,9 +44,11 @@ pub(crate) use context::{
     resolve_outbound_tls_generation as current_outbound_tls_generation, resolve_outbound_tls_state as current_outbound_tls_state,
     resolve_performance_metrics as current_performance_metrics, resolve_ready_iam_handle as current_ready_iam_handle,
     resolve_region as current_region, resolve_replication_pool_handle as current_replication_pool_handle,
-    resolve_replication_stats_handle as current_replication_stats_handle, resolve_runtime_port as current_runtime_port,
-    resolve_s3select_db as current_s3select_db, resolve_scanner_metrics_report as current_scanner_metrics_report,
-    resolve_server_config as current_server_config, resolve_server_config_for_context as current_server_config_for_context,
+    resolve_replication_stats_handle as current_replication_stats_handle,
+    resolve_replication_stats_handle_with as current_replication_stats_handle_for_context,
+    resolve_runtime_port as current_runtime_port, resolve_s3select_db as current_s3select_db,
+    resolve_scanner_metrics_report as current_scanner_metrics_report, resolve_server_config as current_server_config,
+    resolve_server_config_for_context as current_server_config_for_context,
     resolve_tier_config_handle as current_tier_config_handle, resolve_token_signing_key as current_token_signing_key,
 };
 

rustfs/src/storage/rpc/node_service.rs

@@ -1209,9 +1209,28 @@ impl Node for NodeService {
 
     async fn get_bucket_stats(
         &self,
-        _request: Request<GetBucketStatsDataRequest>,
+        request: Request<GetBucketStatsDataRequest>,
     ) -> Result<Response<GetBucketStatsDataResponse>, Status> {
-        Err(unimplemented_rpc("get_bucket_stats"))
+        let bucket = request.into_inner().bucket;
+        if bucket.is_empty() {
+            return Err(Status::invalid_argument("bucket is required"));
+        }
+        let context = self.context.clone().or_else(runtime_sources::current_app_context);
+        let Some(stats) = runtime_sources::current_replication_stats_handle_for_context(context.as_deref()) else {
+            return Ok(Response::new(GetBucketStatsDataResponse {
+                success: false,
+                bucket_stats: Bytes::new(),
+                error_info: Some("replication statistics provider is unavailable".to_string()),
+            }));
+        };
+        let bucket_stats = stats.get_latest_replication_stats(&bucket).await;
+        let bucket_stats =
+            rmp_serde::to_vec_named(&bucket_stats).map_err(|_| Status::internal("failed to serialize replication statistics"))?;
+        Ok(Response::new(GetBucketStatsDataResponse {
+            success: true,
+            bucket_stats: bucket_stats.into(),
+            error_info: None,
+        }))
     }
 
     async fn get_sr_metrics(
@@ -4428,12 +4447,11 @@ mod tests {
                 .await,
             "download_profile_data",
         );
-        assert_unimplemented_status(
-            service
-                .get_bucket_stats(Request::new(GetBucketStatsDataRequest::default()))
-                .await,
-            "get_bucket_stats",
-        );
+        let bucket_stats_err = service
+            .get_bucket_stats(Request::new(GetBucketStatsDataRequest::default()))
+            .await
+            .expect_err("empty bucket statistics request should fail");
+        assert_eq!(bucket_stats_err.code(), tonic::Code::InvalidArgument);
         assert_unimplemented_status(
             service.get_sr_metrics(Request::new(GetSrMetricsDataRequest::default())).await,
             "get_sr_metrics",

rustfs/src/storage/runtime_sources.rs

@@ -39,6 +39,12 @@ pub(crate) fn current_object_store_handle_for_context(context: Option<&AppContex
     root_runtime_sources::current_object_store_handle_for_context(context)
 }
 
+pub(crate) fn current_replication_stats_handle_for_context(
+    context: Option<&AppContext>,
+) -> Option<Arc<crate::storage::storage_api::ReplicationStats>> {
+    root_runtime_sources::current_replication_stats_handle_for_context(context.map(|context| Arc::new(context.clone())))
+}
+
 pub(crate) fn current_buffer_config() -> RustFSBufferConfig {
     root_runtime_sources::current_buffer_config().unwrap_or_default()
 }

rustfs/src/storage/storage_api.rs

@@ -699,6 +699,17 @@ impl StorageReplicationStatsHandle {
         self.inner.get_latest_replication_stats(bucket).await
     }
 
+    pub(crate) async fn aggregate_bucket_replication_stats(
+        &self,
+        bucket: &str,
+        stats: Vec<ecstore_bucket::replication::BucketStats>,
+        expected_node_count: u32,
+    ) -> ecstore_bucket::replication::BucketStats {
+        self.inner
+            .aggregate_bucket_replication_stats(bucket, stats, expected_node_count)
+            .await
+    }
+
     pub(crate) async fn site_metrics_snapshot(&self) -> ReplicationSiteMetricsSnapshot {
         let metrics = self.inner.get_sr_metrics_for_node().await;
         ReplicationSiteMetricsSnapshot {