s3: route non-conditional suspended DELETE off the DLM

A suspended DELETE removes the null version (the main object entry) and adds a
delete marker. Now that suspended PUT routes on the object key, a DELETE left on
the distributed lock wouldn't serialize against it (the per-path entry lock and
the distributed lock are independent). Route it on the object key too.

FinalizeVersionedWrite gains an optional delete_path it removes under the same
lock before flipping the pointer, so "delete null + add marker" is one atomic
step on the owner. createDeleteMarker threads it through (empty for the
versioning-enabled marker path). The DeleteObjectHandler routes a no-versionId
suspended delete when there is no If-Match — that condition targets the main
object, not the .versions pointer the op evaluates, so a conditional suspended
delete stays on the lock path. Object-lock can't apply (it requires versioning
enabled, never suspended).

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) {
     }
 
@@ -222,6 +225,29 @@ message CreateEntryRequest {
     bool is_from_other_cluster = 4;
     repeated int32 signatures = 5;
     bool skip_check_parent_directory = 6;
+    // Optional precondition evaluated against the current entry atomically with
+    // the write, under the filer's per-path lock. The caller must route the
+    // key's writes to this entry's owner filer for the check to be authoritative.
+    WriteCondition condition = 7;
+}
+
+// WriteCondition is a single precondition the filer evaluates against the
+// existing entry before writing. The client reduces request semantics (e.g.
+// RFC 7232 precedence) to one primitive; the filer just compares. A failed
+// condition returns FilerError PRECONDITION_FAILED.
+message WriteCondition {
+    enum Kind {
+        NONE                = 0; // unconditional
+        IF_NOT_EXISTS       = 1; // fail if the entry exists (If-None-Match: *)
+        IF_EXISTS           = 2; // fail if the entry is absent (If-Match: *)
+        IF_ETAG_MATCH       = 3; // fail if absent or etag != expected (If-Match: <etag>)
+        IF_ETAG_NOT_MATCH   = 4; // fail if present and etag == expected (If-None-Match: <etag>)
+        IF_UNMODIFIED_SINCE = 5; // fail if present and mtime > unix_time
+        IF_MODIFIED_SINCE   = 6; // fail if present and mtime <= unix_time
+    }
+    Kind kind = 1;
+    string etag = 2;      // expected ETag for IF_ETAG_* kinds
+    int64 unix_time = 3;  // bound (unix seconds) for IF_*_SINCE kinds
 }
 
 // Structured error codes for filer entry operations.
