Merge pull request #162 from pgsty/codex/replication-reliability-20260909

fix(replication): complete purges, expose MRF drops, and cancel resyncs reliably
2026-09-09 20:14:04 +03:00 · 2026-09-09 14:55:35 +08:00
parent 450dcb8484 66fe61ff65
commit d1105bbb3d
15 changed files with 1050 additions and 134 deletions
@@ -640,10 +640,10 @@ type VersionPurgeStatusType = replication.VersionPurgeStatusType

 type replicationResyncer struct {
 	// map of bucket to their resync status
-	statusMap      map[string]BucketReplicationResyncStatus
-	workerSize     int
-	resyncCancelCh chan struct{}
-	workerCh       chan struct{}
+	statusMap     map[string]BucketReplicationResyncStatus
+	workerSize    int
+	cancelResyncs map[resyncOpts]context.CancelCauseFunc
+	workerCh      chan struct{}
 	sync.RWMutex
 }

@@ -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
@@ -2964,10 +2964,10 @@ const (

 func newresyncer() *replicationResyncer {
 	rs := replicationResyncer{
-		statusMap:      make(map[string]BucketReplicationResyncStatus),
-		workerSize:     resyncWorkerCnt,
-		resyncCancelCh: make(chan struct{}, resyncWorkerCnt),
-		workerCh:       make(chan struct{}, resyncWorkerCnt),
+		statusMap:     make(map[string]BucketReplicationResyncStatus),
+		workerSize:    resyncWorkerCnt,
+		cancelResyncs: make(map[resyncOpts]context.CancelCauseFunc),
+		workerCh:      make(chan struct{}, resyncWorkerCnt),
 	}
 	for i := 0; i < rs.workerSize; i++ {
 		rs.workerCh <- struct{}{}
@@ -2975,13 +2975,67 @@ func newresyncer() *replicationResyncer {
 	return &rs
 }

+var errResyncCanceled = errors.New("replication resync canceled")
+
+// Registration and cancellation share the status lock. A queued run therefore
+// cannot miss cancellation between publishing its status and taking a slot.
+func (s *replicationResyncer) registerResync(parent context.Context, opts resyncOpts) (context.Context, context.CancelCauseFunc, bool) {
+	s.Lock()
+	defer s.Unlock()
+	st, ok := s.statusMap[opts.bucket].TargetsMap[opts.arn]
+	if !ok || st.ResyncID != opts.resyncID {
+		return nil, nil, false
+	}
+	if _, running := s.cancelResyncs[opts]; running {
+		return nil, nil, false
+	}
+	ctx, cancel := context.WithCancelCause(parent)
+	if s.cancelResyncs == nil {
+		s.cancelResyncs = make(map[resyncOpts]context.CancelCauseFunc)
+	}
+	s.cancelResyncs[opts] = cancel
+	if st.ResyncStatus == ResyncCanceled {
+		cancel(errResyncCanceled)
+	}
+	return ctx, cancel, true
+}
+
+func (s *replicationResyncer) cancelResyncID(resyncID string) {
+	s.Lock()
+	defer s.Unlock()
+	for bucket, m := range s.statusMap {
+		for arn, st := range m.TargetsMap {
+			if st.ResyncID == resyncID && (st.ResyncStatus == ResyncPending || st.ResyncStatus == ResyncStarted) {
+				st.ResyncStatus = ResyncCanceled
+				st.LastUpdate = UTCNow()
+				m.TargetsMap[arn] = st
+				m.LastUpdate = st.LastUpdate
+			}
+		}
+		s.statusMap[bucket] = m
+	}
+	for opts, cancel := range s.cancelResyncs {
+		if opts.resyncID == resyncID {
+			cancel(errResyncCanceled)
+		}
+	}
+}
+
 // mark status of replication resync on remote target for the bucket
-func (s *replicationResyncer) markStatus(status ResyncStatusType, opts resyncOpts, objAPI ObjectLayer) {
+func (s *replicationResyncer) markStatus(status ResyncStatusType, opts resyncOpts, objAPI ObjectLayer) ResyncStatusType {
 	s.Lock()
 	defer s.Unlock()

 	m := s.statusMap[opts.bucket]
-	st := m.TargetsMap[opts.arn]
+	st, ok := m.TargetsMap[opts.arn]
+	if !ok || st.ResyncID != opts.resyncID {
+		return NoResync
+	}
+	// A cancel may win the lock after the finalizer checked its context.
+	// Persist that cancellation, never a stale Started/Completed result.
+	if st.ResyncStatus == ResyncCanceled {
+		status = ResyncCanceled
+	}
 	st.LastUpdate = UTCNow()
 	st.ResyncStatus = status
 	m.TargetsMap[opts.arn] = st
@@ -2991,14 +3045,18 @@ func (s *replicationResyncer) markStatus(status ResyncStatusType, opts resyncOpt
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 	saveResyncStatus(ctx, opts.bucket, m, objAPI)
+	return status
 }

 // update replication resync stats for bucket's remote target
-func (s *replicationResyncer) incStats(ts TargetReplicationResyncStatus, opts resyncOpts) {
+func (s *replicationResyncer) incStats(ts TargetReplicationResyncStatus, opts resyncOpts) bool {
 	s.Lock()
 	defer s.Unlock()
 	m := s.statusMap[opts.bucket]
-	st := m.TargetsMap[opts.arn]
+	st, ok := m.TargetsMap[opts.arn]
+	if !ok || st.ResyncID != opts.resyncID || st.ResyncStatus == ResyncCanceled {
+		return false
+	}
 	st.Object = ts.Object
 	st.ReplicatedCount += ts.ReplicatedCount
 	st.FailedCount += ts.FailedCount
@@ -3007,6 +3065,7 @@ func (s *replicationResyncer) incStats(ts TargetReplicationResyncStatus, opts re
 	m.TargetsMap[opts.arn] = st
 	m.LastUpdate = UTCNow()
 	s.statusMap[opts.bucket] = m
+	return true
 }

 // resyncResults consumes the per-object outcomes produced by the resync worker
@@ -3023,8 +3082,9 @@ type resyncResults struct {
 // result into the bucket's resync status.
 func (s *replicationResyncer) newResyncResults(opts resyncOpts) *resyncResults {
 	return startResyncResults(func(r TargetReplicationResyncStatus) {
-		s.incStats(r, opts)
-		globalSiteResyncMetrics.updateMetric(r, opts.resyncID)
+		if s.incStats(r, opts) {
+			globalSiteResyncMetrics.updateMetric(r, opts.resyncID)
+		}
 	})
 }

@@ -3063,29 +3123,22 @@ func (rr *resyncResults) finish(workers []chan ReplicateObjectInfo, workerWg *sy
 }

 // sendResyncResult delivers a worker's computed per-object result to ch,
-// returning false if the worker must stop first. On the resync-cancel signal it
-// records the abort - the already-computed result is dropped - so
-// finalResyncStatus can downgrade a Completed run; on ctx cancellation it stops
-// without recording, since finalResyncStatus's parent-context check covers that.
-func (s *replicationResyncer) sendResyncResult(ctx context.Context, ch chan<- TargetReplicationResyncStatus, st TargetReplicationResyncStatus, workerAborted *atomic.Bool) bool {
+// returning false if the run's context was canceled first.
+func (s *replicationResyncer) sendResyncResult(ctx context.Context, ch chan<- TargetReplicationResyncStatus, st TargetReplicationResyncStatus) bool {
 	select {
 	case <-ctx.Done():
 		return false
-	case <-s.resyncCancelCh:
-		workerAborted.Store(true)
-		return false
 	case ch <- st:
 		return true
 	}
 }

-// finalResyncStatus downgrades a Completed status to Failed when the run could
-// not have observed every object: the parent context was canceled (workers then
-// return without sending their computed result) or a worker dropped a result on
-// the resync-cancel signal. Without this a persisted Completed would misrepresent
-// an incomplete resync.
-func finalResyncStatus(status ResyncStatusType, ctxErr error, workerAborted bool) ResyncStatusType {
-	if status == ResyncCompleted && (ctxErr != nil || workerAborted) {
+// User cancellation is terminal; interrupted completion remains retryable.
+func finalResyncStatus(status ResyncStatusType, cause error) ResyncStatusType {
+	if errors.Is(cause, errResyncCanceled) {
+		return ResyncCanceled
+	}
+	if status == ResyncCompleted && cause != nil {
 		return ResyncFailed
 	}
 	return status
@@ -3156,26 +3209,31 @@ func objectNeedsResyncForARN(roi ReplicateObjectInfo, arn string) bool {
 // resyncBucket resyncs all qualifying objects as per replication rules for the target
 // ARN
 func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI ObjectLayer, heal bool, opts resyncOpts) {
+	ctx, cancel, registered := s.registerResync(ctx, opts)
+	if !registered {
+		return
+	}
+	defer func() {
+		cancel(nil)
+		s.Lock()
+		delete(s.cancelResyncs, opts)
+		s.Unlock()
+	}()
 	select {
 	case <-s.workerCh: // block till a worker is available
 	case <-ctx.Done():
+		if errors.Is(context.Cause(ctx), errResyncCanceled) {
+			status := s.markStatus(ResyncCanceled, opts, objectAPI)
+			globalSiteResyncMetrics.incBucket(opts, status)
+		}
 		return
 	}

 	resyncStatus := ResyncFailed
-	// workerAborted records that a worker dropped an already-computed result on
-	// the resync-cancel signal. With a canceled parent context (which makes
-	// workers return without sending their result), it means a Completed run did
-	// not actually observe every object - see finalResyncStatus below.
-	var workerAborted atomic.Bool
 	defer func() {
-		// Downgrade a Completed status whose counts are incomplete, so the
-		// persisted status is not a misleading Completed. Runs after results.finish
-		// drains (LIFO) and before markStatus persists - markStatus uses its own
-		// background context, so a parent cancellation during the drain would
-		// otherwise still record Completed.
-		resyncStatus = finalResyncStatus(resyncStatus, ctx.Err(), workerAborted.Load())
-		s.markStatus(resyncStatus, opts, objectAPI)
+		// Runs after workers/results drain and before our own deferred cancel.
+		resyncStatus = finalResyncStatus(resyncStatus, context.Cause(ctx))
+		resyncStatus = s.markStatus(resyncStatus, opts, objectAPI)
 		globalSiteResyncMetrics.incBucket(opts, resyncStatus)
 		s.workerCh <- struct{}{}
 	}()
@@ -3210,8 +3268,8 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 		return
 	}
 	// mark resync status as resync started
-	if !heal {
-		s.markStatus(ResyncStarted, opts, objectAPI)
+	if !heal && s.markStatus(ResyncStarted, opts, objectAPI) != ResyncStarted {
+		return
 	}

 	// Walk through all object versions - Walk() is always in ascending order needed to ensure
@@ -3237,7 +3295,14 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 	// cannot race the last incStats. Registered after the markStatus finalizer, so
 	// LIFO runs finish first.
 	results := s.newResyncResults(opts)
-	defer results.finish(workers, &wg)
+	defer func() {
+		if resyncStatus != ResyncCompleted {
+			cancel(nil)
+		}
+		// Exactly one finish: success drains with a live context; errors stop
+		// blocked workers first. Both drain counts before persisting status.
+		results.finish(workers, &wg)
+	}()
 	for i := range resyncParallelRoutines {
 		wg.Add(1)
 		workers[i] = make(chan ReplicateObjectInfo, 100)
@@ -3248,7 +3313,6 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 				select {
 				case <-ctx.Done():
 					return
-				case <-s.resyncCancelCh:
 				default:
 				}
 				traceFn := s.trace(tgt.ResetID, fmt.Sprintf("%s/%s (%s)", opts.bucket, roi.Name, roi.VersionID))
@@ -3295,26 +3359,29 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 					}
 				}
 				traceFn(traceSize, traceErr)
-				if !s.sendResyncResult(ctx, results.ch, st, &workerAborted) {
+				if !s.sendResyncResult(ctx, results.ch, st) {
 					return
 				}
 			}
 		}(ctx, i)
 	}
-	for res := range objInfoCh {
+walkLoop:
+	for {
+		var res itemOrErr[ObjectInfo]
+		select {
+		case <-ctx.Done():
+			return
+		case item, ok := <-objInfoCh:
+			if !ok {
+				break walkLoop
+			}
+			res = item
+		}
 		if res.Err != nil {
 			resyncStatus = ResyncFailed
 			replLogIf(ctx, res.Err)
 			return
 		}
-		select {
-		case <-s.resyncCancelCh:
-			resyncStatus = ResyncCanceled
-			return
-		case <-ctx.Done():
-			return
-		default:
-		}
 		if heal && lastCheckpoint != "" && lastCheckpoint != res.Item.Name {
 			continue
 		}
@@ -3328,14 +3395,11 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 		if !objectNeedsResyncForARN(roi, opts.arn) {
 			continue
 		}
+		h := xxh3.HashString(roi.Bucket + roi.Name)
 		select {
-		case <-s.resyncCancelCh:
-			return
 		case <-ctx.Done():
 			return
-		default:
-			h := xxh3.HashString(roi.Bucket + roi.Name)
-			workers[h%uint64(resyncParallelRoutines)] <- roi
+		case workers[h%uint64(resyncParallelRoutines)] <- roi:
 		}
 	}
 	resyncStatus = ResyncCompleted
@@ -3363,9 +3427,9 @@ func (s *replicationResyncer) start(ctx context.Context, objAPI ObjectLayer, opt
 	if len(tgtArns) == 0 {
 		return fmt.Errorf("arn %s specified for resync not found in replication config", opts.arn)
 	}
-	globalReplicationPool.Get().resyncer.RLock()
-	data, ok := globalReplicationPool.Get().resyncer.statusMap[opts.bucket]
-	globalReplicationPool.Get().resyncer.RUnlock()
+	s.Lock()
+	defer s.Unlock()
+	data, ok := s.statusMap[opts.bucket]
 	if !ok {
 		data, err = loadBucketResyncMetadata(ctx, opts.bucket, objAPI)
 		if err != nil {
@@ -3386,23 +3450,14 @@ func (s *replicationResyncer) start(ctx context.Context, objAPI ObjectLayer, opt
 		ResyncStatus:     ResyncPending,
 		Bucket:           opts.bucket,
 	}
+	data.TargetsMap = data.cloneTgtStats()
 	data.TargetsMap[opts.arn] = status
 	if err = saveResyncStatus(ctx, opts.bucket, data, objAPI); err != nil {
 		return err
 	}

-	globalReplicationPool.Get().resyncer.Lock()
-	defer globalReplicationPool.Get().resyncer.Unlock()
-	brs, ok := globalReplicationPool.Get().resyncer.statusMap[opts.bucket]
-	if !ok {
-		brs = BucketReplicationResyncStatus{
-			Version:    resyncMetaVersion,
-			TargetsMap: make(map[string]TargetReplicationResyncStatus),
-		}
-	}
-	brs.TargetsMap[opts.arn] = status
-	globalReplicationPool.Get().resyncer.statusMap[opts.bucket] = brs
-	go globalReplicationPool.Get().resyncer.resyncBucket(GlobalContext, objAPI, false, opts)
+	s.statusMap[opts.bucket] = data
+	go s.resyncBucket(GlobalContext, objAPI, false, opts)
 	return nil
 }

@@ -3476,6 +3531,9 @@ func (p *ReplicationPool) loadResync(ctx context.Context, buckets []string, objA
 	// Make sure only one node running resync on the cluster.
 	ctx, cancel := globalLeaderLock.GetLock(ctx)
 	defer cancel()
+	var workers sync.WaitGroup
+	// Keep the merged leader context alive until every resumed run exits.
+	defer workers.Wait()

 	for index := range buckets {
 		bucket := buckets[index]
@@ -3489,19 +3547,24 @@ func (p *ReplicationPool) loadResync(ctx context.Context, buckets []string, objA
 		}

 		p.resyncer.Lock()
-		p.resyncer.statusMap[bucket] = meta
-		p.resyncer.Unlock()
-
+		if current, ok := p.resyncer.statusMap[bucket]; ok {
+			// A concurrent start/cancel is newer than the disk snapshot.
+			meta = current
+		} else {
+			p.resyncer.statusMap[bucket] = meta
+		}
 		tgts := meta.cloneTgtStats()
+		p.resyncer.Unlock()
 		for arn, st := range tgts {
 			switch st.ResyncStatus {
 			case ResyncFailed, ResyncStarted, ResyncPending:
-				go p.resyncer.resyncBucket(ctx, objAPI, true, resyncOpts{
+				opts := resyncOpts{
 					bucket:       bucket,
 					arn:          arn,
 					resyncID:     st.ResyncID,
 					resyncBefore: st.ResyncBeforeDate,
-				})
+				}
+				workers.Go(func() { p.resyncer.resyncBucket(ctx, objAPI, true, opts) })
 			}
 		}
 	}
@@ -3820,6 +3883,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 +3898,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)
 		}
 	}
 }
@@ -26,7 +26,6 @@ import (
 	"path"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -328,20 +327,19 @@ func TestReplicationValidationObjectUsesRulePrefix(t *testing.T) {

 func newTestResyncer(bucket, arn string) (*replicationResyncer, resyncOpts) {
 	s := &replicationResyncer{
-		statusMap:      map[string]BucketReplicationResyncStatus{},
-		resyncCancelCh: make(chan struct{}, resyncWorkerCnt),
+		statusMap: map[string]BucketReplicationResyncStatus{},
 	}
 	brs := newBucketResyncStatus(bucket)
-	brs.TargetsMap[arn] = TargetReplicationResyncStatus{ResyncStatus: ResyncStarted}
+	brs.TargetsMap[arn] = TargetReplicationResyncStatus{ResyncStatus: ResyncStarted, ResyncID: "reset-" + bucket}
 	s.statusMap[bucket] = brs
 	return s, resyncOpts{bucket: bucket, arn: arn, resyncID: "reset-" + bucket}
 }

 // TestResyncBucketFinalize round-trips the terminal status through a real
 // ObjectLayer: a clean run persists Completed with every result, while a run
-// whose parent context was canceled during the drain, or in which a worker
-// dropped a result on the cancel signal, is downgraded to Failed so a persisted
-// Completed never misrepresents an incomplete resync.
+// whose parent context was canceled during the drain is downgraded to Failed;
+// a user-canceled run persists Canceled. Completed never misrepresents an
+// incomplete resync.
 func TestResyncBucketFinalize(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -354,9 +352,9 @@ func TestResyncBucketFinalize(t *testing.T) {
 	// persistTerminal applies resyncBucket's finalizer logic (finalResyncStatus
 	// then markStatus, which persists) and reads the status back the way the
 	// resync status API does.
-	persistTerminal := func(t *testing.T, s *replicationResyncer, opts resyncOpts, status ResyncStatusType, ctxErr error, aborted bool) TargetReplicationResyncStatus {
+	persistTerminal := func(t *testing.T, s *replicationResyncer, opts resyncOpts, status ResyncStatusType, cause error) TargetReplicationResyncStatus {
 		t.Helper()
-		s.markStatus(finalResyncStatus(status, ctxErr, aborted), opts, objAPI)
+		s.markStatus(finalResyncStatus(status, cause), opts, objAPI)
 		brs, err := loadBucketResyncMetadata(ctx, opts.bucket, objAPI)
 		if err != nil {
 			t.Fatalf("load persisted resync metadata: %v", err)
@@ -376,7 +374,7 @@ func TestResyncBucketFinalize(t *testing.T) {
 		var wg sync.WaitGroup // no producer workers for this case
 		results.finish(nil, &wg)

-		st := persistTerminal(t, s, opts, ResyncCompleted, nil, false)
+		st := persistTerminal(t, s, opts, ResyncCompleted, nil)
 		if st.ResyncStatus != ResyncCompleted {
 			t.Fatalf("persisted status = %s, want Completed", st.ResyncStatus)
 		}
@@ -398,29 +396,24 @@ func TestResyncBucketFinalize(t *testing.T) {

 		cctx, ccancel := context.WithCancel(context.Background())
 		ccancel()
-		st := persistTerminal(t, s, opts, ResyncCompleted, cctx.Err(), false)
+		st := persistTerminal(t, s, opts, ResyncCompleted, context.Cause(cctx))
 		if st.ResyncStatus != ResyncFailed {
 			t.Fatalf("persisted status = %s, want Failed (parent canceled during drain)", st.ResyncStatus)
 		}
 	})

-	// 3. A worker dropped a computed result on the resync-cancel token (parent
-	//    still alive) -> sendResyncResult records the abort and Completed is
-	//    downgraded to Failed.
-	t.Run("worker abort downgrades to failed", func(t *testing.T) {
+	// 3. A user-canceled worker cannot report a completed resync.
+	t.Run("user cancel persists canceled", func(t *testing.T) {
 		s, opts := newTestResyncer("finalize-worker-abort", "arn1")
-		s.resyncCancelCh <- struct{}{}                 // cancel token waiting
-		ch := make(chan TargetReplicationResyncStatus) // no reader: the send would block
-		var aborted atomic.Bool
-		if s.sendResyncResult(context.Background(), ch, TargetReplicationResyncStatus{Object: "dropped", ReplicatedCount: 1}, &aborted) {
-			t.Fatal("sendResyncResult reported success despite the cancel token")
+		ctx, cancel := context.WithCancelCause(context.Background())
+		cancel(errResyncCanceled)
+		ch := make(chan TargetReplicationResyncStatus)
+		if s.sendResyncResult(ctx, ch, TargetReplicationResyncStatus{Object: "dropped", ReplicatedCount: 1}) {
+			t.Fatal("sendResyncResult reported success after cancellation")
 		}
-		if !aborted.Load() {
-			t.Fatal("worker abort was not recorded")
-		}
-		st := persistTerminal(t, s, opts, ResyncCompleted, nil, aborted.Load())
-		if st.ResyncStatus != ResyncFailed {
-			t.Fatalf("persisted status = %s, want Failed (worker dropped a result)", st.ResyncStatus)
+		st := persistTerminal(t, s, opts, ResyncCompleted, context.Cause(ctx))
+		if st.ResyncStatus != ResyncCanceled {
+			t.Fatalf("persisted status = %s, want Canceled", st.ResyncStatus)
 		}
 	})
 }
@@ -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()
@@ -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() {
@@ -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
 }
@@ -342,6 +342,8 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection {
 			replicationMaxQueuedCountMD,
 			replicationMaxDataTransferRateMD,
 			replicationRecentBacklogCountMD,
+			replicationMRFDroppedOperationsMD,
+			replicationMRFDroppedBytesMD,
 		},
 		loadClusterReplicationMetrics,
 	)
@@ -3170,7 +3170,7 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 	if objInfo.ReplicationStatus == replication.Pending || objInfo.VersionPurgeStatus == replication.VersionPurgePending {
 		dmVersionID := ""
 		versionID := ""
-		if objInfo.DeleteMarker {
+		if objInfo.DeleteMarker && objInfo.VersionPurgeStatus.Empty() {
 			dmVersionID = objInfo.VersionID
 		} else {
 			versionID = objInfo.VersionID
@@ -0,0 +1,241 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio/internal/auth"
+	"github.com/minio/minio/internal/bucket/replication"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio/internal/once"
+)
+
+// Exercise the actual single-object DELETE handler and erasure metadata. A
+// marker purge must delete the target version and finish the source purge;
+// merely seeing a 405 on HEAD is not evidence of a completed permanent delete.
+func TestReplicateDeleteMarkerPurge(t *testing.T) {
+	defer DetectTestLeak(t)()
+	for _, legacy := range []bool{false, true} {
+		t.Run(fmt.Sprintf("recover_legacy_%v", legacy), func(t *testing.T) {
+			ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+				t: t, endpoints: []string{"DeleteObject"},
+				objAPITest: func(obj ObjectLayer, instanceType, bucket string, router http.Handler, creds auth.Credentials, t *testing.T) {
+					testReplicateDeleteMarkerPurge(obj, instanceType, bucket, router, creds, t, legacy)
+				},
+			})
+		})
+	}
+}
+
+func TestReplicateDeleteMarkerTargetSemantics(t *testing.T) {
+	for _, tt := range []struct {
+		name       string
+		purge      bool
+		deleteCode int
+		wantDelete bool
+		wantFailed bool
+	}{
+		{name: "existing marker is idempotent"},
+		{name: "purge removes an existing marker", purge: true, deleteCode: 204, wantDelete: true},
+		{name: "purge forbidden", purge: true, deleteCode: 403, wantDelete: true, wantFailed: true},
+		{name: "purge method rejected", purge: true, deleteCode: 405, wantDelete: true, wantFailed: true},
+		{name: "purge unavailable", purge: true, deleteCode: 503, wantDelete: true, wantFailed: true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			var deletes atomic.Int32
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.Method == http.MethodHead {
+					w.Header().Set(xhttp.AmzDeleteMarker, "true")
+					w.Header().Set(xhttp.AmzVersionID, r.URL.Query().Get("versionId"))
+					w.WriteHeader(http.StatusMethodNotAllowed)
+					return
+				}
+				deletes.Add(1)
+				w.WriteHeader(tt.deleteCode)
+				if tt.wantFailed {
+					code := map[int]string{403: "AccessDenied", 405: "MethodNotAllowed", 503: "ServiceUnavailable"}[tt.deleteCode]
+					fmt.Fprintf(w, `<Error><Code>%s</Code><Message>injected delete rejection</Message></Error>`, code)
+				}
+			}))
+			defer server.Close()
+			client, err := minio.New(strings.TrimPrefix(server.URL, "http://"), &minio.Options{Region: "us-east-1", MaxRetries: 1})
+			if err != nil {
+				t.Fatal(err)
+			}
+			old := globalBucketTargetSys
+			globalBucketTargetSys = &BucketTargetSys{hc: map[string]epHealth{client.EndpointURL().Host: {Online: true}}}
+			defer func() { globalBucketTargetSys = old }()
+			deletion := DeletedObjectReplicationInfo{Bucket: "source", DeletedObject: DeletedObject{
+				ObjectName: "marker", DeleteMarker: true, DeleteMarkerVersionID: mustGetUUID(),
+			}}
+			if tt.purge {
+				deletion.VersionID = deletion.DeleteMarkerVersionID
+				deletion.DeleteMarkerVersionID = ""
+				deletion.ReplicationState.PurgeTargets = map[string]VersionPurgeStatusType{"arn1": replication.VersionPurgePending}
+			}
+			result := replicateDeleteToTarget(t.Context(), deletion, &TargetClient{Client: client, ARN: "arn1", Bucket: "target"})
+			if (deletes.Load() != 0) != tt.wantDelete {
+				t.Fatalf("remote DELETE count = %d, want delete %v", deletes.Load(), tt.wantDelete)
+			}
+			if tt.purge {
+				want := replication.VersionPurgeComplete
+				if tt.wantFailed {
+					want = replication.VersionPurgeFailed
+				}
+				if result.VersionPurgeStatus != want || (result.Err != nil) != tt.wantFailed {
+					t.Errorf("purge result = %+v, want %s", result, want)
+				}
+			} else if result.ReplicationStatus != replication.Completed {
+				t.Errorf("existing marker result = %+v, want Completed", result)
+			}
+		})
+	}
+}
+
+func testReplicateDeleteMarkerPurge(obj ObjectLayer, instanceType, bucket string, router http.Handler, creds auth.Credentials, t *testing.T, legacy bool) {
+	ctx := t.Context()
+	const arn = "arn:minio:replication::af470089-d354-4473-934c-9e1f52f6da89:bucket"
+	const name = "marker"
+	version := mustGetUUID()
+	remoteBucket := getRandomBucketName()
+	if err := obj.MakeBucket(ctx, remoteBucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil {
+		t.Fatal(err)
+	}
+	for _, b := range []string{bucket, remoteBucket} {
+		if _, err := obj.PutObject(ctx, b, name, mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""), ObjectOptions{Versioned: true}); err != nil {
+			t.Fatal(err)
+		}
+		opts := ObjectOptions{VersionID: version, Versioned: true, DeleteMarker: true, ReplicationRequest: true, MTime: UTCNow()}
+		opts.SetReplicaStatus(replication.Replica)
+		if _, err := obj.DeleteObject(ctx, b, name, opts); err != nil {
+			t.Fatalf("%s: seed marker in %s: %v", instanceType, b, err)
+		}
+	}
+	remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		opts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true}
+		switch r.Method {
+		case http.MethodHead:
+			oi, err := obj.GetObjectInfo(r.Context(), remoteBucket, name, opts)
+			if oi.DeleteMarker {
+				w.Header().Set(xhttp.AmzDeleteMarker, "true")
+				w.Header().Set(xhttp.AmzVersionID, oi.VersionID)
+			}
+			if err != nil {
+				writeErrorResponseHeadersOnly(w, toAPIError(r.Context(), err))
+				return
+			}
+			w.WriteHeader(http.StatusOK)
+		case http.MethodDelete:
+			opts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true"
+			opts.SetReplicaStatus(replication.Replica)
+			_, err := obj.DeleteObject(r.Context(), remoteBucket, name, opts)
+			if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+				writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL)
+				return
+			}
+			w.WriteHeader(http.StatusNoContent)
+		default:
+			t.Errorf("unexpected remote method %s", r.Method)
+			w.WriteHeader(http.StatusBadRequest)
+		}
+	}))
+	defer remote.Close()
+	client, err := minio.New(strings.TrimPrefix(remote.URL, "http://"), &minio.Options{Region: "us-east-1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	target := &TargetClient{Client: client, ARN: arn, Bucket: remoteBucket}
+	globalBucketTargetSys.Lock()
+	globalBucketTargetSys.arnRemotesMap[arn] = arnTarget{Client: target, lastRefresh: UTCNow()}
+	globalBucketTargetSys.targetsMap[bucket] = []madmin.BucketTarget{{Arn: arn, TargetBucket: remoteBucket}}
+	globalBucketTargetSys.Unlock()
+	globalBucketTargetSys.hMutex.Lock()
+	globalBucketTargetSys.hc[client.EndpointURL().Host] = epHealth{Online: true}
+	globalBucketTargetSys.hMutex.Unlock()
+	meta, err := globalBucketMetadataSys.Get(bucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cfg := configs[0]
+	cfg.RoleArn = arn
+	meta.replicationConfig = &cfg
+	globalBucketMetadataSys.Set(bucket, meta)
+	worker := make(chan ReplicationWorkerOperation, 1)
+	p := &ReplicationPool{
+		ctx:       ctx,
+		objLayer:  obj,
+		workers:   []chan ReplicationWorkerOperation{worker},
+		stats:     globalReplicationStats.Load(),
+		mrfSaveCh: make(chan MRFReplicateEntry, 1),
+	}
+	oldPool := globalReplicationPool
+	globalReplicationPool = once.NewSingleton[ReplicationPool]()
+	globalReplicationPool.Set(p)
+	defer func() { globalReplicationPool = oldPool }()
+
+	req, err := newTestSignedRequestV4(http.MethodDelete, "/"+bucket+"/"+name+"?versionId="+version, 0, nil, creds.AccessKey, creds.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	w := httptest.NewRecorder()
+	router.ServeHTTP(w, req)
+	if w.Code != http.StatusNoContent {
+		t.Fatalf("DELETE status %d: %s", w.Code, w.Body.String())
+	}
+	var deletion DeletedObjectReplicationInfo
+	select {
+	case op := <-worker:
+		deletion = op.(DeletedObjectReplicationInfo)
+	case <-time.After(time.Second):
+		t.Fatal("DELETE did not schedule replication")
+	}
+	if deletion.VersionID != version || deletion.DeleteMarkerVersionID != "" {
+		t.Errorf("purge scheduled as marker creation: version=%q marker=%q", deletion.VersionID, deletion.DeleteMarkerVersionID)
+	}
+	if legacy {
+		// Reproduce the old producer's state and let the existing scanner/heal
+		// path recover it. Upgrades must also finish purges already left pending.
+		deletion.VersionID, deletion.DeleteMarkerVersionID = "", version
+		replicateDelete(ctx, deletion, obj)
+		oi, _ := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: version, Versioned: true})
+		if oi.VersionPurgeStatus != replication.VersionPurgePending {
+			t.Fatalf("legacy source purge = %s, want PENDING", oi.VersionPurgeStatus)
+		}
+		targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		queueReplicationHeal(ctx, bucket, oi, replicationConfig{Config: &cfg, remotes: targets}, 0)
+		select {
+		case op := <-worker:
+			deletion = op.(DeletedObjectReplicationInfo)
+		case <-time.After(time.Second):
+			t.Fatal("legacy pending purge was not scheduled for healing")
+		}
+	}
+	result := replicateDelete(context.Background(), deletion, obj)
+	if result.VersionPurgeStatus() != replication.VersionPurgeComplete {
+		t.Errorf("remote purge result = %s, want COMPLETE", result.VersionPurgeStatus())
+	}
+	for _, b := range []string{bucket, remoteBucket} {
+		oi, err := obj.GetObjectInfo(ctx, b, name, ObjectOptions{VersionID: version, Versioned: true})
+		if !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+			t.Errorf("%s: marker remains in %s: %+v, err=%v", instanceType, b, oi, err)
+		}
+	}
+}
@@ -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() != replicationCollectorPath.metricPrefix()+"_"+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)
+			}
+		}
+	}
+}
@@ -0,0 +1,442 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio/internal/once"
+)
+
+func TestSiteResyncCancelState(t *testing.T) {
+	globalSiteReplicationSys.Lock()
+	old := globalSiteReplicationSys.enabled
+	globalSiteReplicationSys.enabled = true
+	globalSiteReplicationSys.Unlock()
+	t.Cleanup(func() {
+		globalSiteReplicationSys.Lock()
+		globalSiteReplicationSys.enabled = old
+		globalSiteReplicationSys.Unlock()
+	})
+	rs := newSiteResyncStatus("peer", []BucketInfo{{Name: "one"}, {Name: "two"}})
+	sm := &siteResyncMetrics{
+		resyncStatus:  map[string]SiteResyncStatus{rs.ResyncID: rs.clone()},
+		peerResyncMap: map[string]resyncState{"peer": {resyncID: rs.ResyncID}},
+	}
+	rs.Status = ResyncCanceled
+	if err := sm.updateState(rs); err != nil {
+		t.Fatal(err)
+	}
+	got, err := sm.status("peer")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != ResyncCanceled {
+		t.Fatalf("cancel returned without updating site state: got %s", got.Status)
+	}
+	for _, bucket := range []string{"one", "two"} {
+		sm.incBucket(resyncOpts{bucket: bucket, resyncID: rs.ResyncID}, ResyncCanceled)
+	}
+	got, err = sm.status("peer")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for bucket, status := range got.BucketStatuses {
+		if status != ResyncCanceled {
+			t.Errorf("bucket %s = %s, want Canceled", bucket, status)
+		}
+	}
+	// An already-running worker must not resurrect a canceled site.
+	sm.incBucket(resyncOpts{bucket: "one", resyncID: rs.ResyncID}, ResyncCompleted)
+	got, _ = sm.status("peer")
+	if got.Status != ResyncCanceled || got.BucketStatuses["one"] != ResyncCanceled {
+		t.Fatalf("late completion overwrote canceled state: %+v", got)
+	}
+}
+
+// The fixture runs the real resyncBucket control flow with in-memory metadata
+// persistence. Walk is deliberately controllable so cancellation does not
+// depend on disk speed, network timing or the walker closing its output.
+type resyncCancelObjectLayer struct {
+	ObjectLayer
+	walk  func(context.Context, chan<- itemOrErr[ObjectInfo]) error
+	lock  RWLocker
+	mu    sync.Mutex
+	saved map[string]BucketReplicationResyncStatus
+}
+
+func (o *resyncCancelObjectLayer) NewNSLock(bucket string, objects ...string) RWLocker {
+	return o.lock
+}
+
+func (o *resyncCancelObjectLayer) Walk(ctx context.Context, bucket, prefix string, results chan<- itemOrErr[ObjectInfo], opts WalkOptions) error {
+	return o.walk(ctx, results)
+}
+
+func (o *resyncCancelObjectLayer) PutObject(_ context.Context, bucket, object string, r *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+	data, err := io.ReadAll(r)
+	if err == nil && o.saved != nil {
+		var status BucketReplicationResyncStatus
+		_, err = status.UnmarshalMsg(data[4:])
+		o.mu.Lock()
+		o.saved[object] = status
+		o.mu.Unlock()
+	}
+	return ObjectInfo{}, err
+}
+
+func (o *resyncCancelObjectLayer) GetObjectNInfo(_ context.Context, bucket, object string, _ *HTTPRangeSpec, _ http.Header, opts ObjectOptions) (*GetObjectReader, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	status, ok := o.saved[object]
+	if !ok {
+		return nil, ObjectNotFound{Bucket: bucket, Object: object}
+	}
+	data := make([]byte, 4)
+	binary.LittleEndian.PutUint16(data[:2], resyncMetaFormat)
+	binary.LittleEndian.PutUint16(data[2:], resyncMetaVersion)
+	data, err := status.MarshalMsg(data)
+	if err != nil {
+		return nil, err
+	}
+	return NewGetObjectReaderFromReader(bytes.NewReader(data), ObjectInfo{Size: int64(len(data))}, opts)
+}
+
+func TestResyncRecoveryOwnsLeaderContext(t *testing.T) {
+	for _, loseLeader := range []bool{false, true} {
+		t.Run(fmt.Sprintf("lose_leader_%v", loseLeader), func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				var walkCtx context.Context
+				var output chan<- itemOrErr[ObjectInfo]
+				obj := &resyncCancelObjectLayer{saved: make(map[string]BucketReplicationResyncStatus), walk: func(ctx context.Context, ch chan<- itemOrErr[ObjectInfo]) error {
+					walkCtx, output = ctx, ch
+					return nil
+				}}
+				s, opts := setupResyncCancelTest(t, obj)
+				if err := saveResyncStatus(t.Context(), opts.bucket, s.statusMap[opts.bucket], obj); err != nil {
+					t.Fatal(err)
+				}
+				leaderCtx, lose := context.WithCancel(t.Context())
+				defer lose()
+				leader := &sharedLock{lockContext: make(chan LockContext, 1)}
+				leader.lockContext <- LockContext{ctx: leaderCtx}
+				oldLeader := globalLeaderLock
+				globalLeaderLock = leader
+				defer func() { globalLeaderLock = oldLeader }()
+				done := make(chan error, 1)
+				go func() { done <- (&ReplicationPool{resyncer: s}).loadResync(t.Context(), []string{opts.bucket}, obj) }()
+				synctest.Wait()
+				if walkCtx == nil || walkCtx.Err() != nil {
+					t.Fatal("recovery canceled its worker at startup")
+				}
+				select {
+				case <-done:
+					t.Fatal("recovery released its leader context before the run finished")
+				default:
+				}
+				want := ResyncCompleted
+				if loseLeader {
+					want = ResyncFailed
+					lose()
+				} else {
+					close(output)
+				}
+				synctest.Wait()
+				if err := <-done; err != nil {
+					t.Fatal(err)
+				}
+				if walkCtx.Err() == nil {
+					t.Fatal("finished recovery leaked the Walk context")
+				}
+				if got := s.statusMap[opts.bucket].TargetsMap[opts.arn].ResyncStatus; got != want {
+					t.Fatalf("recovery status = %s, want %s", got, want)
+				}
+			})
+		})
+	}
+}
+
+func TestResyncCancelRouting(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		parent, stop := context.WithCancel(t.Context())
+		defer stop()
+		walks := make(chan context.Context, 4)
+		obj := &resyncCancelObjectLayer{walk: func(ctx context.Context, output chan<- itemOrErr[ObjectInfo]) error {
+			walks <- ctx
+			return nil
+		}}
+		s, opts := setupResyncCancelTest(t, obj)
+		add := func(bucket, id string) resyncOpts {
+			o := opts
+			o.bucket, o.resyncID = bucket, id
+			s.Lock()
+			m := newBucketResyncStatus(bucket)
+			m.TargetsMap[o.arn] = TargetReplicationResyncStatus{ResyncID: id, ResyncStatus: ResyncPending}
+			s.statusMap[bucket] = m
+			s.Unlock()
+			meta, _ := globalBucketMetadataSys.Get(opts.bucket)
+			globalBucketMetadataSys.Set(bucket, meta)
+			globalBucketTargetSys.Lock()
+			globalBucketTargetSys.targetsMap[bucket] = []madmin.BucketTarget{{Arn: o.arn}}
+			globalBucketTargetSys.Unlock()
+			return o
+		}
+		run := func(o resyncOpts) chan struct{} {
+			done := make(chan struct{})
+			go func() { s.resyncBucket(parent, obj, false, o); close(done) }()
+			return done
+		}
+		first := run(opts)
+		synctest.Wait()
+		firstCtx := <-walks
+		queuedOpts := add("queued", opts.resyncID)
+		queued := run(queuedOpts)
+		otherOpts := add("other", "other-id")
+		other := run(otherOpts)
+		synctest.Wait()
+		s.cancelResyncID(opts.resyncID)
+		synctest.Wait()
+		for name, done := range map[string]chan struct{}{"active": first, "queued": queued} {
+			select {
+			case <-done:
+			default:
+				t.Fatalf("%s matching run survived cancellation", name)
+			}
+		}
+		if !errors.Is(context.Cause(firstCtx), errResyncCanceled) {
+			t.Fatal("active Walk did not receive user cancellation")
+		}
+		select {
+		case <-other:
+			t.Fatal("unrelated run was canceled")
+		default:
+		}
+		otherCtx := <-walks
+		if otherCtx.Err() != nil {
+			t.Fatal("unrelated Walk was canceled")
+		}
+		// There is no token/tombstone left for a subsequently registered run.
+		freshOpts := add("fresh", opts.resyncID)
+		fresh := run(freshOpts)
+		synctest.Wait()
+		select {
+		case <-fresh:
+			t.Fatal("fresh run inherited an old cancellation")
+		default:
+		}
+		stop()
+		synctest.Wait()
+		<-other
+		<-fresh
+		s.RLock()
+		defer s.RUnlock()
+		if len(s.cancelResyncs) != 0 || len(s.workerCh) != 1 {
+			t.Fatal("run registrations or worker slot leaked")
+		}
+		if s.statusMap[queuedOpts.bucket].TargetsMap[opts.arn].ResyncStatus != ResyncCanceled {
+			t.Fatal("queued user cancellation was not recorded")
+		}
+		if s.statusMap[freshOpts.bucket].TargetsMap[opts.arn].ResyncStatus != ResyncPending {
+			t.Fatal("shutdown changed a queued run's resumable Pending status")
+		}
+	})
+}
+
+func TestResyncCancellationWinsFinalization(t *testing.T) {
+	s, opts := newTestResyncer("finalize-cancel", "arn1")
+	obj := &resyncCancelObjectLayer{saved: make(map[string]BucketReplicationResyncStatus)}
+	ctx, cancel, registered := s.registerResync(t.Context(), opts)
+	if !registered {
+		t.Fatal("registration failed")
+	}
+	defer cancel(nil)
+	// Simulate cancellation after the finalizer computed Completed but before
+	// it acquired the persistence lock.
+	status := finalResyncStatus(ResyncCompleted, context.Cause(ctx))
+	s.cancelResyncID(opts.resyncID)
+	if got := s.markStatus(status, opts, obj); got != ResyncCanceled {
+		t.Fatalf("final status = %s, want Canceled", got)
+	}
+	for _, saved := range obj.saved {
+		if saved.TargetsMap[opts.arn].ResyncStatus != ResyncCanceled {
+			t.Fatal("persisted Completed over cancellation")
+		}
+	}
+	if len(obj.saved) != 1 {
+		t.Fatal("canceled status was not persisted")
+	}
+	m := s.statusMap[opts.bucket]
+	m.TargetsMap[opts.arn] = TargetReplicationResyncStatus{ResyncID: "new-run", ResyncStatus: ResyncStarted}
+	s.statusMap[opts.bucket] = m
+	if s.markStatus(ResyncCompleted, opts, obj) != NoResync || s.incStats(TargetReplicationResyncStatus{ReplicatedCount: 1}, opts) {
+		t.Fatal("old run changed the replacement run")
+	}
+	if st := s.statusMap[opts.bucket].TargetsMap[opts.arn]; st.ResyncStatus != ResyncStarted || st.ReplicatedCount != 0 {
+		t.Fatalf("replacement state changed: %+v", st)
+	}
+	delete(s.statusMap, opts.bucket)
+	if s.markStatus(ResyncFailed, opts, obj) != NoResync {
+		t.Fatal("deleted bucket was recreated")
+	}
+}
+
+func setupResyncCancelTest(t *testing.T, obj *resyncCancelObjectLayer) (*replicationResyncer, resyncOpts) {
+	t.Helper()
+	s, opts := newTestResyncer("cancel-bucket", "arn1")
+	s.workerCh = make(chan struct{}, 1)
+	s.workerCh <- struct{}{}
+	cfg := configs[0]
+	cfg.RoleArn = opts.arn
+	meta := newBucketMetadata(opts.bucket)
+	meta.replicationConfig = &cfg
+	oldMeta, oldTargets, oldObj := globalBucketMetadataSys, globalBucketTargetSys, newObjectLayerFn()
+	oldPool := globalReplicationPool
+	oldNotifier := globalEventNotifier
+	globalEventNotifier = &EventNotifier{}
+	globalReplicationPool = once.NewSingleton[ReplicationPool]()
+	globalReplicationPool.Set(&ReplicationPool{})
+	globalBucketMetadataSys = NewBucketMetadataSys()
+	globalBucketMetadataSys.Set(opts.bucket, meta)
+	client, err := minio.New("127.0.0.1:1", &minio.Options{Region: "us-east-1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	globalBucketTargetSys = &BucketTargetSys{
+		arnRemotesMap: map[string]arnTarget{opts.arn: {Client: &TargetClient{Client: client, ARN: opts.arn}}},
+		targetsMap:    map[string][]madmin.BucketTarget{opts.bucket: {{Arn: opts.arn}}},
+	}
+	setObjectLayer(obj)
+	t.Cleanup(func() {
+		globalBucketMetadataSys, globalBucketTargetSys = oldMeta, oldTargets
+		globalReplicationPool = oldPool
+		globalEventNotifier = oldNotifier
+		setObjectLayer(oldObj)
+	})
+	return s, opts
+}
+
+type blockedResyncLock struct {
+	RWLocker
+	started chan struct{}
+}
+
+func (l *blockedResyncLock) GetLock(ctx context.Context, _ *dynamicTimeout) (LockContext, error) {
+	close(l.started)
+	<-ctx.Done()
+	return LockContext{}, ctx.Err()
+}
+
+func TestResyncCancelFullWorkerQueue(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		lock := &blockedResyncLock{started: make(chan struct{})}
+		produced := 0
+		walkerDone := make(chan struct{})
+		obj := &resyncCancelObjectLayer{lock: lock}
+		s, opts := setupResyncCancelTest(t, obj)
+		obj.walk = func(ctx context.Context, output chan<- itemOrErr[ObjectInfo]) error {
+			go func() {
+				defer close(walkerDone)
+				defer close(output)
+				for range 120 {
+					select {
+					case <-ctx.Done():
+						return
+					case output <- itemOrErr[ObjectInfo]{Item: ObjectInfo{Bucket: opts.bucket, Name: "same-key", VersionID: mustGetUUID(), DeleteMarker: true, ModTime: time.Now()}}:
+						produced++
+					}
+				}
+			}()
+			return nil
+		}
+		done := make(chan struct{})
+		go func() { s.resyncBucket(ctx, obj, false, opts); close(done) }()
+		synctest.Wait()
+		select {
+		case <-lock.started:
+		default:
+			t.Fatal("worker did not enter replication")
+		}
+		if produced < 101 || produced >= 120 {
+			t.Fatalf("expected a full 100-entry worker queue to block dispatch; Walk produced %d", produced)
+		}
+		cancel()
+		synctest.Wait()
+		select {
+		case <-done:
+		default:
+			t.Fatal("canceled dispatcher deadlocked sending to a full worker queue")
+		}
+		select {
+		case <-walkerDone:
+		default:
+			t.Fatal("canceled Walk producer leaked")
+		}
+		if len(s.workerCh) != 1 {
+			t.Fatal("resync worker slot was not returned")
+		}
+	})
+}
+
+func TestResyncCancelBlockedWalkReceive(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		parent, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		var output chan<- itemOrErr[ObjectInfo]
+		obj := &resyncCancelObjectLayer{walk: func(ctx context.Context, ch chan<- itemOrErr[ObjectInfo]) error {
+			output = ch
+			return nil
+		}}
+		s, opts := setupResyncCancelTest(t, obj)
+		done := make(chan struct{})
+		go func() { s.resyncBucket(parent, obj, false, opts); close(done) }()
+		synctest.Wait()
+		if output == nil {
+			t.Fatal("resync did not reach Walk")
+		}
+		cancel()
+		synctest.Wait()
+		select {
+		case <-done:
+		default:
+			t.Error("resync remained blocked on Walk output after cancellation")
+		}
+		close(output) // release the old implementation on failure
+		synctest.Wait()
+		if len(s.workerCh) != 1 {
+			t.Error("resync worker slot was not released")
+		}
+	})
+}
+
+func TestResyncCancelsOwnedWalkOnError(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		var walkCtx context.Context
+		obj := &resyncCancelObjectLayer{walk: func(ctx context.Context, ch chan<- itemOrErr[ObjectInfo]) error {
+			walkCtx = ctx
+			return errors.New("injected walk failure")
+		}}
+		s, opts := setupResyncCancelTest(t, obj)
+		s.resyncBucket(t.Context(), obj, false, opts)
+		if walkCtx == nil || walkCtx.Err() == nil {
+			t.Fatal("resync exit did not cancel the context it passed to Walk")
+		}
+		if t.Context().Err() != nil {
+			t.Fatal("resync canceled its caller's context")
+		}
+	})
+}
@@ -204,16 +204,24 @@ func (sm *siteResyncMetrics) updateState(s SiteResyncStatus) error {
 	switch s.Status {
 	case ResyncStarted:
 		sm.peerResyncMap[s.DeplID] = resyncState{resyncID: s.ResyncID, LastSaved: time.Time{}}
-		sm.resyncStatus[s.ResyncID] = s
+		sm.resyncStatus[s.ResyncID] = s.clone()
 	case ResyncCompleted, ResyncCanceled, ResyncFailed:
 		st, ok := sm.resyncStatus[s.ResyncID]
 		if ok {
 			st.LastUpdate = s.LastUpdate
 			st.Status = s.Status
+			if s.Status == ResyncCanceled {
+				for bucket, status := range st.BucketStatuses {
+					if status == ResyncPending || status == ResyncStarted {
+						st.BucketStatuses[bucket] = ResyncCanceled
+					}
+				}
+			}
+			sm.resyncStatus[s.ResyncID] = st
 			return nil
 		}
-		sm.resyncStatus[s.ResyncID] = st
-		return saveSiteResyncMetadata(GlobalContext, st, newObjectLayerFn())
+		sm.resyncStatus[s.ResyncID] = s.clone()
+		return saveSiteResyncMetadata(GlobalContext, s, newObjectLayerFn())
 	}
 	return nil
 }
