From 86db60b669844563662a1e7fb4e072d5069923e5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 13:20:36 -0800 Subject: [PATCH] s3api: mint a versioned PUT's id at commit, after the lock wait and the upload (#12454) * s3api: mint a versioned PUT's id after the object write lock putVersionedObject minted the version id before the upload and the lock wait. A DELETE that committed while the PUT waited got a newer id, so a conditional PUT whose If-None-Match passed against the delete marker was stored behind it: accepted, but GET returned NoSuchKey. The finalize now re-mints the id once the lock is held and the precondition passed, so the version orders after everything the check saw. An ambiguous routed attempt keeps its id, since recovery looks for that entry. Co-Authored-By: Claude Opus 5.5 * s3api: mint a routed versioned PUT's id after the upload On the routed path the version id was still the one minted before the upload, and RECOMPUTE_LATEST picks the newest name. A DELETE that committed during a long upload left its marker latest over the accepted PUT. Re-mint right before the routed transaction, as the lock path does. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3api_object_handlers_put.go | 17 ++- ...3api_object_handlers_put_ambiguous_test.go | 4 + weed/s3api/s3api_object_versioned_finalize.go | 10 +- .../s3api_object_versioned_put_order_test.go | 142 ++++++++++++++++++ 4 files changed, 169 insertions(+), 4 deletions(-) create mode 100644 weed/s3api/s3api_object_versioned_put_order_test.go 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) + } +}