@@ -233,6 +259,7 @@ enum FilerError {
     EXISTING_IS_DIRECTORY = 3; // cannot overwrite directory with file
     EXISTING_IS_FILE      = 4; // cannot overwrite file with directory
     ENTRY_ALREADY_EXISTS  = 5; // O_EXCL and entry already exists
+    PRECONDITION_FAILED   = 6; // WriteCondition not satisfied
 }
 
 message CreateEntryResponse {
@@ -252,6 +279,40 @@ 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 delete_path = 14;             // optional entry to delete under the lock first (suspended delete: the null version)
+    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;
@@ -280,11 +341,16 @@ message DeleteEntryRequest {
     bool is_from_other_cluster = 7;
     repeated int32 signatures = 8;
     int64 if_not_modified_after = 9;
+    // Optional precondition evaluated against the current entry atomically with
+    // the delete, under the filer's per-path lock. Only honored for a
+    // non-recursive single-entry delete routed to the entry's owner filer.
+    WriteCondition condition = 10;
 }
 
 message DeleteEntryResponse {
     string error = 1;
     SubscribeMetadataResponse metadata_event = 2;
+    FilerError error_code = 3; // machine-readable error code (e.g. PRECONDITION_FAILED)
 }
 
 message AtomicRenameEntryRequest {
@@ -421,6 +487,7 @@ message SubscribeMetadataRequest {
     repeated string directories = 10;  // exact directory to watch
     bool client_supports_batching = 11; // client can unpack SubscribeMetadataResponse.events
     bool client_supports_metadata_chunks = 12; // client can read log file chunks from volume servers
+    bool client_supports_idle_heartbeat = 13; // server may send empty responses carrying the current time while the client is caught up
 }
 message SubscribeMetadataResponse {
     string directory = 1;

weed/cluster/lock_client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -19,6 +20,14 @@ type LockClient struct {
 	maxLockDuration time.Duration
 	sleepDuration   time.Duration
 	seedFiler       pb.ServerAddress
+
+	// ring is an optional client-side view of the filer lock hash ring. When
+	// populated, a new lock starts at the key's primary filer instead of the
+	// seed filer, avoiding the seed->primary forward hop. A stale view stays
+	// correct: the filer forwards to the real primary as a fallback.
+	ringMu      sync.RWMutex
+	ring        *lock_manager.HashRing
+	ringVersion int64
 }
 
 func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *LockClient {
@@ -30,6 +39,49 @@ func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *
 	}
 }
 
+// SetRing updates the client-side view of the filer lock ring. It mirrors the
+// master's LockRingUpdate broadcasts so the client computes the same primary
+// the filers do. Updates older than the current version are ignored to tolerate
+// reordered broadcasts; version 0 is always accepted (bootstrap).
+func (lc *LockClient) SetRing(servers []pb.ServerAddress, version int64) {
+	lc.ringMu.Lock()
+	defer lc.ringMu.Unlock()
+	if version != 0 && version < lc.ringVersion {
+		return
+	}
+	lc.ringVersion = version
+	if lc.ring == nil {
+		lc.ring = lock_manager.NewHashRing(lock_manager.DefaultVnodeCount)
+	}
+	lc.ring.SetServers(servers)
+}
+
+// hostForKey returns the filer that should own key per the current ring view,
+// falling back to the seed filer when no view has been received yet.
+func (lc *LockClient) hostForKey(key string) pb.ServerAddress {
+	lc.ringMu.RLock()
+	defer lc.ringMu.RUnlock()
+	if lc.ring == nil {
+		return lc.seedFiler
+	}
+	if primary := lc.ring.GetPrimary(key); primary != "" {
+		return primary
+	}
+	return lc.seedFiler
+}
+
+// PrimaryForKey returns the filer that owns key per the current ring view, or
+// "" when no ring view has been received yet. Callers use it to route a key's
+// operations to a single owner so that owner's local lock serializes them.
+func (lc *LockClient) PrimaryForKey(key string) pb.ServerAddress {
+	lc.ringMu.RLock()
+	defer lc.ringMu.RUnlock()
+	if lc.ring == nil {
+		return ""
+	}
+	return lc.ring.GetPrimary(key)
+}
+
 type LiveLock struct {
 	key                 string
 	renewToken          string
@@ -51,7 +103,7 @@ type LiveLock struct {
 func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLock) {
 	lock = &LiveLock{
 		key:            key,
-		hostFiler:      lc.seedFiler,
+		hostFiler:      lc.hostForKey(key),
 		cancelCh:       make(chan struct{}),
 		expireAtNs:     time.Now().Add(5 * time.Second).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,
@@ -72,7 +124,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D
 	}
 	lock := &LiveLock{
 		key:            key,
-		hostFiler:      lc.seedFiler,
+		hostFiler:      lc.hostForKey(key),
 		cancelCh:       make(chan struct{}),
 		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,
@@ -110,7 +162,7 @@ func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.D
 func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string), lockTTL time.Duration) (lock *LiveLock) {
 	lock = &LiveLock{
 		key:            key,
-		hostFiler:      lc.seedFiler,
+		hostFiler:      lc.hostForKey(key),
 		cancelCh:       make(chan struct{}),
 		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
 		grpcDialOption: lc.grpcDialOption,

weed/cluster/lock_client_test.go

@@ -0,0 +1,73 @@
+package cluster
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+// The gateway must resolve a lock key to the same primary the filers do,
+// otherwise it dials the wrong filer and the lock still gets forwarded. Both
+// sides use the same HashRing over the same server set, so for every key the
+// client's hostForKey must equal the filer ring's GetPrimary.
+func TestLockClientHostMatchesFilerRing(t *testing.T) {
+	servers := []pb.ServerAddress{
+		"filer-a:8888", "filer-b:8888", "filer-c:8888", "filer-d:8888",
+	}
+
+	filerRing := lock_manager.NewHashRing(lock_manager.DefaultVnodeCount)
+	filerRing.SetServers(servers)
+
+	lc := NewLockClient(nil, "seed:8888")
+	lc.SetRing(servers, 1)
+
+	for _, key := range []string{
+		"s3.object.write:/buckets/b/obj-0",
+		"s3.object.write:/buckets/b/obj-1",
+		"s3.object.write:/buckets/b/obj-2",
+		"s3.object.write:/buckets/gosbench-0/w0obj-kilo-0877",
+		"some/other/key",
+	} {
+		if got, want := lc.hostForKey(key), filerRing.GetPrimary(key); got != want {
+			t.Errorf("key %q: client host %q != filer primary %q", key, got, want)
+		}
+	}
+}
+
+// Without a ring view, the client falls back to the seed filer (which the filer
+// forwards from), preserving the pre-optimization behavior.
+func TestLockClientHostFallsBackToSeed(t *testing.T) {
+	lc := NewLockClient(nil, "seed:8888")
+	if got := lc.hostForKey("any-key"); got != "seed:8888" {
+		t.Errorf("expected seed fallback, got %q", got)
+	}
+
+	// An empty ring (no members yet) also falls back to the seed.
+	lc.SetRing(nil, 1)
+	if got := lc.hostForKey("any-key"); got != "seed:8888" {
+		t.Errorf("expected seed fallback on empty ring, got %q", got)
+	}
+}
+
+// A stale (older-version) update must not regress a newer ring view, while
+// version 0 always applies as a bootstrap.
+func TestLockClientSetRingVersionGuard(t *testing.T) {
+	lc := NewLockClient(nil, "seed:8888")
+
+	newer := []pb.ServerAddress{"filer-a:8888", "filer-b:8888"}
+	lc.SetRing(newer, 10)
+	primaryAt10 := lc.hostForKey("k")
+
+	// Older version is ignored.
+	lc.SetRing([]pb.ServerAddress{"filer-z:8888"}, 5)
+	if got := lc.hostForKey("k"); got != primaryAt10 {
+		t.Errorf("stale update applied: host changed to %q", got)
+	}
+
+	// version 0 is always accepted.
+	lc.SetRing([]pb.ServerAddress{"filer-z:8888"}, 0)
+	if got := lc.hostForKey("k"); got != "filer-z:8888" {
+		t.Errorf("bootstrap update not applied, got %q", got)
+	}
+}

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) {
     }
 
@@ -222,6 +225,29 @@ message CreateEntryRequest {
     bool is_from_other_cluster = 4;
     repeated int32 signatures = 5;
     bool skip_check_parent_directory = 6;
+    // Optional precondition evaluated against the current entry atomically with
+    // the write, under the filer's per-path lock. The caller must route the
+    // key's writes to this entry's owner filer for the check to be authoritative.
+    WriteCondition condition = 7;
+}
+
+// WriteCondition is a single precondition the filer evaluates against the
+// existing entry before writing. The client reduces request semantics (e.g.
+// RFC 7232 precedence) to one primitive; the filer just compares. A failed
+// condition returns FilerError PRECONDITION_FAILED.
+message WriteCondition {
+    enum Kind {
+        NONE                = 0; // unconditional
+        IF_NOT_EXISTS       = 1; // fail if the entry exists (If-None-Match: *)
+        IF_EXISTS           = 2; // fail if the entry is absent (If-Match: *)
+        IF_ETAG_MATCH       = 3; // fail if absent or etag != expected (If-Match: <etag>)
+        IF_ETAG_NOT_MATCH   = 4; // fail if present and etag == expected (If-None-Match: <etag>)
+        IF_UNMODIFIED_SINCE = 5; // fail if present and mtime > unix_time
+        IF_MODIFIED_SINCE   = 6; // fail if present and mtime <= unix_time
+    }
+    Kind kind = 1;
+    string etag = 2;      // expected ETag for IF_ETAG_* kinds
+    int64 unix_time = 3;  // bound (unix seconds) for IF_*_SINCE kinds
 }
 
 // Structured error codes for filer entry operations.
@@ -233,6 +259,7 @@ enum FilerError {
     EXISTING_IS_DIRECTORY = 3; // cannot overwrite directory with file
     EXISTING_IS_FILE      = 4; // cannot overwrite file with directory
     ENTRY_ALREADY_EXISTS  = 5; // O_EXCL and entry already exists
+    PRECONDITION_FAILED   = 6; // WriteCondition not satisfied
 }
 
 message CreateEntryResponse {
@@ -252,6 +279,40 @@ 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 delete_path = 14;             // optional entry to delete under the lock first (suspended delete: the null version)
+    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;
@@ -280,11 +341,16 @@ message DeleteEntryRequest {
     bool is_from_other_cluster = 7;
     repeated int32 signatures = 8;
     int64 if_not_modified_after = 9;
+    // Optional precondition evaluated against the current entry atomically with
+    // the delete, under the filer's per-path lock. Only honored for a
+    // non-recursive single-entry delete routed to the entry's owner filer.
+    WriteCondition condition = 10;
 }
 
 message DeleteEntryResponse {
     string error = 1;
     SubscribeMetadataResponse metadata_event = 2;
+    FilerError error_code = 3; // machine-readable error code (e.g. PRECONDITION_FAILED)
 }
 
 message AtomicRenameEntryRequest {

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/filer_multipart.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 )
 
@@ -553,9 +554,16 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 	uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
 	entryName, dirName := s3a.getEntryNameAndDir(input)
 	var completionState *multipartCompletionState
-	finalizeCode := s3a.withObjectWriteLock(*input.Bucket, *input.Key, func() s3err.ErrorCode {
-		return s3a.checkConditionalHeaders(r, *input.Bucket, *input.Key)
-	}, func() s3err.ErrorCode {
+	// Non-versioned destination: the final write is a single CreateEntry, so route
+	// it (with the precondition) to the object's owner and skip the distributed
+	// lock. Assembly and the idempotency replay below are read-only and safe
+	// outside any lock. Versioned/suspended completions keep the lock path.
+	mpuCond, mpuCondOk := buildWriteCondition(r)
+	var nvOwner pb.ServerAddress
+	if vs, vErr := s3a.getVersioningState(*input.Bucket); vErr == nil && vs == "" && mpuCondOk {
+		nvOwner, _ = s3a.routedObjectOwner(*input.Bucket, *input.Key)
+	}
+	finalizeBody := func() s3err.ErrorCode {
 		var prepCode s3err.ErrorCode
 		completionState, output, prepCode = s3a.prepareMultipartCompletionState(r, input, uploadDirectory, entryName, dirName, completedPartNumbers, completedPartMap, maxPartNo)
 		if prepCode != s3err.ErrNone || output != nil {
@@ -739,7 +747,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 		}
 
 		// For non-versioned buckets, create main object file
-		if err := s3a.mkFile(dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
+		nvModifier := func(entry *filer_pb.Entry) {
 			if entry.Extended == nil {
 				entry.Extended = make(map[string][]byte)
 			}
@@ -787,7 +795,14 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 			if completionState.entityWithTtl {
 				entry.Extended[s3_constants.SeaweedFSExpiresS3] = []byte("true")
 			}
-		}); err != nil {
+		}
+		if nvOwner != "" {
+			// Routed path: the owner evaluates the precondition and writes under
+			// its local lock, so the lock wrapper below is skipped.
+			if code := s3a.mkFileRouted(nvOwner, dirName, entryName, completionState.finalParts, mpuCond, nvModifier); code != s3err.ErrNone {
+				return code
+			}
+		} else if err := s3a.mkFile(dirName, entryName, completionState.finalParts, nvModifier); err != nil {
 			glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
 			return s3err.ErrInternalError
 		}
@@ -802,7 +817,17 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl
 			ChecksumValue:      completionState.checksumValue,
 		}
 		return s3err.ErrNone
-	})
+	}
+	var finalizeCode s3err.ErrorCode
+	if nvOwner != "" {
+		// Routed non-versioned completion: no distributed lock, no gateway
+		// precondition here — the routed write carries the condition.
+		finalizeCode = finalizeBody()
+	} else {
+		finalizeCode = s3a.withObjectWriteLock(*input.Bucket, *input.Key, func() s3err.ErrorCode {
+			return s3a.checkConditionalHeaders(r, *input.Bucket, *input.Key)
+		}, finalizeBody)
+	}
 	if finalizeCode != s3err.ErrNone {
 		return nil, finalizeCode
 	}

weed/s3api/s3api_internal_lifecycle.go

@@ -71,7 +71,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 		}
 		switch state {
 		case s3_constants.VersioningEnabled:
-			if _, err := s3a.createDeleteMarker(req.Bucket, req.ObjectPath); err != nil {
+			if _, err := s3a.createDeleteMarker(req.Bucket, req.ObjectPath, "", nil, ""); err != nil {
 				return retryLater("TRANSPORT_ERROR: createDeleteMarker: " + err.Error()), nil
 			}
 			return done(), nil
@@ -82,7 +82,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle
 					return retryLater("TRANSPORT_ERROR: deleteNullVersion: " + err.Error()), nil
 				}
 			}
-			if _, err := s3a.createDeleteMarker(req.Bucket, req.ObjectPath); err != nil {
+			if _, err := s3a.createDeleteMarker(req.Bucket, req.ObjectPath, "", nil, ""); err != nil {
 				return retryLater("TRANSPORT_ERROR: createDeleteMarker: " + err.Error()), nil
 			}
 			return done(), nil

weed/s3api/s3api_object_handlers_copy.go

@@ -18,6 +18,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"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"
@@ -182,7 +183,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 			}
 			updatedEntry.Attributes.Mtime = t.Unix()
 
-			dstVersionId, etag, currentErr = s3a.finalizeCopyDestination(dstBucket, dstObject, dstVersioningState, updatedEntry)
+			dstVersionId, etag, currentErr = s3a.finalizeCopyDestination(dstBucket, dstObject, dstVersioningState, updatedEntry, "", nil)
 			if currentErr != nil {
 				return filerErrorToS3Error(currentErr)
 			}
@@ -337,16 +338,49 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 	var dstVersionId string
 	var etag string
 
-	finalizeCode := s3a.withObjectWriteLock(dstBucket, dstObject, func() s3err.ErrorCode {
-		return s3a.checkConditionalHeaders(r, dstBucket, dstObject)
-	}, func() s3err.ErrorCode {
-		var finalizeErr error
-		dstVersionId, etag, finalizeErr = s3a.finalizeCopyDestination(dstBucket, dstObject, dstVersioningState, dstEntry)
-		if finalizeErr != nil {
-			return filerErrorToS3Error(finalizeErr)
+	// Fast path: route the finalize to the destination's owner filer and skip the
+	// distributed lock. A non-versioned destination is a single CreateEntry; a
+	// versioning-enabled destination is a unique version file plus an atomic
+	// latest-pointer update on the .versions owner (FinalizeVersionedWrite).
+	// Suspended versioning (multi-version IsLatest rewrite) keeps the lock path.
+	var finalizeCode s3err.ErrorCode
+	routed := false
+	if cond, condOk := buildWriteCondition(r); condOk {
+		switch dstVersioningState {
+		case "":
+			if owner, ownerOk := s3a.routedObjectOwner(dstBucket, dstObject); ownerOk {
+				if code, e, handled := s3a.routedCopyFinalize(owner, dstBucket, dstObject, dstEntry, cond); handled {
+					finalizeCode, etag, routed = code, e, true
+				}
+			}
+		case s3_constants.VersioningEnabled:
+			if owner := s3a.objectWriteOwner(dstBucket, s3_constants.NormalizeObjectKey(dstObject)); owner != "" {
+				var finalizeErr error
+				dstVersionId, etag, finalizeErr = s3a.finalizeCopyDestination(dstBucket, dstObject, dstVersioningState, dstEntry, owner, cond)
+				switch {
+				case finalizeErr == nil:
+					finalizeCode = s3err.ErrNone
+				case errors.Is(finalizeErr, errVersionedPreconditionFailed):
+					finalizeCode = s3err.ErrPreconditionFailed
+				default:
+					finalizeCode = filerErrorToS3Error(finalizeErr)
+				}
+				routed = true
+			}
 		}
-		return s3err.ErrNone
-	})
+	}
+	if !routed {
+		finalizeCode = s3a.withObjectWriteLock(dstBucket, dstObject, func() s3err.ErrorCode {
+			return s3a.checkConditionalHeaders(r, dstBucket, dstObject)
+		}, func() s3err.ErrorCode {
+			var finalizeErr error
+			dstVersionId, etag, finalizeErr = s3a.finalizeCopyDestination(dstBucket, dstObject, dstVersioningState, dstEntry, "", nil)
+			if finalizeErr != nil {
+				return filerErrorToS3Error(finalizeErr)
+			}
+			return s3err.ErrNone
+		})
+	}
 	if finalizeCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, finalizeCode)
 		return
@@ -412,7 +446,58 @@ func copyEntryToTarget(dst, src *filer_pb.Entry) {
 	dst.WormEnforcedAtTsNs = src.WormEnforcedAtTsNs
 }
 
-func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersioningState string, dstEntry *filer_pb.Entry) (versionId string, etag string, err error) {
+// routedCopyFinalize writes a non-versioned copy destination as a single
+// CreateEntry routed to the destination key's owner filer, building the same
+// entry that finalizeCopyDestination's default branch produces via mkFile. It
+// returns handled=false to fall back to the lock path on a transport error or an
+// unmapped in-band error.
+func (s3a *S3ApiServer) routedCopyFinalize(owner pb.ServerAddress, dstBucket, dstObject string, dstEntry *filer_pb.Entry, cond *filer_pb.WriteCondition) (errCode s3err.ErrorCode, etag string, handled bool) {
+	normalizedObject := s3_constants.NormalizeObjectKey(dstObject)
+	dstPath := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(dstBucket), normalizedObject))
+	dstDir, dstName := dstPath.DirAndName()
+
+	if dstEntry.Attributes == nil {
+		dstEntry.Attributes = &filer_pb.FuseAttributes{}
+	}
+	if dstEntry.Extended == nil {
+		dstEntry.Extended = make(map[string][]byte)
+	}
+	etag = copyEntryETag(dstPath, dstEntry)
+	cleanupVersioningMetadata(dstEntry.Extended)
+	dstEntry.Extended[s3_constants.ExtETagKey] = []byte(etag)
+
+	// Build the destination entry exactly as mkFile + copyEntryToTarget would.
+	entry := &filer_pb.Entry{Name: dstName, Chunks: dstEntry.Chunks}
+	copyEntryToTarget(entry, dstEntry)
+
+	resp, err := s3a.createEntryOnFiler(owner, &filer_pb.CreateEntryRequest{
+		Directory: dstDir,
+		Entry:     entry,
+		Condition: cond,
+	})
+	switch {
+	case err != nil:
+		glog.Warningf("CopyObjectHandler: routed finalize to %s failed for %s/%s, falling back to lock: %v", owner, dstBucket, dstObject, err)
+		return s3err.ErrNone, "", false
+	case resp.ErrorCode != filer_pb.FilerError_OK:
+		if code, mapped := filerErrorCodeToS3Error(resp.ErrorCode); mapped {
+			return code, "", true
+		}
+		glog.Warningf("CopyObjectHandler: routed finalize to %s returned code %v for %s/%s, falling back to lock", owner, resp.ErrorCode, dstBucket, dstObject)
+		return s3err.ErrNone, "", false
+	case resp.Error != "":
+		glog.Warningf("CopyObjectHandler: routed finalize to %s returned %q for %s/%s, falling back to lock", owner, resp.Error, dstBucket, dstObject)
+		return s3err.ErrNone, "", false
+	default:
+		return s3err.ErrNone, etag, true
+	}
+}
+
+// finalizeCopyDestination writes the copy destination. When routedOwner is set
+// (versioning-enabled destination routed off the distributed lock), the latest-
+// pointer update goes through FinalizeVersionedWrite on that owner, which also
+// evaluates cond; otherwise it uses the lock-based updateLatestVersionInDirectory.
+func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersioningState string, dstEntry *filer_pb.Entry, routedOwner pb.ServerAddress, cond *filer_pb.WriteCondition) (versionId string, etag string, err error) {
 	normalizedObject := s3_constants.NormalizeObjectKey(dstObject)
 	dstPath := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(dstBucket), normalizedObject))
 	dstDir, dstName := dstPath.DirAndName()
