From 46e82eb54df3e8e420ebe421335f613e5b0de681 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sun, 6 Sep 2026 18:35:13 +0800 Subject: [PATCH] fix(replication): count resync success by outcome, not target existence The resync worker classified each object by whether the target version merely existed (a tgt.StatObject HEAD), ignoring the outcome of the replicateObject/replicateDelete call it had just made. A quota-rejected update leaves the old version in place, so StatObject succeeded and the resync recorded a false success - reported as Completed / N success / 0 failed and persisted across restart (issue #139). #134's SSE-C HEAD marker made StatObject succeed for SSE-C too, exposing it there. The delete path had the mirror flaw (a failed delete leaves the object, so the HEAD succeeded), and FailedSize was never incremented (a failed 196,608-byte object counted as 1 failed / 0 bytes). replicateObject and replicateDelete already build the per-target replicatedInfos (each replicatedTargetInfo carries Arn, ReplicationStatus and Err) but discarded it. Return it (callers that only trigger replication ignore the value - a Go call statement discards it, so the queue paths are unchanged) and classify the resync from the target whose Arn == opts.arn via a small pure helper: - Completed without error -> replicated (+ that target's size, falling back to the object size). - Failed or errored -> failed (+ the object size, fixing FailedSize). - opts.arn absent from the result (not attempted) -> failed; a resync that cannot confirm the object reached the target is not a success. The StatObject-existence block (including the delete-marker/MethodNotAllowed special case, now subsumed by the delete outcome) is removed. #134's SSE-C HEAD marker is left intact - it is needed for genuine SSE-C success. Adds a table-driven regression for the classifier covering a completed update, a failed update over an existing version, an errored-but-Completed result, a delete failure, a delete-marker success (zero bytes) and an un-attempted ARN. Classifying by existence makes the failed cases count success and fails the test. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01L7qJqWwy8oFA6aCXWRzXQe Signed-off-by: Feng Ruohang --- cmd/bucket-replication.go | 113 ++++++++++++++++++++++--------- cmd/bucket-replication_test.go | 119 +++++++++++++++++++++++++++++++++ 2 files changed, 201 insertions(+), 31 deletions(-) diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go index 447124efe..7f52b06ac 100644 --- a/cmd/bucket-replication.go +++ b/cmd/bucket-replication.go @@ -418,7 +418,12 @@ func checkReplicateDelete(ctx context.Context, bucket string, dobj ObjectToDelet // target cluster, the object version is marked deleted on the source and hidden from listing. It is permanently // deleted from the source when the VersionPurgeStatus changes to "Complete", i.e after replication succeeds // on target. -func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, objectAPI ObjectLayer) { +// replicateDelete replicates a delete (delete marker or version purge) to all +// applicable targets and returns the per-target replication outcome. Callers +// that only trigger replication may ignore the return value; the resync path +// uses it to classify success/failure per target rather than inferring it from +// the mere presence or absence of the target version. +func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, objectAPI ObjectLayer) replicatedInfos { var replicationStatus replication.StatusType bucket := dobj.Bucket versionID := dobj.DeleteMarkerVersionID @@ -453,7 +458,7 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj Host: globalLocalNodeName, EventName: event.ObjectReplicationNotTracked, }) - return + return replicatedInfos{} } dsc, err := parseReplicateDecision(ctx, bucket, dobj.ReplicationState.ReplicateDecisionStr) if err != nil { @@ -471,7 +476,7 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj Host: globalLocalNodeName, EventName: event.ObjectReplicationNotTracked, }) - return + return replicatedInfos{} } // Lock the object name before starting replication operation. @@ -492,7 +497,7 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj Host: globalLocalNodeName, EventName: event.ObjectReplicationNotTracked, }) - return + return replicatedInfos{} } ctx = lkctx.Context() defer lk.Unlock(lkctx) @@ -597,6 +602,7 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj EventName: eventName, }) } + return rinfos } func replicateDeleteToTarget(ctx context.Context, dobj DeletedObjectReplicationInfo, tgt *TargetClient) (rinfo replicatedTargetInfo) { @@ -1035,7 +1041,12 @@ func getReplicationAction(oi1 ObjectInfo, oi2 minio.ObjectInfo, opType replicati // replicateObject replicates the specified version of the object to destination bucket // The source object is then updated to reflect the replication status. -func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI ObjectLayer) { +// replicateObject replicates a single object version to all applicable targets +// and returns the per-target replication outcome. Callers that only trigger +// replication may ignore the return value; the resync path uses it to classify +// success/failure per target rather than inferring it from the mere existence +// of the target version. +func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI ObjectLayer) replicatedInfos { var replicationStatus replication.StatusType defer func() { if replicationStatus.Empty() { @@ -1068,7 +1079,7 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje UserAgent: "Internal: [Replication]", Host: globalLocalNodeName, }) - return + return replicatedInfos{} } tgtArns := cfg.FilterTargetArns(replication.ObjectOpts{ Name: object, @@ -1088,7 +1099,7 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje Host: globalLocalNodeName, }) globalReplicationPool.Get().queueMRFSave(ri.ToMRFEntry()) - return + return replicatedInfos{} } ctx = lkctx.Context() defer lk.Unlock(lkctx) @@ -1194,6 +1205,7 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje ri.RetryCount++ globalReplicationPool.Get().queueMRFSave(ri.ToMRFEntry()) } + return rinfos } // replicateObject replicates object data for specified version of the object to destination bucket @@ -2968,6 +2980,55 @@ func finalResyncStatus(status ResyncStatusType, ctxErr error, workerAborted bool return status } +// resyncTargetSucceeded reports whether this object (or delete) actually +// replicated to the target, from the target's own outcome. A version purge +// reports success through VersionPurgeStatus, not ReplicationStatus. For an +// object or delete marker, success requires a Completed status; a retained +// error is a real failure unless it is the benign duplicate 412 the +// destination returns when it already holds this exact ETag and version, which +// replicateAll deliberately keeps Completed. +func resyncTargetSucceeded(t replicatedTargetInfo, roi ReplicateObjectInfo) bool { + if !roi.VersionPurgeStatus.Empty() { + return t.VersionPurgeStatus == replication.VersionPurgeComplete + } + if t.ReplicationStatus != replication.Completed { + return false + } + return t.Err == nil || minio.ToErrorResponse(t.Err).Code == "PreconditionFailed" +} + +// resyncResultFor derives the resync outcome for target arn from the aggregate +// replication result of a single object (or delete). The target counts as a +// success only when its own replication Completed without error - not when the +// target version merely exists. A target that Failed, errored, or was not +// attempted for this object (its arn absent from the result) counts as a +// failure, and the failed byte count is recorded (previously always zero). +func resyncResultFor(rinfos replicatedInfos, arn string, roi ReplicateObjectInfo) TargetReplicationResyncStatus { + st := TargetReplicationResyncStatus{Object: roi.Name, Bucket: roi.Bucket} + for _, t := range rinfos.Targets { + if t.Arn != arn { + continue + } + if resyncTargetSucceeded(t, roi) { + sz := t.Size + if sz == 0 { + sz = roi.Size + } + st.ReplicatedCount++ + st.ReplicatedSize += sz + } else { + st.FailedCount++ + st.FailedSize += roi.Size + } + return st + } + // arn was not attempted for this object: a resync that cannot confirm the + // object reached the target is not a success. + st.FailedCount++ + st.FailedSize += roi.Size + return st +} + // 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) { @@ -3067,6 +3128,7 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object default: } traceFn := s.trace(tgt.ResetID, fmt.Sprintf("%s/%s (%s)", opts.bucket, roi.Name, roi.VersionID)) + var rinfos replicatedInfos if roi.DeleteMarker || !roi.VersionPurgeStatus.Empty() { versionID := "" dmVersionID := "" @@ -3089,37 +3151,26 @@ func (s *replicationResyncer) resyncBucket(ctx context.Context, objectAPI Object OpType: replication.ExistingObjectReplicationType, EventType: ReplicateExistingDelete, } - replicateDelete(ctx, doi, objectAPI) + rinfos = replicateDelete(ctx, doi, objectAPI) } else { roi.OpType = replication.ExistingObjectReplicationType roi.EventType = ReplicateExisting - replicateObject(ctx, roi, objectAPI) + rinfos = replicateObject(ctx, roi, objectAPI) } - st := TargetReplicationResyncStatus{ - Object: roi.Name, - Bucket: roi.Bucket, - } - - _, err := tgt.StatObject(ctx, tgt.Bucket, roi.Name, minio.StatObjectOptions{ - VersionID: roi.VersionID, - Internal: minio.AdvancedGetOptions{ - ReplicationProxyRequest: "false", - }, - }) - sz := roi.Size - if err != nil { - if roi.DeleteMarker && isErrMethodNotAllowed(ErrorRespToObjectError(err, opts.bucket, roi.Name)) { - st.ReplicatedCount++ - } else { - st.FailedCount++ + // Classify success/failure from the actual replication outcome + // for this target, not from whether the target version merely + // exists (a rejected update leaves the old version in place). + st := resyncResultFor(rinfos, opts.arn, roi) + var traceSize int64 + var traceErr error + for i := range rinfos.Targets { + if rinfos.Targets[i].Arn == opts.arn { + traceSize, traceErr = rinfos.Targets[i].Size, rinfos.Targets[i].Err + break } - sz = 0 - } else { - st.ReplicatedCount++ - st.ReplicatedSize += roi.Size } - traceFn(sz, err) + traceFn(traceSize, traceErr) if !s.sendResyncResult(ctx, results.ch, st, &workerAborted) { return } diff --git a/cmd/bucket-replication_test.go b/cmd/bucket-replication_test.go index 9d6e5f79d..8e2bb102c 100644 --- a/cmd/bucket-replication_test.go +++ b/cmd/bucket-replication_test.go @@ -29,6 +29,7 @@ import ( "time" "github.com/minio/madmin-go/v3" + "github.com/minio/minio-go/v7" "github.com/minio/minio/internal/bucket/replication" xhttp "github.com/minio/minio/internal/http" ) @@ -528,3 +529,121 @@ func TestResyncFinishWaitsForInflightWorker(t *testing.T) { } }) } + +// TestResyncResultFor asserts the resync worker classifies a target from the +// actual replication outcome, not from whether the target version merely exists. +// The key regression is the "failed update over an existing version" case: a +// quota-rejected update leaves the old version in place, and counting existence +// (the previous behavior) would score it a success. It also checks a genuine +// success, an errored-but-Completed result, a delete failure, a delete-marker +// success (zero bytes), and an ARN that was never attempted. +func TestResyncResultFor(t *testing.T) { + const arn = "arn:minio:replication::id:bucket" + obj := ReplicateObjectInfo{Name: "obj", Bucket: "bucket", Size: 196608} + deleteMarker := ReplicateObjectInfo{Name: "dm", Bucket: "bucket", Size: 0, DeleteMarker: true} + + tests := []struct { + name string + roi ReplicateObjectInfo + rinfos replicatedInfos + wantRepl, wantReplSize, wantFail, wantFailSize int64 + }{ + { + name: "completed update", + roi: obj, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, ReplicationStatus: replication.Completed, Size: 196608}, + }}, + wantRepl: 1, wantReplSize: 196608, + }, + { + name: "failed update over existing version", + roi: obj, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, ReplicationStatus: replication.Failed, Err: fmt.Errorf("quota exceeded"), Size: 196608}, + }}, + wantFail: 1, wantFailSize: 196608, + }, + { + name: "completed but errored is a failure", + roi: obj, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, ReplicationStatus: replication.Completed, Err: fmt.Errorf("boom"), Size: 196608}, + }}, + wantFail: 1, wantFailSize: 196608, + }, + { + name: "delete failed", + roi: deleteMarker, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, ReplicationStatus: replication.Failed}, + }}, + wantFail: 1, wantFailSize: 0, + }, + { + name: "delete marker replicated counts zero bytes", + roi: deleteMarker, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, ReplicationStatus: replication.Completed}, + }}, + wantRepl: 1, wantReplSize: 0, + }, + { + name: "arn not attempted is a failure", + roi: obj, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: "arn:minio:replication::id2:bucket", ReplicationStatus: replication.Completed, Size: 196608}, + }}, + wantFail: 1, wantFailSize: 196608, + }, + { + name: "completed with zero size falls back to object size", + roi: obj, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, ReplicationStatus: replication.Completed, Size: 0}, + }}, + wantRepl: 1, wantReplSize: 196608, + }, + { + name: "version purge complete is a success", + roi: ReplicateObjectInfo{Name: "purge", Bucket: "bucket", Size: 196608, VersionPurgeStatus: replication.VersionPurgePending}, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + // a successful purge sets only VersionPurgeStatus; ReplicationStatus stays empty. + {Arn: arn, VersionPurgeStatus: replication.VersionPurgeComplete}, + }}, + wantRepl: 1, wantReplSize: 196608, + }, + { + name: "version purge failed is a failure", + roi: ReplicateObjectInfo{Name: "purge", Bucket: "bucket", Size: 196608, VersionPurgeStatus: replication.VersionPurgePending}, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + {Arn: arn, VersionPurgeStatus: replication.VersionPurgeFailed, Err: fmt.Errorf("quota exceeded")}, + }}, + wantFail: 1, wantFailSize: 196608, + }, + { + name: "benign duplicate 412 is a success", + roi: obj, + rinfos: replicatedInfos{Targets: []replicatedTargetInfo{ + // the destination answers PreconditionFailed for an exact duplicate; + // replicateAll keeps Completed but retains the error. + {Arn: arn, ReplicationStatus: replication.Completed, Err: minio.ErrorResponse{Code: "PreconditionFailed"}, Size: 196608}, + }}, + wantRepl: 1, wantReplSize: 196608, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + st := resyncResultFor(tc.rinfos, arn, tc.roi) + if st.Object != tc.roi.Name || st.Bucket != tc.roi.Bucket { + t.Fatalf("object/bucket = %s/%s, want %s/%s", st.Object, st.Bucket, tc.roi.Name, tc.roi.Bucket) + } + if st.ReplicatedCount != tc.wantRepl || st.ReplicatedSize != tc.wantReplSize || + st.FailedCount != tc.wantFail || st.FailedSize != tc.wantFailSize { + t.Fatalf("resyncResultFor = {replicated:%d/%d failed:%d/%d}, want {%d/%d %d/%d}", + st.ReplicatedCount, st.ReplicatedSize, st.FailedCount, st.FailedSize, + tc.wantRepl, tc.wantReplSize, tc.wantFail, tc.wantFailSize) + } + }) + } +}