From 3a579ece3beac6d03eab3a1d3bf3617c4322f9d7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 09:10:10 -0800 Subject: [PATCH] s3: commit the delete marker in the owner's finalize transaction (#12432) * s3: commit the delete marker in the owner's finalize transaction createDeleteMarker wrote the marker with mkFile to the gateway's current filer and then sent RECOMPUTE_LATEST to the object's ring owner. When the two differ, the owner scans .versions before peer metadata sync delivers the marker, re-derives the pointer to the previous version, and the marker stays non-current on every filer. routedVersionedFinalize now takes an optional version entry and PUTs it in the same ObjectTransaction ahead of the recompute, the way PutObject and CompleteMultipartUpload already do. The copy destination keeps passing nil: its version file is already written on the owner. Co-Authored-By: Claude Opus 5.5 * s3: test a versioned delete when the gateway's filer is not the owner Co-Authored-By: Claude Opus 5.5 * s3: assert the delete marker and the pointer flip share one transaction Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3api_object_handlers_copy.go | 2 +- .../s3api_object_handlers_delete_test.go | 68 +++++++++++++++++++ weed/s3api/s3api_object_versioned_finalize.go | 18 +++-- weed/s3api/s3api_object_versioning.go | 30 +++++--- 4 files changed, 105 insertions(+), 13 deletions(-) diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index a7b5aa6d27..fa991cca91 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -757,7 +757,7 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion // Route the pointer flip to the owner filer when known (off the // distributed lock); RECOMPUTE_LATEST picks the just-written version. if owner != "" { - if code := s3a.routedVersionedFinalize(owner, dstBucket, normalizedObject, isNewFormatVersionId(versionId)); code != s3err.ErrNone { + if code := s3a.routedVersionedFinalize(owner, dstBucket, normalizedObject, isNewFormatVersionId(versionId), nil); code != s3err.ErrNone { if rollbackErr := s3a.rollbackCopyVersion(owner, dstBucket, normalizedObject, versionObjectPath); rollbackErr != nil { glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, dstBucket, normalizedObject, rollbackErr) } diff --git a/weed/s3api/s3api_object_handlers_delete_test.go b/weed/s3api/s3api_object_handlers_delete_test.go index dd680d6c0e..6c7c956af0 100644 --- a/weed/s3api/s3api_object_handlers_delete_test.go +++ b/weed/s3api/s3api_object_handlers_delete_test.go @@ -3,10 +3,19 @@ package s3api import ( "context" "errors" + "net/http" + "net/http/httptest" + "strings" "testing" + "time" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -107,3 +116,62 @@ func TestDeleteUnversionedObjectWithClient_PropagatesEntryAttributesIrrelevant(t require.NotNil(t, client.deleteReq) assert.False(t, client.deleteReq.IsDeleteData) } + +// The gateway's own filer is not the object's write owner and has not seen the +// owner's writes yet. The delete marker must still commit on the owner, where +// the latest-pointer recompute scans .versions, so the marker becomes current. +func TestVersionedDeleteMarkerCommitsOnOwner(t *testing.T) { + util_http.InitGlobalHttpClient() + volume := startFakeVolumeServer(t) + owner := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + replica := &ambiguousPutFiler{volume: volume, apply: true, entries: map[string]*filer_pb.Entry{}} + ownerAddr := startFakeFiler(t, owner) + replicaAddr := startFakeFiler(t, replica) + s3a := newPutTestServer(t, replicaAddr) + s3a.iam = &IdentityAccessManagement{} + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("bucket", &BucketConfig{Name: "bucket", Versioning: s3_constants.VersioningEnabled}) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, ownerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{ownerAddr}, 1) + + put := httptest.NewRequest(http.MethodPut, "/bucket/key", strings.NewReader("data")) + put = mux.SetURLVars(put, map[string]string{"bucket": "bucket", "object": "key"}) + putResult := httptest.NewRecorder() + s3a.PutObjectHandler(putResult, put) + require.Equal(t, http.StatusOK, putResult.Code, putResult.Body.String()) + dataVersionId := putResult.Header().Get("x-amz-version-id") + require.NotEmpty(t, dataVersionId) + + var deleteTxns []*filer_pb.ObjectTransactionRequest + owner.mu.Lock() + owner.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { deleteTxns = append(deleteTxns, req) } + owner.mu.Unlock() + + del := httptest.NewRequest(http.MethodDelete, "/bucket/key", nil) + del = mux.SetURLVars(del, map[string]string{"bucket": "bucket", "object": "key"}) + delResult := httptest.NewRecorder() + s3a.DeleteObjectHandler(delResult, del) + require.Equal(t, http.StatusNoContent, delResult.Code, delResult.Body.String()) + markerId := delResult.Header().Get("x-amz-version-id") + require.NotEmpty(t, markerId) + require.NotEqual(t, dataVersionId, markerId) + + markerKey := "/buckets/bucket/key.versions/v_" + markerId + owner.mu.Lock() + require.Len(t, deleteTxns, 1, "the marker and the pointer flip must share one transaction") + mutations := deleteTxns[0].Mutations + require.Len(t, mutations, 2) + require.Equal(t, filer_pb.ObjectMutation_PUT, mutations[0].Type) + require.Equal(t, "/buckets/bucket/key.versions", mutations[0].Directory) + require.Equal(t, "v_"+markerId, mutations[0].Entry.GetName()) + require.Equal(t, filer_pb.ObjectMutation_RECOMPUTE_LATEST, mutations[1].Type) + require.NotNil(t, owner.entries[markerKey], "the marker must commit on the owner") + pointer := owner.entries["/buckets/bucket/key.versions"] + require.NotNil(t, pointer) + require.Equal(t, markerId, string(pointer.Extended[s3_constants.ExtLatestVersionIdKey])) + require.Equal(t, "true", string(pointer.Extended[s3_constants.ExtLatestVersionIsDeleteMarker])) + owner.mu.Unlock() + replica.mu.Lock() + require.Nil(t, replica.entries[markerKey], "the replica must still be behind") + replica.mu.Unlock() +} diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index c9ec86251a..abd2e378b3 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -66,13 +66,23 @@ func (s3a *S3ApiServer) latestPointerRecompute(bucket, object string, useInverte // routedVersionedFinalize flips the .versions pointer to the newest version and // demotes the prior latest, atomically under the object's per-path lock on the -// owner filer, via a single RECOMPUTE_LATEST. The version file is already -// written; the owner re-derives the pointer by scanning the directory. -func (s3a *S3ApiServer) routedVersionedFinalize(owner pb.ServerAddress, bucket, object string, useInvertedFormat bool) s3err.ErrorCode { +// owner filer, via a single RECOMPUTE_LATEST. versionEntry, when set, is PUT +// into .versions in the same transaction, so the recompute's scan sees it even +// when the gateway's own filer is not the owner; otherwise the version file is +// already written on the owner. +func (s3a *S3ApiServer) routedVersionedFinalize(owner pb.ServerAddress, bucket, object string, useInvertedFormat bool, versionEntry *filer_pb.Entry) s3err.ErrorCode { + mutations := []*filer_pb.ObjectMutation{s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true)} + if versionEntry != nil { + mutations = append([]*filer_pb.ObjectMutation{{ + Type: filer_pb.ObjectMutation_PUT, + Directory: s3a.toFilerPath(bucket, object+s3_constants.VersionsFolder), + Entry: versionEntry, + }}, mutations...) + } req := &filer_pb.ObjectTransactionRequest{ LockKey: s3a.toFilerPath(bucket, object), RouteKey: s3a.objectRouteKey(bucket, object), - Mutations: []*filer_pb.ObjectMutation{s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true)}, + Mutations: mutations, } resp, err := s3a.objectTxnOnFiler(owner, req) switch { diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index ae6c3b98fd..a251d1153c 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -215,13 +215,33 @@ func (s3a *S3ApiServer) createDeleteMarker(bucket, object string) (string, error bucketDir := s3a.bucketDir(bucket) versionsDir := bucketDir + "/" + cleanObject + s3_constants.VersionsFolder - // Create the delete marker entry in the .versions directory deleteMarkerMtime := time.Now().Unix() deleteMarkerExtended := map[string][]byte{ s3_constants.ExtVersionIdKey: []byte(versionId), s3_constants.ExtDeleteMarkerKey: []byte("true"), } + // On the owner filer the marker and the pointer flip commit in one + // transaction; RECOMPUTE_LATEST picks the marker as the new latest. + if owner := s3a.objectWriteOwner(bucket, cleanObject); owner != "" { + deleteMarkerEntry := &filer_pb.Entry{ + Name: versionFileName, + Attributes: &filer_pb.FuseAttributes{ + Mtime: deleteMarkerMtime, + Crtime: deleteMarkerMtime, + FileMode: uint32(0770), + Uid: filer_pb.OS_UID, + Gid: filer_pb.OS_GID, + }, + Extended: deleteMarkerExtended, + } + if code := s3a.routedVersionedFinalize(owner, bucket, cleanObject, useInvertedFormat, deleteMarkerEntry); code != s3err.ErrNone { + return "", fmt.Errorf("createDeleteMarker: routed finalize failed for %s/%s: code %d", bucket, object, code) + } + glog.V(2).Infof("createDeleteMarker: successfully created delete marker %s for %s/%s", versionId, bucket, object) + return versionId, nil + } + err := s3a.mkFile(versionsDir, versionFileName, nil, func(entry *filer_pb.Entry) { entry.IsDirectory = false if entry.Attributes == nil { @@ -249,13 +269,7 @@ func (s3a *S3ApiServer) createDeleteMarker(bucket, object string) (string, error }, Extended: deleteMarkerExtended, } - // Route the pointer flip to the owner filer when known (off the distributed - // lock); RECOMPUTE_LATEST picks the just-written marker as the new latest. - if owner := s3a.objectWriteOwner(bucket, cleanObject); owner != "" { - if code := s3a.routedVersionedFinalize(owner, bucket, cleanObject, useInvertedFormat); code != s3err.ErrNone { - return "", fmt.Errorf("createDeleteMarker: routed finalize failed for %s/%s: code %d", bucket, object, code) - } - } else if err = s3a.updateLatestVersionInDirectory(bucket, cleanObject, versionId, versionFileName, deleteMarkerEntry); err != nil { + if err = s3a.updateLatestVersionInDirectory(bucket, cleanObject, versionId, versionFileName, deleteMarkerEntry); err != nil { glog.Errorf("createDeleteMarker: failed to update latest version in directory: %v", err) return "", fmt.Errorf("failed to update latest version in directory: %w", err) }