@@ -444,6 +529,18 @@ func (s3a *S3ApiServer) finalizeCopyDestination(dstBucket, dstObject, dstVersion
 			return "", "", err
 		}
 
+		if routedOwner != "" {
+			// Routed: precondition + demote + pointer flip run atomically on the
+			// .versions owner. Roll back the version file if it does not succeed.
+			if code := s3a.routedVersionedFinalize(routedOwner, dstBucket, normalizedObject, versionId, versionFileName, dstEntry, cond, ""); code != s3err.ErrNone {
+				if rollbackErr := s3a.rollbackCopyVersion(bucketDir, versionObjectPath); rollbackErr != nil {
+					glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, dstBucket, normalizedObject, rollbackErr)
+				}
+				return "", "", finalizeCodeToError(code)
+			}
+			return versionId, etag, nil
+		}
+
 		if err = s3a.updateLatestVersionInDirectory(dstBucket, normalizedObject, versionId, versionFileName, dstEntry); err != nil {
 			if rollbackErr := s3a.rollbackCopyVersion(bucketDir, versionObjectPath); rollbackErr != nil {
 				glog.Errorf("CopyObjectHandler: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, dstBucket, normalizedObject, rollbackErr)

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_delete.go

@@ -133,7 +133,7 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v
 		return result, s3err.ErrNone
 
 	case versioningState == s3_constants.VersioningEnabled:
-		deleteMarkerVersionId, err := s3a.createDeleteMarker(bucket, object)
+		deleteMarkerVersionId, err := s3a.createDeleteMarker(bucket, object, "", nil, "")
 		if err != nil {
 			glog.Errorf("deleteVersionedObject: failed to create delete marker for %s/%s: %v", bucket, object, err)
 			return result, s3err.ErrInternalError
@@ -152,7 +152,7 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v
 			glog.Errorf("deleteVersionedObject: failed to delete null version for %s/%s: %v", bucket, object, err)
 			return result, s3err.ErrInternalError
 		}
-		deleteMarkerVersionId, err := s3a.createDeleteMarker(bucket, object)
+		deleteMarkerVersionId, err := s3a.createDeleteMarker(bucket, object, "", nil, "")
 		if err != nil {
 			glog.Errorf("deleteVersionedObject: failed to create delete marker for suspended versioning %s/%s: %v", bucket, object, err)
 			return result, s3err.ErrInternalError
@@ -214,33 +214,104 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 	}
 
 	var deleteResult deleteMutationResult
-	deleteCode := s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
-		return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed)
-	}, func() s3err.ErrorCode {
-		if versioningConfigured {
-			result, errCode := s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState)
-			if errCode != s3err.ErrNone {
-				return errCode
+	var deleteCode s3err.ErrorCode
+
+	// Fast path: route an unversioned object delete to its owner filer, which
+	// serializes it and evaluates the If-Match precondition under the per-path
+	// lock, avoiding the distributed lock. Falls back below for versioned /
+	// object-lock buckets, unsupported conditions, or an unreachable owner.
+	deleteHandled := false
+	if !versioningConfigured {
+		if cond, condOk := buildDeleteCondition(r); condOk {
+			if owner, ownerOk := s3a.routedObjectOwner(bucket, object); ownerOk {
+				resp, err := s3a.deleteEntryOnFiler(owner, bucket, object, cond)
+				switch {
+				case err != nil:
+					glog.Warningf("DeleteObjectHandler: routed delete to %s failed for %s/%s, falling back to lock: %v", owner, bucket, object, err)
+				case resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED:
+					deleteCode, deleteHandled = s3err.ErrPreconditionFailed, true
+				case resp.Error != "":
+					// Fall back to the lock path, which carries extra handling the
+					// raw DeleteEntry lacks (e.g. demoting a directory marker that
+					// still has children instead of failing on a non-empty folder).
+					glog.Warningf("DeleteObjectHandler: routed delete to %s returned %q for %s/%s, falling back to lock", owner, resp.Error, bucket, object)
+				default:
+					deleteCode, deleteHandled = s3err.ErrNone, true
+				}
 			}
-			deleteResult = result
+		}
+	}
+
+	// Fast path: a versioning-enabled delete (no specific version) creates a
+	// delete marker — a unique marker file plus an atomic latest-pointer flip. We
+	// route it to the .versions owner, off the distributed lock. Suspended
+	// versioning (multi-step) and version-specific deletes keep the lock path.
+	if !deleteHandled && versionId == "" && versioningState == s3_constants.VersioningEnabled {
+		if cond, condOk := buildDeleteCondition(r); condOk {
+			if owner := s3a.objectWriteOwner(bucket, s3_constants.NormalizeObjectKey(object)); owner != "" {
+				vid, err := s3a.createDeleteMarker(bucket, object, owner, cond, "")
+				switch {
+				case err == nil:
+					deleteResult = deleteMutationResult{versionId: vid, deleteMarker: true}
+					deleteCode, deleteHandled = s3err.ErrNone, true
+				case errors.Is(err, errVersionedPreconditionFailed):
+					deleteCode, deleteHandled = s3err.ErrPreconditionFailed, true
+				default:
+					glog.Warningf("DeleteObjectHandler: routed delete marker for %s/%s failed, falling back to lock: %v", bucket, object, err)
+				}
+			}
+		}
+	}
+
+	// Fast path: a suspended-versioning delete (no specific version) removes the
+	// null version (the main object entry) and adds a delete marker — the op does
+	// both atomically under the object-key lock, so it serializes with the (also
+	// routed) suspended PUT. Routed only without If-Match: that condition targets
+	// the main object, not the .versions pointer the op evaluates, so a
+	// conditional suspended delete stays on the lock path.
+	if !deleteHandled && versionId == "" && versioningState == s3_constants.VersioningSuspended {
+		if cond, condOk := buildDeleteCondition(r); condOk && cond.Kind == filer_pb.WriteCondition_NONE {
+			if owner := s3a.objectWriteOwner(bucket, s3_constants.NormalizeObjectKey(object)); owner != "" {
+				nullPath := s3a.toFilerPath(bucket, s3_constants.NormalizeObjectKey(object))
+				vid, err := s3a.createDeleteMarker(bucket, object, owner, cond, nullPath)
+				if err == nil {
+					deleteResult = deleteMutationResult{versionId: vid, deleteMarker: true}
+					deleteCode, deleteHandled = s3err.ErrNone, true
+				} else {
+					glog.Warningf("DeleteObjectHandler: routed suspended delete for %s/%s failed, falling back to lock: %v", bucket, object, err)
+				}
+			}
+		}
+	}
+	if !deleteHandled {
+		deleteCode = s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
+			return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed)
+		}, func() s3err.ErrorCode {
+			if versioningConfigured {
+				result, errCode := s3a.deleteVersionedObject(r, bucket, object, versionId, versioningState)
+				if errCode != s3err.ErrNone {
+					return errCode
+				}
+				deleteResult = result
+				return s3err.ErrNone
+			}
+
+			governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
+			if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
+				glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
+				return s3err.ErrAccessDenied
+			}
+
+			if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+				return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false)
+			}); err != nil {
+				glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err)
+				return s3err.ErrInternalError
+			}
+
 			return s3err.ErrNone
