s3: route versioned PutObject finalize off the DLM (object-key serialization)

A versioned write's only contended mutation is the .versions directory's latest
pointer; the version file itself goes to a unique <object>/.versions/<versionId>
path. Add a FinalizeVersionedWrite filer op that, under one exclusive lock on the
object key, evaluates the precondition against the current latest, stamps the
previous latest noncurrent (before the pointer flip so the lifecycle router
observes it), then merges the latest pointer / cached metadata into the .versions
entry. Key names are passed in, so the filer carries no S3 semantics.

Routing and the lock are keyed on the object (objectWriteOwner + lock_key), the
same key normal and suspended writes use, so all writes to one object resolve the
same owner and serialize on the same lock regardless of versioning state — a
versioned and a non-versioned write to the same object can't race on different
owners during a versioning-state change.

The gateway routes a versioned PutObject's finalize to that owner and tells
putToFiler the version path is unique, so it skips the object write lock and the
gateway precondition (the op does both atomically). When the owner is unknown or
the condition can't reduce to one primitive, it stays on the lock path; on op
error it returns InternalError. Versioned COPY, delete markers, suspended
versioning, and multipart completion still use the lock and adopt the same op as
follow-ups.

other/java/client/src/main/proto/filer.proto

@@ -22,6 +22,9 @@ service SeaweedFiler {
     rpc UpdateEntry (UpdateEntryRequest) returns (UpdateEntryResponse) {
     }
 
+    rpc FinalizeVersionedWrite (FinalizeVersionedWriteRequest) returns (FinalizeVersionedWriteResponse) {
+    }
+
     rpc TouchAccessTime (TouchAccessTimeRequest) returns (TouchAccessTimeResponse) {
     }
 
@@ -276,6 +279,39 @@ message UpdateEntryResponse {
     SubscribeMetadataResponse metadata_event = 1;
 }
 
+// FinalizeVersionedWrite atomically finalizes an S3 versioned write on the
+// object's owner filer, after the caller has created the new version file: under
+// one exclusive lock on lock_key (the object path — the same key all of this
+// object's writes serialize on, regardless of versioning state) it evaluates the
+// precondition against the current latest, stamps the previously-latest version
+// as noncurrent (before the pointer flip, so the lifecycle router observes it),
+// then merges the latest-pointer / cached metadata into the .versions directory
+// entry. The caller routes the object's writes to this owner so the whole
+// sequence shares one local lock — no distributed lock. Key names are passed in
+// so the filer stays free of S3 semantics; the new version's file name is read
+// from set_extended[prior_latest_key].
+message FinalizeVersionedWriteRequest {
+    string lock_key = 13;                // object path; the per-path lock all of this object's writes share
+    string versions_dir = 1;             // full path of <object>/.versions
+    map<string, bytes> set_extended = 3; // merge into the .versions directory entry
+    repeated string delete_extended = 4; // remove from the .versions directory entry
+    string prior_latest_key = 5;         // .versions entry key holding the latest file name (prior value = demote target)
+    string noncurrent_since_key = 6;     // key to stamp on the demoted previous version
+    int64 noncurrent_since_ns = 7;       // stamp value; <=0 skips the demote stamp
+    bool is_from_other_cluster = 8;
+    repeated int32 signatures = 9;
+    // Optional precondition, evaluated against the current latest version
+    // (derived from the .versions directory entry's cached metadata) before
+    // anything is written, so the check and the write are atomic on this owner.
+    WriteCondition condition = 10;
+    string latest_etag_key = 11;          // .versions entry key holding the latest version's ETag
+    string latest_delete_marker_key = 12; // .versions entry key; "true" means the latest is a delete marker
+}
+message FinalizeVersionedWriteResponse {
+    string error = 1;
+    FilerError error_code = 2;
+}
+
 message TouchAccessTimeRequest {
     string directory = 1;
     string name = 2;

weed/pb/filer.proto

@@ -22,6 +22,9 @@ service SeaweedFiler {
     rpc UpdateEntry (UpdateEntryRequest) returns (UpdateEntryResponse) {
     }
 
+    rpc FinalizeVersionedWrite (FinalizeVersionedWriteRequest) returns (FinalizeVersionedWriteResponse) {
+    }
+
     rpc TouchAccessTime (TouchAccessTimeRequest) returns (TouchAccessTimeResponse) {
     }
 
@@ -276,6 +279,39 @@ message UpdateEntryResponse {
     SubscribeMetadataResponse metadata_event = 1;
 }
 
+// FinalizeVersionedWrite atomically finalizes an S3 versioned write on the
+// object's owner filer, after the caller has created the new version file: under
+// one exclusive lock on lock_key (the object path — the same key all of this
+// object's writes serialize on, regardless of versioning state) it evaluates the
+// precondition against the current latest, stamps the previously-latest version
+// as noncurrent (before the pointer flip, so the lifecycle router observes it),
+// then merges the latest-pointer / cached metadata into the .versions directory
+// entry. The caller routes the object's writes to this owner so the whole
+// sequence shares one local lock — no distributed lock. Key names are passed in
+// so the filer stays free of S3 semantics; the new version's file name is read
+// from set_extended[prior_latest_key].
+message FinalizeVersionedWriteRequest {
+    string lock_key = 13;                // object path; the per-path lock all of this object's writes share
+    string versions_dir = 1;             // full path of <object>/.versions
+    map<string, bytes> set_extended = 3; // merge into the .versions directory entry
+    repeated string delete_extended = 4; // remove from the .versions directory entry
+    string prior_latest_key = 5;         // .versions entry key holding the latest file name (prior value = demote target)
+    string noncurrent_since_key = 6;     // key to stamp on the demoted previous version
+    int64 noncurrent_since_ns = 7;       // stamp value; <=0 skips the demote stamp
+    bool is_from_other_cluster = 8;
+    repeated int32 signatures = 9;
+    // Optional precondition, evaluated against the current latest version
+    // (derived from the .versions directory entry's cached metadata) before
+    // anything is written, so the check and the write are atomic on this owner.
+    WriteCondition condition = 10;
+    string latest_etag_key = 11;          // .versions entry key holding the latest version's ETag
+    string latest_delete_marker_key = 12; // .versions entry key; "true" means the latest is a delete marker
+}
+message FinalizeVersionedWriteResponse {
+    string error = 1;
+    FilerError error_code = 2;
+}
+
 message TouchAccessTimeRequest {
     string directory = 1;
     string name = 2;

weed/pb/filer_pb/filer_grpc.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.6.2
-// - protoc             v7.34.1
+// - protoc             v6.33.4
 // source: filer.proto
 
 package filer_pb
@@ -23,6 +23,7 @@ const (
 	SeaweedFiler_ListEntries_FullMethodName                     = "/filer_pb.SeaweedFiler/ListEntries"
 	SeaweedFiler_CreateEntry_FullMethodName                     = "/filer_pb.SeaweedFiler/CreateEntry"
 	SeaweedFiler_UpdateEntry_FullMethodName                     = "/filer_pb.SeaweedFiler/UpdateEntry"
+	SeaweedFiler_FinalizeVersionedWrite_FullMethodName          = "/filer_pb.SeaweedFiler/FinalizeVersionedWrite"
 	SeaweedFiler_TouchAccessTime_FullMethodName                 = "/filer_pb.SeaweedFiler/TouchAccessTime"
 	SeaweedFiler_AppendToEntry_FullMethodName                   = "/filer_pb.SeaweedFiler/AppendToEntry"
 	SeaweedFiler_DeleteEntry_FullMethodName                     = "/filer_pb.SeaweedFiler/DeleteEntry"
@@ -59,6 +60,7 @@ type SeaweedFilerClient interface {
 	ListEntries(ctx context.Context, in *ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListEntriesResponse], error)
 	CreateEntry(ctx context.Context, in *CreateEntryRequest, opts ...grpc.CallOption) (*CreateEntryResponse, error)
 	UpdateEntry(ctx context.Context, in *UpdateEntryRequest, opts ...grpc.CallOption) (*UpdateEntryResponse, error)
+	FinalizeVersionedWrite(ctx context.Context, in *FinalizeVersionedWriteRequest, opts ...grpc.CallOption) (*FinalizeVersionedWriteResponse, error)
 	TouchAccessTime(ctx context.Context, in *TouchAccessTimeRequest, opts ...grpc.CallOption) (*TouchAccessTimeResponse, error)
 	AppendToEntry(ctx context.Context, in *AppendToEntryRequest, opts ...grpc.CallOption) (*AppendToEntryResponse, error)
 	DeleteEntry(ctx context.Context, in *DeleteEntryRequest, opts ...grpc.CallOption) (*DeleteEntryResponse, error)
@@ -147,6 +149,16 @@ func (c *seaweedFilerClient) UpdateEntry(ctx context.Context, in *UpdateEntryReq
 	return out, nil
 }
 
+func (c *seaweedFilerClient) FinalizeVersionedWrite(ctx context.Context, in *FinalizeVersionedWriteRequest, opts ...grpc.CallOption) (*FinalizeVersionedWriteResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(FinalizeVersionedWriteResponse)
+	err := c.cc.Invoke(ctx, SeaweedFiler_FinalizeVersionedWrite_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *seaweedFilerClient) TouchAccessTime(ctx context.Context, in *TouchAccessTimeRequest, opts ...grpc.CallOption) (*TouchAccessTimeResponse, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	out := new(TouchAccessTimeResponse)
@@ -454,6 +466,7 @@ type SeaweedFilerServer interface {
 	ListEntries(*ListEntriesRequest, grpc.ServerStreamingServer[ListEntriesResponse]) error
 	CreateEntry(context.Context, *CreateEntryRequest) (*CreateEntryResponse, error)
 	UpdateEntry(context.Context, *UpdateEntryRequest) (*UpdateEntryResponse, error)
+	FinalizeVersionedWrite(context.Context, *FinalizeVersionedWriteRequest) (*FinalizeVersionedWriteResponse, error)
 	TouchAccessTime(context.Context, *TouchAccessTimeRequest) (*TouchAccessTimeResponse, error)
 	AppendToEntry(context.Context, *AppendToEntryRequest) (*AppendToEntryResponse, error)
 	DeleteEntry(context.Context, *DeleteEntryRequest) (*DeleteEntryResponse, error)
@@ -505,6 +518,9 @@ func (UnimplementedSeaweedFilerServer) CreateEntry(context.Context, *CreateEntry
 func (UnimplementedSeaweedFilerServer) UpdateEntry(context.Context, *UpdateEntryRequest) (*UpdateEntryResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method UpdateEntry not implemented")
 }
+func (UnimplementedSeaweedFilerServer) FinalizeVersionedWrite(context.Context, *FinalizeVersionedWriteRequest) (*FinalizeVersionedWriteResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method FinalizeVersionedWrite not implemented")
+}
 func (UnimplementedSeaweedFilerServer) TouchAccessTime(context.Context, *TouchAccessTimeRequest) (*TouchAccessTimeResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method TouchAccessTime not implemented")
 }
@@ -669,6 +685,24 @@ func _SeaweedFiler_UpdateEntry_Handler(srv interface{}, ctx context.Context, dec
 	return interceptor(ctx, in, info, handler)
 }
 
+func _SeaweedFiler_FinalizeVersionedWrite_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(FinalizeVersionedWriteRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SeaweedFilerServer).FinalizeVersionedWrite(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: SeaweedFiler_FinalizeVersionedWrite_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SeaweedFilerServer).FinalizeVersionedWrite(ctx, req.(*FinalizeVersionedWriteRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _SeaweedFiler_TouchAccessTime_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(TouchAccessTimeRequest)
 	if err := dec(in); err != nil {
@@ -1117,6 +1151,10 @@ var SeaweedFiler_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "UpdateEntry",
 			Handler:    _SeaweedFiler_UpdateEntry_Handler,
 		},
+		{
+			MethodName: "FinalizeVersionedWrite",
+			Handler:    _SeaweedFiler_FinalizeVersionedWrite_Handler,
+		},
 		{
 			MethodName: "TouchAccessTime",
 			Handler:    _SeaweedFiler_TouchAccessTime_Handler,

weed/s3api/s3api_object_handlers_copy_part_sse.go

@@ -397,7 +397,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption(
 	filePath := s3a.genPartUploadPath(dstBucket, uploadID, partID)
 	// Copy-part is an MPU part write under .uploads/<id>/<n>; lifecycle
 	// TTL only applies to the eventual completed object. Pass 0.
-	tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil)
+	tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false)
 	if code != s3err.ErrNone {
 		return "", SSEResponseMetadata{}, code
 	}

weed/s3api/s3api_object_handlers_multipart.go

@@ -457,7 +457,7 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 	// transient .uploads/<id>/<n> path, and a part write would otherwise
 	// start the TTL clock before CompleteMultipartUpload ever assembled
 	// the object.
-	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil)
+	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil, false)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("PutObjectPart: putToFiler failed with error code %v for bucket=%s, object=%s, partNumber=%d",
 			errCode, bucket, object, partID)

weed/s3api/s3api_object_handlers_postpolicy.go

@@ -135,7 +135,7 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R
 	// fields and boundaries inflates ContentLength relative to the
 	// object body, which would mis-evaluate any size-filtered rule.
 	ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, fileSize)
-	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil)
+	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, fileBody, bucket, object, 1, ttlSec, nil, false)
 
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)

weed/s3api/s3api_object_handlers_put.go

@@ -297,7 +297,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			}
 
 			ttlSec := s3a.lifecycleTTLForObjectWrite(bucket, object, r.ContentLength)
-			etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil)
+			etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, object, 1, ttlSec, nil, false)
 
 			if errCode != s3err.ErrNone {
 				s3err.WriteErrorResponse(w, r, errCode)
@@ -359,7 +359,12 @@ func (s3a *S3ApiServer) withObjectWriteLock(bucket, object string, preconditionF
 // pass 0 because their own keys aren't the user-visible object the rule
 // targets and a part write would otherwise bind a TTL clock starting
 // before CompleteMultipartUpload.
-func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
+// uniqueVersionPath marks writes whose target path never collides (an S3 object
+// version file under <object>/.versions/<versionId>). For those, putToFiler
+// skips the object write lock and the gateway precondition: the afterCreate hook
+// finalizes the version atomically on the .versions owner (precondition + demote
+// + pointer flip) and rolls back the file if that fails.
+func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader io.Reader, bucket string, object string, partNumber int, lifecycleTTLSec int32, afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode, uniqueVersionPath bool) (etag string, code s3err.ErrorCode, sseMetadata SSEResponseMetadata) {
 	// NEW OPTIMIZATION: Write directly to volume servers, bypassing filer proxy
 	// This eliminates the filer proxy overhead for PUT operations
 	// Note: filePath is now passed directly instead of URL (no parsing needed)
@@ -840,29 +845,38 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 	// owner is unreachable.
 	var createCode s3err.ErrorCode
 	routed := false
-	if owner, cond, ok := s3a.routedObjectWrite(r, bucket, object, afterCreate != nil); ok {
-		req := &filer_pb.CreateEntryRequest{
-			Directory: path.Dir(filePath),
-			Entry:     entry,
-			Condition: cond,
-		}
-		resp, err := s3a.createEntryOnFiler(owner, req)
-		switch {
-		case err != nil:
-			glog.Warningf("putToFiler: routed create to %s failed for %s, falling back to lock: %v", owner, filePath, err)
-		case resp.ErrorCode != filer_pb.FilerError_OK:
-			// Map known filer error codes to the same S3 errors the lock path
-			// would produce; fall back for any code this does not recognize.
-			if code, mapped := filerErrorCodeToS3Error(resp.ErrorCode); mapped {
-				createCode, routed = code, true
-			} else {
-				glog.Warningf("putToFiler: routed create to %s returned code %v for %s, falling back to lock", owner, resp.ErrorCode, filePath)
+	if uniqueVersionPath {
+		// The version-file path never collides, so no object write lock is needed.
+		// createUnderLock creates the file and runs afterCreate, which finalizes
+		// the version (precondition + demote + pointer flip) atomically on the
+		// .versions owner and rolls back the file on failure.
+		createCode, routed = createUnderLock(), true
+	}
+	if !routed {
+		if owner, cond, ok := s3a.routedObjectWrite(r, bucket, object, afterCreate != nil); ok {
+			req := &filer_pb.CreateEntryRequest{
+				Directory: path.Dir(filePath),
+				Entry:     entry,
+				Condition: cond,
+			}
+			resp, err := s3a.createEntryOnFiler(owner, req)
+			switch {
+			case err != nil:
+				glog.Warningf("putToFiler: routed create to %s failed for %s, falling back to lock: %v", owner, filePath, err)
+			case resp.ErrorCode != filer_pb.FilerError_OK:
+				// Map known filer error codes to the same S3 errors the lock path
+				// would produce; fall back for any code this does not recognize.
+				if code, mapped := filerErrorCodeToS3Error(resp.ErrorCode); mapped {
+					createCode, routed = code, true
+				} else {
+					glog.Warningf("putToFiler: routed create to %s returned code %v for %s, falling back to lock", owner, resp.ErrorCode, filePath)
+				}
+			case resp.Error != "":
+				// In-band error without a code: fall back so the lock path maps it.
+				glog.Warningf("putToFiler: routed create to %s returned %q for %s, falling back to lock", owner, resp.Error, filePath)
+			default:
+				entryCreated, createCode, routed = true, s3err.ErrNone, true
 			}
-		case resp.Error != "":
-			// In-band error without a code: fall back so the lock path maps it.
-			glog.Warningf("putToFiler: routed create to %s returned %q for %s, falling back to lock", owner, resp.Error, filePath)
-		default:
-			entryCreated, createCode, routed = true, s3err.ErrNone, true
 		}
 	}
 	if !routed {
@@ -1323,7 +1337,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
 			glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
 		}
 		return s3err.ErrNone
-	})
+	}, false)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
 		return "", errCode, SSEResponseMetadata{}
@@ -1489,13 +1503,13 @@ func (s3a *S3ApiServer) putVersionedObject(r *http.Request, bucket, object strin
 	// Versioned bucket: resolver returns 0 by construction. Pass 0
 	// 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, func(versionEntry *filer_pb.Entry) s3err.ErrorCode {
-		if err := s3a.updateLatestVersionInDirectory(bucket, normalizedObject, versionId, versionFileName, versionEntry); err != nil {
-			glog.Errorf("putVersionedObject: failed to update latest version in directory: %v", err)
-			return s3err.ErrInternalError
-		}
-		return s3err.ErrNone
-	})
+	//
+	// The version file lives at a unique <object>/.versions/<versionId> path, so
+	// when the .versions owner is known the finalize (precondition + demote +
+	// pointer flip) routes there to run atomically under one local lock, and
+	// putToFiler skips the object write lock. Otherwise it stays on the lock path.
+	finalize, routedVersioned := s3a.versionedAfterCreate(r, bucket, normalizedObject, versionId, versionFileName)
+	etag, errCode, sseMetadata = s3a.putToFiler(r, versionFilePath, body, bucket, normalizedObject, 1, 0, finalize, routedVersioned)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("putVersionedObject: failed to upload version: %v", errCode)
 		return "", "", errCode, SSEResponseMetadata{}

weed/s3api/s3api_object_versioned_finalize.go

@@ -0,0 +1,132 @@
+package s3api
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"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"
+)
+
+// objectWriteOwner returns the filer that owns all of an object's writes, or ""
+// when no ring view is available. It hashes the object key with the same prefix
+// the non-versioned path uses (routedObjectOwner), so normal, suspended, and
+// versioned writes to the same object all resolve to one owner and serialize on
+// the same lock — regardless of versioning state. The whole object subtree
+// (object, .versions dir, version files) belongs to this owner.
+func (s3a *S3ApiServer) objectWriteOwner(bucket, object string) pb.ServerAddress {
+	if s3a.objectWriteLockClient == nil {
+		return ""
+	}
+	return s3a.objectWriteLockClient.PrimaryForKey("s3.object.write:" + s3a.toFilerPath(bucket, object))
+}
+
+// versionedPointerExtended computes the Extended keys to set and to clear on the
+// .versions directory entry for a new latest version, mirroring
+// setCachedListMetadata (clear cached fields, then set id/filename + cached
+// size/mtime/etag/owner/delete-marker). The filer applies deletes before sets.
+func versionedPointerExtended(versionId, versionFileName string, versionEntry *filer_pb.Entry) (set map[string][]byte, del []string) {
+	set = map[string][]byte{
+		s3_constants.ExtLatestVersionIdKey:       []byte(versionId),
+		s3_constants.ExtLatestVersionFileNameKey: []byte(versionFileName),
+	}
+	if versionEntry.Attributes != nil {
+		set[s3_constants.ExtLatestVersionSizeKey] = []byte(strconv.FormatUint(versionEntry.Attributes.FileSize, 10))
+		set[s3_constants.ExtLatestVersionMtimeKey] = []byte(strconv.FormatInt(versionEntry.Attributes.Mtime, 10))
+	}
+	isDeleteMarker := []byte("false")
+	if versionEntry.Extended != nil {
+		if v, ok := versionEntry.Extended[s3_constants.ExtETagKey]; ok {
+			set[s3_constants.ExtLatestVersionETagKey] = v
+		}
+		if v, ok := versionEntry.Extended[s3_constants.ExtAmzOwnerKey]; ok {
+			set[s3_constants.ExtLatestVersionOwnerKey] = v
+		}
+		if v, ok := versionEntry.Extended[s3_constants.ExtDeleteMarkerKey]; ok {
+			isDeleteMarker = v
+		}
+	}
+	set[s3_constants.ExtLatestVersionIsDeleteMarker] = isDeleteMarker
+	del = []string{
+		s3_constants.ExtLatestVersionSizeKey,
+		s3_constants.ExtLatestVersionMtimeKey,
+		s3_constants.ExtLatestVersionETagKey,
+		s3_constants.ExtLatestVersionOwnerKey,
+		s3_constants.ExtLatestVersionIsDeleteMarker,
+	}
+	return set, del
+}
+
+// routedVersionedFinalize finalizes a versioned write by routing the precondition
+// check, the demote of the previous latest, and the pointer flip to the owner of
+// the object's .versions directory, where they run atomically under one local
+// lock. The caller (putToFiler with uniqueVersionPath) has already created the
+// version file and skips the object write lock and gateway precondition. A
+// transport/in-band error maps to InternalError (the client retries); there is
+// no silent non-atomic fallback.
+func (s3a *S3ApiServer) routedVersionedFinalize(owner pb.ServerAddress, bucket, object, versionId, versionFileName string, versionEntry *filer_pb.Entry, cond *filer_pb.WriteCondition) s3err.ErrorCode {
+	set, del := versionedPointerExtended(versionId, versionFileName, versionEntry)
+	req := &filer_pb.FinalizeVersionedWriteRequest{
+		LockKey:               s3a.toFilerPath(bucket, object),
+		VersionsDir:           s3a.toFilerPath(bucket, object+s3_constants.VersionsFolder),
+		SetExtended:           set,
+		DeleteExtended:        del,
+		PriorLatestKey:        s3_constants.ExtLatestVersionFileNameKey,
+		NoncurrentSinceKey:    s3_constants.ExtNoncurrentSinceNsKey,
+		NoncurrentSinceNs:     time.Now().UnixNano(),
+		Condition:             cond,
+		LatestEtagKey:         s3_constants.ExtLatestVersionETagKey,
+		LatestDeleteMarkerKey: s3_constants.ExtLatestVersionIsDeleteMarker,
+	}
+	var resp *filer_pb.FinalizeVersionedWriteResponse
+	err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		var e error
+		resp, e = client.FinalizeVersionedWrite(context.Background(), req)
+		return e
+	})
+	switch {
+	case err != nil:
+		glog.Errorf("routedVersionedFinalize: %s/%s on %s: %v", bucket, object, owner, err)
+		return s3err.ErrInternalError
+	case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED:
+		return s3err.ErrPreconditionFailed
+	case resp.ErrorCode != filer_pb.FilerError_OK:
+		if code, ok := filerErrorCodeToS3Error(resp.ErrorCode); ok {
+			return code
+		}
+		glog.Errorf("routedVersionedFinalize: %s/%s unexpected code %v", bucket, object, resp.ErrorCode)
+		return s3err.ErrInternalError
+	case resp.Error != "":
+		glog.Errorf("routedVersionedFinalize: %s/%s: %s", bucket, object, resp.Error)
+		return s3err.ErrInternalError
+	default:
+		return s3err.ErrNone
+	}
+}
+
+// versionedAfterCreate returns the putToFiler afterCreate hook for a versioned
+// write: the routed atomic finalize when the .versions owner is known and the
+// precondition reduces to a single primitive, otherwise the existing lock-based
+// updateLatestVersionInDirectory path. The bool reports whether the routed path
+// was selected, which the caller passes as putToFiler's uniqueVersionPath.
+func (s3a *S3ApiServer) versionedAfterCreate(r *http.Request, bucket, object, versionId, versionFileName string) (func(*filer_pb.Entry) s3err.ErrorCode, bool) {
+	owner := s3a.objectWriteOwner(bucket, object)
+	cond, condOk := buildWriteCondition(r)
+	if owner != "" && condOk {
+		return func(versionEntry *filer_pb.Entry) s3err.ErrorCode {
+			return s3a.routedVersionedFinalize(owner, bucket, object, versionId, versionFileName, versionEntry, cond)
+		}, true
+	}
+	return func(versionEntry *filer_pb.Entry) s3err.ErrorCode {
+		if err := s3a.updateLatestVersionInDirectory(bucket, object, versionId, versionFileName, versionEntry); err != nil {
+			glog.Errorf("putVersionedObject: failed to update latest version in directory: %v", err)
+			return s3err.ErrInternalError
+		}
+		return s3err.ErrNone
+	}, false
+}