@@ -230,7 +238,15 @@ func (sm *siteResyncMetrics) incBucket(o resyncOpts, bktStatus ResyncStatusType)
 		if st.BucketStatuses == nil {
 			st.BucketStatuses = map[string]ResyncStatusType{}
 		}
+		if st.BucketStatuses[o.bucket] == ResyncCanceled {
+			return
+		}
 		switch bktStatus {
+		case ResyncCanceled:
+			st.BucketStatuses[o.bucket] = ResyncCanceled
+			st.Status = ResyncCanceled
+			st.LastUpdate = UTCNow()
+			sm.resyncStatus[o.resyncID] = st
 		case ResyncCompleted:
 			st.BucketStatuses[o.bucket] = ResyncCompleted
 			st.Status = siteResyncStatus(st.Status, st.BucketStatuses)
@@ -202,6 +202,7 @@ func wrapSRErr(err error) SRError {
 // SiteReplicationSys - manages cluster-level replication.
 type SiteReplicationSys struct {
 	sync.RWMutex
+	resyncMu sync.Mutex // serialize site resync configuration start/cancel

 	enabled bool

@@ -6202,6 +6203,8 @@ func (c *SiteReplicationSys) getPeerForUpload(deplID string) (pi srPeerInfo, loc
 // is maintained in .minio.sys/buckets/site-replication/resync/<deployment-id.meta>, while collecting
 // individual bucket resync status in .minio.sys/buckets/<bucket-name>/replication/resync.bin
 func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) {
+	c.resyncMu.Lock()
+	defer c.resyncMu.Unlock()
 	if !c.isEnabled() {
 		return res, errSRNotEnabled
 	}
@@ -6321,6 +6324,8 @@ func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer

 // cancelResync stops an ongoing site level resync for the peer specified.
 func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) {
+	c.resyncMu.Lock()
+	defer c.resyncMu.Unlock()
 	if !c.isEnabled() {
 		return res, errSRNotEnabled
 	}
@@ -6386,34 +6391,22 @@ func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLaye
 				})
 				continue
 			}
-			// update resync state for the bucket
-			globalReplicationPool.Get().resyncer.Lock()
-			m, ok := globalReplicationPool.Get().resyncer.statusMap[bucket]
-			if !ok {
-				m = newBucketResyncStatus(bucket)
-			}
-			if st, ok := m.TargetsMap[t.Arn]; ok {
-				st.LastUpdate = UTCNow()
-				st.ResyncStatus = ResyncCanceled
-				m.TargetsMap[t.Arn] = st
-				m.LastUpdate = UTCNow()
-			}
-			globalReplicationPool.Get().resyncer.statusMap[bucket] = m
-			globalReplicationPool.Get().resyncer.Unlock()
 		}
 	}

+	// Configuration errors must not leave active or queued buckets running.
+	globalReplicationPool.Get().resyncer.cancelResyncID(rs.ResyncID)
 	rs.Status = ResyncCanceled
 	rs.LastUpdate = UTCNow()
+	for bucket, status := range rs.BucketStatuses {
+		if status == ResyncPending || status == ResyncStarted {
+			rs.BucketStatuses[bucket] = ResyncCanceled
+		}
+	}
+	globalSiteResyncMetrics.updateState(rs)
 	if err := saveSiteResyncMetadata(ctx, rs, objAPI); err != nil {
 		return res, err
 	}
-	select {
-	case globalReplicationPool.Get().resyncer.resyncCancelCh <- struct{}{}:
-	case <-ctx.Done():
-	}
-
-	globalSiteResyncMetrics.updateState(rs)

 	res.Status = rs.Status.String()
 	return res, nil
@@ -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

@@ -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                                                |