-		}
-
-		governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
-		if err := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); err != nil {
-			glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, err)
-			return s3err.ErrAccessDenied
-		}
-
-		if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-			return s3a.deleteUnversionedObjectWithClient(client, bucket, object, false)
-		}); err != nil {
-			glog.Errorf("DeleteObjectHandler: failed to delete %s/%s: %v", bucket, object, err)
-			return s3err.ErrInternalError
-		}
-
-		return s3err.ErrNone
-	})
+		})
+	}
 	if deleteCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, deleteCode)
 		return

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)
@@ -802,7 +807,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 		}
 		return s3a.checkConditionalHeaders(r, bucket, object)
 	}
-	createCode := s3a.withObjectWriteLock(bucket, object, preconditionFn, func() s3err.ErrorCode {
+	createUnderLock := func() s3err.ErrorCode {
 		createErr = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 			req := &filer_pb.CreateEntryRequest{
 				Directory: path.Dir(filePath),
@@ -831,7 +836,52 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 			}
 		}
 		return s3err.ErrNone
-	})
+	}
+
+	// Fast path: route the create to the object key's owner filer, which
+	// serializes it with its local per-path lock and evaluates the precondition,
+	// avoiding the distributed lock. Restricted to the safe subset by
+	// routedObjectWrite; falls back to the lock for everything else and if the
+	// owner is unreachable.
+	var createCode s3err.ErrorCode
+	routed := false
+	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
+			}
+		}
+	}
+	if !routed {
+		createCode = s3a.withObjectWriteLock(bucket, object, preconditionFn, createUnderLock)
+	}
 	if createCode != s3err.ErrNone {
 		if createErr != nil {
 			glog.Errorf("putToFiler: failed to create entry for %s: %v", filePath, createErr)
@@ -1280,18 +1330,22 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
 	// it runs while withObjectWriteLock is still held in putToFiler.
 	// Doing it after putToFiler returns would race a concurrent PUT
 	// promoting a newer latest, which we'd then incorrectly wipe.
-	etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, func(_ *filer_pb.Entry) s3err.ErrorCode {
-		if err := s3a.updateIsLatestFlagsForSuspendedVersioning(bucket, normalizedObject); err != nil {
-			// Best-effort: a stale IsLatest flag is recoverable on the
-			// next list-versions resync, so don't fail the PUT.
-			glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
-		}
-		return s3err.ErrNone
-	})
+	// The null version is written to the main object path, so this is a
+	// single-entry object write — route it on the object key like a normal PUT
+	// (no afterCreate, so putToFiler can take the route-by-key path and skip the
+	// distributed lock). The IsLatest flag rewrite over existing versions is
+	// best-effort bookkeeping, so it runs after the write rather than inside the
+	// atomic boundary.
+	etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, nil, false)
 	if errCode != s3err.ErrNone {
 		glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
 		return "", errCode, SSEResponseMetadata{}
 	}
+	if err := s3a.updateIsLatestFlagsForSuspendedVersioning(bucket, normalizedObject); err != nil {
+		// Best-effort: a stale IsLatest flag is recoverable on the next
+		// list-versions resync, so don't fail the PUT.
+		glog.Warningf("putSuspendedVersioningObject: failed to update IsLatest flags: %v", err)
+	}
 
 	glog.V(2).Infof("putSuspendedVersioningObject: successfully created null version for %s/%s", bucket, object)
 
