diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index def8a74f8..ce7fc69b0 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -2354,9 +2354,9 @@ func (p *ReplicationPool) queueReplicaTask(ri ReplicateObjectInfo) {
switch ri.OpType {
case replication.HealReplicationType, replication.ExistingObjectReplicationType:
ch = p.mrfReplicaCh
- healCh = p.getWorkerCh(ri.Name, ri.Bucket, ri.Size)
+ healCh = p.getWorkerCh(ri.Bucket, ri.Name, ri.Size)
default:
- ch = p.getWorkerCh(ri.Name, ri.Bucket, ri.Size)
+ ch = p.getWorkerCh(ri.Bucket, ri.Name, ri.Size)
}
if ch == nil && healCh == nil {
return
@@ -3820,6 +3820,7 @@ func (p *ReplicationPool) queueMRFSave(entry MRFReplicateEntry) {
if entry.RetryCount > mrfRetryLimit { // let scanner catch up if retry count exceeded
atomic.AddUint64(&p.stats.mrfStats.TotalDroppedCount, 1)
atomic.AddUint64(&p.stats.mrfStats.TotalDroppedBytes, uint64(entry.sz))
+ replLogOnceIf(GlobalContext, errors.New("Replication MRF retry limit reached; further repair is deferred to the scanner"), "replication-mrf-retry-limit", logger.WarningKind)
return
}
@@ -3834,6 +3835,7 @@ func (p *ReplicationPool) queueMRFSave(entry MRFReplicateEntry) {
default:
atomic.AddUint64(&p.stats.mrfStats.TotalDroppedCount, 1)
atomic.AddUint64(&p.stats.mrfStats.TotalDroppedBytes, uint64(entry.sz))
+ replLogOnceIf(GlobalContext, errors.New("Replication MRF queue is full; dropped entries will need scanner repair"), "replication-mrf-queue-full", logger.WarningKind)
}
}
}
diff --git a/cmd/bucket-stats.go b/cmd/bucket-stats.go
index e2eb00aa4..573c1caf1 100644
--- a/cmd/bucket-stats.go
+++ b/cmd/bucket-stats.go
@@ -319,7 +319,9 @@ func (r *ReplicationStats) getNodeQueueStats(bucket string) (qs ReplQNodeStats)
qs.QStats = r.qCache.getBucketStats(bucket)
qs.TgtXferStats = make(map[string]map[RMetricName]XferStats)
qs.MRFStats = ReplicationMRFStats{
- LastFailedCount: atomic.LoadUint64(&r.mrfStats.LastFailedCount),
+ LastFailedCount: atomic.LoadUint64(&r.mrfStats.LastFailedCount),
+ TotalDroppedCount: atomic.LoadUint64(&r.mrfStats.TotalDroppedCount),
+ TotalDroppedBytes: atomic.LoadUint64(&r.mrfStats.TotalDroppedBytes),
}
r.RLock()
@@ -410,7 +412,9 @@ func (r *ReplicationStats) getNodeQueueStatsSummary() (qs ReplQNodeStats) {
qs.XferStats = make(map[RMetricName]XferStats)
qs.QStats = r.qCache.getSiteStats()
qs.MRFStats = ReplicationMRFStats{
- LastFailedCount: atomic.LoadUint64(&r.mrfStats.LastFailedCount),
+ LastFailedCount: atomic.LoadUint64(&r.mrfStats.LastFailedCount),
+ TotalDroppedCount: atomic.LoadUint64(&r.mrfStats.TotalDroppedCount),
+ TotalDroppedBytes: atomic.LoadUint64(&r.mrfStats.TotalDroppedBytes),
}
r.RLock()
defer r.RUnlock()
diff --git a/cmd/metrics-v2.go b/cmd/metrics-v2.go
index b8d5c1290..120db3494 100644
--- a/cmd/metrics-v2.go
+++ b/cmd/metrics-v2.go
@@ -992,6 +992,26 @@ func getClusterReplMRFFailedOperationsMD() MetricDescription {
}
}
+func getClusterReplMRFDroppedOperationsMD() MetricDescription {
+ return MetricDescription{
+ Namespace: nodeMetricNamespace,
+ Subsystem: replicationSubsystem,
+ Name: "mrf_dropped_operations_total",
+ Help: "Total number of replication MRF entries dropped since server start; entries may refer to the same object",
+ Type: counterMetric,
+ }
+}
+
+func getClusterReplMRFDroppedBytesMD() MetricDescription {
+ return MetricDescription{
+ Namespace: nodeMetricNamespace,
+ Subsystem: replicationSubsystem,
+ Name: "mrf_dropped_bytes_total",
+ Help: "Total known bytes of replication MRF entries dropped since server start; delete entries count as zero bytes",
+ Type: counterMetric,
+ }
+}
+
func getClusterRepCredentialErrorsMD(namespace MetricNamespace) MetricDescription {
return MetricDescription{
Namespace: namespace,
@@ -2423,6 +2443,8 @@ func getReplicationNodeMetrics(opts MetricsGroupOpts) *MetricsGroupV2 {
avgTransferRate,
maxTransferRate,
mrfCount,
+ {Description: getClusterReplMRFDroppedOperationsMD(), Value: float64(qs.MRFStats.TotalDroppedCount)},
+ {Description: getClusterReplMRFDroppedBytesMD(), Value: float64(qs.MRFStats.TotalDroppedBytes)},
}
}
for ep, health := range globalBucketTargetSys.healthStats() {
diff --git a/cmd/metrics-v3-replication.go b/cmd/metrics-v3-replication.go
index 44a8e87ae..adaf59ccd 100644
--- a/cmd/metrics-v3-replication.go
+++ b/cmd/metrics-v3-replication.go
@@ -35,6 +35,8 @@ const (
replicationMaxQueuedCount = "max_queued_count"
replicationMaxDataTransferRate = "max_data_transfer_rate"
replicationRecentBacklogCount = "recent_backlog_count"
+ replicationMRFDroppedOperations = "mrf_dropped_operations_total"
+ replicationMRFDroppedBytes = "mrf_dropped_bytes_total"
)
var (
@@ -64,6 +66,10 @@ var (
"Maximum replication data transfer rate in bytes/sec seen since server start")
replicationRecentBacklogCountMD = NewGaugeMD(replicationRecentBacklogCount,
"Total number of objects seen in replication backlog in the last 5 minutes")
+ replicationMRFDroppedOperationsMD = NewCounterMD(replicationMRFDroppedOperations,
+ "Total number of replication MRF entries dropped since server start; entries may refer to the same object")
+ replicationMRFDroppedBytesMD = NewCounterMD(replicationMRFDroppedBytes,
+ "Total known bytes of replication MRF entries dropped since server start; delete entries count as zero bytes")
)
// loadClusterReplicationMetrics - `MetricsLoaderFn` for cluster replication metrics
@@ -96,6 +102,8 @@ func loadClusterReplicationMetrics(ctx context.Context, m MetricValues, c *metri
m.Set(replicationMaxDataTransferRate, tots.Peak)
}
m.Set(replicationRecentBacklogCount, float64(qs.MRFStats.LastFailedCount))
+ m.Set(replicationMRFDroppedOperations, float64(qs.MRFStats.TotalDroppedCount))
+ m.Set(replicationMRFDroppedBytes, float64(qs.MRFStats.TotalDroppedBytes))
return nil
}
diff --git a/cmd/metrics-v3.go b/cmd/metrics-v3.go
index 5b578448d..501589a26 100644
--- a/cmd/metrics-v3.go
+++ b/cmd/metrics-v3.go
@@ -342,6 +342,8 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection {
replicationMaxQueuedCountMD,
replicationMaxDataTransferRateMD,
replicationRecentBacklogCountMD,
+ replicationMRFDroppedOperationsMD,
+ replicationMRFDroppedBytesMD,
},
loadClusterReplicationMetrics,
)
diff --git a/cmd/replication-mrf-observability_test.go b/cmd/replication-mrf-observability_test.go
new file mode 100644
index 000000000..5d0c5af7b
--- /dev/null
+++ b/cmd/replication-mrf-observability_test.go
@@ -0,0 +1,126 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package cmd
+
+import (
+ "encoding/json"
+ "fmt"
+ "sync/atomic"
+ "testing"
+
+ "github.com/minio/minio/internal/bucket/replication"
+ "github.com/prometheus/client_golang/prometheus"
+)
+
+// A full MRF queue and an exhausted retry budget drop queue entries, not the
+// source objects. Both drops must remain visible in the admin API snapshots.
+func TestReplicationMRFDropsVisible(t *testing.T) {
+ stats := NewReplicationStats(t.Context(), nil)
+ old := globalReplicationStats.Swap(stats)
+ t.Cleanup(func() { globalReplicationStats.Store(old) })
+ p := &ReplicationPool{
+ objLayer: &replicationMRFTestObjectLayer{},
+ stats: stats,
+ mrfSaveCh: make(chan MRFReplicateEntry, 1),
+ mrfStopCh: make(chan struct{}),
+ }
+ p.queueMRFSave(MRFReplicateEntry{sz: 10})
+ p.queueMRFSave(MRFReplicateEntry{sz: 20})
+ p.queueMRFSave(MRFReplicateEntry{sz: 30, RetryCount: mrfRetryLimit + 1})
+ if len(p.mrfSaveCh) != 1 {
+ t.Fatalf("queue length = %d, want 1", len(p.mrfSaveCh))
+ }
+ if got := atomic.LoadUint64(&stats.mrfStats.TotalDroppedCount); got != 2 {
+ t.Fatalf("dropped entries = %d, want 2", got)
+ }
+ for name, qs := range map[string]ReplQNodeStats{
+ "bucket": stats.getNodeQueueStats("test-bucket"),
+ "node": stats.getNodeQueueStatsSummary(),
+ } {
+ t.Run(name, func(t *testing.T) {
+ data, err := json.Marshal(qs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var decoded ReplQNodeStats
+ if err := json.Unmarshal(data, &decoded); err != nil {
+ t.Fatal(err)
+ }
+ if got := decoded.MRFStats; got.TotalDroppedCount != 2 || got.TotalDroppedBytes != 50 {
+ t.Fatalf("API MRF stats = %+v, want dropped count 2 and bytes 50", got)
+ }
+ })
+ }
+ oldTargets := globalBucketTargetSys
+ globalBucketTargetSys = &BucketTargetSys{}
+ t.Cleanup(func() { globalBucketTargetSys = oldTargets })
+ want := map[string]float64{"mrf_dropped_operations_total": 2, "mrf_dropped_bytes_total": 50}
+ seen := make(map[string]bool)
+ for _, metric := range getReplicationNodeMetrics(MetricsGroupOpts{}).Get() {
+ name := string(metric.Description.Name)
+ if value, ok := want[name]; ok {
+ seen[name] = true
+ if metric.Value != value || metric.Description.Type != counterMetric {
+ t.Errorf("v2 %s = %+v, want counter %v", name, metric, value)
+ }
+ }
+ }
+ for name := range want {
+ if !seen[name] {
+ t.Errorf("v2 does not expose %s", name)
+ }
+ }
+ groups := newMetricGroups(prometheus.NewRegistry())
+ families, err := groups.mgGatherers[replicationCollectorPath].Gather()
+ if err != nil {
+ t.Fatal(err)
+ }
+ seen = make(map[string]bool)
+ for _, family := range families {
+ for name, value := range want {
+ if family.GetName() != "minio_replication_"+name {
+ continue
+ }
+ seen[name] = true
+ if len(family.Metric) != 1 || family.Metric[0].Counter == nil || family.Metric[0].GetCounter().GetValue() != value {
+ t.Errorf("v3 %s = %v, want counter %v", name, family, value)
+ }
+ }
+ }
+ for name := range want {
+ if !seen[name] {
+ t.Errorf("v3 registry does not expose %s", name)
+ }
+ }
+}
+
+type replicationMRFTestObjectLayer struct{ ObjectLayer }
+
+func TestReplicationObjectDeleteWorkerAffinity(t *testing.T) {
+ p := &ReplicationPool{ctx: t.Context(), workers: make([]chan ReplicationWorkerOperation, 8)}
+ for i := range p.workers {
+ p.workers[i] = make(chan ReplicationWorkerOperation, 2)
+ }
+ for _, op := range []replication.Type{replication.ObjectReplicationType, replication.HealReplicationType, replication.ExistingObjectReplicationType} {
+ for i := range 10 {
+ name := fmt.Sprintf("object-%d", i)
+ p.queueReplicaTask(ReplicateObjectInfo{Bucket: "bucket", Name: name, OpType: op})
+ p.queueReplicaDeleteTask(DeletedObjectReplicationInfo{Bucket: "bucket", DeletedObject: DeletedObject{ObjectName: name}})
+ objectWorker, deleteWorker := -1, -1
+ for idx, ch := range p.workers {
+ for len(ch) > 0 {
+ switch (<-ch).(type) {
+ case ReplicateObjectInfo:
+ objectWorker = idx
+ case DeletedObjectReplicationInfo:
+ deleteWorker = idx
+ }
+ }
+ }
+ if objectWorker < 0 || objectWorker != deleteWorker {
+ t.Fatalf("operation %d, %s: object worker %d != delete worker %d", op, name, objectWorker, deleteWorker)
+ }
+ }
+ }
+}
diff --git a/docs/metrics/prometheus/list.md b/docs/metrics/prometheus/list.md
index 050e2577b..acd977825 100644
--- a/docs/metrics/prometheus/list.md
+++ b/docs/metrics/prometheus/list.md
@@ -132,6 +132,8 @@ For deployments with [bucket](https://silo.pgsty.com/administration/bucket-repli
| `minio_node_replication_max_queued_bytes` | Maximum number of bytes queued for replication seen since server start |
| `minio_node_replication_max_queued_count` | Maximum number of objects queued for replication seen since server start |
| `minio_node_replication_recent_backlog_count` | Total number of objects seen in replication backlog in the last 5 minutes |
+| `minio_node_replication_mrf_dropped_operations_total` | Cumulative MRF entries dropped due to queue capacity or retry exhaustion; may count the same object more than once. Scanner repair remains available. |
+| `minio_node_replication_mrf_dropped_bytes_total` | Cumulative known bytes of dropped MRF entries; delete entries count as zero bytes. |
## Healing Metrics
diff --git a/docs/metrics/v3.md b/docs/metrics/v3.md
index 192c2fda0..7583680f3 100644
--- a/docs/metrics/v3.md
+++ b/docs/metrics/v3.md
@@ -278,6 +278,8 @@ Metrics about Silo site and bucket replication.
| `minio_replication_max_queued_count` | Maximum number of objects queued for replication since server start.
Type: gauge | `server` |
| `minio_replication_max_data_transfer_rate` | Maximum replication data transfer rate in bytes/sec since server start.
Type: gauge | `server` |
| `minio_replication_recent_backlog_count` | Total number of objects seen in replication backlog in the last 5 minutes
Type: gauge | `server` |
+| `minio_replication_mrf_dropped_operations_total` | MRF entries dropped since server start due to queue capacity or retry exhaustion; entries may refer to the same object. Source objects remain eligible for scanner repair.
Type: counter | `server` |
+| `minio_replication_mrf_dropped_bytes_total` | Known bytes of dropped MRF entries since server start; delete entries count as zero bytes.
Type: counter | `server` |
#### `/bucket/replication`
| Name | Description | Labels |