diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index f06360883d..2e895d3e47 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -503,6 +503,10 @@ type putFinalize struct { // when the write path is absent — a suspended-versioning write points it // at the latest version so check and write stay under the filer lock conditionAltKey string + // newVersion re-mints a versioned write's id at commit time, so the + // version orders after every write committed during its upload or lock + // wait. It renames entry and returns the new write path. + newVersion func(entry *filer_pb.Entry) string } const ( @@ -1095,6 +1099,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader if finalize != nil { condAltKey = finalize.conditionAltKey } + remintVersion := func() { + if finalize != nil && finalize.newVersion != nil { + filePath = finalize.newVersion(entry) + } + } if owner = s3a.routableWriteOwner(bucket, object); owner != "" { if cond, ok := routeWriteCondition(r, uniqueWritePath && condAltKey == ""); ok { // Routed mutations ride in the PUT's transaction (committing atomically), @@ -1104,6 +1113,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader lockKey, finalizeMutations = finalize.lockKey, finalize.mutations } routeKey := s3a.objectRouteKey(bucket, object) + remintVersion() resp, err := s3a.routedPut(owner, routeKey, lockKey, filePath, entry, cond, "", condAltKey, false, nil, "", finalizeMutations) if err == nil && resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED && condAltKey != "" { // A filer that ignores condition_alt_key evaluated the empty @@ -1143,7 +1153,10 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } } if !routed { - createFn := func() s3err.ErrorCode { return createUnderLock(nil) } + createFn := func() s3err.ErrorCode { + remintVersion() + return createUnderLock(nil) + } if routedAmbiguous { createFn = func() s3err.ErrorCode { return s3a.createAfterAmbiguousRoute(filePath, bucket, object, owner, entry, chunkResult.FileChunks, finalize, &entryCreated, s3a.uploadedChunksExist, createUnderLock) @@ -2185,7 +2198,7 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin // directly — versioned objects sit on regular volumes and the // lifecycle worker handles their expiration. etag, errCode, sseMetadata = s3a.putToFiler(r, versionFilePath, body, bucket, normalizedObject, 1, 0, - s3a.versionedFinalize(bucket, normalizedObject, versionId, versionFileName, useInvertedFormat), true, "") + s3a.versionedFinalize(bucket, normalizedObject, &versionId, useInvertedFormat), true, "") if errCode != s3err.ErrNone { glog.Errorf("putVersionedObject: failed to upload version: %v", errCode) return "", "", errCode, SSEResponseMetadata{} diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 796b6e7ef5..43d7e311b2 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -156,6 +156,7 @@ type ambiguousPutFiler struct { // just-committed entry before recovery reads it. beforeCreate func() beforeLookup func() + beforeAssign func() stores int conds []*filer_pb.WriteCondition } @@ -163,6 +164,9 @@ type ambiguousPutFiler struct { func (f *ambiguousPutFiler) AssignVolume(context.Context, *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) { f.mu.Lock() defer f.mu.Unlock() + if f.beforeAssign != nil { + f.beforeAssign() + } f.nextKey++ return &filer_pb.AssignVolumeResponse{ FileId: fmt.Sprintf("3,%016x%08x", f.nextKey, uint32(f.nextKey)), diff --git a/weed/s3api/s3api_object_versioned_finalize.go b/weed/s3api/s3api_object_versioned_finalize.go index abd2e378b3..5104b9610f 100644 --- a/weed/s3api/s3api_object_versioned_finalize.go +++ b/weed/s3api/s3api_object_versioned_finalize.go @@ -276,12 +276,18 @@ func (s3a *S3ApiServer) routedDeleteNullVersion(owner pb.ServerAddress, bucket, // on the routed path RECOMPUTE_LATEST rides in the version file's PUT transaction, // committing atomically under the object's per-path lock; off the ring // updateLatestVersionInDirectory does it under the object write lock. -func (s3a *S3ApiServer) versionedFinalize(bucket, object, versionId, versionFileName string, useInvertedFormat bool) *putFinalize { +func (s3a *S3ApiServer) versionedFinalize(bucket, object string, versionId *string, useInvertedFormat bool) *putFinalize { return &putFinalize{ lockKey: s3a.toFilerPath(bucket, object), mutations: []*filer_pb.ObjectMutation{s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true)}, + newVersion: func(entry *filer_pb.Entry) string { + *versionId = generateVersionId(useInvertedFormat) + entry.Name = s3a.getVersionFileName(*versionId) + entry.Extended[s3_constants.ExtVersionIdKey] = []byte(*versionId) + return s3a.toFilerPath(bucket, object+s3_constants.VersionsFolder+"/"+entry.Name) + }, afterCreate: func(versionEntry *filer_pb.Entry) s3err.ErrorCode { - if err := s3a.updateLatestVersionInDirectory(bucket, object, versionId, versionFileName, versionEntry); err != nil { + if err := s3a.updateLatestVersionInDirectory(bucket, object, *versionId, s3a.getVersionFileName(*versionId), versionEntry); err != nil { glog.Errorf("putVersionedObject: failed to update latest version in directory: %v", err) return s3err.ErrInternalError } diff --git a/weed/s3api/s3api_object_versioned_put_order_test.go b/weed/s3api/s3api_object_versioned_put_order_test.go new file mode 100644 index 0000000000..aa5e3b68ab --- /dev/null +++ b/weed/s3api/s3api_object_versioned_put_order_test.go @@ -0,0 +1,142 @@ +package s3api + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "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" +) + +type noopObjectWriteLock struct{} + +func (noopObjectWriteLock) StopShortLivedLock() error { return nil } + +func newVersionedPutFiler(t *testing.T) (*ambiguousPutFiler, *S3ApiServer) { + t.Helper() + firstId := generateVersionId(true) + firstName := "v_" + firstId + filerImpl := &ambiguousPutFiler{ + volume: startFakeVolumeServer(t), + entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)}, + }, + "/buckets/b/k.versions": { + Name: "k.versions", + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte(firstId), + s3_constants.ExtLatestVersionFileNameKey: []byte(firstName), + }, + }, + "/buckets/b/k.versions/" + firstName: { + Name: firstName, + Attributes: &filer_pb.FuseAttributes{FileSize: 1}, + Extended: map[string][]byte{ + s3_constants.ExtETagKey: []byte("abc123"), + s3_constants.ExtVersionIdKey: []byte(firstId), + }, + }, + }, + apply: true, + } + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + return filerImpl, s3a +} + +func latestVersionId(f *ambiguousPutFiler) string { + f.mu.Lock() + defer f.mu.Unlock() + return string(f.entries["/buckets/b/k.versions"].Extended[s3_constants.ExtLatestVersionIdKey]) +} + +// A conditional PUT that waits on the object write lock while a DELETE commits +// must still become the latest version once its condition passes. +func TestVersionedConditionalPutAfterLockWaitBecomesLatest(t *testing.T) { + filerImpl, s3a := newVersionedPutFiler(t) + var markerId string + s3a.newObjectWriteLock = func(bucket, object string) (objectWriteLock, error) { + if markerId == "" { + var err error + if markerId, err = s3a.createDeleteMarker(bucket, object); err != nil { + t.Fatalf("create delete marker: %v", err) + } + } + return noopObjectWriteLock{}, nil + } + + r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new")) + r.Header.Set(s3_constants.IfNoneMatch, "*") + versionId, _, code, _ := s3a.putVersionedObject(r, "b", "k", r.Body, "") + if code != s3err.ErrNone { + t.Fatalf("If-None-Match PUT after the key was deleted = %v, want success", code) + } + if markerId == "" || latestVersionId(filerImpl) == markerId { + t.Fatalf("latest stayed on delete marker %q, want the accepted PUT", markerId) + } + if got := latestVersionId(filerImpl); got != versionId { + t.Fatalf("latest version = %q, want the returned version %q", got, versionId) + } + if compareVersionIds(markerId, versionId) <= 0 { + t.Fatalf("version %q does not order after delete marker %q", versionId, markerId) + } + filerImpl.mu.Lock() + stored := filerImpl.entries["/buckets/b/k.versions/"+s3a.getVersionFileName(versionId)] + filerImpl.mu.Unlock() + if stored == nil || string(stored.Extended[s3_constants.ExtVersionIdKey]) != versionId { + t.Fatalf("version file for %q missing or carries another id", versionId) + } +} + +// A routed PUT whose upload overlaps a DELETE commits after it, so the delete +// marker must not stay latest over the accepted write. +func TestVersionedRoutedPutAfterConcurrentDeleteBecomesLatest(t *testing.T) { + filerImpl, s3a := newVersionedPutFiler(t) + filerAddr := s3a.option.Filers[0] + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + markerId := "" + filerImpl.beforeAssign = func() { + if markerId != "" { + return + } + markerId = generateVersionId(true) + markerName := "v_" + markerId + filerImpl.entries["/buckets/b/k.versions/"+markerName] = &filer_pb.Entry{ + Name: markerName, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte(markerId), + s3_constants.ExtDeleteMarkerKey: []byte("true"), + }, + } + pointer := filerImpl.entries["/buckets/b/k.versions"] + pointer.Extended[s3_constants.ExtLatestVersionIdKey] = []byte(markerId) + pointer.Extended[s3_constants.ExtLatestVersionFileNameKey] = []byte(markerName) + } + + r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new")) + versionId, _, code, _ := s3a.putVersionedObject(r, "b", "k", r.Body, "") + if code != s3err.ErrNone { + t.Fatalf("PUT = %v", code) + } + if filerImpl.txnCount != 1 { + t.Fatalf("PUT should route one transaction, got %d", filerImpl.txnCount) + } + if got := latestVersionId(filerImpl); got != versionId { + t.Fatalf("latest version = %q, want the returned version %q (delete marker %q)", got, versionId, markerId) + } +}