Merge pull request #138 from pgsty/fix/issue-136-resync-counter-flush

fix: persist an honest resync terminal status about object counts

cmd/bucket-replication.go

@@ -2886,6 +2886,88 @@ func (s *replicationResyncer) incStats(ts TargetReplicationResyncStatus, opts re
 	s.statusMap[opts.bucket] = m
 }
 
+// resyncResults consumes the per-object outcomes produced by the resync worker
+// pool and applies each to the in-memory resync status via apply. It centralizes
+// the finalization ordering so a status persisted after finish() returns always
+// reflects every result.
+type resyncResults struct {
+	ch    chan TargetReplicationResyncStatus
+	apply func(TargetReplicationResyncStatus)
+	wg    sync.WaitGroup
+}
+
+// newResyncResults starts the result-consuming goroutine that folds each worker
+// 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)
+	})
+}
+
+// startResyncResults starts a goroutine that applies every received result with
+// apply. Injecting the apply action keeps the shutdown ordering in finish()
+// testable.
+func startResyncResults(apply func(TargetReplicationResyncStatus)) *resyncResults {
+	rr := &resyncResults{
+		ch:    make(chan TargetReplicationResyncStatus, 1),
+		apply: apply,
+	}
+	rr.wg.Add(1)
+	go func() {
+		defer rr.wg.Done()
+		for r := range rr.ch {
+			rr.apply(r)
+		}
+	}()
+	return rr
+}
+
+// finish shuts the resync pipeline down in an order that guarantees a status
+// persisted afterwards reflects every result. It first closes the worker input
+// channels and waits for the producer workers to exit, so none can send on a
+// closed result channel (a hazard on early-return paths) and every submitted
+// result is delivered (a result a worker discards on cancellation is
+// intentionally not); only then does it close the result channel and wait for
+// the consumer to apply the last buffered result.
+func (rr *resyncResults) finish(workers []chan ReplicateObjectInfo, workerWg *sync.WaitGroup) {
+	for i := range workers {
+		xioutil.SafeClose(workers[i])
+	}
+	workerWg.Wait()
+	xioutil.SafeClose(rr.ch)
+	rr.wg.Wait()
+}
+
+// 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 {
+	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) {
+		return ResyncFailed
+	}
+	return status
+}
+
 // 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) {
@@ -2896,7 +2978,18 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 	}
 
 	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)
 		globalSiteResyncMetrics.incBucket(opts, resyncStatus)
 		s.workerCh <- struct{}{}
@@ -2952,16 +3045,14 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 		lastCheckpoint = st.Object
 	}
 	workers := make([]chan ReplicateObjectInfo, resyncParallelRoutines)
-	resultCh := make(chan TargetReplicationResyncStatus, 1)
-	defer xioutil.SafeClose(resultCh)
-	go func() {
-		for r := range resultCh {
-			s.incStats(r, opts)
-			globalSiteResyncMetrics.updateMetric(r, opts.resyncID)
-		}
-	}()
-
 	var wg sync.WaitGroup
+	// results consumes each worker's per-object outcome and folds it into the
+	// in-memory status. finish() (deferred below) stops the workers and drains
+	// every result before the deferred markStatus persists, so a Completed status
+	// cannot race the last incStats. Registered after the markStatus finalizer, so
+	// LIFO runs finish first.
+	results := s.newResyncResults(opts)
+	defer results.finish(workers, &wg)
 	for i := range resyncParallelRoutines {
 		wg.Add(1)
 		workers[i] = make(chan ReplicateObjectInfo, 100)
@@ -3029,12 +3120,8 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 					st.ReplicatedSize += roi.Size
 				}
 				traceFn(sz, err)
-				select {
-				case <-ctx.Done():
+				if !s.sendResyncResult(ctx, results.ch, st, &workerAborted) {
 					return
-				case <-s.resyncCancelCh:
-					return
-				case resultCh <- st:
 				}
 			}
 		}(ctx, i)
@@ -3071,10 +3158,6 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object
 			workers[h%uint64(resyncParallelRoutines)] <- roi
 		}
 	}
-	for i := range resyncParallelRoutines {
-		xioutil.SafeClose(workers[i])
-	}
-	wg.Wait()
 	resyncStatus = ResyncCompleted
 }
 