weed/server/filer_grpc_server.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -276,6 +277,122 @@ func normalizeETag(etag string) string {
 	return strings.Trim(etag, `"`)
 }
 
+// versionedConditionSatisfied evaluates a precondition against the current
+// latest version, derived from the .versions directory entry's cached metadata
+// (whether a latest exists and isn't a delete marker, and its ETag). It mirrors
+// writeConditionSatisfied but reads the cached pointer fields instead of a plain
+// object entry.
+func versionedConditionSatisfied(req *filer_pb.FinalizeVersionedWriteRequest, versionsEntry *filer.Entry) bool {
+	exists := false
+	etag := ""
+	if versionsEntry != nil && versionsEntry.Extended != nil {
+		latestFile := string(versionsEntry.Extended[req.PriorLatestKey])
+		isDeleteMarker := req.LatestDeleteMarkerKey != "" && string(versionsEntry.Extended[req.LatestDeleteMarkerKey]) == "true"
+		exists = latestFile != "" && !isDeleteMarker
+		if req.LatestEtagKey != "" {
+			etag = normalizeETag(string(versionsEntry.Extended[req.LatestEtagKey]))
+		}
+	}
+	want := normalizeETag(req.Condition.Etag)
+	switch req.Condition.Kind {
+	case filer_pb.WriteCondition_IF_NOT_EXISTS:
+		return !exists
+	case filer_pb.WriteCondition_IF_EXISTS:
+		return exists
+	case filer_pb.WriteCondition_IF_ETAG_MATCH:
+		return exists && etag == want
+	case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH:
+		return !exists || etag != want
+	default:
+		return true
+	}
+}
+
+// FinalizeVersionedWrite performs an S3 versioned-write finalize atomically on
+// the owner of the object's .versions directory: it creates the new version
+// file, stamps the previous latest as noncurrent (before the pointer flip so the
+// lifecycle router observes it), then merges the latest-pointer / cached
+// metadata into the .versions directory entry. A single exclusive lock on the
+// .versions directory serializes the whole sequence against concurrent versioned
+// writes to the same object, replacing the distributed lock. Key names come from
+// the caller so the filer carries no S3 versioning semantics.
+func (fs *FilerServer) FinalizeVersionedWrite(ctx context.Context, req *filer_pb.FinalizeVersionedWriteRequest) (*filer_pb.FinalizeVersionedWriteResponse, error) {
+	resp := &filer_pb.FinalizeVersionedWriteResponse{}
+	versionsDir := util.FullPath(req.VersionsDir)
+
+	// Serialize on the object key — the same lock all of this object's writes
+	// (normal, suspended, versioned) share — not the .versions directory, so a
+	// versioned write and a non-versioned write to the same object can't run on
+	// different owners during a versioning-state change. Fall back to the
+	// .versions directory for older callers that don't set lock_key.
+	lockKey := util.FullPath(req.LockKey)
+	if lockKey == "" {
+		lockKey = versionsDir
+	}
+	lock := fs.entryLockTable.AcquireLock("FinalizeVersionedWrite", lockKey, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(lockKey, lock)
+
+	// The caller already created the new version file under this .versions
+	// directory, so it exists; read the directory entry that holds the pointer.
+	versionsEntry, err := fs.filer.FindEntry(ctx, versionsDir)
+	if err != nil {
+		resp.Error = err.Error()
+		return resp, nil
+	}
+
+	// 1. Evaluate the optional precondition against the current latest version
+	// (the .versions directory's cached pointer metadata) before the flip, so the
+	// check and the flip are atomic under this lock.
+	if req.Condition != nil && req.Condition.Kind != filer_pb.WriteCondition_NONE {
+		if !versionedConditionSatisfied(req, versionsEntry) {
+			glog.V(3).InfofCtx(ctx, "FinalizeVersionedWrite %s: precondition %v failed", versionsDir, req.Condition.Kind)
+			resp.Error = "precondition failed"
+			resp.ErrorCode = filer_pb.FilerError_PRECONDITION_FAILED
+			return resp, nil
+		}
+	}
+
+	// 2. Stamp the previously-latest version as noncurrent BEFORE the pointer
+	// flip (so the lifecycle router observes it). Best-effort.
+	newFileName := string(req.SetExtended[req.PriorLatestKey])
+	if req.NoncurrentSinceNs > 0 && req.PriorLatestKey != "" && req.NoncurrentSinceKey != "" {
+		prior := string(versionsEntry.Extended[req.PriorLatestKey])
+		if prior != "" && prior != newFileName {
+			demotedPath := util.NewFullPath(req.VersionsDir, prior)
+			if demoted, derr := fs.filer.FindEntry(ctx, demotedPath); derr == nil {
+				if demoted.Extended == nil {
+					demoted.Extended = make(map[string][]byte)
+				}
+				demoted.Extended[req.NoncurrentSinceKey] = []byte(strconv.FormatInt(req.NoncurrentSinceNs, 10))
+				if uerr := fs.filer.UpdateEntry(ctx, demoted, demoted); uerr != nil {
+					glog.Warningf("FinalizeVersionedWrite: demote stamp %s: %v", demotedPath, uerr)
+				}
+			} else {
+				glog.V(2).InfofCtx(ctx, "FinalizeVersionedWrite: demote target %s not found: %v", demotedPath, derr)
+			}
+		}
+	}
+
+	// 3. Flip the latest pointer: merge the requested Extended keys into the
+	// .versions directory entry and write it back (emits the meta-log event the
+	// lifecycle router consumes).
+	if versionsEntry.Extended == nil {
+		versionsEntry.Extended = make(map[string][]byte)
+	}
+	for _, k := range req.DeleteExtended {
+		delete(versionsEntry.Extended, k)
+	}
+	for k, v := range req.SetExtended {
+		versionsEntry.Extended[k] = v
+	}
+	if err := fs.filer.CreateEntry(ctx, versionsEntry, false, req.IsFromOtherCluster, req.Signatures, true, fs.filer.MaxFilenameLength); err != nil {
+		resp.Error = err.Error()
+		return resp, nil
+	}
+
+	return resp, nil
+}
+
 func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
 
 	glog.V(4).InfofCtx(ctx, "UpdateEntry %v", req)

weed/server/filer_grpc_server_versioned_finalize_test.go

@@ -0,0 +1,95 @@
+package weed_server
+
+import (
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func newFinalizeTestServer(store *renameTestStore) *FilerServer {
+	f := newRenameTestFiler(store)
+	f.DirBucketsPath = "/buckets"
+	return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}
+}
+
+func seedVersions(store *renameTestStore, latestFile, latestEtag string) {
+	store.entries["/buckets/b/obj/.versions"] = &filer.Entry{
+		FullPath: "/buckets/b/obj/.versions",
+		Attr:     filer.Attr{Inode: 10, Mode: 0755 | (1 << 31)},
+		Extended: map[string][]byte{
+			s3_constants.ExtLatestVersionFileNameKey: []byte(latestFile),
+			s3_constants.ExtLatestVersionETagKey:     []byte(latestEtag),
+		},
+	}
+	store.entries["/buckets/b/obj/.versions/"+latestFile] = &filer.Entry{
+		FullPath: util.FullPath("/buckets/b/obj/.versions/" + latestFile),
+		Attr:     filer.Attr{Inode: 11},
+		Extended: map[string][]byte{},
+	}
+}
+
+func finalizeReq(cond *filer_pb.WriteCondition) *filer_pb.FinalizeVersionedWriteRequest {
+	return &filer_pb.FinalizeVersionedWriteRequest{
+		VersionsDir: "/buckets/b/obj/.versions",
+		SetExtended: map[string][]byte{
+			s3_constants.ExtLatestVersionFileNameKey: []byte("v2.ver"),
+			s3_constants.ExtLatestVersionIdKey:       []byte("vid2"),
+		},
+		DeleteExtended:        []string{s3_constants.ExtLatestVersionETagKey},
+		PriorLatestKey:        s3_constants.ExtLatestVersionFileNameKey,
+		NoncurrentSinceKey:    s3_constants.ExtNoncurrentSinceNsKey,
+		NoncurrentSinceNs:     77777,
+		Condition:             cond,
+		LatestEtagKey:         s3_constants.ExtLatestVersionETagKey,
+		LatestDeleteMarkerKey: s3_constants.ExtLatestVersionIsDeleteMarker,
+	}
+}
+
+// The finalize flips the latest pointer and stamps the previously-latest version
+// noncurrent, atomically under the .versions lock.
+func TestFinalizeVersionedWriteFlipAndDemote(t *testing.T) {
+	store := newRenameTestStore()
+	seedVersions(store, "v1.ver", "etag1")
+	fs := newFinalizeTestServer(store)
+
+	resp, err := fs.FinalizeVersionedWrite(context.Background(), finalizeReq(&filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_NONE}))
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if resp.Error != "" {
+		t.Fatalf("unexpected resp error: %s", resp.Error)
+	}
+
+	if got := string(store.entries["/buckets/b/obj/.versions"].Extended[s3_constants.ExtLatestVersionFileNameKey]); got != "v2.ver" {
+		t.Errorf("latest pointer not flipped: %q", got)
+	}
+	if got := string(store.entries["/buckets/b/obj/.versions/v1.ver"].Extended[s3_constants.ExtNoncurrentSinceNsKey]); got != "77777" {
+		t.Errorf("previous latest not demoted: %q", got)
+	}
+}
+
+// A failing precondition is reported and nothing is flipped or demoted.
+func TestFinalizeVersionedWritePreconditionFailed(t *testing.T) {
+	store := newRenameTestStore()
+	seedVersions(store, "v1.ver", "etag1")
+	fs := newFinalizeTestServer(store)
+
+	// IF_NOT_EXISTS against an existing latest must fail.
+	resp, err := fs.FinalizeVersionedWrite(context.Background(), finalizeReq(&filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}))
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
+		t.Fatalf("want PRECONDITION_FAILED, got %v (%q)", resp.ErrorCode, resp.Error)
+	}
+	if got := string(store.entries["/buckets/b/obj/.versions"].Extended[s3_constants.ExtLatestVersionFileNameKey]); got != "v1.ver" {
+		t.Errorf("pointer should be unchanged on precondition failure, got %q", got)
+	}
+	if _, stamped := store.entries["/buckets/b/obj/.versions/v1.ver"].Extended[s3_constants.ExtNoncurrentSinceNsKey]; stamped {
+		t.Errorf("previous latest should not be demoted on precondition failure")
+	}
+}