@@ -1453,13 +1507,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_routed_write.go

@@ -0,0 +1,235 @@
+package s3api
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"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"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// routedObjectOwner returns the filer that owns this object's metadata for
+// route-by-key on a single-entry object write, or ok=false when the write must
+// keep the distributed lock. Only versioning-*enabled* buckets are excluded:
+// their writes go to <obj>/.versions and flip the latest pointer (the versioned
+// finalize path handles those). Suspended and unversioned writes both go to the
+// main object path, so they route here. Object-lock buckets stay on the lock
+// path. On any lookup error it falls back to be safe.
+func (s3a *S3ApiServer) routedObjectOwner(bucket, object string) (pb.ServerAddress, bool) {
+	if object == "" {
+		return "", false
+	}
+	if enabled, err := s3a.isVersioningEnabled(bucket); err != nil || enabled {
+		return "", false
+	}
+	// objectWriteOwner already excludes object-lock buckets (they stay fully on
+	// the distributed lock), returning "" for them.
+	owner := s3a.objectWriteOwner(bucket, object)
+	if owner == "" {
+		return "", false
+	}
+	return owner, true
+}
+
+// routedObjectWrite decides whether an object PUT can take the route-by-key
+// fast path: the metadata write goes straight to the key's owner filer, which
+// serializes it with its local per-path lock and evaluates the precondition,
+// instead of acquiring a distributed lock. It returns the owner filer and the
+// reduced precondition, or ok=false to fall back to withObjectWriteLock.
+//
+// The fast path additionally requires no post-create hook and a precondition
+// that reduces to one primitive.
+func (s3a *S3ApiServer) routedObjectWrite(r *http.Request, bucket, object string, hasAfterCreate bool) (owner pb.ServerAddress, cond *filer_pb.WriteCondition, ok bool) {
+	if hasAfterCreate {
+		return "", nil, false
+	}
+	cond, condOk := buildWriteCondition(r)
+	if !condOk {
+		return "", nil, false
+	}
+	owner, ok = s3a.routedObjectOwner(bucket, object)
+	if !ok {
+		return "", nil, false
+	}
+	return owner, cond, true
+}
+
+// buildWriteCondition reduces the request's conditional headers to a single
+// WriteCondition primitive the filer can evaluate. It only handles the
+// unambiguous single-condition cases; anything else (header combinations,
+// time-based conditions, ETag lists, weak ETags) returns ok=false so the caller
+// keeps the existing gateway-side evaluation under the distributed lock.
+func buildWriteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) {
+	headers, errCode := parseConditionalHeaders(r)
+	if errCode != s3err.ErrNone {
+		return nil, false
+	}
+	if !headers.isSet {
+		return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_NONE}, true
+	}
+	// Time-based conditions are rare on writes; let the lock path handle them.
+	if !headers.ifModifiedSince.IsZero() || !headers.ifUnmodifiedSince.IsZero() {
+		return nil, false
+	}
+	hasMatch := headers.ifMatch != ""
+	hasNoneMatch := headers.ifNoneMatch != ""
+	switch {
+	case hasMatch && !hasNoneMatch:
+		if headers.ifMatch == "*" {
+			return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_EXISTS}, true
+		}
+		if etag, single := singleStrongETag(headers.ifMatch); single {
+			return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etag: etag}, true
+		}
+		return nil, false
+	case hasNoneMatch && !hasMatch:
+		if headers.ifNoneMatch == "*" {
+			return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}, true
+		}
+		if etag, single := singleStrongETag(headers.ifNoneMatch); single {
+			return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, Etag: etag}, true
+		}
+		return nil, false
+	default:
+		// Both If-Match and If-None-Match present — leave to the lock path.
+		return nil, false
+	}
+}
+
+// singleStrongETag returns the normalized ETag when the header carries exactly
+// one strong ETag, and false for ETag lists or weak ("W/") ETags, which the
+// fast path does not replicate.
+func singleStrongETag(v string) (string, bool) {
+	v = strings.TrimSpace(v)
+	if strings.Contains(v, ",") {
+		return "", false
+	}
+	if strings.HasPrefix(v, "W/") || strings.HasPrefix(v, "w/") {
+		return "", false
+	}
+	return strings.Trim(v, `"`), true
+}
+
+// mkFileRouted builds a file entry the way filer_pb.MkFile does, applies fn, and
+// routes the CreateEntry (with an optional precondition) to the given owner filer
+// so its local per-path lock serializes the write. Used by callers that build an
+// object via a modifier (e.g. multipart completion) but want the routed,
+// distributed-lock-free write path.
+func (s3a *S3ApiServer) mkFileRouted(owner pb.ServerAddress, dir, name string, chunks []*filer_pb.FileChunk, cond *filer_pb.WriteCondition, fn func(*filer_pb.Entry)) s3err.ErrorCode {
+	now := time.Now().Unix()
+	entry := &filer_pb.Entry{
+		Name:        name,
+		IsDirectory: false,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:    now,
+			Crtime:   now,
+			FileMode: uint32(0770),
+			Uid:      filer_pb.OS_UID,
+			Gid:      filer_pb.OS_GID,
+		},
+		Chunks: chunks,
+	}
+	if fn != nil {
+		fn(entry)
+	}
+	resp, err := s3a.createEntryOnFiler(owner, &filer_pb.CreateEntryRequest{
+		Directory: dir,
+		Entry:     entry,
+		Condition: cond,
+	})
+	switch {
+	case err != nil:
+		glog.Errorf("mkFileRouted: %s/%s on %s: %v", dir, name, 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("mkFileRouted: %s/%s unexpected code %v", dir, name, resp.ErrorCode)
+		return s3err.ErrInternalError
+	case resp.Error != "":
+		glog.Errorf("mkFileRouted: %s/%s: %s", dir, name, resp.Error)
+		return s3err.ErrInternalError
+	default:
+		return s3err.ErrNone
+	}
+}
+
+// createEntryOnFiler sends a CreateEntry directly to the given owner filer so
+// its local per-path lock serializes the write. The raw response is returned so
+// the caller can distinguish PRECONDITION_FAILED from other outcomes.
+func (s3a *S3ApiServer) createEntryOnFiler(owner pb.ServerAddress, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	var resp *filer_pb.CreateEntryResponse
+	err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		var e error
+		resp, e = client.CreateEntry(context.Background(), req)
+		return e
+	})
+	return resp, err
+}
+
+// filerErrorCodeToS3Error maps a routed response's machine-readable FilerError
+// to the same S3 error the lock path produces via filerErrorToS3Error, so the
+// fast path keeps identical semantics. ok is false for codes it does not map,
+// signalling the caller to fall back to the lock path for exact behavior.
+func filerErrorCodeToS3Error(code filer_pb.FilerError) (s3err.ErrorCode, bool) {
+	switch code {
+	case filer_pb.FilerError_PRECONDITION_FAILED:
+		return s3err.ErrPreconditionFailed, true
+	case filer_pb.FilerError_ENTRY_NAME_TOO_LONG:
+		return s3err.ErrKeyTooLongError, true
+	case filer_pb.FilerError_PARENT_IS_FILE, filer_pb.FilerError_EXISTING_IS_FILE:
+		return s3err.ErrExistingObjectIsFile, true
+	case filer_pb.FilerError_EXISTING_IS_DIRECTORY:
+		return s3err.ErrExistingObjectIsDirectory, true
+	default:
+		return s3err.ErrNone, false
+	}
+}
+
+// buildDeleteCondition reduces a DeleteObject's If-Match header to a primitive.
+// DeleteObject only honors If-Match (matching checkDeleteIfMatch), so other
+// conditional headers are ignored here as they are on the existing path.
+func buildDeleteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) {
+	ifMatch := strings.TrimSpace(r.Header.Get(s3_constants.IfMatch))
+	switch {
+	case ifMatch == "":
+		return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_NONE}, true
+	case ifMatch == "*":
+		return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_EXISTS}, true
+	default:
+		if etag, single := singleStrongETag(ifMatch); single {
+			return &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etag: etag}, true
+		}
+		return nil, false
+	}
+}
+
+// deleteEntryOnFiler sends a non-recursive DeleteEntry for the object directly
+// to its owner filer, which serializes it and evaluates the precondition under
+// the per-path lock. Flags mirror the unversioned delete path (doDeleteEntry).
+func (s3a *S3ApiServer) deleteEntryOnFiler(owner pb.ServerAddress, bucket, object string, cond *filer_pb.WriteCondition) (*filer_pb.DeleteEntryResponse, error) {
+	dir, name := util.NewFullPath(s3a.bucketDir(bucket), object).DirAndName()
+	var resp *filer_pb.DeleteEntryResponse
+	err := pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		var e error
+		resp, e = client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{
+			Directory:            dir,
+			Name:                 name,
+			IsDeleteData:         true,
+			IsRecursive:          false,
+			IgnoreRecursiveError: true,
+			Condition:            cond,
+		})
+		return e
+	})
+	return resp, err
+}

