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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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

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

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
 			}

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