cmd/bucket-replication_test.go

@@ -18,10 +18,14 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"path"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/minio/madmin-go/v3"
@@ -307,3 +311,220 @@ func TestReplicationValidationObjectUsesRulePrefix(t *testing.T) {
 		})
 	}
 }
+
+// The resync-finalization tests below exercise the real result sink, the
+// finish() shutdown ordering, and the sendResyncResult / finalResyncStatus
+// helpers, plus (for the persistence cases) markStatus with on-disk
+// round-tripping. resyncBucket cannot be driven end to end in a unit test
+// because its workers call a live remote target (StatObject), so the helpers it
+// uses are exercised directly. The blocking-order assertions run under
+// testing/synctest so a removed wait fails deterministically, with no timing
+// windows.
+
+func newTestResyncer(bucket, arn string) (*replicationResyncer, resyncOpts) {
+	s := &replicationResyncer{
+		statusMap:      map[string]BucketReplicationResyncStatus{},
+		resyncCancelCh: make(chan struct{}, resyncWorkerCnt),
+	}
+	brs := newBucketResyncStatus(bucket)
+	brs.TargetsMap[arn] = TargetReplicationResyncStatus{ResyncStatus: ResyncStarted}
+	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.
+func TestResyncBucketFinalize(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	objAPI, fsDirs, err := prepareErasure16(ctx)
+	if err != nil {
+		t.Fatalf("prepare erasure backend: %v", err)
+	}
+	defer removeRoots(fsDirs)
+
+	// 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 {
+		t.Helper()
+		s.markStatus(finalResyncStatus(status, ctxErr, aborted), opts, objAPI)
+		brs, err := loadBucketResyncMetadata(ctx, opts.bucket, objAPI)
+		if err != nil {
+			t.Fatalf("load persisted resync metadata: %v", err)
+		}
+		return brs.TargetsMap[opts.arn]
+	}
+
+	// 1. Clean completion: every result - including the failed object - is folded
+	//    into the persisted status, which stays Completed.
+	t.Run("persists complete counts", func(t *testing.T) {
+		s, opts := newTestResyncer("finalize-counts", "arn1")
+		results := s.newResyncResults(opts)
+		results.ch <- TargetReplicationResyncStatus{Object: "ok-1", ReplicatedCount: 1, ReplicatedSize: 100}
+		results.ch <- TargetReplicationResyncStatus{Object: "ok-2", ReplicatedCount: 1, ReplicatedSize: 200}
+		results.ch <- TargetReplicationResyncStatus{Object: "bad", FailedCount: 1, FailedSize: 300}
+
+		var wg sync.WaitGroup // no producer workers for this case
+		results.finish(nil, &wg)
+
+		st := persistTerminal(t, s, opts, ResyncCompleted, nil, false)
+		if st.ResyncStatus != ResyncCompleted {
+			t.Fatalf("persisted status = %s, want Completed", st.ResyncStatus)
+		}
+		if st.ReplicatedCount != 2 || st.ReplicatedSize != 300 || st.FailedCount != 1 || st.FailedSize != 300 {
+			t.Fatalf("persisted counts = {replicated:%d/%d failed:%d/%d}, want {2/300 1/300}",
+				st.ReplicatedCount, st.ReplicatedSize, st.FailedCount, st.FailedSize)
+		}
+	})
+
+	// 2. Parent context canceled during the drain -> Completed downgraded to
+	//    Failed (markStatus persists under its own context, so nothing else stops
+	//    a bare Completed from being recorded).
+	t.Run("parent cancel during drain downgrades to failed", func(t *testing.T) {
+		s, opts := newTestResyncer("finalize-parent-cancel", "arn1")
+		results := s.newResyncResults(opts)
+		results.ch <- TargetReplicationResyncStatus{Object: "ok-1", ReplicatedCount: 1, ReplicatedSize: 100}
+		var wg sync.WaitGroup
+		results.finish(nil, &wg)
+
+		cctx, ccancel := context.WithCancel(context.Background())
+		ccancel()
+		st := persistTerminal(t, s, opts, ResyncCompleted, cctx.Err(), false)
+		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) {
+		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")
+		}
+		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)
+		}
+	})
+}
+
+// TestResyncFinishDrainsResults asserts finish() does not return until the
+// consumer has applied the final result (the #136 defect). A gated apply holds
+// the last result unapplied; under synctest finish() must stay durably blocked
+// until it is released - if rr.wg.Wait() is removed, finish() returns early and
+// the test fails deterministically.
+func TestResyncFinishDrainsResults(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, opts := newTestResyncer("drain", "arn1")
+		reachedFinal := make(chan struct{})
+		release := make(chan struct{})
+		results := startResyncResults(func(r TargetReplicationResyncStatus) {
+			if r.Object == "final" {
+				close(reachedFinal)
+				<-release
+			}
+			s.incStats(r, opts)
+		})
+		results.ch <- TargetReplicationResyncStatus{Object: "ok-1", ReplicatedCount: 1, ReplicatedSize: 100}
+		results.ch <- TargetReplicationResyncStatus{Object: "final", FailedCount: 1, FailedSize: 200}
+		<-reachedFinal // consumer received "final" but is gated before incStats(final)
+
+		var wg sync.WaitGroup
+		finishDone := make(chan struct{})
+		go func() {
+			results.finish(nil, &wg)
+			close(finishDone)
+		}()
+
+		synctest.Wait()
+		select {
+		case <-finishDone:
+			close(release)
+			synctest.Wait()
+			t.Fatal("finish() returned before the final result was drained (drain wait missing)")
+		default:
+			// finish() is durably blocked in rr.wg.Wait() - correct.
+		}
+
+		close(release)
+		synctest.Wait()
+		<-finishDone
+		st := s.statusMap[opts.bucket].TargetsMap[opts.arn]
+		if st.ReplicatedCount != 1 || st.FailedCount != 1 || st.FailedSize != 200 {
+			t.Fatalf("status after finish = {replicated:%d failed:%d/%d}, want {1 1/200}",
+				st.ReplicatedCount, st.FailedCount, st.FailedSize)
+		}
+	})
+}
+
+// TestResyncFinishWaitsForInflightWorker asserts finish() stops the producer
+// workers before it closes the result channel, so an in-flight worker (as on an
+// early-return path) never sends on a closed channel and its result is not lost.
+// A gated worker stays in flight past the shutdown request; under synctest
+// finish() must stay durably blocked until the worker is released - if
+// workerWg.Wait() is removed, finish() returns early and the test fails
+// deterministically.
+func TestResyncFinishWaitsForInflightWorker(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, opts := newTestResyncer("workers", "arn1")
+		results := startResyncResults(func(r TargetReplicationResyncStatus) { s.incStats(r, opts) })
+
+		workers := []chan ReplicateObjectInfo{make(chan ReplicateObjectInfo, 1)}
+		var wg sync.WaitGroup
+		gotRoi := make(chan struct{})
+		release := make(chan struct{})
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for roi := range workers[0] {
+				close(gotRoi)
+				<-release
+				// Mirror the real worker's send; recover so that if finish()
+				// wrongly closed the result channel first, the test fails via the
+				// assertion below instead of crashing on send-on-closed.
+				func() {
+					defer func() { _ = recover() }()
+					results.ch <- TargetReplicationResyncStatus{Object: roi.Name, ReplicatedCount: 1, ReplicatedSize: 500}
+				}()
+			}
+		}()
+		workers[0] <- ReplicateObjectInfo{Name: "inflight"}
+		<-gotRoi // worker holds a result in flight, not yet delivered
+
+		finishDone := make(chan struct{})
+		go func() {
+			results.finish(workers, &wg)
+			close(finishDone)
+		}()
+
+		synctest.Wait()
+		select {
+		case <-finishDone:
+			close(release)
+			synctest.Wait()
+			t.Fatal("finish() closed the result channel before the in-flight worker finished (worker wait missing)")
+		default:
+			// finish() is durably blocked in workerWg.Wait() - correct.
+		}
+
+		close(release)
+		synctest.Wait()
+		<-finishDone
+		st := s.statusMap[opts.bucket].TargetsMap[opts.arn]
+		if st.ReplicatedCount != 1 || st.ReplicatedSize != 500 {
+			t.Fatalf("status after finish = {replicated:%d/%d}, want {1/500}", st.ReplicatedCount, st.ReplicatedSize)
+		}
+	})
+}