weed/s3api/s3api_object_routed_write_test.go

@@ -0,0 +1,83 @@
+package s3api
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+func reqWith(headers map[string]string) *http.Request {
+	r, _ := http.NewRequest(http.MethodPut, "/b/o", nil)
+	for k, v := range headers {
+		r.Header.Set(k, v)
+	}
+	return r
+}
+
+func TestBuildWriteCondition(t *testing.T) {
+	cases := []struct {
+		name    string
+		headers map[string]string
+		wantOk  bool
+		want    filer_pb.WriteCondition_Kind
+		wantTag string
+	}{
+		{"none", nil, true, filer_pb.WriteCondition_NONE, ""},
+		{"ifnonematch-star", map[string]string{s3_constants.IfNoneMatch: "*"}, true, filer_pb.WriteCondition_IF_NOT_EXISTS, ""},
+		{"ifmatch-star", map[string]string{s3_constants.IfMatch: "*"}, true, filer_pb.WriteCondition_IF_EXISTS, ""},
+		{"ifmatch-etag", map[string]string{s3_constants.IfMatch: `"abc"`}, true, filer_pb.WriteCondition_IF_ETAG_MATCH, "abc"},
+		{"ifnonematch-etag", map[string]string{s3_constants.IfNoneMatch: `"abc"`}, true, filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, "abc"},
+		// Cases that must fall back to the lock path:
+		{"both", map[string]string{s3_constants.IfMatch: "*", s3_constants.IfNoneMatch: "*"}, false, 0, ""},
+		{"etag-list", map[string]string{s3_constants.IfMatch: `"a","b"`}, false, 0, ""},
+		{"weak-etag", map[string]string{s3_constants.IfMatch: `W/"abc"`}, false, 0, ""},
+		{"time-based", map[string]string{s3_constants.IfUnmodifiedSince: "Wed, 21 Oct 2015 07:28:00 GMT"}, false, 0, ""},
+	}
+	for _, tc := range cases {
+		cond, ok := buildWriteCondition(reqWith(tc.headers))
+		if ok != tc.wantOk {
+			t.Errorf("%s: ok=%v want %v", tc.name, ok, tc.wantOk)
+			continue
+		}
+		if ok {
+			if cond.Kind != tc.want {
+				t.Errorf("%s: kind=%v want %v", tc.name, cond.Kind, tc.want)
+			}
+			if cond.Etag != tc.wantTag {
+				t.Errorf("%s: etag=%q want %q", tc.name, cond.Etag, tc.wantTag)
+			}
+		}
+	}
+}
+
+func TestBuildDeleteCondition(t *testing.T) {
+	cases := []struct {
+		name    string
+		ifMatch string
+		wantOk  bool
+		want    filer_pb.WriteCondition_Kind
+		wantTag string
+	}{
+		{"none", "", true, filer_pb.WriteCondition_NONE, ""},
+		{"star", "*", true, filer_pb.WriteCondition_IF_EXISTS, ""},
+		{"etag", `"abc"`, true, filer_pb.WriteCondition_IF_ETAG_MATCH, "abc"},
+		{"list", `"a","b"`, false, 0, ""},
+		{"weak", `W/"abc"`, false, 0, ""},
+	}
+	for _, tc := range cases {
+		h := map[string]string{}
+		if tc.ifMatch != "" {
+			h[s3_constants.IfMatch] = tc.ifMatch
+		}
+		cond, ok := buildDeleteCondition(reqWith(h))
+		if ok != tc.wantOk {
+			t.Errorf("%s: ok=%v want %v", tc.name, ok, tc.wantOk)
+			continue
+		}
+		if ok && (cond.Kind != tc.want || cond.Etag != tc.wantTag) {
+			t.Errorf("%s: got kind=%v etag=%q want kind=%v etag=%q", tc.name, cond.Kind, cond.Etag, tc.want, tc.wantTag)
+		}
+	}
+}

weed/s3api/s3api_object_versioned_finalize.go

