fix(replication): expose bounded MRF queue drops

Signed-off-by: Feng Ruohang <rh@vonng.com>

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)
 		}
 	}
 }

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()

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() {

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
 }

cmd/metrics-v3.go

@@ -342,6 +342,8 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection {
 			replicationMaxQueuedCountMD,
 			replicationMaxDataTransferRateMD,
 			replicationRecentBacklogCountMD,
+			replicationMRFDroppedOperationsMD,
+			replicationMRFDroppedBytesMD,
 		},
 		loadClusterReplicationMetrics,
 	)

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)
+			}
+		}
+	}
+}

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
 

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. <br><br>Type: gauge    | `server` |
 | `minio_replication_max_data_transfer_rate`        | Maximum replication data transfer rate in bytes/sec since server start. <br><br>Type: gauge | `server` |
 | `minio_replication_recent_backlog_count`          | Total number of objects seen in replication backlog in the last 5 minutes <br><br>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. <br><br>Type: counter | `server` |
+| `minio_replication_mrf_dropped_bytes_total` | Known bytes of dropped MRF entries since server start; delete entries count as zero bytes. <br><br>Type: counter | `server` |
 #### `/bucket/replication`
 
 | Name                                                                | Description                                                                                                     | Labels                                                |