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) }