@@ -0,0 +1,162 @@
+package s3api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"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"
+)
+
+// errVersionedPreconditionFailed is returned by versioned finalize helpers whose
+// contract is (…, error) when the routed FinalizeVersionedWrite reports a failed
+// precondition, so callers can surface it as 412 rather than a generic 500.
+var errVersionedPreconditionFailed = errors.New("versioned write precondition failed")
+
+// finalizeCodeToError adapts a routedVersionedFinalize result to an error: nil on
+// success, the precondition sentinel on 412, else a generic error.
+func finalizeCodeToError(code s3err.ErrorCode) error {
+	switch code {
+	case s3err.ErrNone:
+		return nil
+	case s3err.ErrPreconditionFailed:
+		return errVersionedPreconditionFailed
+	default:
+		return fmt.Errorf("versioned finalize failed: code %v", code)
+	}
+}
+
+// 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 ""
+	}
+	// Object-lock (WORM) buckets keep ALL of their writes on the distributed
+	// lock. Retention enforcement is a gateway-side check that the per-path-locked
+	// filer ops don't perform, so routing only some of such a bucket's writes
+	// would split serialization between the entry lock and the distributed lock
+	// (they don't serialize against each other). Excluding the bucket entirely
+	// keeps it internally consistent.
+	if locked, err := s3a.isObjectLockEnabled(bucket); err != nil || locked {
+		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, deletePath string) s3err.ErrorCode {
+	set, del := versionedPointerExtended(versionId, versionFileName, versionEntry)
+	req := &filer_pb.FinalizeVersionedWriteRequest{
+		LockKey:               s3a.toFilerPath(bucket, object),
+		DeletePath:            deletePath,
+		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/s3api/s3api_object_versioning.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
@@ -191,7 +192,13 @@ type ObjectVersion struct {
 }
 
 // createDeleteMarker creates a delete marker for versioned delete operations
-func (s3a *S3ApiServer) createDeleteMarker(bucket, object string) (string, error) {
+// createDeleteMarker writes a delete-marker version and makes it the latest.
+// When routedOwner is set, the latest-pointer update (and the optional cond) run
+// atomically via FinalizeVersionedWrite on that owner, off the distributed lock;
+// otherwise it uses the lock-based updateLatestVersionInDirectory. A failed
+// routed finalize rolls back the marker file and returns
+// errVersionedPreconditionFailed for a precondition miss.
+func (s3a *S3ApiServer) createDeleteMarker(bucket, object string, routedOwner pb.ServerAddress, cond *filer_pb.WriteCondition, deleteNullPath string) (string, error) {
 	// Clean up the object path first
 	cleanObject := strings.TrimPrefix(object, "/")
 
@@ -242,6 +249,19 @@ func (s3a *S3ApiServer) createDeleteMarker(bucket, object string) (string, error
 		},
 		Extended: deleteMarkerExtended,
 	}
+	if routedOwner != "" {
+		// Routed: precondition + demote + pointer flip run atomically on the
+		// .versions owner. Roll back the marker file if it does not succeed.
+		if code := s3a.routedVersionedFinalize(routedOwner, bucket, cleanObject, versionId, versionFileName, deleteMarkerEntry, cond, deleteNullPath); code != s3err.ErrNone {
+			if rbErr := s3a.rmObject(versionsDir, versionFileName, true, false); rbErr != nil {
+				glog.Errorf("createDeleteMarker: rollback %s/%s: %v", versionsDir, versionFileName, rbErr)
+			}
+			return "", finalizeCodeToError(code)
+		}
+		glog.V(2).Infof("createDeleteMarker: routed delete marker %s for %s/%s", versionId, bucket, object)
+		return versionId, nil
+	}
+
 	err = s3a.updateLatestVersionInDirectory(bucket, cleanObject, versionId, versionFileName, deleteMarkerEntry)
 	if err != nil {
 		glog.Errorf("createDeleteMarker: failed to update latest version in directory: %v", err)

weed/s3api/s3api_server.go

@@ -25,6 +25,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/iam/policy"
 	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
@@ -95,6 +96,11 @@ type S3ApiServer struct {
 	stsHandlers           *STSHandlers    // STS HTTP handlers for AssumeRoleWithWebIdentity
 	cipher                bool            // encrypt data on volume servers
 	newObjectWriteLock    func(bucket, object string) objectWriteLock
+	// objectWriteLockClient holds the lock-ring view used to route an object's
+	// metadata writes to its owner filer (route-by-key), so that owner's local
+	// per-path lock serializes them without acquiring a distributed lock. Nil
+	// when no filers are configured.
+	objectWriteLockClient *cluster.LockClient
 	// Shared ReaderCache used by the S3 GET streaming path. It lives for the
 	// lifetime of the server so that concurrent and repeat reads share a
 	// single in-flight download per chunk, and so that no per-request
@@ -152,6 +158,8 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 	// Uses the battle-tested vidMap with filer-based lookups
 	// Supports multiple filer addresses with automatic failover for high availability
 	var filerClient *wdclient.FilerClient
+	var masterClient *wdclient.MasterClient
+	var objectWriteLockClient *cluster.LockClient
 	if len(option.Masters) > 0 {
 		// Enable filer discovery via master
 		masterMap := make(map[string]pb.ServerAddress)
@@ -162,7 +170,22 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 		if clientHost == "0.0.0.0" || clientHost == "" {
 			clientHost = util.DetectedHostAddress()
 		}
-		masterClient := wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), "", "", *pb.NewServiceDiscoveryFromMap(masterMap))
+		masterClient = wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), "", "", *pb.NewServiceDiscoveryFromMap(masterMap))
+		// Build the object-write lock client and subscribe to the master's
+		// lock-ring updates BEFORE starting the master loop, so the initial
+		// LockRingUpdate sent on connect isn't dropped (the master only delivers
+		// it once per connect). The masterClient already filters updates to this
+		// server's filer group.
+		if len(option.Filers) > 0 {
+			objectWriteLockClient = cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
+			masterClient.SetOnLockRingUpdateFn(func(update *master_pb.LockRingUpdate) {
+				servers := make([]pb.ServerAddress, 0, len(update.Servers))
+				for _, s := range update.Servers {
+					servers = append(servers, pb.ServerAddress(s))
+				}
+				objectWriteLockClient.SetRing(servers, update.Version)
+			})
+		}
 		// Start the master client connection loop - required for GetMaster() to work
 		go masterClient.KeepConnectedToMaster(context.Background())
 
@@ -263,7 +286,12 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl
 	}
 
 	if len(option.Filers) > 0 {
-		objectWriteLockClient := cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
+		// Reuse the lock client built in the masters block (already subscribed to
+		// ring updates); create a plain one when no masters are configured.
+		if objectWriteLockClient == nil {
+			objectWriteLockClient = cluster.NewLockClient(option.GrpcDialOption, option.Filers[0])
+		}
+		s3ApiServer.objectWriteLockClient = objectWriteLockClient
 		s3ApiServer.newObjectWriteLock = func(bucket, object string) objectWriteLock {
 			lockKey := fmt.Sprintf("s3.object.write:%s", s3ApiServer.toFilerPath(bucket, object))
 			owner := fmt.Sprintf("s3api-%d", s3ApiServer.randomClientId)

weed/server/filer_grpc_server.go

@@ -7,15 +7,16 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"time"
 
-	"github.com/seaweedfs/seaweedfs/weed/cluster"
-
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
@@ -186,6 +187,32 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr
 		newEntry.TtlSec = 0
 	}
 
+	// Serialize concurrent mutations to the same path on this filer so the
+	// read (existence/condition) and the write are atomic. Callers route a
+	// key's writes to this owner filer, making this local lock sufficient.
+	fullpath := util.NewFullPath(req.Directory, req.Entry.Name)
+	pathLock := fs.entryLockTable.AcquireLock("CreateEntry", fullpath, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(fullpath, pathLock)
+
+	// Evaluate the optional precondition against the current entry while the
+	// path lock is held, so the check and the write are atomic on this filer.
+	if req.Condition != nil && req.Condition.Kind != filer_pb.WriteCondition_NONE {
+		current, findErr := fs.filer.FindEntry(ctx, fullpath)
+		if findErr != nil && findErr != filer_pb.ErrNotFound {
+			return &filer_pb.CreateEntryResponse{}, fmt.Errorf("CreateEntry condition check %s: %w", fullpath, findErr)
+		}
+		if findErr == filer_pb.ErrNotFound {
+			current = nil
+		}
+		if !writeConditionSatisfied(req.Condition, current) {
+			glog.V(3).InfofCtx(ctx, "CreateEntry %s: precondition %v failed", fullpath, req.Condition.Kind)
+			return &filer_pb.CreateEntryResponse{
+				Error:     "precondition failed",
+				ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED,
+			}, nil
+		}
+	}
+
 	ctx, eventSink := filer.WithMetadataEventSink(ctx)
 	createErr := fs.filer.CreateEntry(ctx, newEntry, req.OExcl, req.IsFromOtherCluster, req.Signatures, req.SkipCheckParentDirectory, so.MaxFileNameLength)
 
@@ -212,6 +239,170 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr
 	return
 }
 
+// writeConditionSatisfied reports whether the precondition holds against the
+// current entry (nil if absent). The caller reduces request semantics to one
+// primitive; each kind is a single comparison evaluated under the path lock.
+func writeConditionSatisfied(cond *filer_pb.WriteCondition, current *filer.Entry) bool {
+	exists := current != nil
+	switch cond.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 && storedEntryETag(current) == normalizeETag(cond.Etag)
+	case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH:
+		return !exists || storedEntryETag(current) != normalizeETag(cond.Etag)
+	case filer_pb.WriteCondition_IF_UNMODIFIED_SINCE:
+		return !exists || current.Attr.Mtime.Unix() <= cond.UnixTime
+	case filer_pb.WriteCondition_IF_MODIFIED_SINCE:
+		return !exists || current.Attr.Mtime.Unix() > cond.UnixTime
+	default:
+		return true
+	}
+}
+
+// storedEntryETag mirrors the S3 gateway's ETag precedence (the stored
+// Seaweed ETag extended attribute, then the chunk/Md5 fallback) so conditional
+// comparisons match what the gateway computes, without coupling the filer to
+// S3 request handling.
+func storedEntryETag(entry *filer.Entry) string {
+	if v, ok := entry.Extended[s3_constants.ExtETagKey]; ok && len(v) > 0 {
+		return normalizeETag(string(v))
+	}
+	return normalizeETag(filer.ETagEntry(entry))
+}
+
+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
+		}
+	}
+
+	// Optional: delete an entry under the same lock first (suspended delete
+	// removes the "null" version at the main object path before adding the
+	// marker). Not-found is fine (idempotent).
+	if req.DeletePath != "" {
+		if derr := fs.filer.DeleteEntryMetaAndData(ctx, util.FullPath(req.DeletePath), false, false, true, req.IsFromOtherCluster, req.Signatures, 0); derr != nil && derr != filer_pb.ErrNotFound {
+			resp.Error = derr.Error()
+			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)
@@ -328,9 +519,11 @@ func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendTo
 	glog.V(4).InfofCtx(ctx, "AppendToEntry %v", req)
 	fullpath := util.NewFullPath(req.Directory, req.EntryName)
 
-	lockClient := cluster.NewLockClient(fs.grpcDialOption, fs.option.Host)
-	lock := lockClient.NewShortLivedLock(string(fullpath), string(fs.option.Host))
-	defer lock.StopShortLivedLock()
+	// Serialize the read-modify-write against concurrent mutations to the same
+	// path on this filer. The append must route to this entry's owner filer for
+	// this local lock to be authoritative.
+	pathLock := fs.entryLockTable.AcquireLock("AppendToEntry", fullpath, util.ExclusiveLock)
+	defer fs.entryLockTable.ReleaseLock(fullpath, pathLock)
 
 	var offset int64 = 0
 	entry, err := fs.filer.FindEntry(ctx, fullpath)
@@ -375,9 +568,35 @@ func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntr
 
 	glog.V(4).InfofCtx(ctx, "DeleteEntry %v", req)
 
-	ctx, eventSink := filer.WithMetadataEventSink(ctx)
-	err = fs.filer.DeleteEntryMetaAndData(ctx, util.JoinPath(req.Directory, req.Name), req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData, req.IsFromOtherCluster, req.Signatures, req.IfNotModifiedAfter)
+	fullpath := util.JoinPath(req.Directory, req.Name)
 	resp = &filer_pb.DeleteEntryResponse{}
+
+	// A single-entry delete serializes against concurrent mutations to the same
+	// path so a conditional delete's check and the removal are atomic. Recursive
+	// deletes span many paths and keep the prior unlocked behavior.
+	if !req.IsRecursive {
+		pathLock := fs.entryLockTable.AcquireLock("DeleteEntry", fullpath, util.ExclusiveLock)
+		defer fs.entryLockTable.ReleaseLock(fullpath, pathLock)
+
+		if req.Condition != nil && req.Condition.Kind != filer_pb.WriteCondition_NONE {
+			current, findErr := fs.filer.FindEntry(ctx, fullpath)
+			if findErr != nil && findErr != filer_pb.ErrNotFound {
+				return &filer_pb.DeleteEntryResponse{}, fmt.Errorf("DeleteEntry condition check %s: %w", fullpath, findErr)
+			}
+			if findErr == filer_pb.ErrNotFound {
+				current = nil
+			}
+			if !writeConditionSatisfied(req.Condition, current) {
+				glog.V(3).InfofCtx(ctx, "DeleteEntry %s: precondition %v failed", fullpath, req.Condition.Kind)
+				resp.Error = "precondition failed"
+				resp.ErrorCode = filer_pb.FilerError_PRECONDITION_FAILED
+				return resp, nil
+			}
+		}
+	}
+
+	ctx, eventSink := filer.WithMetadataEventSink(ctx)
+	err = fs.filer.DeleteEntryMetaAndData(ctx, fullpath, req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData, req.IsFromOtherCluster, req.Signatures, req.IfNotModifiedAfter)
 	if err != nil && err != filer_pb.ErrNotFound {
 		resp.Error = err.Error()
 	} else {

weed/server/filer_grpc_server_condition_test.go

@@ -0,0 +1,108 @@
+package weed_server
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"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 entryWithETag(etag string, mtime time.Time) *filer.Entry {
+	return &filer.Entry{
+		FullPath: "/test/obj",
+		Attr:     filer.Attr{Mtime: mtime},
+		Extended: map[string][]byte{s3_constants.ExtETagKey: []byte(etag)},
+	}
+}
+
+func TestWriteConditionSatisfied(t *testing.T) {
+	base := time.Unix(1700000000, 0)
+	present := entryWithETag("abc", base)
+
+	cases := []struct {
+		name string
+		cond *filer_pb.WriteCondition
+		cur  *filer.Entry
+		want bool
+	}{
+		{"none-absent", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_NONE}, nil, true},
+		{"ifnotexists-absent", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}, nil, true},
+		{"ifnotexists-present", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}, present, false},
+		{"ifexists-absent", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_EXISTS}, nil, false},
+		{"ifexists-present", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_EXISTS}, present, true},
+		{"etagmatch-hit", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etag: `"abc"`}, present, true},
+		{"etagmatch-miss", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etag: `"zzz"`}, present, false},
+		{"etagmatch-absent", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etag: `"abc"`}, nil, false},
+		{"etagnotmatch-hit", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, Etag: `"abc"`}, present, false},
+		{"etagnotmatch-miss", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, Etag: `"zzz"`}, present, true},
+		{"etagnotmatch-absent", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, Etag: `"abc"`}, nil, true},
+		{"unmodsince-ok", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_UNMODIFIED_SINCE, UnixTime: base.Unix()}, present, true},
+		{"unmodsince-fail", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_UNMODIFIED_SINCE, UnixTime: base.Unix() - 1}, present, false},
+		{"modsince-ok", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_MODIFIED_SINCE, UnixTime: base.Unix() - 1}, present, true},
+		{"modsince-fail", &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_MODIFIED_SINCE, UnixTime: base.Unix()}, present, false},
+	}
+	for _, tc := range cases {
+		if got := writeConditionSatisfied(tc.cond, tc.cur); got != tc.want {
+			t.Errorf("%s: got %v want %v", tc.name, got, tc.want)
+		}
+	}
+}
+
+// storedEntryETag prefers the stored Seaweed ETag attribute and falls back to
+// the Md5-derived ETag, matching the S3 gateway.
+func TestStoredEntryETag(t *testing.T) {
+	withExt := entryWithETag("explicit", time.Unix(0, 0))
+	if got := storedEntryETag(withExt); got != "explicit" {
+		t.Errorf("extended etag: got %q", got)
+	}
+	md5Only := &filer.Entry{Attr: filer.Attr{Md5: []byte{0xab, 0xcd}}}
+	if got := storedEntryETag(md5Only); got != "abcd" {
+		t.Errorf("md5 fallback: got %q", got)
+	}
+}
+
+// The CreateEntry handler enforces the precondition atomically: a matching
+// If-Match overwrites, a non-matching one returns PRECONDITION_FAILED.
+func TestCreateEntryConditionEnforced(t *testing.T) {
+	store := newRenameTestStore()
+	store.entries["/test/obj"] = &filer.Entry{
+		FullPath: "/test/obj",
+		Attr:     filer.Attr{Inode: 1, Mtime: time.Unix(1700000000, 0)},
+		Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc")},
+	}
+	f := newRenameTestFiler(store)
+	f.DirBucketsPath = "/buckets"
+	fs := &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}
+
+	req := func(etag string) *filer_pb.CreateEntryRequest {
+		return &filer_pb.CreateEntryRequest{
+			Directory:                "/test",
+			SkipCheckParentDirectory: true,
+			Entry: &filer_pb.Entry{
+				Name:       "obj",
+				Attributes: &filer_pb.FuseAttributes{Mtime: 1700000001, FileMode: 0644, Inode: 2},
+			},
+			Condition: &filer_pb.WriteCondition{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etag: etag},
+		}
+	}
+
+	resp, err := fs.CreateEntry(context.Background(), req(`"zzz"`))
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
+		t.Fatalf("mismatched etag: want PRECONDITION_FAILED, got %v (%q)", resp.ErrorCode, resp.Error)
+	}
+
+	resp, err = fs.CreateEntry(context.Background(), req(`"abc"`))
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if resp.Error != "" {
+		t.Fatalf("matching etag should overwrite, got error %q", resp.Error)
+	}
+}

