From 9d7094b770fcb7a38399a8a6a1eb956df852d4d7 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 9 Sep 2026 14:32:09 +0800 Subject: [PATCH] fix(replication): complete single-object delete marker purges Signed-off-by: Feng Ruohang --- cmd/object-handlers.go | 2 +- cmd/replication-delete-marker_test.go | 236 ++++++++++++++++++++++++++ 2 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 cmd/replication-delete-marker_test.go diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 83d312ba3..f4cb876b3 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -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 diff --git a/cmd/replication-delete-marker_test.go b/cmd/replication-delete-marker_test.go new file mode 100644 index 000000000..778d1823b --- /dev/null +++ b/cmd/replication-delete-marker_test.go @@ -0,0 +1,236 @@ +// 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, `%sinjected delete rejection`, 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::00000000-0000-4000-8000-000000000001:replica" + 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) + } + } +}