filer: route exclusive and conditional creates to the entry's ring owner (#11109)

* proto: resync the java copy of filer.proto

The Makefile keeps other/java/client/src/main/proto/filer.proto a verbatim copy,
but AssignVolumeResponse.fsync and SubscribeMetadataResponse.flushed_ts_ns
landed without it. Copy them over; no behaviour change.

Claude-Session: https://claude.ai/code/session_01Fx1Hx8RqsJqHpbfbgTf4WJ

* filer: route exclusive and conditional creates to the entry's ring owner

CreateEntry with o_excl is a FindEntry-then-Insert. The per-path lock added for
it makes that atomic only on the filer running it, and the store's insert is an
upsert on every backend, so two filers both pass the existence check and both
report success. mkdir(2) then succeeds twice for the same path. The same hole
sits under the condition precondition, whose comment already told callers to
route the key's writes to the owner filer themselves.

Do it on the server instead, with the mechanism ObjectTransaction already uses:
resolve the entry's ring owner and forward one hop, bounded by is_moved. The
ring's membership comes from the master, so it tolerates a stale view and
reassigns when a filer dies, neither of which a client's configured filer list
can do. Every creator gets this — mount, S3, the filer's own HTTP surface, the
Java client — not only the ones that opted in.

Plain creates are upserts whoever applies them, so they stay local and pay
nothing. The route key shares the S3 gateway's namespace so an object's
ObjectTransaction and its CreateEntry land on the same filer's per-path lock.

Claude-Session: https://claude.ai/code/session_01Fx1Hx8RqsJqHpbfbgTf4WJ

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

@@ -250,9 +250,11 @@ message CreateEntryRequest {
     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.
+    // the write, under the filer's per-path lock.
     WriteCondition condition = 7;
+    // Set on a create a filer forwarded to the entry's ring owner, so the owner
+    // applies it locally instead of forwarding again. Clients leave it unset.
+    bool is_moved = 8;
 }
 
 // WriteCondition is the precondition the filer evaluates against the existing