weed/server/filer_grpc_server_create_lock_test.go

@@ -0,0 +1,59 @@
+package weed_server
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// Concurrent OExcl creates for the same path must yield exactly one winner. The
+// filer's CreateEntry is a FindEntry-then-Insert; without the per-path lock both
+// racers observe "not found" and both insert. The exclusive entry lock makes the
+// check-then-act atomic so the losers see ErrEntryAlreadyExists.
+func TestCreateEntryOExclSerialized(t *testing.T) {
+	store := newRenameTestStore()
+	store.findDelay = 5 * time.Millisecond
+	f := newRenameTestFiler(store)
+	f.DirBucketsPath = "/buckets"
+
+	fs := &FilerServer{
+		filer:          f,
+		option:         &FilerOption{},
+		entryLockTable: util.NewLockTable[util.FullPath](),
+	}
+
+	const racers = 8
+	var success int32
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for i := 0; i < racers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			resp, err := fs.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
+				Directory:                "/test",
+				OExcl:                    true,
+				SkipCheckParentDirectory: true,
+				Entry: &filer_pb.Entry{
+					Name:       "obj",
+					Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileMode: 0644, Inode: 1},
+				},
+			})
+			if err == nil && resp.Error == "" {
+				atomic.AddInt32(&success, 1)
+			}
+		}()
+	}
+	close(start)
+	wg.Wait()
+
+	if success != 1 {
+		t.Fatalf("expected exactly 1 OExcl winner, got %d", success)
+	}
+}

weed/server/filer_grpc_server_rename_test.go

@@ -29,6 +29,7 @@ type renameTestStore struct {
 	findCalls map[string]int
 	commitErr error
 	deleteErr error
+	findDelay time.Duration // optional: widen check-then-act windows in tests
 }
 
 func newRenameTestStore() *renameTestStore {
@@ -69,6 +70,9 @@ func (s *renameTestStore) UpdateEntry(_ context.Context, entry *filer.Entry) err
 }
 
 func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer.Entry, error) {
+	if s.findDelay > 0 {
+		time.Sleep(s.findDelay)
+	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.findCalls[string(p)]++

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

weed/server/filer_server.go

@@ -124,6 +124,13 @@ type FilerServer struct {
 	// mountPeerRegistry backs the MountRegister / MountList RPCs for peer
 	// chunk sharing (tier 1). Always populated.
 	mountPeerRegistry *filer.MountPeerRegistry
+
+	// entryLockTable serializes mutations to the same entry path on this filer.
+	// It is the local serialization point for read-modify-write operations
+	// (conditional create, append) once writers for a key are routed to this
+	// node, replacing the distributed lock for that purpose. Idle keys are
+	// evicted automatically, so the table stays bounded.
+	entryLockTable *util.LockTable[util.FullPath]
 }
 
 func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) {
@@ -162,6 +169,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
 		recentCopyRequests:    make(map[string]recentCopyRequest),
 		CredentialManager:     option.CredentialManager,
+		entryLockTable:        util.NewLockTable[util.FullPath](),
 	}
 	fs.mountPeerRegistry = filer.NewMountPeerRegistry()
 	go fs.runMountPeerRegistrySweeper()

weed/server/master_grpc_server.go

@@ -393,7 +393,13 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 }
 
 func (ms *MasterServer) initialLockRingUpdate(clientType string, filerGroup string) *master_pb.KeepConnectedResponse {
-	if clientType != cluster.FilerType || ms.LockRingManager == nil {
+	if ms.LockRingManager == nil {
+		return nil
+	}
+	// Filers are ring members; S3 gateways are lock clients that need the same
+	// view to dial a key's primary directly. Both get the initial snapshot;
+	// later membership changes already broadcast to every connected client.
+	if clientType != cluster.FilerType && clientType != cluster.S3Type {
 		return nil
 	}