@@ -561,6 +563,9 @@ message AssignVolumeResponse {
     string error = 8;
     Location location = 9;
     repeated Location replicas = 10;
+    // fsync is the storage rule's fsync decision for the assigned path, so the
+    // client can carry it onto the volume server upload request.
+    bool fsync = 11;
 }
 
 message LookupVolumeRequest {
@@ -576,7 +581,6 @@ message Location {
     string public_url = 2;
     uint32 grpc_port = 3;
     string data_center = 4;
-    bool data_in_remote = 5;
 }
 message LookupVolumeResponse {
     map<string, Locations> locations_map = 1;

weed/pb/filer.proto

@@ -250,9 +250,11 @@ message CreateEntryRequest {
     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.
+    // the write, under the filer's per-path lock.
     WriteCondition condition = 7;
+    // Set on a create a filer forwarded to the entry's ring owner, so the owner
+    // applies it locally instead of forwarding again. Clients leave it unset.
+    bool is_moved = 8;
 }
 
 // WriteCondition is the precondition the filer evaluates against the existing

weed/pb/filer_pb/filer.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.36.6
-// 	protoc        v7.35.1
+// 	protoc        v7.35.0
 // source: filer.proto
 
 package filer_pb
@@ -1374,9 +1374,11 @@ type CreateEntryRequest struct {
 	Signatures               []int32                `protobuf:"varint,5,rep,packed,name=signatures,proto3" json:"signatures,omitempty"`
 	SkipCheckParentDirectory bool                   `protobuf:"varint,6,opt,name=skip_check_parent_directory,json=skipCheckParentDirectory,proto3" json:"skip_check_parent_directory,omitempty"`
 	// 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.
+	// the write, under the filer's per-path lock.
 	Condition *WriteCondition `protobuf:"bytes,7,opt,name=condition,proto3" json:"condition,omitempty"`
+	// Set on a create a filer forwarded to the entry's ring owner, so the owner
+	// applies it locally instead of forwarding again. Clients leave it unset.
+	IsMoved       bool `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -1460,6 +1462,13 @@ func (x *CreateEntryRequest) GetCondition() *WriteCondition {
 	return nil
 }
 
+func (x *CreateEntryRequest) GetIsMoved() bool {
+	if x != nil {
+		return x.IsMoved
+	}
+	return false
+}
+
 // WriteCondition is the precondition the filer evaluates against the existing
 // entry before writing, under the per-path lock. A failed condition returns
 // FilerError PRECONDITION_FAILED. The client maps request semantics (e.g. RFC
@@ -7075,7 +7084,7 @@ const file_filer_proto_rawDesc = "" +
 	"\bctime_ns\x18\x14 \x01(\x05R\actimeNs\x12\x1b\n" +
 	"\tcrtime_ns\x18\x15 \x01(\x05R\bcrtimeNs\x12\x14\n" +
 	"\x05atime\x18\x16 \x01(\x03R\x05atime\x12\x19\n" +
-	"\batime_ns\x18\x17 \x01(\x05R\aatimeNs\"\xba\x02\n" +
+	"\batime_ns\x18\x17 \x01(\x05R\aatimeNs\"\xd5\x02\n" +
 	"\x12CreateEntryRequest\x12\x1c\n" +
 	"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" +
 	"\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12\x15\n" +
@@ -7085,7 +7094,8 @@ const file_filer_proto_rawDesc = "" +
 	"signatures\x18\x05 \x03(\x05R\n" +
 	"signatures\x12=\n" +
 	"\x1bskip_check_parent_directory\x18\x06 \x01(\bR\x18skipCheckParentDirectory\x126\n" +
-	"\tcondition\x18\a \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\"\xbc\x04\n" +
+	"\tcondition\x18\a \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12\x19\n" +
+	"\bis_moved\x18\b \x01(\bR\aisMoved\"\xbc\x04\n" +
 	"\x0eWriteCondition\x129\n" +
 	"\aclauses\x18\x01 \x03(\v2\x1f.filer_pb.WriteCondition.ClauseR\aclauses\x1a\x91\x02\n" +
 	"\x06Clause\x121\n" +

weed/pb/filer_pb/filer_vtproto.pb.go

@@ -1014,6 +1014,16 @@ func (m *CreateEntryRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
 		i -= len(m.unknownFields)
 		copy(dAtA[i:], m.unknownFields)
 	}
+	if m.IsMoved {
+		i--
+		if m.IsMoved {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x40
+	}
 	if m.Condition != nil {
 		size, err := m.Condition.MarshalToSizedBufferVT(dAtA[:i])
 		if err != nil {
@@ -6687,6 +6697,9 @@ func (m *CreateEntryRequest) SizeVT() (n int) {
 		l = m.Condition.SizeVT()
 		n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
 	}
+	if m.IsMoved {
+		n += 2
+	}
 	n += len(m.unknownFields)
 	return n
 }
@@ -11746,6 +11759,26 @@ func (m *CreateEntryRequest) UnmarshalVT(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 8:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field IsMoved", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return protohelpers.ErrIntOverflow
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.IsMoved = bool(v != 0)
 		default:
 			iNdEx = preIndex
 			skippy, err := protohelpers.Skip(dAtA[iNdEx:])

weed/server/filer_grpc_server.go

@@ -200,6 +200,36 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr
 
 	resp = &filer_pb.CreateEntryResponse{}
 
+	// An exclusive or conditional create is a read-then-write that the per-path
+	// lock below only makes atomic on this filer, while the store's insert is an
+	// upsert. Route it to the entry's ring owner so one filer's lock arbitrates
+	// every creator cluster-wide; is_moved bounds this to one hop. Plain creates
+	// are upserts either way and stay local.
+	if !req.IsMoved && (req.OExcl || conditionIsSet(req.Condition)) {
+		fullpath := util.NewFullPath(req.Directory, req.Entry.Name)
+		// Held apart from the named resp, which the local path below writes into:
+		// a failed forward must not leave it nil.
+		var ownerResp *filer_pb.CreateEntryResponse
+		handled, forwardErr := fs.forwardToWriteOwner(ctx, entryRouteKey(fullpath), func(owner pb.ServerAddress) error {
+			glog.V(2).InfofCtx(ctx, "CreateEntry %s: forwarding to owner %s", fullpath, owner)
+			req.IsMoved = true
+			return pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+				forwarded, e := client.CreateEntry(ctx, req)
+				if e != nil {
+					return e
+				}
+				ownerResp = forwarded
+				return nil
+			})
+		})
+		if handled {
+			if forwardErr != nil {
+				return &filer_pb.CreateEntryResponse{}, forwardErr
+			}
+			return ownerResp, nil
+		}
+	}
+
 	chunks, garbage, err2 := fs.cleanupChunks(ctx, util.Join(req.Directory, req.Entry.Name), nil, req.Entry)
 	if err2 != nil {
 		return &filer_pb.CreateEntryResponse{}, fmt.Errorf("CreateEntry cleanupChunks %s %s: %v", req.Directory, req.Entry.Name, err2)

weed/server/filer_grpc_server_create_route_test.go

@@ -0,0 +1,175 @@
+package weed_server
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// createRouteServer builds a filer whose ring holds two unreachable peers, so a
+// forwarded create fails at the dial and is distinguishable from a local apply.
+func createRouteServer(t *testing.T) (*FilerServer, *renameTestStore) {
+	t.Helper()
+	const self = pb.ServerAddress("127.0.0.1:18888")
+	store := newRenameTestStore()
+	f := newRenameTestFiler(t, store)
+	f.DirBucketsPath = "/buckets"
+	f.Dlm = lock_manager.NewDistributedLockManager(self)
+	f.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{self, "127.0.0.1:18889"}, 1)
+	return &FilerServer{
+		filer:          f,
+		option:         &FilerOption{Host: self},
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		entryLockTable: util.NewLockTable[util.FullPath](),
+	}, store
+}
+
+// peerOwnedName returns a child name under /test whose entry the ring assigns to
+// the peer, so a routed create must leave this filer.
+func peerOwnedName(t *testing.T, fs *FilerServer) string {
+	t.Helper()
+	for i := 0; i < 4000; i++ {
+		name := fmt.Sprintf("obj-%d", i)
+		key := entryRouteKey(util.NewFullPath("/test", name))
+		if fs.filer.Dlm.LockRing.GetPrimary(key) != fs.option.Host {
+			return name
+		}
+	}
+	t.Skip("no name owned by the peer")
+	return ""
+}
+
+func createReq(name string, oExcl bool) *filer_pb.CreateEntryRequest {
+	return &filer_pb.CreateEntryRequest{
+		Directory:                "/test",
+		OExcl:                    oExcl,
+		SkipCheckParentDirectory: true,
+		Entry: &filer_pb.Entry{
+			Name:       name,
+			Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileMode: 0644, Inode: 1},
+		},
+	}
+}
+
+// An exclusive create for a peer-owned path must leave this filer rather than
+// take the local lock, whose arbitration would not bind the other filers.
+func TestCreateEntryExclusiveLeavesNonOwner(t *testing.T) {
+	fs, store := createRouteServer(t)
+	name := peerOwnedName(t, fs)
+
+	_, err := fs.CreateEntry(context.Background(), createReq(name, true))
+	if err == nil {
+		t.Fatal("exclusive create on a non-owner must not be applied locally")
+	}
+	if _, found := store.entries[string(util.NewFullPath("/test", name))]; found {
+		t.Fatal("forwarded create must not also write locally")
+	}
+}
+
+// Plain creates are upserts whichever filer applies them, so routing them would
+// buy nothing and cost a hop.
+func TestCreateEntryPlainStaysLocal(t *testing.T) {
+	fs, store := createRouteServer(t)
+	name := peerOwnedName(t, fs)
+
+	resp, err := fs.CreateEntry(context.Background(), createReq(name, false))
+	if err != nil || resp.Error != "" {
+		t.Fatalf("plain create must be applied locally, got err=%v resp=%v", err, resp.Error)
+	}
+	if _, found := store.entries[string(util.NewFullPath("/test", name))]; !found {
+		t.Fatal("plain create must reach the local store")
+	}
+}
+
+// is_moved bounds forwarding to one hop: the owner applies it even if its own
+// ring view says someone else owns the key.
+func TestCreateEntryMovedAppliesLocally(t *testing.T) {
+	fs, store := createRouteServer(t)
+	name := peerOwnedName(t, fs)
+
+	req := createReq(name, true)
+	req.IsMoved = true
+	resp, err := fs.CreateEntry(context.Background(), req)
+	if err != nil || resp.Error != "" {
+		t.Fatalf("forwarded create must be applied locally, got err=%v resp=%v", err, resp.Error)
+	}
+	if _, found := store.entries[string(util.NewFullPath("/test", name))]; !found {
+		t.Fatal("forwarded create must reach the local store")
+	}
+}
+
+// An exclusive create this filer owns is applied locally, under its per-path lock.
+func TestCreateEntryExclusiveAppliedByOwner(t *testing.T) {
+	fs, store := createRouteServer(t)
+	var name string
+	for i := 0; i < 4000 && name == ""; i++ {
+		candidate := fmt.Sprintf("obj-%d", i)
+		key := entryRouteKey(util.NewFullPath("/test", candidate))
+		if fs.filer.Dlm.LockRing.GetPrimary(key) == fs.option.Host {
+			name = candidate
+		}
+	}
+	if name == "" {
+		t.Skip("no name owned by this filer")
+	}
+
+	resp, err := fs.CreateEntry(context.Background(), createReq(name, true))
+	if err != nil || resp.Error != "" {
+		t.Fatalf("owner must apply the create, got err=%v resp=%v", err, resp.Error)
+	}
+	if _, found := store.entries[string(util.NewFullPath("/test", name))]; !found {
+		t.Fatal("owner's create must reach the local store")
+	}
+}
+
+// A key still inside the cooling-off window belongs to its prior owner. If that
+// filer cannot be reached the create fails: it may be partitioned rather than
+// down, and applying here would be the second serialization point on the key.
+func TestCreateEntryFailsWhenPriorOwnerIsUnreachable(t *testing.T) {
+	const self = pb.ServerAddress("127.0.0.1:28888")
+	const peer = pb.ServerAddress("127.0.0.1:28889")
+	store := newRenameTestStore()
+	f := newRenameTestFiler(t, store)
+	f.DirBucketsPath = "/buckets"
+	f.Dlm = lock_manager.NewDistributedLockManager(self)
+	// Dropping the peer moves every key it owned to this filer, with the peer as
+	// their prior owner — and nothing is listening on it.
+	f.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{self, peer}, 1)
+	f.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{self}, 2)
+	fs := &FilerServer{
+		filer:          f,
+		option:         &FilerOption{Host: self},
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		entryLockTable: util.NewLockTable[util.FullPath](),
+	}
+
+	var name string
+	for i := 0; i < 4000 && name == ""; i++ {
+		candidate := fmt.Sprintf("obj-%d", i)
+		if fs.filer.Dlm.LockRing.PriorOwner(entryRouteKey(util.NewFullPath("/test", candidate))) == peer {
+			name = candidate
+		}
+	}
+	if name == "" {
+		t.Skip("no key whose prior owner is the departed peer")
+	}
+
+	resp, err := fs.CreateEntry(context.Background(), createReq(name, true))
+	if err == nil {
+		t.Fatalf("unreachable prior owner must fail the create, got resp=%v", resp)
+	}
+	if resp == nil {
+		t.Fatal("a failed create must still answer with a response, not nil")
+	}
+	if _, found := store.entries[string(util.NewFullPath("/test", name))]; found {
+		t.Fatal("a create routed away must not also write locally")
+	}
+}

weed/server/filer_grpc_server_route.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 // writeOwner returns the filer that serializes writes to key, or "" when this
@@ -46,3 +48,10 @@ func (fs *FilerServer) forwardToWriteOwner(ctx context.Context, key string, send
 	}
 	return true, nil
 }
+
+// entryRouteKey is the ring key for an entry's writes. It shares the S3
+// gateway's namespace so an object's ObjectTransaction and its CreateEntry
+// resolve to the same owner, and land on that filer's one per-path lock.
+func entryRouteKey(fullpath util.FullPath) string {
+	return s3_constants.ObjectWriteRouteKeyPrefix + string(fullpath)
+}